diff --git "a/docs/\350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214/0. \350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214.md" "b/docs/\350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214/0. \350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214.md" index 34dc3042..1e7b8018 100644 --- "a/docs/\350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214/0. \350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214.md" +++ "b/docs/\350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214/0. \350\256\276\350\256\241\345\206\263\347\255\226\344\270\216\350\276\271\347\225\214.md" @@ -38,8 +38,17 @@ API 细节见教程站点()。 ### 观测方式:事件驱动,不要跨线程读字段 -- 值的**唯一受支持的观测方式**是事件:`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 diff --git a/src/StdUnit.Tags.Core/Projects/ITagsProject.cs b/src/StdUnit.Tags.Core/Projects/ITagsProject.cs index 6a2e9c07..1a4dacb5 100644 --- a/src/StdUnit.Tags.Core/Projects/ITagsProject.cs +++ b/src/StdUnit.Tags.Core/Projects/ITagsProject.cs @@ -68,6 +68,55 @@ public interface ITagsProject : IDisposable event RunnerCrashed? RunnerCrashed; + /// + /// 为指定入口设置"轮次处理完成"回调(读取输入 + 逻辑处理 之后、刷写输出 之前触发), + /// 等价于该入口的 。
+ /// 每个入口最多一个回调;重复设置是替换(不是追加)。需要多个观察者时,请自行组合成一个 + /// 再设置——这样调用顺序由你在组合处显式决定,比"按注册顺序"更可控:
+ /// + /// proj.SetTurnProcessHandler("g1", async (entry, channel, ct) => + /// { + /// await 遥测(entry, channel, ct); + /// await 推送(entry, channel, ct); + /// }); + /// + ///
+ /// 只能在项目运行()之前设置:入口列表在 + /// 之后才确定,运行期不再支持增删(与 "只能在运行前设置"的约定一致)。 + /// 典型设置点是 + /// 的 hook,或 的"已启动"事件——两者都在轮询开始前同步执行。
+ /// 需要在运行期"动态开关观测"时,请在已设置的回调内部判断(例如先检查自己的订阅者集合是否为空再直接返回), + /// 而不是反复设置/清除。
+ ///
+ /// 回调运行在轮询线程上,必须尽快返回:不要阻塞,也不要在回调里做 IO / 序列化。 + /// 需要较重的处理时,请只把必要信息投递到自己的队列,由独立的后台循环消费。
+ ///
+ /// 不同入口的回调是并行的(各入口由各自的运行器并行驱动)。同一个回调实例若被设置到多个入口, + /// 必须自行保证线程安全。
+ ///
+ /// 回调签名是 (entry, channel, ct):ct 是项目的生命周期取消令牌。 + /// 回调里发起的异步操作请把它透传下去,否则"停止项目"要等到 的等待超时才收场。
+ ///
+ /// 回调抛出的异常不会被吞掉:它等价于该入口本轮失败(与 抛异常同义), + /// 会进入 的恢复路径。观察者请自行 try/catch,不要让观测逻辑影响轮询。
+ ///
+ /// 注意: + /// + /// 必须在 之后、 之前调用; + /// 入口名不存在时抛 (与 WriteIntent 的校验一致); + /// 不需要清除:回调的作用域就是项目实例本身,项目停止后实例被释放,回调随之消失; + /// 只覆盖"产生了轮次"的情形:入口被禁用、项目未启动或已停止时不会产生轮次,也就不会调用; + /// 生命周期请用 与 。 + /// + ///
+ /// 入口测点组的名称(即 entry.TagName()) + /// 回调;传 null 表示清除该入口已设置的回调 + /// 为 null + /// 未找到指定的入口测点组 + /// 项目已经在运行(本方法只能在 之前调用) + void SetTurnProcessHandler(string entry, TurnProcess? handler); + + /// /// 初始化,如果root为空,则默认取 projRoot下的index.xml文件 /// diff --git a/src/StdUnit.Tags.Core/Projects/TagsProject.cs b/src/StdUnit.Tags.Core/Projects/TagsProject.cs index cdf81b92..a3a9aa12 100644 --- a/src/StdUnit.Tags.Core/Projects/TagsProject.cs +++ b/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(); Parallel.ForEach(entries, entry => { @@ -235,7 +238,7 @@ 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) { @@ -243,7 +246,15 @@ public virtual Task RunAsync(CancellationToken ct) { 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) /// public event RunnerStarted? RunnerStarted; + #region TurnProcess handlers + /// + /// 入口名 → 该入口的回调(每个入口最多一个)。只在运行前写入、运行期只读 + /// (见 的"只能在运行前设置"约定),因此轮询线程读取时不需要任何同步。 + /// + private readonly Dictionary _entryTurnHandlers = new(StringComparer.Ordinal); + + /// + /// 是否已进入运行期( 已开始)。运行期不再接受设置。 + /// + private volatile bool _running; + + /// + 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; diff --git a/src/StdUnit.Tags.Core/TagGrpRunner/ITagGrpRunner.cs b/src/StdUnit.Tags.Core/TagGrpRunner/ITagGrpRunner.cs index 156161aa..96ea5a3f 100644 --- a/src/StdUnit.Tags.Core/TagGrpRunner/ITagGrpRunner.cs +++ b/src/StdUnit.Tags.Core/TagGrpRunner/ITagGrpRunner.cs @@ -12,8 +12,13 @@ namespace StdUnit.Tags; /// 入口下若还挂了其它通道,运行器会自动为它们建连,但不会在这里逐个暴露; /// 需要用到辅通道时,请自行从 上解析(如 grp.SelectGrp("grp2").SearchChannel())。 /// +/// +/// 项目的生命周期取消令牌(与本轮读/写用的是同一个)。
+/// 回调里发起的异步操作应把它透传下去,这样"停止项目"能及时打断, +/// 不至于拖到 的等待超时。 +/// /// -public delegate Task TurnProcess(ITagGrp grp, ITagChannel? channel); +public delegate Task TurnProcess(ITagGrp grp, ITagChannel? channel, CancellationToken ct); /// /// 中有轮训启动 diff --git a/src/StdUnit.Tags/TagGrpRunners/TagGrpRunner.cs b/src/StdUnit.Tags/TagGrpRunners/TagGrpRunner.cs index 035c96f9..777f7663 100644 --- a/src/StdUnit.Tags/TagGrpRunners/TagGrpRunner.cs +++ b/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); diff --git a/tests/StdUnit.Tags.Tests/Core/Logicets/LogicetLoaderAotTests.cs b/tests/StdUnit.Tags.Tests/Core/Logicets/LogicetLoaderAotTests.cs index a59a68d1..789f5403 100644 --- a/tests/StdUnit.Tags.Tests/Core/Logicets/LogicetLoaderAotTests.cs +++ b/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() { } diff --git a/tests/StdUnit.Tags.Tests/Core/Logicets/ProjectAddLogicetTests.cs b/tests/StdUnit.Tags.Tests/Core/Logicets/ProjectAddLogicetTests.cs index e9b18d92..c56814e2 100644 --- a/tests/StdUnit.Tags.Tests/Core/Logicets/ProjectAddLogicetTests.cs +++ b/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(); } /// diff --git a/tests/StdUnit.Tags.Tests/Core/Projects/TagsProjectTurnProcessTests.cs b/tests/StdUnit.Tags.Tests/Core/Projects/TagsProjectTurnProcessTests.cs new file mode 100644 index 00000000..9d1d5dd1 --- /dev/null +++ b/tests/StdUnit.Tags.Tests/Core/Projects/TagsProjectTurnProcessTests.cs @@ -0,0 +1,485 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using System.Xml.Linq; +using Microsoft.Extensions.DependencyInjection; +using StdUnit.Tags.Tests.Fakes; +using Xunit; + +namespace StdUnit.Tags.Tests.Core.Projects; + +/// +/// 验证 :
+/// - 按入口设置,某入口每一轮触发一次,且只触发该入口的回调;
+/// - 排在 Logicets 之后(回调看到的是逻辑处理完成后的值);
+/// - 每个入口最多一个回调:重复设置是替换,多个观察者由调用方自己组合(顺序显式);
+/// - 只能在项目运行之前设置(运行后抛 ),传 null 表示清除;
+/// - 回调抛出的异常不被吞掉(走 的恢复路径);
+/// - 入口被禁用时不会产生轮次,因此不会触发;
+/// - 未知入口 / null 入口按契约抛异常。 +///
+public class TagsProjectTurnProcessTests : IDisposable +{ + private const int TimeoutMs = 10_000; + private const int ScanIntervalMs = 10; + + private readonly ServiceProvider _root; + private readonly string _projectRoot; + + public TagsProjectTurnProcessTests() + { + var services = new ServiceCollection(); + services.AddLogging(); + services.AddTagsProjectServices(b => + { + b.AddFakedSupport(); + b.AddFakedTagSupport(); + }); + + _root = services.BuildServiceProvider(); + _projectRoot = TestPaths.TempPath("TagsProjectTurnProcessTests"); + Directory.CreateDirectory(_projectRoot); + } + + public void Dispose() + { + _root.Dispose(); + } + + /// 每轮把指定测点写成固定值的逻辑组件。 + private sealed class SetValueLogicet : LogicetBase + { + private readonly ITag _tag; + private readonly object _value; + + public SetValueLogicet(ITagsProject project, ITag tag, object value) : base(project) + { + _tag = tag; + _value = value; + } + + public override int Order => 0; + + public override bool MatchEntry(ITagGrp entry) => true; + + public override Task ProcessAsync(ITagGrp entry, ITagChannel? thisChannel, CancellationToken ct) + { + _tag.Value = _value; + return Task.CompletedTask; + } + } + + private static XElement Channel(string name) => + new("Channel", new XAttribute("name", name), new XAttribute("driver", "fake")); + + private static XElement Entry(string name, string channelName, bool isEnabled = true) => + new("TagGrp", + new XAttribute("name", name), + new XAttribute("isEntry", "true"), + new XAttribute("isEnabled", isEnabled ? "true" : "false"), + new XAttribute("channel", channelName), + new XAttribute("scanInterval", ScanIntervalMs), + new XElement("Tag", + new XAttribute("name", "v"), + new XAttribute("type", "INT16"), + new XAttribute("access", "RW"), + new XAttribute("address", "0"))); + + private ITagsProject CreateProject(IServiceProvider sp, XElement xml) => + sp.GetRequiredService().Create(_projectRoot, xml); + + private ITagsProject CreateSingleEntryProject(IServiceScope scope) => + CreateProject(scope.ServiceProvider, + new XElement("Project", Channel("fake"), Entry("g1", "fake"))); + + private static async Task WaitOrFailAsync(Task task, string what) + { + var completed = await Task.WhenAny(task, Task.Delay(TimeoutMs)).ConfigureAwait(false); + Assert.True(ReferenceEquals(completed, task), $"等待 {what} 超时({TimeoutMs}ms)"); + } + + private static async Task StopAsync(CancellationTokenSource cts, Task run) + { + cts.Cancel(); + + var finished = await Task.WhenAny(run, Task.Delay(TimeoutMs)).ConfigureAwait(false); + Assert.True(ReferenceEquals(finished, run), "取消后轮询循环仍未退出"); + + try + { + await run.ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // 取消时轮询循环以 OCE 退出,这是既有设计(见 LogicetCancellationTests)。 + } + } + + [Fact] + public async Task Handler_FiresEveryTurn_ForRegisteredEntry() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + var count = new int[1]; + var threeTurns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + proj.SetTurnProcessHandler("g1", (entry, channel, _) => + { + Assert.Equal("g1", entry.TagName()); + if (Interlocked.Increment(ref count[0]) >= 3) + { + threeTurns.TrySetResult(true); + } + return Task.CompletedTask; + }); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(threeTurns.Task, "连续三轮回调"); + Assert.True(Volatile.Read(ref count[0]) >= 3); + + await StopAsync(cts, run); + } + + [Fact] + public async Task Handler_IsScopedToItsEntry() + { + // 通道不能跨入口共用(一个通道 = 一个轮询回路),所以每个入口用自己的通道实例。 + using var scope = _root.CreateScope(); + using var proj = CreateProject(scope.ServiceProvider, new XElement("Project", + Channel("fake1"), + Channel("fake2"), + Entry("g1", "fake1"), + Entry("g2", "fake2"))); + + var g1Calls = new int[1]; + var g2Calls = new int[1]; + var g1Turns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var g2Turns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + proj.SetTurnProcessHandler("g1", (entry, channel, _) => + { + if (Interlocked.Increment(ref g1Calls[0]) >= 2) + { + g1Turns.TrySetResult(true); + } + return Task.CompletedTask; + }); + proj.SetTurnProcessHandler("g2", (entry, channel, _) => + { + if (Interlocked.Increment(ref g2Calls[0]) >= 2) + { + g2Turns.TrySetResult(true); + } + return Task.CompletedTask; + }); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(g1Turns.Task, "g1 的连续两轮回调"); + await WaitOrFailAsync(g2Turns.Task, "g2 的连续两轮回调"); + + Assert.True(Volatile.Read(ref g1Calls[0]) >= 2); + Assert.True(Volatile.Read(ref g2Calls[0]) >= 2); + + await StopAsync(cts, run); + } + + [Fact] + public async Task Handler_OnlyRegisteredEntryIsNotified() + { + using var scope = _root.CreateScope(); + using var proj = CreateProject(scope.ServiceProvider, new XElement("Project", + Channel("fake1"), + Channel("fake2"), + Entry("g1", "fake1"), + Entry("g2", "fake2"))); + + var entries = new ConcurrentDictionary(StringComparer.Ordinal); + var g1Turns = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + proj.SetTurnProcessHandler("g1", (entry, channel, _) => + { + entries.AddOrUpdate(entry.TagName(), 1, (_, old) => old + 1); + g1Turns.TrySetResult(true); + return Task.CompletedTask; + }); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(g1Turns.Task, "g1 的回调"); + await Task.Delay(ScanIntervalMs * 3); + + // 只为 g1 注册:无论 g2 跑了多少轮,都不应收到 g2 的通知。 + Assert.True(entries.ContainsKey("g1")); + Assert.DoesNotContain("g2", entries.Keys); + + await StopAsync(cts, run); + } + + [Fact] + public async Task Handler_RunsAfterLogicets() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + var tag = proj.Tags.Descendant("g1/v").AsTag(); + const short written = 1234; + + proj.Logicets.Add(new SetValueLogicet(proj, tag, written)); + + var observed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + proj.SetTurnProcessHandler("g1", (entry, channel, _) => + { + observed.TrySetResult(tag.Value); + return Task.CompletedTask; + }); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(observed.Task, "回调观察到逻辑组件的写入"); + var value = await observed.Task; + Assert.Equal(written, Assert.IsType(value)); + + await StopAsync(cts, run); + } + + [Fact] + public async Task Handler_NotRegistered_DoesNotThrow() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + // 让它跑几轮:没有注册回调时,内部必须走"查不到就跳过"而不是空引用。 + await Task.Delay(ScanIntervalMs * 5); + + await StopAsync(cts, run); + } + + [Fact] + public async Task Set_SecondCall_ReplacesPreviousHandler() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + var first = new int[1]; + var second = new int[1]; + var secondCalled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + proj.SetTurnProcessHandler("g1", (entry, channel, _) => + { + Interlocked.Increment(ref first[0]); + return Task.CompletedTask; + }); + // 替换语义:后设置的生效,前一个不再被调用。 + proj.SetTurnProcessHandler("g1", (entry, channel, _) => + { + Interlocked.Increment(ref second[0]); + secondCalled.TrySetResult(true); + return Task.CompletedTask; + }); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(secondCalled.Task, "后设置的回调被调用"); + + Assert.True(Volatile.Read(ref second[0]) >= 1); + Assert.Equal(0, Volatile.Read(ref first[0])); + + await StopAsync(cts, run); + } + + [Fact] + public async Task Set_ComposedHandler_InvokesObserversInDeclaredOrder() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + var order = new List(); + var composed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + // 多个观察者由调用方自己组合:顺序在组合处显式可见,而不是依赖"注册顺序"。 + proj.SetTurnProcessHandler("g1", async (entry, channel, ct) => + { + await Task.CompletedTask; + lock (order) { order.Add("first"); } + + await Task.CompletedTask; + lock (order) { order.Add("second"); } + + composed.TrySetResult(true); + }); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(composed.Task, "组合回调被执行"); + + string[] snapshot; + lock (order) { snapshot = order.ToArray(); } + Assert.Equal(new[] { "first", "second" }, snapshot); + + await StopAsync(cts, run); + } + + [Fact] + public async Task Set_Null_ClearsHandler() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + var count = new int[1]; + proj.SetTurnProcessHandler("g1", (entry, channel, _) => + { + Interlocked.Increment(ref count[0]); + return Task.CompletedTask; + }); + proj.SetTurnProcessHandler("g1", null); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await Task.Delay(ScanIntervalMs * 5); + Assert.Equal(0, Volatile.Read(ref count[0])); + + await StopAsync(cts, run); + } + + [Fact] + public async Task SetTurnProcessHandler_AfterRunStarted_Throws() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + // 运行期不再接受注册:入口列表在运行前就已固定,动态开关观测应在回调内部判断。 + var ex = Assert.Throws( + () => proj.SetTurnProcessHandler("g1", (entry, channel, _) => Task.CompletedTask)); + Assert.Contains("运行之前", ex.Message); + + await StopAsync(cts, run); + } + + [Fact] + public async Task Handler_SubscriberThrows_EntersRunnerCrashed() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + proj.SetTurnProcessHandler( + "g1", + (entry, channel, _) => throw new InvalidOperationException("observer boom")); + + var crashed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + proj.RunnerCrashed += (entry, channel, ex) => + { + crashed.TrySetResult(ex); + return Task.CompletedTask; + }; + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(crashed.Task, "回调的异常进入 RunnerCrashed"); + var captured = await crashed.Task; + var ex = Assert.IsType(captured); + Assert.Equal("observer boom", ex.Message); + + await StopAsync(cts, run); + } + + [Fact] + public async Task Handler_DisabledEntry_NotFired() + { + using var scope = _root.CreateScope(); + using var proj = CreateProject(scope.ServiceProvider, new XElement("Project", + Channel("fake"), + Entry("g1", "fake", isEnabled: false))); + + var count = new int[1]; + proj.SetTurnProcessHandler("g1", (entry, channel, _) => + { + Interlocked.Increment(ref count[0]); + return Task.CompletedTask; + }); + + using var cts = new CancellationTokenSource(); + var run = proj.RunAsync(cts.Token); + + // 被禁用的入口在 TagGrpRunner 里直接 Task.Delay(500) 跳过,根本不会产生轮次。 + await Task.Delay(300); + Assert.Equal(0, Volatile.Read(ref count[0])); + + await StopAsync(cts, run); + } + + [Fact] + public async Task Handler_ReceivesProjectCancellationToken() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + using var cts = new CancellationTokenSource(); + var observed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + proj.SetTurnProcessHandler("g1", (entry, channel, ct) => + { + observed.TrySetResult(ct); + return Task.CompletedTask; + }); + + var run = proj.RunAsync(cts.Token); + + await WaitOrFailAsync(observed.Task, "回调收到取消令牌"); + var captured = await observed.Task; + Assert.True(captured.CanBeCanceled); + Assert.Equal(cts.Token, captured); + + await StopAsync(cts, run); + } + + [Fact] + public void SetTurnProcessHandler_UnknownEntry_Throws() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + Assert.Throws( + () => proj.SetTurnProcessHandler("nope", (entry, channel, _) => Task.CompletedTask)); + } + + [Fact] + public void SetTurnProcessHandler_NullHandler_IsAllowed() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + // null = 清除;在从未设置过的入口上清除也是合法的空操作。 + proj.SetTurnProcessHandler("g1", null); + proj.SetTurnProcessHandler("g1", (entry, channel, _) => Task.CompletedTask); + proj.SetTurnProcessHandler("g1", null); + } + + [Fact] + public void SetTurnProcessHandler_NullEntry_Throws() + { + using var scope = _root.CreateScope(); + using var proj = CreateSingleEntryProject(scope); + + Assert.Throws(() => proj.SetTurnProcessHandler(null!, (entry, channel, _) => Task.CompletedTask)); + } +} diff --git a/tests/StdUnit.Tags.Tests/Core/TagGrpRunners/TagGrpRunnerTests.cs b/tests/StdUnit.Tags.Tests/Core/TagGrpRunners/TagGrpRunnerTests.cs index 690336bd..2d6abd13 100644 --- a/tests/StdUnit.Tags.Tests/Core/TagGrpRunners/TagGrpRunnerTests.cs +++ b/tests/StdUnit.Tags.Tests/Core/TagGrpRunners/TagGrpRunnerTests.cs @@ -60,7 +60,7 @@ public async Task StartAsync_NormalLoop_CallsReadWriteAndEvents() turnStartedCalled = true; return Task.CompletedTask; }; - runner.TurnProcess += (grp, ch) => + runner.TurnProcess += (grp, ch, _) => { turnProcessCalled = true; cts.Cancel(); @@ -137,7 +137,7 @@ public async Task StartAsync_TurnProcessEvent_FiresAfterRead() entry.OnRead = () => processOrder.Add("Read"); entry.OnWrite = () => processOrder.Add("Write"); - runner.TurnProcess += (grp, ch) => + runner.TurnProcess += (grp, ch, _) => { processOrder.Add("Process"); cts.Cancel(); @@ -171,7 +171,7 @@ public async Task StartAsync_IsDirty_CallsWriteBeforeRead() entry.OnWrite = () => callOrder.Add("Write"); entry.OnRead = () => callOrder.Add("Read"); - runner.TurnProcess += (_, _) => + runner.TurnProcess += (_, _, _) => { cts.Cancel(); return Task.CompletedTask; @@ -283,7 +283,7 @@ public async Task StartAsync_Cancellation_StillDisconnectsChannel() using var cts = new CancellationTokenSource(); // 在 TurnProcess 中取消(模拟正常轮询中用户停止) - runner.TurnProcess += (_, _) => + runner.TurnProcess += (_, _, _) => { cts.Cancel(); return Task.CompletedTask; @@ -319,7 +319,7 @@ public async Task StartAsync_WhenDisconnectThrows_SwallowsButLogs() var runner = new TagGrpRunner(new MockProject(), loggerFactory.CreateLogger()); using var cts = new CancellationTokenSource(); - runner.TurnProcess += (_, _) => + runner.TurnProcess += (_, _, _) => { cts.Cancel(); return Task.CompletedTask; @@ -356,7 +356,7 @@ public async Task StartAsync_WhenDisconnectStrategyThrows_SwallowsButLogs() disconnectStrategy: strategy); using var cts = new CancellationTokenSource(); - runner.TurnProcess += (_, _) => + runner.TurnProcess += (_, _, _) => { cts.Cancel(); return Task.CompletedTask; @@ -392,7 +392,7 @@ public async Task StartAsync_Cancellation_SlowDisconnect_TimesOut_AndStillExits( var runner = new TagGrpRunner(project, NullLogger.Instance); using var cts = new CancellationTokenSource(); - runner.TurnProcess += (_, _) => + runner.TurnProcess += (_, _, _) => { cts.Cancel(); return Task.CompletedTask; @@ -432,7 +432,7 @@ public async Task StartAsync_Cancellation_CustomDisconnectStrategy_IsInvoked() var runner = new TagGrpRunner(project, NullLogger.Instance, disconnectStrategy: strategy); using var cts = new CancellationTokenSource(); - runner.TurnProcess += (_, _) => + runner.TurnProcess += (_, _, _) => { cts.Cancel(); return Task.CompletedTask; @@ -467,7 +467,7 @@ public async Task StartAsync_MultiChannelEntry_EnsuresAllSubtreeChannelsConnecte var runner = new TagGrpRunner(project, NullLogger.Instance); using var cts = new CancellationTokenSource(); - runner.TurnProcess += (_, _) => + runner.TurnProcess += (_, _, _) => { cts.Cancel(); return Task.CompletedTask; @@ -500,7 +500,7 @@ public async Task StartAsync_MultiChannelEntry_DisconnectsAllSubtreeChannels() var runner = new TagGrpRunner(project, NullLogger.Instance); using var cts = new CancellationTokenSource(); - runner.TurnProcess += (_, _) => + runner.TurnProcess += (_, _, _) => { cts.Cancel(); return Task.CompletedTask; @@ -536,7 +536,7 @@ public async Task StartAsync_ProcessIntents() return default; }, out var intentTask); - runner.TurnProcess += (_, _) => + runner.TurnProcess += (_, _, _) => { cts.Cancel(); return Task.CompletedTask; @@ -576,7 +576,7 @@ public async Task StartAsync_ProcessMultipleIntents_CompletesAllWithSequentialIn return default; }, out var second); - runner.TurnProcess += (_, _) => + runner.TurnProcess += (_, _, _) => { cts.Cancel(); return Task.CompletedTask; @@ -739,7 +739,7 @@ public async Task StartAsync_WithRealTagGrp_RespectsIsEnabledOfChildren() var runner = new TagGrpRunner(project, NullLogger.Instance); using var cts = new CancellationTokenSource(); - runner.TurnProcess += (_, _) => + runner.TurnProcess += (_, _, _) => { cts.Cancel(); return Task.CompletedTask; @@ -921,6 +921,7 @@ public void Dispose() { } public event RunnerStarted? RunnerStarted; public event RunnerCrashed? RunnerCrashed; + public void SetTurnProcessHandler(string entry, TurnProcess? handler) => throw new NotImplementedException(); } /// diff --git a/tests/StdUnit.Tags.Tests/Core/TagsProjectCtrls/MockTagsProject.cs b/tests/StdUnit.Tags.Tests/Core/TagsProjectCtrls/MockTagsProject.cs index 12348ce4..638a2eff 100644 --- a/tests/StdUnit.Tags.Tests/Core/TagsProjectCtrls/MockTagsProject.cs +++ b/tests/StdUnit.Tags.Tests/Core/TagsProjectCtrls/MockTagsProject.cs @@ -48,6 +48,7 @@ internal class MockTagsProject : ITagsProject public event RunnerStarted? RunnerStarted; public event RunnerCrashed? RunnerCrashed; + public void SetTurnProcessHandler(string entry, TurnProcess? handler) => throw new NotImplementedException(); public void Initialize(string projRoot, XElement? root = null) { InitializeCallCount++;