diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/Pages/SessionDetailsPage.razor b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/Pages/SessionDetailsPage.razor index d05ef32..af50f53 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/Pages/SessionDetailsPage.razor +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Components/Pages/SessionDetailsPage.razor @@ -1,11 +1,11 @@ @page "/sessions/{SessionId}" @inherits DashboardPageBase @implements IAsyncDisposable -@using Microsoft.AspNetCore.SignalR.Client @using ZB.MOM.WW.MxGateway.Contracts.Proto @using ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs @inject AuthenticationStateProvider AuthenticationStateProvider @inject IDashboardSessionAdminService SessionAdminService +@inject IDashboardEventBroadcaster EventBroadcaster Dashboard Session @@ -157,7 +157,8 @@ else private DashboardSessionSummary? CurrentSession => Snapshot?.Sessions.FirstOrDefault(session => string.Equals(session.SessionId, SessionId, StringComparison.Ordinal)); - private HubConnection? _eventsHub; + private IDashboardEventSubscription? _eventSubscription; + private CancellationTokenSource? _eventPumpCancellation; private bool _eventsConnected; private string? _subscribedSessionId; private readonly LinkedList _recentEvents = new(); @@ -179,13 +180,17 @@ else CanManage = SessionAdminService.CanManage(authenticationState.User); } - protected override async Task OnParametersSetAsync() + protected override Task OnParametersSetAsync() { + // Attach/detach are synchronous now that the feed is in-process; the override + // stays on the async lifecycle member so the base class's contract is untouched. if (!string.Equals(_subscribedSessionId, SessionId, StringComparison.Ordinal)) { - await DetachEventsHubAsync().ConfigureAwait(false); - await AttachEventsHubAsync().ConfigureAwait(false); + DetachEvents(); + AttachEvents(); } + + return Task.CompletedTask; } private PendingConfirm? PendingAction { get; set; } @@ -261,68 +266,91 @@ else string ConfirmButtonClass, Func> Action); - private async Task AttachEventsHubAsync() + // The dashboard runs in the same process as the broadcaster, so this page reads + // the session's mirrored events straight from it. It used to open a loopback + // SignalR connection to /hubs/events — mint a hub token, negotiate, hold a + // WebSocket, serialize every event — to reach data already sitting in memory. + // The subscription still registers with EventsHubViewerRegistry, so the + // broadcaster's "nobody is watching" gate keeps working for both audiences. + // ACL posture is unchanged from the hub path: any dashboard Viewer may watch + // any session (SEC-25 tracks the per-session ACL for both seams). + private void AttachEvents() { - if (string.IsNullOrWhiteSpace(SessionId)) + if (string.IsNullOrWhiteSpace(SessionId) || EventBroadcaster is not IDashboardSessionEventSubscriber subscriber) { return; } - _eventsHub = HubFactory.Create("/hubs/events"); - _eventsHub.On(EventsHub.EventMessage, async mxEvent => - { - _recentEvents.AddFirst(mxEvent); - while (_recentEvents.Count > MaxRecentEvents) - { - _recentEvents.RemoveLast(); - } + _eventSubscription = subscriber.Subscribe(SessionId); + _eventPumpCancellation = new CancellationTokenSource(); + _eventsConnected = true; + _subscribedSessionId = SessionId; - await InvokeAsync(StateHasChanged).ConfigureAwait(false); - }); - - _eventsHub.Closed += _ => - { - _eventsConnected = false; - return InvokeAsync(StateHasChanged); - }; - _eventsHub.Reconnected += _ => - { - _eventsConnected = true; - return InvokeAsync(StateHasChanged); - }; + _ = PumpEventsAsync(_eventSubscription, _eventPumpCancellation.Token); + } + private async Task PumpEventsAsync(IDashboardEventSubscription subscription, CancellationToken cancellationToken) + { try { - await _eventsHub.StartAsync().ConfigureAwait(false); - await _eventsHub.SendAsync("SubscribeSession", SessionId).ConfigureAwait(false); - _eventsConnected = true; - _subscribedSessionId = SessionId; + while (await subscription.Reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false)) + { + // Drain what is queued and render once: a burst costs one render pass, + // not one per event. Reading past the display cap would be wasted work, + // and anything left queued is picked up on the next pass. + List batch = new(); + while (batch.Count < MaxRecentEvents && subscription.Reader.TryRead(out MxEvent? mxEvent)) + { + batch.Add(mxEvent); + } + + if (batch.Count == 0) + { + continue; + } + + await InvokeAsync(() => + { + foreach (MxEvent mxEvent in batch) + { + _recentEvents.AddFirst(mxEvent); + } + + while (_recentEvents.Count > MaxRecentEvents) + { + _recentEvents.RemoveLast(); + } + + StateHasChanged(); + }).ConfigureAwait(false); + } } - catch + catch (OperationCanceledException) { - _eventsConnected = false; + // The page navigated to another session or was disposed. + } + catch (ObjectDisposedException) + { + // The renderer went away while a batch was being dispatched. } } - private async Task DetachEventsHubAsync() + private void DetachEvents() { - HubConnection? hub = _eventsHub; - _eventsHub = null; + IDashboardEventSubscription? subscription = _eventSubscription; + CancellationTokenSource? cancellation = _eventPumpCancellation; + _eventSubscription = null; + _eventPumpCancellation = null; _eventsConnected = false; _subscribedSessionId = null; _recentEvents.Clear(); - if (hub is not null) - { - try - { - await hub.DisposeAsync().ConfigureAwait(false); - } - catch - { - // Disposal-time errors are best-effort. - } - } + // Cancel first so the pump stops touching the renderer, then dispose the + // subscription — that is what releases the viewer registration and lets the + // broadcaster go back to skipping mirror work for this session. + cancellation?.Cancel(); + cancellation?.Dispose(); + subscription?.Dispose(); } private static string EventStatusLabel(MxEvent evt) @@ -334,7 +362,7 @@ else public new async ValueTask DisposeAsync() { - await DetachEventsHubAsync().ConfigureAwait(false); + DetachEvents(); await base.DisposeAsync().ConfigureAwait(false); } } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs index 33597d6..cf94e1d 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs @@ -1,3 +1,4 @@ +using System.Threading.Channels; using Microsoft.AspNetCore.SignalR; using Microsoft.Extensions.Options; using ZB.MOM.WW.MxGateway.Contracts.Proto; @@ -6,11 +7,13 @@ using ZB.MOM.WW.MxGateway.Server.Configuration; namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; /// -/// Broadcasts MxEvents to clients subscribed to the -/// session's group. Fire-and-forget: we hand the send to the hub context -/// and return immediately so the source gRPC stream is never blocked. -/// Errors are logged once and dropped — keeping the SignalR mirror best-effort -/// preserves the gRPC contract that exists today. +/// Broadcasts MxEvents to the two dashboard audiences for a session: remote +/// clients subscribed to the session's group, and +/// in-process subscribers opened through +/// . Fire-and-forget: we +/// hand the send to the hub context and return immediately so the source gRPC +/// stream is never blocked. Errors are logged once and dropped — keeping the +/// SignalR mirror best-effort preserves the gRPC contract that exists today. /// /// /// When MxGateway:Dashboard:ShowTagValues is false (the default), tag @@ -23,7 +26,10 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; /// /// Hub context used to send to the session's group. /// -/// Live-subscriber registry consulted before any per-event work is done. +/// Live-subscriber registry consulted before any per-event work is done. Both +/// audiences register here — hub connections by their SignalR connection id, +/// in-process subscriptions by a synthetic one — so the gate stays a single +/// source of truth. /// /// Gateway options supplying Dashboard:ShowTagValues. /// Logger for best-effort mirror failures. @@ -31,10 +37,36 @@ public sealed class DashboardEventBroadcaster( IHubContext hubContext, EventsHubViewerRegistry viewerRegistry, IOptions options, - ILogger logger) : IDashboardEventBroadcaster + ILogger logger) : IDashboardEventBroadcaster, IDashboardSessionEventSubscriber { + /// + /// Queue depth per in-process subscriber. The consumer is a Blazor page + /// rendering the newest handful of events, so a burst it cannot keep up with + /// is dropped oldest-first rather than allowed to grow — same best-effort + /// contract the SignalR mirror already has. + /// + private const int InProcessQueueCapacity = 256; + private readonly bool _showTagValues = options.Value.Dashboard.ShowTagValues; + private readonly object _syncRoot = new(); + + /// + /// In-process subscribers per session. Values are treated as immutable once + /// stored: a subscribe or dispose swaps in a new array under + /// , so can grab the reference + /// and write to it after releasing the lock. + /// + private readonly Dictionary _inProcessSubscribers = + new(StringComparer.Ordinal); + + /// + /// Total live in-process subscribers, read without the lock so the common + /// case — nobody has a session-details page open — never contends on it. + /// Written only under . + /// + private int _inProcessSubscriberCount; + /// public void Publish(string sessionId, MxEvent mxEvent) { @@ -55,6 +87,10 @@ public sealed class DashboardEventBroadcaster( MxEvent outbound = _showTagValues ? mxEvent : RedactValues(mxEvent); + // In-process delivery first: it is synchronous, cannot throw, and must not be + // skipped by the early return the hub send's guard clause takes. + DeliverInProcess(sessionId, outbound); + // Wrap the Task acquisition in a try/catch so a hypothetical synchronous throw // from SendAsync (e.g. an implementation that throws before returning the Task) // cannot escape Publish. The interface contract is never-throw; fire-and-forget. @@ -85,6 +121,117 @@ public sealed class DashboardEventBroadcaster( } } + /// + public IDashboardEventSubscription Subscribe(string sessionId) + { + ArgumentException.ThrowIfNullOrWhiteSpace(sessionId); + + // A synthetic connection id keeps the registry's per-connection bookkeeping + // usable for a subscriber that has no SignalR connection behind it. The + // "inproc-" prefix cannot collide with a SignalR connection id and makes the + // origin obvious in a debugger. + string connectionId = "inproc-" + Guid.NewGuid().ToString("N"); + InProcessSubscription subscription = new(this, sessionId, connectionId, InProcessQueueCapacity); + + // Register before the subscriber becomes a delivery target, exactly as + // EventsHub.SubscribeSession registers before joining the group: the reverse + // order would leave a window in which this subscriber is a delivery target but + // Publish's gate still reports the session unwatched, silently dropping events + // it should receive. The cost of this order is at worst a redaction clone that + // reaches nobody for the width of the window. + viewerRegistry.AddViewer(connectionId, sessionId); + + lock (_syncRoot) + { + _inProcessSubscribers[sessionId] = + _inProcessSubscribers.TryGetValue(sessionId, out InProcessSubscription[]? existing) + ? [.. existing, subscription] + : [subscription]; + + Volatile.Write(ref _inProcessSubscriberCount, _inProcessSubscriberCount + 1); + } + + return subscription; + } + + /// + /// Hands the already-redacted event to every in-process subscriber of the + /// session. Writes are non-blocking and lossy by construction, so this never + /// stalls the caller's event pipeline. + /// + /// Session the event belongs to. + /// The event as the dashboard should see it. + private void DeliverInProcess(string sessionId, MxEvent outbound) + { + // The gate above admits hub-only viewers too, so check for in-process + // subscribers before touching the lock at all. + if (Volatile.Read(ref _inProcessSubscriberCount) == 0) + { + return; + } + + InProcessSubscription[] subscribers; + lock (_syncRoot) + { + if (!_inProcessSubscribers.TryGetValue(sessionId, out InProcessSubscription[]? found)) + { + return; + } + + subscribers = found; + } + + // The array is never mutated in place, so the writes happen outside the lock. + foreach (InProcessSubscription subscriber in subscribers) + { + subscriber.TryWrite(outbound); + } + } + + /// + /// Removes a disposed subscription from the delivery map and releases its + /// viewer registration. Called at most once per subscription. + /// + /// The subscription being disposed. + private void Unsubscribe(InProcessSubscription subscription) + { + // Drop the delivery target first and deregister after, mirroring + // EventsHub.UnsubscribeSession: the mirror stays enabled for the brief overlap + // rather than dropping events still owed to the session's other subscribers. + lock (_syncRoot) + { + if (_inProcessSubscribers.TryGetValue(subscription.SessionId, out InProcessSubscription[]? existing)) + { + InProcessSubscription[] remaining = + [.. existing.Where(candidate => !ReferenceEquals(candidate, subscription))]; + + // Equal lengths mean it was never in this bucket, so the counter it + // would decrement is not its own to release. + if (remaining.Length != existing.Length) + { + if (remaining.Length == 0) + { + // Drop the key so the map does not grow one entry per session ever viewed. + _inProcessSubscribers.Remove(subscription.SessionId); + } + else + { + _inProcessSubscribers[subscription.SessionId] = remaining; + } + + Volatile.Write(ref _inProcessSubscriberCount, _inProcessSubscriberCount - 1); + } + } + } + + viewerRegistry.RemoveViewer(subscription.ConnectionId, subscription.SessionId); + + // The synthetic connection id is used once and never reconnects, so nothing + // else will ever call ReleaseConnection for it; without this the registry + // would retain an empty per-connection entry per subscription ever opened. + viewerRegistry.ReleaseConnection(subscription.ConnectionId); + } + /// /// Produces a deep clone of with every tag-value /// field cleared, leaving tag reference, quality, status, and timestamps @@ -107,4 +254,75 @@ public sealed class DashboardEventBroadcaster( return redacted; } + + /// + /// One in-process subscriber's feed: a bounded, drop-oldest channel plus the + /// registry bookkeeping that keeps 's viewer gate honest + /// while the feed is live. + /// + private sealed class InProcessSubscription : IDashboardEventSubscription + { + private readonly DashboardEventBroadcaster _owner; + + private readonly Channel _channel; + + private int _disposed; + + /// Initializes a new instance of the class. + /// Broadcaster to deregister from on disposal. + /// Session whose events this subscription carries. + /// Synthetic connection id registered with the viewer registry. + /// Queue depth before the oldest queued event is dropped. + internal InProcessSubscription( + DashboardEventBroadcaster owner, + string sessionId, + string connectionId, + int capacity) + { + _owner = owner; + SessionId = sessionId; + ConnectionId = connectionId; + _channel = Channel.CreateBounded(new BoundedChannelOptions(capacity) + { + // DropOldest, not Wait: a write must never block the gRPC event + // pipeline that calls Publish, and the newest events are the ones a + // live view wants. + FullMode = BoundedChannelFullMode.DropOldest, + SingleReader = true, + SingleWriter = false, + }); + } + + /// + public ChannelReader Reader => _channel.Reader; + + /// Gets the session this subscription is watching. + internal string SessionId { get; } + + /// Gets the synthetic connection id held in the viewer registry. + internal string ConnectionId { get; } + + /// + /// Queues an event for the subscriber, dropping the oldest queued event when + /// the reader has fallen behind. Never blocks and never throws. + /// + /// The event to queue. + internal void TryWrite(MxEvent mxEvent) => _channel.Writer.TryWrite(mxEvent); + + /// + /// Deregisters the subscription and completes its channel so a reader's + /// loop ends. Idempotent — a second call does nothing, so it can never + /// release a viewer count that a sibling subscription owns. + /// + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + + _owner.Unsubscribe(this); + _channel.Writer.TryComplete(); + } + } } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/IDashboardEventSubscription.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/IDashboardEventSubscription.cs new file mode 100644 index 0000000..9b6685b --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/IDashboardEventSubscription.cs @@ -0,0 +1,26 @@ +using System.Threading.Channels; +using ZB.MOM.WW.MxGateway.Contracts.Proto; + +namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; + +/// +/// A live in-process feed of one session's dashboard-mirrored MxEvents, handed +/// out by . Server-side +/// Blazor components read it directly instead of looping back through +/// over a loopback SignalR connection. +/// +/// +/// Events are delivered exactly as a hub client would see them — the same +/// redacted clone the group send carries, so MxGateway:Dashboard:ShowTagValues +/// governs both paths identically. The feed is a bounded, lossy queue: a +/// consumer that falls behind loses the oldest queued events, matching the +/// best-effort contract the SignalR mirror already has. Disposing the +/// subscription deregisters it, which is what lets the broadcaster go back to +/// skipping all mirror work for a session nobody is watching — so callers must +/// dispose. Dispose is idempotent. +/// +public interface IDashboardEventSubscription : IDisposable +{ + /// Gets the reader delivering this session's mirrored events. + ChannelReader Reader { get; } +} diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/IDashboardSessionEventSubscriber.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/IDashboardSessionEventSubscriber.cs new file mode 100644 index 0000000..b60ea9e --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/IDashboardSessionEventSubscriber.cs @@ -0,0 +1,34 @@ +namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs; + +/// +/// In-process subscription seam on the dashboard event mirror. Implemented by +/// alongside +/// . +/// +/// +/// The interactive-server dashboard runs in the same process as the broadcaster, +/// so a session-details page has no reason to open a loopback SignalR connection +/// back to — mint a hub token, negotiate, hold a +/// WebSocket, and serialize every event — just to read events the broadcaster +/// already holds. It subscribes here instead. The registry gate stays honest +/// either way: an in-process subscription registers a synthetic connection id +/// with exactly as the hub registers a real +/// one, so keeps skipping the +/// redaction clone for sessions nobody is watching. +/// +/// It is a separate interface rather than a member of +/// because publishing and consuming are +/// different roles: the session pipeline only ever publishes, and its test +/// doubles should not have to implement a subscription feed. +/// +/// +public interface IDashboardSessionEventSubscriber +{ + /// Opens an in-process feed of the session's mirrored events. + /// Session id whose events the caller wants. + /// + /// The subscription. Dispose it to stop the feed and release the viewer + /// registration that keeps the mirror enabled for this session. + /// + IDashboardEventSubscription Subscribe(string sessionId); +} diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardEventBroadcasterTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardEventBroadcasterTests.cs index 701573c..55aa942 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardEventBroadcasterTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardEventBroadcasterTests.cs @@ -121,6 +121,140 @@ public sealed class DashboardEventBroadcasterTests Assert.NotNull(sent.OnAlarmTransition.LimitValue); } + /// + /// An in-process subscriber gets the same redacted clone the hub group gets, + /// and the shared source event is still left untouched. + /// + [Fact] + public void Subscribe_WhenShowTagValuesFalse_DeliversRedactedCloneWithoutMutatingSource() + { + CapturingHubContext hubContext = new(); + EventsHubViewerRegistry viewers = new(); + DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + using IDashboardEventSubscription subscription = broadcaster.Subscribe("session-1"); + MxEvent source = BuildEventWithValue(); + + broadcaster.Publish("session-1", source); + + MxEvent received = ReadOne(subscription); + Assert.Null(received.Value); + Assert.Null(received.OnAlarmTransition.CurrentValue); + Assert.Null(received.OnAlarmTransition.LimitValue); + Assert.Equal("Tank01.Level.HiHi", received.OnAlarmTransition.AlarmFullReference); + + // One clone feeds both audiences — the hub group and the in-process feed. + Assert.Same(hubContext.LastArgument, received); + + // The source is shared with the gRPC stream and the replay ring. + Assert.NotSame(source, received); + Assert.NotNull(source.Value); + Assert.Equal(42.5, source.Value.DoubleValue); + Assert.NotNull(source.OnAlarmTransition.CurrentValue); + Assert.NotNull(source.OnAlarmTransition.LimitValue); + } + + /// An in-process subscription opens the viewer gate the same way a hub client does. + [Fact] + public void Subscribe_OpensTheViewerGate() + { + CapturingHubContext hubContext = new(); + EventsHubViewerRegistry viewers = new(); + DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + + broadcaster.Publish("session-1", BuildEventWithValue()); + Assert.Equal(0, hubContext.SendCount); + Assert.Null(hubContext.LastArgument); + + using IDashboardEventSubscription subscription = broadcaster.Subscribe("session-1"); + + Assert.True(viewers.HasViewers("session-1")); + + broadcaster.Publish("session-1", BuildEventWithValue()); + + Assert.Equal(1, hubContext.SendCount); + Assert.NotNull(hubContext.LastArgument); + } + + /// Disposing the last in-process subscription restores the no-viewers short-circuit. + [Fact] + public void Dispose_OfLastInProcessSubscription_RestoresTheShortCircuit() + { + CapturingHubContext hubContext = new(); + EventsHubViewerRegistry viewers = new(); + DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + IDashboardEventSubscription subscription = broadcaster.Subscribe("session-1"); + + broadcaster.Publish("session-1", BuildEventWithValue()); + Assert.Equal(1, hubContext.SendCount); + Assert.Equal("session-1", ReadOne(subscription).SessionId); + + subscription.Dispose(); + + Assert.False(viewers.HasViewers("session-1")); + + broadcaster.Publish("session-1", BuildEventWithValue()); + + Assert.Equal(1, hubContext.SendCount); + Assert.False(subscription.Reader.TryRead(out _)); + } + + /// Hub viewers and in-process subscribers are audiences of their own session only. + [Fact] + public void Publish_DeliversOnlyToTheSubscribedSessionsAudience() + { + CapturingHubContext hubContext = new(); + EventsHubViewerRegistry viewers = new(); + viewers.AddViewer("conn-1", "session-1"); + DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + using IDashboardEventSubscription subscription = broadcaster.Subscribe("session-2"); + + // The hub viewer's session must not spill into the in-process feed. + broadcaster.Publish("session-1", BuildEventWithValue()); + + Assert.Equal(1, hubContext.SendCount); + Assert.False(subscription.Reader.TryRead(out _)); + + broadcaster.Publish("session-2", BuildEventWithValue("session-2")); + + Assert.Equal("session-2", ReadOne(subscription).SessionId); + + // A session with no audience at all still short-circuits. + broadcaster.Publish("session-3", BuildEventWithValue("session-3")); + + Assert.Equal(2, hubContext.SendCount); + } + + /// Disposing twice is a no-op and cannot release a sibling subscription's registration. + [Fact] + public void Dispose_CalledTwice_IsSafeAndLeavesSiblingSubscriptionsAlone() + { + CapturingHubContext hubContext = new(); + EventsHubViewerRegistry viewers = new(); + DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers); + IDashboardEventSubscription first = broadcaster.Subscribe("session-1"); + using IDashboardEventSubscription second = broadcaster.Subscribe("session-1"); + + first.Dispose(); + first.Dispose(); + + Assert.True(viewers.HasViewers("session-1")); + + broadcaster.Publish("session-1", BuildEventWithValue()); + + Assert.Equal(1, hubContext.SendCount); + Assert.False(first.Reader.TryRead(out _)); + Assert.Equal("session-1", ReadOne(second).SessionId); + } + + /// Reads exactly one event from a subscription, failing the test if none is queued. + /// The subscription to read from. + /// The event that was read. + private static MxEvent ReadOne(IDashboardEventSubscription subscription) + { + Assert.True(subscription.Reader.TryRead(out MxEvent? received)); + return Assert.IsType(received); + } + private static DashboardEventBroadcaster Create( CapturingHubContext hubContext, bool showTagValues, @@ -147,12 +281,15 @@ public sealed class DashboardEventBroadcasterTests return viewers; } - private static MxEvent BuildEventWithValue() + /// Builds a value-bearing alarm-transition event for the given session. + /// Session id stamped on the event. + /// The event. + private static MxEvent BuildEventWithValue(string sessionId = "session-1") { return new MxEvent { Family = MxEventFamily.OnAlarmTransition, - SessionId = "session-1", + SessionId = sessionId, ServerHandle = 7, ItemHandle = 11, Quality = 192,