增加TurnProcess Handler

docs/设计决策与边界/0. 设计决策与边界.md

@@ -38,8 +38,17 @@ API 细节见教程站点（<http://tags.doc.stdunit.com>）。
 
 ### 观测方式：事件驱动，不要跨线程读字段
 
-- 值的**唯一受支持的观测方式**是事件：`ITag.OnTagRead` / `ITag.OnTagWritten`，或 Rx / R3 的 `Watch()` / `ObserveOnRead()`；启停同理用 `ObserveStartedOrStopped()`。
-- 事件在**轮询线程**上同步触发，回调里拿到的 `TagSyncEventArgs` 已经是本轮最终结果（值 + 时间戳），不需要、也不应该再去读字段做同步。
+- 值的**受支持的观测方式**有两种：
+  1. **按测点**：`ITag.OnTagRead` / `ITag.OnTagWritten`，或 Rx / R3 的 `Watch()` / `ObserveOnRead()`；
+  2. **按入口、按轮次**：`ITagsProject.SetTurnProcessHandler(entry, handler)`——某入口一轮结束（读取输入 + 逻辑处理之后、刷写输出之前）触发一次，适合"一次拿到该入口当前值"的观察者（如界面推送流）。回调签名是 `(entry, channel, ct)`，`ct` 是项目生命周期令牌，回调里的异步操作应透传它。
+- `SetTurnProcessHandler` **每个入口最多一个回调，重复设置是替换**（传 `null` 清除）。要多个观察者就自己组装成一个 `TurnProcess`——顺序在组合处显式可见，比"按注册顺序"更可控：
+  ```csharp
+  proj.SetTurnProcessHandler("g1", async (entry, channel, ct) => { await 遥测(...); await 推送(...); });
+  ```
+- `SetTurnProcessHandler` **只能在项目运行前设置**（启动 hook 或 `StartedOrStopped` 的"已启动"事件里；运行后调用抛 `InvalidOperationException`），与 `IntentCapacity` "只能在运行前设置"同构。设置无需清除：回调的作用域就是项目实例，项目停止即随之释放。需要运行期动态开关观测时，在已设置的回调内部判断（一次集合查找），不要反复设置/清除。
+- 启停用 `ObserveStartedOrStopped()` / `RunnerStarted`；注意 `SetTurnProcessHandler` **只覆盖"产生了轮次"**的情形，入口被禁用、项目未启动/已停止时不产生轮次，也就不会触发。
+- 事件/回调都在**轮询线程**上同步触发，拿到的已经是本轮最终结果（值 + 时间戳），不需要、也不应该再去读字段做同步。
+- **不同入口的回调并行**（各入口由各自的运行器驱动）；回调必须尽快返回（要干重活就只投递到自己的队列，由独立循环消费）。回调抛出的异常不会被吞掉——等价于该入口本轮失败，会进入 `RunnerCrashed` 的恢复路径，观察者请自行 try/catch。
 - 直接读 `ITag.Value` 只适用于"同一轮询线程内"或"接受最终一致性"的场景（例如 `Watch(startWithCurrent: true)` 拿首帧）。**界面不要轮询字段**——那是事件该干的事。
 
 ### 时间戳：一律 UTC

src/StdUnit.Tags.Core/Projects/ITagsProject.cs

@@ -68,6 +68,55 @@ public interface ITagsProject : IDisposable
     event RunnerCrashed? RunnerCrashed;
 
 
