diff --git a/CHANGELOG.md b/CHANGELOG.md index c04c049..1969f53 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,8 @@ - 写命令：`POST /stdunit.tags/value`（改测点值）、`POST /stdunit.tags/rescan`（标记未扫描）。 - 注册：`services.AddTagsStream()` / `app.MapTagsStream()`；多项目场景用 `AddKeyedTagsStream(key)` 注册具名流（路径形如 `/stdunit.tags/{key}/stream`）。 只允许改**路由前缀**（`app.MapTagsStream("/internal/tags")`），相对路径是线协议的一部分。 + - 启动时机：观测流在 `app.MapTagsStream()` 映射端点时启动（每路一次、幂等），**不注册成 `IHostedService`**， + 因此不依赖 Generic Host 的启动顺序；启动时会按 `ITagsProjectCtrl.Project` 对账，项目先跑起来也能补上（值流从下一轮生效）。 - **不内置任何鉴权**，暴露到非受信网络前请自行加策略（如 `app.MapTagsStream().RequireAuthorization()`）。 - **`AddKeyedTagsProjectCtrl(key)`**（`StdUnit.Tags`，`IServiceCollection` 与 `TagsProjectServiceBuilder` 上各一个）： 以指定 key 注册 `ITagsProjectCtrl`；配合 `AddKeyedTagsStream(key)` 可以在一个宿主里同时监控多个项目。 diff --git a/codecov.yml b/codecov.yml index 4969d05..7ec55a9 100644 --- a/codecov.yml +++ b/codecov.yml @@ -71,9 +71,9 @@ component_management: paths: - "src/StdUnit.Tags.SimpleFiles/**" - component_id: stdunit_tags_webui - name: "WebUI" + name: "WebUi" paths: - - "src/StdUnit.Tags.WebUI/**" + - "src/StdUnit.Tags.WebUi/**" ignore: - "**/*.md" diff --git a/samples/WebMonitor/Tags/TagBackgroundService.cs b/samples/WebMonitor/Tags/TagBackgroundService.cs index e8586eb..ed68bcc 100644 --- a/samples/WebMonitor/Tags/TagBackgroundService.cs +++ b/samples/WebMonitor/Tags/TagBackgroundService.cs @@ -7,8 +7,8 @@ namespace WebMonitor.Tags; /// /// 启停两路测点项目，并跑一个只用于演示的值模拟器。
-/// 与模板 tags.web 的 TagBackgroundService 同一形状：轮询必须在这里（宿主启动之后）开始， -/// 观测流的 hub 才能先拿到"项目已启动"事件并挂上"轮次完成"回调。 +/// 与模板 tags.web 的 TagBackgroundService 同一形状：观测流的 hub 在 MapTagsStream() 时 +/// （Program.cs，早于本服务）就已经启动，所以这里只管轮询；哪一路先起来都不影响观测。 /// internal sealed class TagBackgroundService : BackgroundService { diff --git a/src/StdUnit.Tags.WebUi/Core/Registry/TagsStreamEntry.cs b/src/StdUnit.Tags.WebUi/Core/Registry/TagsStreamEntry.cs index c61ac77..1440a13 100644 --- a/src/StdUnit.Tags.WebUi/Core/Registry/TagsStreamEntry.cs +++ b/src/StdUnit.Tags.WebUi/Core/Registry/TagsStreamEntry.cs @@ -2,16 +2,16 @@ namespace StdUnit.Tags.WebUi; /// /// 一路观测流：它观测的控制器、配置与 hub。
-/// 端点、命令、hosted service 都通过它拿到"同一个"hub —— 这是"接口服务 = hosted service = 端点看到的实例"的保证。 +/// 端点、命令、hub 启动点都通过它拿到"同一个"hub —— 这是"接口服务 = 端点看到的实例"的保证。 /// internal sealed class TagsStreamEntry { - public TagsStreamEntry(string? key, TagsStreamOptions options, ITagsProjectCtrl ctrl, TagsStreamHostedService hub) + public TagsStreamEntry(string? key, TagsStreamOptions options, ITagsProjectCtrl ctrl, TagsStreamHub hub) { this.Key = key; this.Options = options; this.Ctrl = ctrl; - this.HostedService = hub; + this.Hub = hub; } public string? Key { get; } @@ -21,5 +21,5 @@ public TagsStreamEntry(string? key, TagsStreamOptions options, ITagsProjectCtrl /// 这一路流观测的控制器（写命令也用它，保证"看哪个项目就改哪个项目"）。 public ITagsProjectCtrl Ctrl { get; } - internal TagsStreamHostedService HostedService { get; } + internal TagsStreamHub Hub { get; } } diff --git a/src/StdUnit.Tags.WebUi/Core/Registry/TagsStreamRegistry.cs b/src/StdUnit.Tags.WebUi/Core/Registry/TagsStreamRegistry.cs index 47cf86d..7be8706 100644 --- a/src/StdUnit.Tags.WebUi/Core/Registry/TagsStreamRegistry.cs +++ b/src/StdUnit.Tags.WebUi/Core/Registry/TagsStreamRegistry.cs @@ -5,7 +5,7 @@ namespace StdUnit.Tags.WebUi; /// /// 所有观测流的注册表：key（不具名用 null）→ 一路独立的 hub + 控制器 + 配置。
-/// 每个 key 恰好创建一个 hub，hosted service 与端点共用同一实例。 +/// 每个 key 恰好创建一个 hub，端点与 hub 启动点共用同一实例。 /// internal sealed class TagsStreamRegistry { @@ -18,7 +18,7 @@ public TagsStreamRegistry(IServiceProvider services, IEnumerable>()); + var hub = new TagsStreamHub(ctrl, descriptor.Options, services.GetService>()); var entry = new TagsStreamEntry(descriptor.Key, descriptor.Options, ctrl, hub); this._entries.Add(entry); @@ -46,6 +46,18 @@ key is null public IReadOnlyList All => this._entries; + /// + /// 启动全部流的 hub（幂等）。这是观测流的唯一启动点，由 MapTagsStream() 在映射端点时调用： + /// 映射必然早于任何请求，也早于 StartPollAsync，所以不需要 Generic Host 的 IHostedService。 + /// + public void StartAll() + { + foreach (var entry in this._entries) + { + entry.Hub.Start(); + } + } + private static ITagsProjectCtrl ResolveCtrl(IServiceProvider services, string? key) { if (key is null) diff --git a/src/StdUnit.Tags.WebUi/Core/TagsStreamHub.cs b/src/StdUnit.Tags.WebUi/Core/TagsStreamHub.cs index e26d452..b15cb07 100644 --- a/src/StdUnit.Tags.WebUi/Core/TagsStreamHub.cs +++ b/src/StdUnit.Tags.WebUi/Core/TagsStreamHub.cs @@ -1,5 +1,4 @@ using System.Threading.Channels; -using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using StdUnit.Tags.WebUi.Internal; @@ -10,19 +9,25 @@ namespace StdUnit.Tags.WebUi; /// - 跟踪 的启停，维护每次启动的 epoch 与结构索引；
/// - 为每个入口用 挂一个"轮次完成"回调，做值 diff；
/// - 把结果序列化成 SSE 帧，扇出给各订阅者（每帧只序列化一次）。
-/// 同时是 （随宿主启停）与订阅者集合的持有者：订阅相关的成员只出现在内部的 +/// 同时是订阅者集合的持有者：订阅相关的成员只出现在内部的 /// 上，只服务同一程序集里的端点与 。 /// +/// 启动时机：由 MapTagsStream()（）启动—— +/// 映射端点必然发生在任何请求之前，也早于 StartPollAsync，因此不依赖 Generic Host 的 +/// IHostedService 机制；启动后运行期的状态迁移一律以 为准 +/// （见 ）。 +/// +/// /// 线程模型：轮次回调在轮询线程上执行，只做 O(改动的点) 的字典写入 + 一次 TryWrite， /// 不做序列化、不做 IO、不阻塞；序列化与扇出在单独的发送循环里，每个订阅者另有自己的写循环与有界队列。 /// 任何一环满了都只丢帧并触发重同步，不会反压到轮询。 /// /// -internal sealed class TagsStreamHostedService : ITagsStreamHub, IHostedService, IDisposable +internal sealed class TagsStreamHub : ITagsStreamHub, IDisposable { private readonly ITagsProjectCtrl _ctrl; private readonly TagsStreamOptions _options; - private readonly ILogger? _logger; + private readonly ILogger? _logger; /// 订阅者集合与状态发布。_transitionGate → _gate 是唯一允许的加锁顺序。 private readonly object _gate = new(); @@ -53,7 +58,7 @@ internal sealed class TagsStreamHostedService : ITagsStreamHub, IHostedService, /// 测点项目控制器（需由 AddTagsProjectServices 注册） /// 配置 /// 日志 - internal TagsStreamHostedService(ITagsProjectCtrl ctrl, TagsStreamOptions options, ILogger? logger = null) + internal TagsStreamHub(ITagsProjectCtrl ctrl, TagsStreamOptions options, ILogger? logger = null) { this._ctrl = ctrl ?? throw new ArgumentNullException(nameof(ctrl)); this._options = options ?? throw new ArgumentNullException(nameof(options)); @@ -87,12 +92,23 @@ internal TagsStreamHostedService(ITagsProjectCtrl ctrl, TagsStreamOptions option /// public long Epoch => this._state.Epoch; - /// - public Task StartAsync(CancellationToken cancellationToken) + /// + /// 幂等启动：订阅 、拉起发送循环与心跳循环， + /// 并按 对账（补上"启动之前项目就已经在跑"的索引与轮次回调）。
+ /// 由 MapTagsStream() 统一调用（）； + /// 启动之后一直订阅到 /——项目停止时不能退订， + /// 否则接不到下一次"已启动"。 + /// + /// 同步：内部只有"订阅事件 + 起两个后台循环 + 一次对账"，没有 IO，全部同步完成； + /// 所以刻意不返回 ——调用方（同步的 MapTagsStream）不必也不需要阻塞等待， + /// 从而不给"同步等待异步"留下死锁口子。 不同，它要等循环真正退出。 + /// + /// + public void Start() { if (Interlocked.Exchange(ref this._started, 1) != 0) { - return Task.CompletedTask; + return; } var cts = new CancellationTokenSource(); @@ -103,10 +119,12 @@ public Task StartAsync(CancellationToken cancellationToken) // 订阅之前项目可能就已经在跑（hub 启动得晚）：先对一次账。 this.SyncWithController(); - return Task.CompletedTask; } - /// + /// + /// 停止观测：退订事件、取消并等待发送/心跳循环退出。
+ /// 正常停机由 DI 容器释放单例时的 兜底（只取消、不等待）。 + /// public async Task StopAsync(CancellationToken cancellationToken) { this._ctrl.StartedOrStopped -= this.OnStartedOrStopped; diff --git a/src/StdUnit.Tags.WebUi/Sse/StreamEndpoint.cs b/src/StdUnit.Tags.WebUi/Sse/StreamEndpoint.cs index 8508897..009437a 100644 --- a/src/StdUnit.Tags.WebUi/Sse/StreamEndpoint.cs +++ b/src/StdUnit.Tags.WebUi/Sse/StreamEndpoint.cs @@ -26,7 +26,7 @@ public static async Task HandleAsync(HttpContext context) context.Response.Headers["X-Accel-Buffering"] = "no"; // nginx var entries = ParseEntries(context.Request.Query["entries"]); - using var piper = entry.HostedService.Subscribe(entries); + using var piper = entry.Hub.Subscribe(entries); try { diff --git a/src/StdUnit.Tags.WebUi/Tree/TreeEndpoint.cs b/src/StdUnit.Tags.WebUi/Tree/TreeEndpoint.cs index 03cb9eb..86ad559 100644 --- a/src/StdUnit.Tags.WebUi/Tree/TreeEndpoint.cs +++ b/src/StdUnit.Tags.WebUi/Tree/TreeEndpoint.cs @@ -18,7 +18,7 @@ public static async Task HandleAsync(HttpContext context) } // 自己序列化：不走 MVC / 无参序列化，AOT/裁剪下没有反射路径。 - var bytes = JsonSerializer.SerializeToUtf8Bytes(entry.HostedService.GetTree(), TagsStreamJsonContext.Default.TreeDto); + var bytes = JsonSerializer.SerializeToUtf8Bytes(entry.Hub.GetTree(), TagsStreamJsonContext.Default.TreeDto); context.Response.StatusCode = StatusCodes.Status200OK; context.Response.ContentType = "application/json; charset=utf-8"; diff --git a/src/StdUnit.Tags.WebUi/_Extensions_Endpoints.cs b/src/StdUnit.Tags.WebUi/_Extensions_Endpoints.cs index 25c062a..b690602 100644 --- a/src/StdUnit.Tags.WebUi/_Extensions_Endpoints.cs +++ b/src/StdUnit.Tags.WebUi/_Extensions_Endpoints.cs @@ -1,5 +1,6 @@ using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Routing; +using Microsoft.Extensions.DependencyInjection; using StdUnit.Tags.WebUi.Internal; namespace StdUnit.Tags.WebUi; @@ -39,6 +40,7 @@ public static class TagsStreamExtensions_Endpoints /// 设计器元数据：GET {prefix}/catalog 把核心 + 驱动 schema 内省成 JSON， /// GET {prefix}/designer 就是用它渲染的设计器页面。两者都与具体观测流无关， /// 只有这一份、没有具名版本；只读，不涉及项目数据。 + ///
/// /// 端点构建器 /// @@ -48,6 +50,11 @@ public static class TagsStreamExtensions_Endpoints /// 路由组：可继续 RequireAuthorization() / WithTags() 等整组配置。 public static RouteGroupBuilder MapTagsStream(this IEndpointRouteBuilder endpoints, string prefix = EndpointConstants.DefaultPrefix) { + // 观测流的**唯一启动点**：映射端点必然早于任何请求，也早于 StartPollAsync， + // 所以 hub 不需要注册成 IHostedService，也不依赖 Generic Host 的启动顺序。 + // 只 map 不 AddTagsStream 时是 no-op（端点会在运行时回 404）。 + endpoints.ServiceProvider.GetService()?.StartAll(); + var group = endpoints.MapGroup(prefix.Length == 0 ? "/" : prefix); // 不具名流：{prefix}/tree|stream|value|rescan|ui diff --git a/src/StdUnit.Tags.WebUi/_Extensions_ServiceCollection.cs b/src/StdUnit.Tags.WebUi/_Extensions_ServiceCollection.cs index 85fea89..4f87be5 100644 --- a/src/StdUnit.Tags.WebUi/_Extensions_ServiceCollection.cs +++ b/src/StdUnit.Tags.WebUi/_Extensions_ServiceCollection.cs @@ -1,7 +1,6 @@ using StdUnit.Tags; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection.Extensions; -using Microsoft.Extensions.Hosting; using StdUnit.Tags.WebUi.Internal; @@ -18,10 +17,10 @@ public static class TagsStreamExtensions_ServiceCollection /// 端点落在 {prefix}/tree|stream|value|rescan|ui。
/// 一个容器里只能有一路不具名流；要多路请用 。
///
- /// 启动顺序：hub 以 注册，随宿主一起启动。它在启动时按 - /// 对账，所以晚启动也能补上：即使项目已在轮询， - /// 结构树与值流都正常（轮次回调运行期也能挂，从下一轮开始生效）。 - /// 想尽早拿到结构树，就把轮询放在宿主启动之后（例如在请求里/更晚注册的 hosted service 里 StartPollAsync）。 + /// 启动时机：hub 在 MapTagsStream() 映射端点时启动（那必然早于任何请求，也早于 StartPollAsync）， + /// 不依赖 Generic Host 的 IHostedService。DI 注册本来就在 Build() 之前完成、映射在其后， + /// 所以这个顺序天然满足；启动时它会按 对账， + /// 项目已经在轮询也能补上（轮次回调运行期也能挂，从下一轮开始生效）。 /// /// 服务集合 /// 这一路流的配置回调 @@ -39,19 +38,12 @@ public static IServiceCollection AddTagsStream(this IServiceCollection services, services.AddSingleton(new TagsStreamDescriptor(key: null, options)); - // 同一实例注册为 IHostedService：用工厂而不是 AddHostedService()——后者会另开 - // IHostedService → T 的描述符再 new 一个，宿主启动的 hub 与端点看到的不是同一份数据。 - services.AddSingleton( - sp => sp.GetRequiredService() - .GetRequired(null) - .HostedService - ); - // 不具名那一路按 ITagsStreamHub 解析（具名那一路是 keyed 服务）。 + // 启动点是 MapTagsStream()：见 TagsStreamRegistry.StartAll()。 services.TryAddSingleton( sp => sp.GetRequiredService() .GetRequired(null) - .HostedService + .Hub ); return services; @@ -60,7 +52,7 @@ public static IServiceCollection AddTagsStream(this IServiceCollection services, /// /// 注册一路具名观测流：观测由 AddKeyedTagsProjectCtrl(key) 注册的 ， /// 端点落在 {prefix}/{key}/tree|stream|value|rescan|ui（监控页各一份）。
- /// 可与 并存；同一个 key 只能注册一路。启动顺序等约束同 。 + /// 可与 并存；同一个 key 只能注册一路。启动时机等约束同 。 /// /// 服务集合 /// @@ -78,17 +70,12 @@ public static IServiceCollection AddKeyedTagsStream( this IServiceCollection ser services.AddSingleton(new TagsStreamDescriptor(projctrlKey, options)); - // 同一实例注册为 IHostedService / keyed ITagsStreamHub（原因见 AddTagsStream 里的注释）。 - services.AddSingleton( - sp => sp.GetRequiredService() - .GetRequired(projctrlKey) - .HostedService - ); + // 同一实例注册为 keyed ITagsStreamHub（原因见 AddTagsStream 里的注释）。 services.AddKeyedSingleton( projctrlKey, (sp, _) => sp.GetRequiredService() .GetRequired(projctrlKey) - .HostedService + .Hub ); return services; diff --git a/src/StdUnit.Tags.WebUi/_Extensions_TagsProjectServiceBuilder.cs b/src/StdUnit.Tags.WebUi/_Extensions_TagsProjectServiceBuilder.cs index eb67efc..0101602 100644 --- a/src/StdUnit.Tags.WebUi/_Extensions_TagsProjectServiceBuilder.cs +++ b/src/StdUnit.Tags.WebUi/_Extensions_TagsProjectServiceBuilder.cs @@ -13,9 +13,9 @@ public static class TagsStreamExtensions_TagsProjectServiceBuilder /// 和不具名的 关联；一个容器里最多只有一路不具名的观测流。 /// /// - /// 启动顺序：hub 以 IHostedService 注册，随宿主一起启动；它在启动时按 ITagsProjectCtrl.Project 对账， + /// 启动时机：hub 在 MapTagsStream() 映射端点时启动（早于任何请求，也早于 StartPollAsync）， + /// 不依赖 Generic Host 的 IHostedService。它在启动时按 ITagsProjectCtrl.Project 对账， /// 所以晚启动也能补上——即使项目已在轮询，结构树与值流都正常（轮次回调运行期也能挂，从下一轮开始生效）。 - /// 想尽早拿到结构树，就把轮询放在宿主启动之后（例如在请求里/更晚注册的 hosted service 里 StartPollAsync）。 /// /// /// 这一路流的配置回调 @@ -33,7 +33,7 @@ public static TagsProjectServiceBuilder AddTagsStream(this TagsProjectServiceBui /// /// /// 时序约束同 ： - /// hub 在启动时对账，晚启动也能补上，顺序只影响"多早开始观测"。 + /// hub 在 MapTagsStream() 时启动，端口映射发生在任何请求之前，所以顺序只影响"多早开始观测"。 /// /// /// diff --git a/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/EndpointTests.cs b/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/EndpointTests.cs index b3cc302..1206288 100644 --- a/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/EndpointTests.cs +++ b/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/EndpointTests.cs @@ -412,10 +412,10 @@ public ITagsProject StartProject(XElement xml) return project; } - public TagsStreamHostedService StartHub(string? key = null) + public TagsStreamHub StartHub(string? key = null) { - var hub = this.Root.GetRequiredService().GetRequired(key).HostedService; - hub.StartAsync(default).GetAwaiter().GetResult(); + var hub = this.Root.GetRequiredService().GetRequired(key).Hub; + hub.Start(); return hub; } diff --git a/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamHubEdgeCaseTests.cs b/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamHubEdgeCaseTests.cs index 55fbcba..07ff55f 100644 --- a/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamHubEdgeCaseTests.cs +++ b/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamHubEdgeCaseTests.cs @@ -11,7 +11,7 @@ namespace StdUnit.Tags.Tests.NetCoreOnly.WebUi; /// -/// 的边界行为：参数校验、启停幂等、无订阅者时的零开销、 +/// 的边界行为：参数校验、启停幂等、无订阅者时的零开销、 /// 心跳、按 拆帧、容器标志单独成帧， /// 以及 队列溢出时的"丢弃 + 重同步"（绝不反压）。 /// @@ -47,26 +47,26 @@ public void Ctor_RejectsNullAndOutOfRangeArguments() { var ok = Options(); - Assert.Throws(() => new TagsStreamHostedService(null!, ok)); - Assert.Throws(() => new TagsStreamHostedService(new FakeProjectCtrl(), null!)); + Assert.Throws(() => new TagsStreamHub(null!, ok)); + Assert.Throws(() => new TagsStreamHub(new FakeProjectCtrl(), null!)); Assert.Throws(() => - new TagsStreamHostedService(new FakeProjectCtrl(), new TagsStreamOptions { ReconcileInterval = TimeSpan.Zero })); + new TagsStreamHub(new FakeProjectCtrl(), new TagsStreamOptions { ReconcileInterval = TimeSpan.Zero })); Assert.Throws(() => - new TagsStreamHostedService(new FakeProjectCtrl(), new TagsStreamOptions { HeartbeatInterval = TimeSpan.Zero })); + new TagsStreamHub(new FakeProjectCtrl(), new TagsStreamOptions { HeartbeatInterval = TimeSpan.Zero })); Assert.Throws(() => - new TagsStreamHostedService(new FakeProjectCtrl(), new TagsStreamOptions { PiperQueueCapacity = 0 })); + new TagsStreamHub(new FakeProjectCtrl(), new TagsStreamOptions { PiperQueueCapacity = 0 })); Assert.Throws(() => - new TagsStreamHostedService(new FakeProjectCtrl(), new TagsStreamOptions { PendingQueueCapacity = 0 })); + new TagsStreamHub(new FakeProjectCtrl(), new TagsStreamOptions { PendingQueueCapacity = 0 })); } [Fact] public async Task StartAndStop_AreIdempotent() { - using var hub = new TagsStreamHostedService(new FakeProjectCtrl(), Options()); + using var hub = new TagsStreamHub(new FakeProjectCtrl(), Options()); - await hub.StartAsync(default); - await hub.StartAsync(default); // 第二次什么也不做 + hub.Start(); + hub.Start(); // 第二次什么也不做 await hub.StopAsync(default); await hub.StopAsync(default); // 第二次时 pump/heartbeat 已为 null } @@ -82,8 +82,8 @@ public async Task OnTurn_WithoutSubscribers_AndAfterStop_IsIgnored() var ctrl = new FakeProjectCtrl(); ctrl.RaiseStarted(project); - using var hub = new TagsStreamHostedService(ctrl, Options()); - await hub.StartAsync(default); + using var hub = new TagsStreamHub(ctrl, Options()); + hub.Start(); using var runCts = new CancellationTokenSource(); var run = project.RunAsync(runCts.Token); @@ -116,8 +116,8 @@ public async Task Heartbeat_SendsPingToSubscribers() var options = Options(); options.HeartbeatInterval = TimeSpan.FromMilliseconds(50); - using var hub = new TagsStreamHostedService(new FakeProjectCtrl(), options); - await hub.StartAsync(default); + using var hub = new TagsStreamHub(new FakeProjectCtrl(), options); + hub.Start(); using var output = new RecordingStream(); using var piperCts = new CancellationTokenSource(); @@ -157,8 +157,8 @@ public async Task Delta_IsSplitByMaxValuesPerFrame() var ctrl = new FakeProjectCtrl(); ctrl.RaiseStarted(project); - using var hub = new TagsStreamHostedService(ctrl, options); - await hub.StartAsync(default); + using var hub = new TagsStreamHub(ctrl, options); + hub.Start(); using var output = new RecordingStream(); using var piperCts = new CancellationTokenSource(); @@ -218,8 +218,8 @@ public async Task ContainerFlagChange_IsPublishedWithoutValues() var ctrl = new FakeProjectCtrl(); ctrl.RaiseStarted(project); - using var hub = new TagsStreamHostedService(ctrl, Options()); - await hub.StartAsync(default); + using var hub = new TagsStreamHub(ctrl, Options()); + hub.Start(); using var output = new RecordingStream(); using var piperCts = new CancellationTokenSource(); diff --git a/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamHubTests.cs b/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamHubTests.cs index e7949b8..bac41ba 100644 --- a/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamHubTests.cs +++ b/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamHubTests.cs @@ -10,7 +10,7 @@ namespace StdUnit.Tags.Tests.NetCoreOnly.WebUi; /// -/// 的端到端行为：状态（Replay 1）、全量基线、增量、按入口过滤、迟到订阅者。
+/// 的端到端行为：状态（Replay 1）、全量基线、增量、按入口过滤、迟到订阅者。
/// 用真实的 TagsProject + 真实 TagGrpRunner（假驱动），只把 换成可控的假实现， /// 以便精确决定"什么时候启动/停止"。 /// @@ -54,8 +54,8 @@ public async Task Hub_PushesStateBaselineAndDelta() { using var project = CreateSingleEntryProject(this._scope); var ctrl = new FakeProjectCtrl(); - using var hub = new TagsStreamHostedService(ctrl, FastOptions()); - await hub.StartAsync(default); + using var hub = new TagsStreamHub(ctrl, FastOptions()); + hub.Start(); using var output = new RecordingStream(); using var piperCts = new CancellationTokenSource(); @@ -102,8 +102,8 @@ public async Task Subscribe_FiltersByEntry() StreamTestContext.EntryGroup("g2", "fake2", StreamTestContext.Tag("a")))); var ctrl = new FakeProjectCtrl(); - using var hub = new TagsStreamHostedService(ctrl, FastOptions()); - await hub.StartAsync(default); + using var hub = new TagsStreamHub(ctrl, FastOptions()); + hub.Start(); using var output = new RecordingStream(); using var piperCts = new CancellationTokenSource(); @@ -135,8 +135,8 @@ public async Task SubscribeAfterStart_ReceivesBaselineImmediately() { using var project = CreateSingleEntryProject(this._scope); var ctrl = new FakeProjectCtrl(); - using var hub = new TagsStreamHostedService(ctrl, FastOptions()); - await hub.StartAsync(default); + using var hub = new TagsStreamHub(ctrl, FastOptions()); + hub.Start(); // 先启动（不跑轮询），再订阅：订阅者应立即拿到状态 + 全量基线。 ctrl.RaiseStarted(project); @@ -164,8 +164,8 @@ public async Task Hub_StartedAfterProject_StillAttachesTurnHandlers() using var runCts = new CancellationTokenSource(); var run = project.RunAsync(runCts.Token); - using var hub = new TagsStreamHostedService(ctrl, FastOptions()); - await hub.StartAsync(default); + using var hub = new TagsStreamHub(ctrl, FastOptions()); + hub.Start(); Assert.True(hub.GetTree().IsStarted); using var output = new RecordingStream(); diff --git a/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamMappingTests.cs b/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamMappingTests.cs index 2cfb3e8..864b53f 100644 --- a/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamMappingTests.cs +++ b/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamMappingTests.cs @@ -2,11 +2,14 @@ using System.Linq; using System.Text; using System.Threading.Tasks; +using System.Xml.Linq; using Microsoft.AspNetCore.Authorization; using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Routing; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; +using StdUnit.Tags.Tests.Fakes; using StdUnit.Tags.WebUi; using Xunit; @@ -72,6 +75,50 @@ public async Task MapTagsStream_PutsEveryEndpointUnderThePrefix(string prefix) } } + /// + /// MapTagsStream 是观测流的唯一启动点：映射之后（不需要宿主启动、也不需要任何请求） + /// hub 就已经在观测，并且能补上"映射之前项目就已经在跑"的状态。 + /// + [Fact] + public async Task MapTagsStream_StartsEveryHub() + { + var builder = WebApplication.CreateSlimBuilder(); + builder.Logging.ClearProviders(); + + var unnamed = new FakeProjectCtrl(); + var named = new FakeProjectCtrl(); + builder.Services.AddTagsProjectServices(b => + { + b.AddFakedSupport(); + b.AddFakedTagSupport(); + }); + builder.Services.AddSingleton(unnamed); // 后注册的覆盖默认实现 + builder.Services.AddKeyedSingleton("dev", named); + builder.Services.AddTagsStream(); + builder.Services.AddKeyedTagsStream("dev"); + + using var app = builder.Build(); + using var project = app.Services.GetRequiredService().Create( + StreamTestContext.ProjectRoot, + new XElement("Project", + StreamTestContext.Channel("fake"), + StreamTestContext.EntryGroup("g1", StreamTestContext.Tag("a")))); + + // 项目先跑起来，之后才挂端点。 + unnamed.RaiseStarted(project); + named.RaiseStarted(project); + + app.MapTagsStream(); // ← 唯一启动点 + + var hub = app.Services.GetRequiredService(); + Assert.True(hub.GetTree().IsStarted); + Assert.NotEmpty(hub.GetTree().Nodes); + + var devHub = app.Services.GetRequiredKeyedService("dev"); + Assert.True(devHub.GetTree().IsStarted); + Assert.NotEmpty(devHub.GetTree().Nodes); + } + [Theory] // 路径 → 期望的 （页面自己的 URL 去掉最后一段） [InlineData("/stdunit.tags/ui", "", "/stdunit.tags/")] diff --git a/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamRegistrationTests.cs b/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamRegistrationTests.cs index d249f48..5affab2 100644 --- a/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamRegistrationTests.cs +++ b/tests/StdUnit.Tags.Tests.NetCoreOnly/WebUi/TagsStreamRegistrationTests.cs @@ -3,7 +3,6 @@ using System.Threading.Tasks; using System.Xml.Linq; using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Hosting; using StdUnit.Tags.Tests.Fakes; using StdUnit.Tags.WebUi; using Xunit; @@ -11,8 +10,9 @@ namespace StdUnit.Tags.Tests.NetCoreOnly.WebUi; /// -/// AddTagsStream：对外只暴露 ，且它与 hosted service 是同一个实例； -/// 具名流用 AddKeyedTagsStream(key) 决定观测哪个 。 +/// AddTagsStream：对外只暴露 ，且它就是 registry 里那一路的同一个实例； +/// 具名流用 AddKeyedTagsStream(key) 决定观测哪个 。
+/// 启动点是 MapTagsStream()（）——这里直接调 registry 模拟。 /// public class TagsStreamRegistrationTests { @@ -51,8 +51,8 @@ public void AddTagsStream_ExposesOnlyTheInterfaceAndReusesTheSameInstance() var hub = provider.GetRequiredService(); Assert.Same(hub, provider.GetRequiredService()); // Singleton - // 同一个实例既是接口服务，又是随宿主启停的 IHostedService。 - Assert.Contains(provider.GetServices(), h => ReferenceEquals(h, hub)); + // 同一个实例：接口服务 = registry 里那一路（端点与启动点共用）。 + Assert.Same(hub, provider.GetRequiredService().GetRequired(null).Hub); Assert.Equal(0, hub.Epoch); var tree = hub.GetTree(); @@ -75,9 +75,10 @@ public async Task AddTagsStream_BindsToTheKeyedControllerWhenKeyIsGiven() using var project = CreateProject(provider); var hub = provider.GetRequiredKeyedService("dev"); - var hosted = Assert.Single(provider.GetServices()); - Assert.Same(hub, hosted); - await hosted.StartAsync(default); + var registry = provider.GetRequiredService(); + var entry = registry.GetRequired("dev"); + Assert.Same(hub, entry.Hub); + registry.StartAll(); try { // 不具名的那个被启动，本 hub 不应有任何反应（它绑的是 "dev"）。 @@ -93,7 +94,7 @@ public async Task AddTagsStream_BindsToTheKeyedControllerWhenKeyIsGiven() } finally { - await hosted.StopAsync(default); + await entry.Hub.StopAsync(default); } } @@ -116,15 +117,11 @@ public async Task AddTagsStream_RegistersOneIndependentHubPerStream() var devHub = provider.GetRequiredKeyedService("dev"); Assert.NotSame(defaultHub, devHub); - // 两路各一个 hosted service，且都指向各自的 hub。 - var hosted = provider.GetServices().ToArray(); - Assert.Equal(2, hosted.Length); - Assert.Contains(hosted, h => ReferenceEquals(h, defaultHub)); - Assert.Contains(hosted, h => ReferenceEquals(h, devHub)); - foreach (var service in hosted) - { - await service.StartAsync(default); - } + // 两路各一个 hub，且接口服务指向各自的那一个。 + var registry = provider.GetRequiredService(); + Assert.Same(defaultHub, registry.GetRequired(null).Hub); + Assert.Same(devHub, registry.GetRequired("dev").Hub); + registry.StartAll(); try { @@ -139,9 +136,9 @@ public async Task AddTagsStream_RegistersOneIndependentHubPerStream() } finally { - foreach (var service in hosted) + foreach (var entry in registry.All) { - await service.StopAsync(default); + await entry.Hub.StopAsync(default); } } } @@ -230,8 +227,9 @@ public async Task AddTagsStream_BindsToTheUnnamedControllerByDefault() using var project = CreateProject(provider); var hub = provider.GetRequiredService(); - var hosted = Assert.Single(provider.GetServices()); - await hosted.StartAsync(default); + var registry = provider.GetRequiredService(); + var entry = registry.GetRequired(null); + registry.StartAll(); try { named.RaiseStarted(project); @@ -243,7 +241,7 @@ public async Task AddTagsStream_BindsToTheUnnamedControllerByDefault() } finally { - await hosted.StopAsync(default); + await entry.Hub.StopAsync(default); } } @@ -268,7 +266,7 @@ public void AddTagsStream_OnTheProjectBuilder_RegistersTheUnnamedStream() var hub = provider.GetRequiredService(); Assert.Same(hub, provider.GetRequiredService()); - Assert.Contains(provider.GetServices(), h => ReferenceEquals(h, hub)); + Assert.Same(hub, provider.GetRequiredService().GetRequired(null).Hub); // configure 落在这一路流自己的描述符上（不是被丢掉的临时对象）。 var descriptor = Assert.Single(provider.GetServices()); @@ -296,8 +294,7 @@ public void AddKeyedTagsStream_OnTheProjectBuilder_RegistersTheKeyedStream() using var provider = services.BuildServiceProvider(); var hub = provider.GetRequiredKeyedService("dev"); - var hosted = Assert.Single(provider.GetServices()); - Assert.Same(hub, hosted); + Assert.Same(hub, provider.GetRequiredService().GetRequired("dev").Hub); var descriptor = Assert.Single(provider.GetServices()); Assert.Equal("dev", descriptor.Key);