+    /// <summary>
+    /// 为<b>指定入口</b>设置"轮次处理完成"回调（<b>读取输入 + 逻辑处理</b> 之后、<b>刷写输出</b> 之前触发），
+    /// 等价于该入口的 <see cref="ITagGrpRunner.TurnProcess"/>。<br/>
+    /// <b>每个入口最多一个回调；重复设置是替换（不是追加）。</b>需要多个观察者时，请自行组合成一个
+    /// <see cref="TurnProcess"/> 再设置——这样调用顺序由你在组合处显式决定，比"按注册顺序"更可控：<br/>
+    /// <code>
+    /// proj.SetTurnProcessHandler("g1", async (entry, channel, ct) =>
+    /// {
+    ///     await 遥测(entry, channel, ct);
+    ///     await 推送(entry, channel, ct);
+    /// });
+    /// </code>
+    /// <br/>
+    /// <b>只能在项目运行（<see cref="RunAsync"/>）之前设置</b>：入口列表在 <see cref="Initialize(string, XElement?)"/>
+    /// 之后才确定，运行期不再支持增删（与 <see cref="IntentCapacity"/>"只能在运行前设置"的约定一致）。
+    /// 典型设置点是 <see cref="ITagsProjectCtrl.StartPollAsync(string?, XElement?, Func{ITagsProject, IServiceProvider, CancellationToken, Task})"/>
+    /// 的 <c>hook</c>，或 <see cref="ITagsProjectCtrl.StartedOrStopped"/> 的"已启动"事件——两者都在轮询开始前同步执行。<br/>
+    /// 需要在运行期"动态开关观测"时，请在已设置的回调内部判断（例如先检查自己的订阅者集合是否为空再直接返回），
+    /// 而不是反复设置/清除。<br/>
+    /// <br/>
+    /// <b>回调运行在轮询线程上，必须尽快返回</b>：不要阻塞，也不要在回调里做 IO / 序列化。
+    /// 需要较重的处理时，请只把必要信息投递到自己的队列，由独立的后台循环消费。<br/>
+    /// <br/>
+    /// <b>不同入口的回调是并行的</b>（各入口由各自的运行器并行驱动）。同一个回调实例若被设置到多个入口，
+    /// 必须自行保证线程安全。<br/>
+    /// <br/>
+    /// 回调签名是 <c>(entry, channel, ct)</c>：<c>ct</c> 是项目的生命周期取消令牌。
+    /// 回调里发起的异步操作请把它透传下去，否则"停止项目"要等到 <see cref="ITagsProjectCtrl.StopAsync"/> 的等待超时才收场。<br/>
+    /// <br/>
+    /// 回调抛出的异常<b>不会被吞掉</b>：它等价于该入口本轮失败（与 <see cref="ILogicet"/> 抛异常同义），
+    /// 会进入 <see cref="RunnerCrashed"/> 的恢复路径。观察者请自行 try/catch，不要让观测逻辑影响轮询。<br/>
+    /// <br/>
+    /// 注意：
+    /// <list type="number">
+    /// <item>必须在 <see cref="Initialize(string, XElement?)"/> 之后、<see cref="RunAsync"/> 之前调用；</item>
+    /// <item>入口名不存在时抛 <see cref="KeyNotFoundException"/>（与 <c>WriteIntent</c> 的校验一致）；</item>
+    /// <item>不需要清除：回调的作用域就是项目实例本身，项目停止后实例被释放，回调随之消失；</item>
+    /// <item>只覆盖"产生了轮次"的情形：入口被禁用、项目未启动或已停止时不会产生轮次，也就不会调用；
+    ///       生命周期请用 <see cref="ITagsProjectCtrl.StartedOrStopped"/> 与 <see cref="RunnerStarted"/>。</item>
+    /// </list>
+    /// </summary>
+    /// <param name="entry">入口测点组的名称（即 <c>entry.TagName()</c>）</param>
+    /// <param name="handler">回调；传 <c>null</c> 表示清除该入口已设置的回调</param>
+    /// <exception cref="ArgumentNullException"><paramref name="entry"/> 为 null</exception>
+    /// <exception cref="KeyNotFoundException">未找到指定的入口测点组</exception>
+    /// <exception cref="InvalidOperationException">项目已经在运行（本方法只能在 <see cref="RunAsync"/> 之前调用）</exception>
+    void SetTurnProcessHandler(string entry, TurnProcess? handler);
+
+
     /// <summary>
     /// 初始化，如果root为空，则默认取 projRoot下的index.xml文件
     /// </summary>

src/StdUnit.Tags.Core/Projects/TagsProject.cs

@@ -225,6 +225,9 @@ public virtual Task RunAsync(CancellationToken ct)
             throw new InvalidOperationException("未配置入口测点组");
         }
 
+        // 进入运行期：此后不再接受 TurnProcess 回调注册（见 AddTurnProcessHandler）。
+        this._running = true;
+
         var tasks = new ConcurrentBag<Task>();
         Parallel.ForEach(entries, entry =>
         {
@@ -235,15 +238,23 @@ public virtual Task RunAsync(CancellationToken ct)
                 .ToList();
             var runner = this._tagGrpRunnerFactory.Create(this);
             runner.RunnerStarted += RunnerStarted;
-            runner.TurnProcess += async (entry, ch) =>
+            runner.TurnProcess += async (entry, ch, turnCt) =>
             {
                 foreach (var l in logicets)
                 {
                     if (!l.Enabled)
                     {
                         continue;
                     }
-                    await l.ProcessAsync(entry, ch, ct);
+                    await l.ProcessAsync(entry, ch, turnCt);
+                }
+
+                // 公共回调排在 Logicets 之后：订阅者看到的是逻辑处理完成后的值。
+                // 有意不 catch：回调抛异常等价于本轮失败（与 Logicet 抛异常同义），见 ITagsProject.SetTurnProcessHandler 的文档。
+                // 回调表在运行期只读（设置在运行前完成），因此这里不需要任何同步。
+                if (this._entryTurnHandlers.TryGetValue(entry.TagName(), out var handler))
+                {
+                    await handler(entry, ch, turnCt);
                 }
             };
             runner.RunnerCrashed += RunnerCrashed;
@@ -339,6 +350,50 @@ protected virtual void CompleteIntentChannels(string disposeMsg)
     /// <inheritdoc/>
     public event RunnerStarted? RunnerStarted;
 
+    #region TurnProcess handlers
+    /// <summary>
+    /// 入口名 → 该入口的回调（每个入口最多一个）。<b>只在运行前写入、运行期只读</b>
+    /// （见 <see cref="SetTurnProcessHandler"/> 的"只能在运行前设置"约定），因此轮询线程读取时不需要任何同步。
+    /// </summary>
+    private readonly Dictionary<string, TurnProcess> _entryTurnHandlers = new(StringComparer.Ordinal);
+
+    /// <summary>
+    /// 是否已进入运行期（<see cref="RunAsync"/> 已开始）。运行期不再接受设置。
+    /// </summary>
+    private volatile bool _running;
+
+    /// <inheritdoc/>
+    public void SetTurnProcessHandler(string entry, TurnProcess? handler)
+    {
+        if (entry is null)
+        {
+            throw new ArgumentNullException(nameof(entry));
+        }
+        if (this._running)
+        {
+            throw new InvalidOperationException(
+                $"TurnProcess 回调只能在项目运行之前设置（入口={entry}）：请在启动 hook 或 StartedOrStopped 的\"已启动\"事件里设置，" +
+                "运行期如需动态开关观测，请在已设置的回调内部判断。");
+        }
+
+        // 与 WriteIntent 的入口校验保持一致：只允许向合法的入口设置
+        if (!this.GetEntries().Any(e => e.TagName() == entry))
+        {
+            throw new KeyNotFoundException($"未找到指定的入口测点组: {entry}");
+        }
+
+        // 替换语义：需要组合多个观察者时，由调用方自己组装成一个 TurnProcess。
+        if (handler is null)
+        {
+            this._entryTurnHandlers.Remove(entry);
+        }
+        else
+        {
+            this._entryTurnHandlers[entry] = handler;
+        }
+    }
+    #endregion
+
 
     #region
     private bool _disposed;

src/StdUnit.Tags.Core/TagGrpRunner/ITagGrpRunner.cs

@@ -12,8 +12,13 @@ namespace StdUnit.Tags;
 /// 入口下若还挂了其它通道，运行器会自动为它们建连，但不会在这里逐个暴露；
 /// 需要用到辅通道时，请自行从 <paramref name="grp"/> 上解析（如 <c>grp.SelectGrp("grp2").SearchChannel()</c>）。
 /// </param>
+/// <param name="ct">
+/// 项目的生命周期取消令牌（与本轮读/写用的是同一个）。<br/>
+/// 回调里发起的异步操作应把它透传下去，这样"停止项目"能及时打断，
+/// 不至于拖到 <see cref="ITagsProjectCtrl.StopAsync"/> 的等待超时。
+/// </param>
 /// <returns></returns>
-public delegate Task TurnProcess(ITagGrp grp, ITagChannel? channel);
+public delegate Task TurnProcess(ITagGrp grp, ITagChannel? channel, CancellationToken ct);
 
 /// <summary>
 /// <see cref="ITagGrpRunner"/>中有轮训启动

src/StdUnit.Tags/TagGrpRunners/TagGrpRunner.cs

@@ -95,7 +95,7 @@ public virtual async Task StartAsync(ITagGrp entry, CancellationToken ct)
                     await entry.ReadAsync(TraversalMode.RespectEnabled, ct);
                     if (TurnProcess is not null)
                     {
-                        await TurnProcess(entry, channel);
+                        await TurnProcess(entry, channel, ct);
                     }
                     await entry.WriteAsync(TraversalMode.RespectEnabled, ct);
 

tests/StdUnit.Tags.Tests/Core/Logicets/LogicetLoaderAotTests.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Threading.Channels;
 using System.Xml.Linq;
@@ -42,6 +42,7 @@ private sealed class DummyProject : ITagsProject
 
         public event RunnerStarted? RunnerStarted;
         public event RunnerCrashed? RunnerCrashed;
+        public void SetTurnProcessHandler(string entry, TurnProcess? handler) => throw new NotImplementedException();
 
         public void Dispose() { }
 

tests/StdUnit.Tags.Tests/Core/Logicets/ProjectAddLogicetTests.cs

@@ -215,6 +215,7 @@ public void Initialize(string projRoot, XElement? root = null) { }
 
         public event RunnerStarted? RunnerStarted;
         public event RunnerCrashed? RunnerCrashed;
+        public void SetTurnProcessHandler(string entry, TurnProcess? handler) => throw new NotImplementedException();
     }
 
     /// <summary>