dohertj2/mxaccessgw
1
0
Fork 0
Code Issues 3 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(dashboard): bound EventsHub delivery with one drop-oldest pump per session

Browse Source 
Publish fired a group SendAsync per event and returned, so a slow remote
viewer accumulated unbounded pending sends plus cloned MxEvents. Hub delivery
now goes through a 256-deep DropOldest channel per hub-watched session with a
single drain task and time-bounded sends. The registry counts hub viewers
separately and raises HubViewersReleased so the pump is torn down with the
last remote viewer; in-process-only sessions start no pump.
This commit is contained in:
Joseph Doherty
2026-10-22 06:13:05 -04:00
parent 488649ce99
commit f1db002b87
5 changed files with 638 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/BoundedHubSend.cs
+52
View File
@@ -0,0 +1,52 @@
using Microsoft.AspNetCore.SignalR;
namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
/// <summary>
/// Time-bounded SignalR send shared by the dashboard hub publishers. A group or
/// broadcast send completes only once every target connection has accepted the
/// frame, so one remote viewer with a full transport buffer would otherwise hold
/// the publisher's single drain loop for as long as that viewer stays slow.
/// </summary>
internal static class BoundedHubSend
{
/// <summary>
/// Sends <paramref name="argument"/> as <paramref name="method"/> and waits at most
/// <paramref name="timeout"/> for it. The timeout cancels the send's own token, so
/// the lifetime manager abandons the pending write to the slow connection, and the
/// wait itself is bounded too, so a proxy that ignores the token cannot stall the
/// caller either.
/// </summary>
/// <param name="proxy">Target client proxy (a group or all connections).</param>
/// <param name="method">Client method name.</param>
/// <param name="argument">The single message argument.</param>
/// <param name="timeout">Upper bound on the send.</param>
/// <param name="stoppingToken">Host shutdown token; its cancellation propagates.</param>
/// <returns>
/// <see langword="true"/> when the send completed in time; <see langword="false"/>
/// when it was abandoned at the timeout.
/// </returns>
/// <exception cref="OperationCanceledException"><paramref name="stoppingToken"/> was cancelled.</exception>
public static async Task<bool> SendAsync(
IClientProxy proxy,
string method,
object argument,
TimeSpan timeout,
CancellationToken stoppingToken)
{
using CancellationTokenSource bounded = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
bounded.CancelAfter(timeout);
try
{
await proxy.SendAsync(method, argument, bounded.Token)
.WaitAsync(bounded.Token)
.ConfigureAwait(false);
return true;
}
catch (OperationCanceledException) when (!stoppingToken.IsCancellationRequested)
{
return false;
}
}
}
src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs
+256 -51
View File
@@ -10,10 +10,11 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
/// Broadcasts MxEvents to the two dashboard audiences for a session: remote
/// <see cref="EventsHub"/> clients subscribed to the session's group, and
/// in-process subscribers opened through
/// <see cref="IDashboardSessionEventSubscriber.Subscribe"/>. 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.
/// <see cref="IDashboardSessionEventSubscriber.Subscribe"/>. <see cref="Publish"/>
/// never blocks: both audiences are fed through bounded drop-oldest queues, so the
/// source gRPC stream is never held up by a slow dashboard viewer. Errors are
/// logged and dropped — keeping the SignalR mirror best-effort preserves the gRPC
/// contract.
/// </summary>
/// <remarks>
/// When <c>MxGateway:Dashboard:ShowTagValues</c> is false (the default), tag
@@ -26,31 +27,45 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
/// <see cref="EventsHub"/>), while the redaction decides what a permitted
/// subscriber sees — so the value-leak seam stays closed whatever the ACL
/// admits.
/// <para>
/// Hub delivery goes through one <see cref="HubPump"/> per watched session: a
/// <see cref="HubQueueCapacity"/>-deep drop-oldest channel drained by a single task
/// that sends one event at a time, each send time-bounded. An earlier version fired
/// a group send per event and returned, so a slow remote viewer accumulated an
/// unbounded number of pending sends — each holding a cloned event — inside the
/// SignalR lifetime manager. The pump is started by the first event published while
/// a hub connection watches the session and torn down when
/// <see cref="EventsHubViewerRegistry.HubViewersReleased"/> reports the last one gone.
/// </para>
/// </remarks>
/// <param name="hubContext">Hub context used to send to the session's group.</param>
/// <param name="viewerRegistry">
/// 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.
/// </param>
/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
/// <param name="logger">Logger for best-effort mirror failures.</param>
public sealed class DashboardEventBroadcaster(
IHubContext<EventsHub> hubContext,
EventsHubViewerRegistry viewerRegistry,
IOptions<GatewayOptions> options,
ILogger<DashboardEventBroadcaster> logger) : IDashboardEventBroadcaster, IDashboardSessionEventSubscriber
public sealed class DashboardEventBroadcaster : IDashboardEventBroadcaster, IDashboardSessionEventSubscriber, IDisposable
{
/// <summary>
/// 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.
/// contract the SignalR mirror has.
/// </summary>
private const int InProcessQueueCapacity = 256;
private readonly bool _showTagValues = options.Value.Dashboard.ShowTagValues;
/// <summary>
/// Queue depth of a session's hub-delivery pump. Same depth and drop policy as
/// the in-process feed: the remote viewer is a live view, so a burst it cannot
/// keep up with loses its oldest events rather than buffering without bound.
/// </summary>
internal const int HubQueueCapacity = 256;
private static readonly TimeSpan DefaultHubSendTimeout = TimeSpan.FromSeconds(5);
private readonly IHubContext<EventsHub> _hubContext;
private readonly EventsHubViewerRegistry _viewerRegistry;
private readonly ILogger<DashboardEventBroadcaster> _logger;
private readonly TimeSpan _hubSendTimeout;
private readonly bool _showTagValues;
private readonly object _syncRoot = new();
@@ -63,6 +78,9 @@ public sealed class DashboardEventBroadcaster(
private readonly Dictionary<string, InProcessSubscription[]> _inProcessSubscribers =
new(StringComparer.Ordinal);
/// <summary>Live hub-delivery pumps per session. Guarded by <see cref="_syncRoot"/>.</summary>
private readonly Dictionary<string, HubPump> _hubPumps = new(StringComparer.Ordinal);
/// <summary>
/// Total live in-process subscribers, read without the lock so the common
/// case — nobody has a session-details page open — never contends on it.
@@ -70,6 +88,61 @@ public sealed class DashboardEventBroadcaster(
/// </summary>
private int _inProcessSubscriberCount;
private bool _disposed;
/// <summary>Initializes a new instance of the <see cref="DashboardEventBroadcaster"/> class.</summary>
/// <param name="hubContext">Hub context used to send to the session's group.</param>
/// <param name="viewerRegistry">
/// 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.
/// </param>
/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
/// <param name="logger">Logger for best-effort mirror failures.</param>
public DashboardEventBroadcaster(
IHubContext<EventsHub> hubContext,
EventsHubViewerRegistry viewerRegistry,
IOptions<GatewayOptions> options,
ILogger<DashboardEventBroadcaster> logger)
: this(hubContext, viewerRegistry, options, logger, DefaultHubSendTimeout)
{
}
/// <summary>Initializes a new instance with a custom hub send timeout (test hook).</summary>
/// <param name="hubContext">Hub context used to send to the session's group.</param>
/// <param name="viewerRegistry">Live-subscriber registry.</param>
/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
/// <param name="logger">Logger for best-effort mirror failures.</param>
/// <param name="hubSendTimeout">Upper bound on one group send before it is abandoned.</param>
internal DashboardEventBroadcaster(
IHubContext<EventsHub> hubContext,
EventsHubViewerRegistry viewerRegistry,
IOptions<GatewayOptions> options,
ILogger<DashboardEventBroadcaster> logger,
TimeSpan hubSendTimeout)
{
ArgumentNullException.ThrowIfNull(options);
_hubContext = hubContext ?? throw new ArgumentNullException(nameof(hubContext));
_viewerRegistry = viewerRegistry ?? throw new ArgumentNullException(nameof(viewerRegistry));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_hubSendTimeout = hubSendTimeout;
_showTagValues = options.Value.Dashboard.ShowTagValues;
_viewerRegistry.HubViewersReleased += OnHubViewersReleased;
}
/// <summary>Gets the number of live hub-delivery pumps (one per hub-watched session).</summary>
internal int ActiveHubPumpCount
{
get
{
lock (_syncRoot)
{
return _hubPumps.Count;
}
}
}
/// <inheritdoc />
public void Publish(string sessionId, MxEvent mxEvent)
{
@@ -81,46 +154,126 @@ public sealed class DashboardEventBroadcaster(
// Every session's dashboard-mirror subscriber calls Publish for every event,
// whether or not a browser is on that session's page. Without this gate the
// steady state — no dashboard viewer at all — still paid a deep protobuf
// clone (redaction is on by default) plus a send to an empty SignalR group
// per event. Bail before both.
if (!viewerRegistry.HasViewers(sessionId))
// clone (redaction is on by default) per event. Bail before it.
if (!_viewerRegistry.HasViewers(sessionId))
{
return;
}
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.
Task send;
// In-process-only viewers need no hub send at all; the pump exists only
// while a remote connection is in the session's group.
if (_viewerRegistry.HasHubViewers(sessionId))
{
EnqueueForHub(sessionId, outbound);
}
}
/// <summary>
/// Stops every hub pump and detaches from the viewer registry. The singleton is
/// disposed with the host; a pump's in-flight send is bounded by its timeout.
/// </summary>
public void Dispose()
{
HubPump[] pumps;
lock (_syncRoot)
{
if (_disposed)
{
return;
}
_disposed = true;
pumps = [.. _hubPumps.Values];
_hubPumps.Clear();
}
_viewerRegistry.HubViewersReleased -= OnHubViewersReleased;
foreach (HubPump pump in pumps)
{
pump.Complete();
}
}
/// <summary>
/// Queues an event on the session's hub pump, starting the pump on first use.
/// The write happens under the lock so it cannot land on a pump that a
/// concurrent teardown has already completed.
/// </summary>
/// <param name="sessionId">Session the event belongs to.</param>
/// <param name="outbound">The event as the dashboard should see it.</param>
private void EnqueueForHub(string sessionId, MxEvent outbound)
{
lock (_syncRoot)
{
if (_disposed)
{
return;
}
if (!_hubPumps.TryGetValue(sessionId, out HubPump? pump))
{
pump = new HubPump(this, sessionId, HubQueueCapacity);
_hubPumps[sessionId] = pump;
}
pump.TryWrite(outbound);
}
}
/// <summary>
/// Tears down a session's hub pump once its last hub viewer has gone. Re-checks
/// the registry under the lock: a viewer that re-subscribed in between keeps the
/// pump, and one that arrives after the teardown gets a fresh pump on the next
/// event.
/// </summary>
/// <param name="sessionId">Session whose last hub viewer left.</param>
private void OnHubViewersReleased(string sessionId)
{
HubPump? pump;
lock (_syncRoot)
{
if (_viewerRegistry.HasHubViewers(sessionId) || !_hubPumps.Remove(sessionId, out pump))
{
return;
}
}
pump.Complete();
}
/// <summary>
/// Sends one event to the session's group, bounded by the hub send timeout.
/// Never throws: a failed or abandoned send is logged and the pump moves on.
/// </summary>
/// <param name="sessionId">Session whose group receives the event.</param>
/// <param name="outbound">The event to send.</param>
/// <returns>A task that completes when the send finished or was abandoned.</returns>
private async Task SendToHubAsync(string sessionId, MxEvent outbound)
{
try
{
send = hubContext.Clients
.Group(EventsHub.GroupName(sessionId))
.SendAsync(EventsHub.EventMessage, outbound);
bool sent = await BoundedHubSend.SendAsync(
_hubContext.Clients.Group(EventsHub.GroupName(sessionId)),
EventsHub.EventMessage,
outbound,
_hubSendTimeout,
CancellationToken.None)
.ConfigureAwait(false);
if (!sent)
{
_logger.LogDebug(
"Dashboard event mirror to session {SessionId} timed out after {Timeout}; continuing.",
sessionId,
_hubSendTimeout);
}
}
catch (Exception ex)
{
logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} threw synchronously.", sessionId);
return;
}
if (!send.IsCompletedSuccessfully)
{
_ = send.ContinueWith(
t =>
{
if (t.Exception is { } ex)
{
logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} failed.", sessionId);
}
},
TaskScheduler.Default);
_logger.LogDebug(ex, "Dashboard event mirror to session {SessionId} failed.", sessionId);
}
}
@@ -133,7 +286,7 @@ public sealed class DashboardEventBroadcaster(
// 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");
string connectionId = EventsHubViewerRegistry.InProcessConnectionIdPrefix + Guid.NewGuid().ToString("N");
InProcessSubscription subscription = new(this, sessionId, connectionId, InProcessQueueCapacity);
// Register before the subscriber becomes a delivery target, exactly as
@@ -142,7 +295,7 @@ public sealed class DashboardEventBroadcaster(
// 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);
_viewerRegistry.AddViewer(connectionId, sessionId);
lock (_syncRoot)
{
@@ -227,12 +380,12 @@ public sealed class DashboardEventBroadcaster(
}
}
viewerRegistry.RemoveViewer(subscription.ConnectionId, subscription.SessionId);
_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);
_viewerRegistry.ReleaseConnection(subscription.ConnectionId);
}
/// <summary>
@@ -328,4 +481,56 @@ public sealed class DashboardEventBroadcaster(
_channel.Writer.TryComplete();
}
}
/// <summary>
/// One session's hub-delivery pump: a bounded drop-oldest channel and the single
/// task that drains it into the session's SignalR group, one bounded send at a
/// time. The drain is the only reader, so at most one send per session is ever
/// pending in the lifetime manager.
/// </summary>
private sealed class HubPump
{
private readonly Channel<MxEvent> _channel;
/// <summary>Initializes a new instance of the <see cref="HubPump"/> class and starts its drain.</summary>
/// <param name="owner">Broadcaster whose send path the drain uses.</param>
/// <param name="sessionId">Session whose group the pump feeds.</param>
/// <param name="capacity">Queue depth before the oldest queued event is dropped.</param>
internal HubPump(DashboardEventBroadcaster owner, string sessionId, int capacity)
{
_channel = Channel.CreateBounded<MxEvent>(new BoundedChannelOptions(capacity)
{
// DropOldest, not Wait: Publish runs on the gRPC event pipeline and
// must never block, and the newest events are the ones a live view wants.
FullMode = BoundedChannelFullMode.DropOldest,
SingleReader = true,
SingleWriter = false,
});
// Task.Run so the drain never starts inline on the publishing thread.
Completion = Task.Run(() => DrainAsync(owner, sessionId));
}
/// <summary>Gets the drain task; it completes once the channel is completed and emptied.</summary>
internal Task Completion { get; }
/// <summary>Queues an event, dropping the oldest queued one when full. Never blocks.</summary>
/// <param name="mxEvent">The event to queue.</param>
internal void TryWrite(MxEvent mxEvent) => _channel.Writer.TryWrite(mxEvent);
/// <summary>Completes the channel so the drain exits once it is empty.</summary>
internal void Complete() => _channel.Writer.TryComplete();
private async Task DrainAsync(DashboardEventBroadcaster owner, string sessionId)
{
ChannelReader<MxEvent> reader = _channel.Reader;
while (await reader.WaitToReadAsync().ConfigureAwait(false))
{
while (reader.TryRead(out MxEvent? next))
{
await owner.SendToHubAsync(sessionId, next).ConfigureAwait(false);
}
}
}
}
}
src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/EventsHubViewerRegistry.cs
+67 -9
View File
@@ -19,11 +19,27 @@ namespace ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs;
/// simply goes away never calls <c>UnsubscribeSession</c>; the hub's
/// <c>OnDisconnectedAsync</c> releases everything the connection held.
/// </para>
/// <para>
/// Hub viewers are also counted on their own, separately from the in-process
/// subscribers that register under an <see cref="InProcessConnectionIdPrefix"/>
/// id: <see cref="DashboardEventBroadcaster"/> runs one bounded hub-delivery pump
/// per session only while a remote connection is watching it, and tears that pump
/// down on <see cref="HubViewersReleased"/>.
/// </para>
/// </remarks>
public sealed class EventsHubViewerRegistry
{
/// <summary>
/// Prefix of the synthetic connection ids in-process subscribers register under.
/// A SignalR connection id never carries it, so it is what tells the two
/// audiences apart for the hub-only count.
/// </summary>
public const string InProcessConnectionIdPrefix = "inproc-";
private readonly ConcurrentDictionary<string, int> _viewersBySession = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, int> _hubViewersBySession = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, ConcurrentDictionary<string, byte>> _sessionsByConnection =
new(StringComparer.Ordinal);
@@ -53,8 +69,19 @@ public sealed class EventsHubViewerRegistry
}
_viewersBySession.AddOrUpdate(sessionId, 1, static (_, count) => count + 1);
if (IsHubConnection(connectionId))
{
_hubViewersBySession.AddOrUpdate(sessionId, 1, static (_, count) => count + 1);
}
}
/// <summary>
/// Raised with a session id when its last remote hub viewer leaves (unsubscribe
/// or disconnect). In-process subscribers do not raise it. Handlers run on the
/// releasing thread and must not throw or block.
/// </summary>
public event Action<string>? HubViewersReleased;
/// <summary>
/// Records that <paramref name="connectionId"/> stopped watching
/// <paramref name="sessionId"/>. A removal with no matching
@@ -75,7 +102,7 @@ public sealed class EventsHubViewerRegistry
return;
}
ReleaseSession(sessionId);
Release(connectionId, sessionId);
}
/// <summary>
@@ -107,7 +134,7 @@ public sealed class EventsHubViewerRegistry
// same detached set must not let the session be decremented twice.
if (sessions.TryRemove(sessionId, out _))
{
ReleaseSession(sessionId);
Release(connectionId, sessionId);
}
}
}
@@ -120,32 +147,63 @@ public sealed class EventsHubViewerRegistry
&& _viewersBySession.TryGetValue(sessionId, out int count)
&& count > 0;
/// <summary>
/// Gets a value indicating whether any remote <see cref="EventsHub"/> connection —
/// as opposed to an in-process subscriber — is watching the session.
/// </summary>
/// <param name="sessionId">Session id to test.</param>
/// <returns><see langword="true"/> when at least one hub connection is subscribed.</returns>
public bool HasHubViewers(string sessionId) =>
!string.IsNullOrEmpty(sessionId)
&& _hubViewersBySession.TryGetValue(sessionId, out int count)
&& count > 0;
private static bool IsHubConnection(string connectionId) =>
!connectionId.StartsWith(InProcessConnectionIdPrefix, StringComparison.Ordinal);
/// <summary>
/// Releases one subscription from both counts it was recorded in, and raises
/// <see cref="HubViewersReleased"/> when it was the session's last hub viewer.
/// </summary>
/// <param name="connectionId">Connection that held the subscription.</param>
/// <param name="sessionId">Session id whose counts are released.</param>
private void Release(string connectionId, string sessionId)
{
ReleaseSession(_viewersBySession, sessionId);
if (IsHubConnection(connectionId) && ReleaseSession(_hubViewersBySession, sessionId))
{
HubViewersReleased?.Invoke(sessionId);
}
}
/// <summary>
/// Decrements the session's viewer count, dropping the entry entirely at
/// zero so the dictionary does not grow one key per session ever viewed.
/// The compare-and-swap loop keeps the decrement correct against a
/// concurrent <see cref="AddViewer"/> on the same session.
/// </summary>
/// <param name="counts">The count map to decrement.</param>
/// <param name="sessionId">Session id whose count is released.</param>
private void ReleaseSession(string sessionId)
/// <returns><see langword="true"/> when this release dropped the count to zero.</returns>
private static bool ReleaseSession(ConcurrentDictionary<string, int> counts, string sessionId)
{
while (true)
{
if (!_viewersBySession.TryGetValue(sessionId, out int count))
if (!counts.TryGetValue(sessionId, out int count))
{
return;
return false;
}
if (count <= 1)
{
if (_viewersBySession.TryRemove(new KeyValuePair<string, int>(sessionId, count)))
if (counts.TryRemove(new KeyValuePair<string, int>(sessionId, count)))
{
return;
return true;
}
}
else if (_viewersBySession.TryUpdate(sessionId, count - 1, count))
else if (counts.TryUpdate(sessionId, count - 1, count))
{
return;
return false;
}
}
}
src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardEventBroadcasterTests.cs
+219 -54
View File
@@ -12,7 +12,8 @@ namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Dashboard;
/// <c>MxGateway:Dashboard:ShowTagValues</c>: tag values are stripped
/// from the mirrored copy when the flag is off, present when it is on, and the
/// shared source event is never mutated. Also verifies the viewer gate — the
/// mirror does no work at all for a session nobody is watching.
/// mirror does no work at all for a session nobody is watching, and that hub
/// delivery goes through one bounded drop-oldest pump per hub-watched session.
/// </summary>
public sealed class DashboardEventBroadcasterTests
{
@@ -22,11 +23,12 @@ public sealed class DashboardEventBroadcasterTests
{
CapturingHubContext hubContext = new();
EventsHubViewerRegistry viewers = new();
DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
MxEvent source = BuildEventWithValue();
broadcaster.Publish("session-1", source);
Assert.Equal(0, broadcaster.ActiveHubPumpCount);
Assert.Equal(0, hubContext.SendCount);
Assert.Null(hubContext.LastArgument);
}
@@ -38,40 +40,105 @@ public sealed class DashboardEventBroadcasterTests
CapturingHubContext hubContext = new();
EventsHubViewerRegistry viewers = new();
viewers.AddViewer("conn-1", "session-2");
DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
broadcaster.Publish("session-1", BuildEventWithValue());
Assert.Equal(0, broadcaster.ActiveHubPumpCount);
Assert.Equal(0, hubContext.SendCount);
}
/// <summary>Once the last viewer leaves, the mirror stops sending again.</summary>
/// <summary>Once the last hub viewer leaves, the session's pump is torn down and sending stops.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public void Publish_AfterLastViewerLeaves_StopsSending()
public async Task Publish_AfterLastViewerLeaves_StopsSending()
{
CapturingHubContext hubContext = new();
EventsHubViewerRegistry viewers = new();
viewers.AddViewer("conn-1", "session-1");
DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
broadcaster.Publish("session-1", BuildEventWithValue());
Assert.Equal(1, hubContext.SendCount);
await hubContext.WaitForSendCountAsync(1);
Assert.Equal(1, broadcaster.ActiveHubPumpCount);
viewers.RemoveViewer("conn-1", "session-1");
Assert.Equal(0, broadcaster.ActiveHubPumpCount);
broadcaster.Publish("session-1", BuildEventWithValue());
Assert.Equal(0, broadcaster.ActiveHubPumpCount);
Assert.Equal(1, hubContext.SendCount);
}
/// <summary>
/// A hub viewer that cannot keep up costs a bounded queue, not an unbounded pile of
/// pending sends: while one send is stuck, further events drop oldest-first at the
/// pump's capacity, only one send is ever in flight, and the newest event survives.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task Publish_WhenHubSendStalls_BuffersAtMostTheCapacityAndKeepsTheNewest()
{
CapturingHubContext hubContext = new();
hubContext.Proxy.Gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
EventsHubViewerRegistry viewers = WatchedSession1();
using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: true, viewers);
const int published = 2000;
broadcaster.Publish("session-1", BuildSequencedEvent(0));
await hubContext.WaitForSendCountAsync(1);
for (int i = 1; i < published; i++)
{
broadcaster.Publish("session-1", BuildSequencedEvent(i));
}
// One send stuck in flight, nothing else started behind it.
Assert.Equal(1, hubContext.SendCount);
Assert.Equal(1, hubContext.Proxy.MaxConcurrentSends);
hubContext.Proxy.Gate.SetResult();
await hubContext.WaitForLastItemHandleAsync(published - 1);
// The stuck event plus at most one full queue — the rest were dropped oldest-first.
Assert.InRange(hubContext.SendCount, 2, DashboardEventBroadcaster.HubQueueCapacity + 1);
Assert.Equal(1, hubContext.Proxy.MaxConcurrentSends);
}
/// <summary>
/// A send that never completes is abandoned at the send timeout, so one stuck
/// connection cannot stop the pump delivering later events.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task Publish_WhenHubSendNeverCompletes_AbandonsItAtTheTimeoutAndContinues()
{
CapturingHubContext hubContext = new();
hubContext.Proxy.Gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
hubContext.Proxy.IgnoreCancellation = true;
using DashboardEventBroadcaster broadcaster = Create(
hubContext,
showTagValues: true,
WatchedSession1(),
hubSendTimeout: TimeSpan.FromMilliseconds(50));
broadcaster.Publish("session-1", BuildSequencedEvent(1));
broadcaster.Publish("session-1", BuildSequencedEvent(2));
await hubContext.WaitForSendCountAsync(2);
}
/// <summary>Values are stripped from the mirror when ShowTagValues is off; metadata survives.</summary>
[Fact]
public void Publish_WhenShowTagValuesFalse_RedactsValuesButKeepsMetadata()
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task Publish_WhenShowTagValuesFalse_RedactsValuesButKeepsMetadata()
{
CapturingHubContext hubContext = new();
DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, WatchedSession1());
using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, WatchedSession1());
MxEvent source = BuildEventWithValue();
broadcaster.Publish("session-1", source);
await hubContext.WaitForSendCountAsync(1);
MxEvent sent = Assert.IsType<MxEvent>(hubContext.LastArgument);
Assert.Null(sent.Value);
@@ -88,13 +155,15 @@ public sealed class DashboardEventBroadcasterTests
/// <summary>Redaction applies to a clone, so the shared source event keeps its values.</summary>
[Fact]
public void Publish_WhenShowTagValuesFalse_DoesNotMutateSourceEvent()
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task Publish_WhenShowTagValuesFalse_DoesNotMutateSourceEvent()
{
CapturingHubContext hubContext = new();
DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, WatchedSession1());
using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, WatchedSession1());
MxEvent source = BuildEventWithValue();
broadcaster.Publish("session-1", source);
await hubContext.WaitForSendCountAsync(1);
// The redaction must apply to a clone; the shared source keeps its values.
Assert.NotNull(source.Value);
@@ -106,13 +175,15 @@ public sealed class DashboardEventBroadcasterTests
/// <summary>Values pass through unredacted when ShowTagValues is on.</summary>
[Fact]
public void Publish_WhenShowTagValuesTrue_KeepsValues()
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task Publish_WhenShowTagValuesTrue_KeepsValues()
{
CapturingHubContext hubContext = new();
DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: true, WatchedSession1());
using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: true, WatchedSession1());
MxEvent source = BuildEventWithValue();
broadcaster.Publish("session-1", source);
await hubContext.WaitForSendCountAsync(1);
MxEvent sent = Assert.IsType<MxEvent>(hubContext.LastArgument);
Assert.NotNull(sent.Value);
@@ -125,16 +196,18 @@ public sealed class DashboardEventBroadcasterTests
/// An in-process subscriber gets the same redacted clone the hub group gets,
/// and the shared source event is still left untouched.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public void Subscribe_WhenShowTagValuesFalse_DeliversRedactedCloneWithoutMutatingSource()
public async Task Subscribe_WhenShowTagValuesFalse_DeliversRedactedCloneWithoutMutatingSource()
{
CapturingHubContext hubContext = new();
EventsHubViewerRegistry viewers = new();
DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
EventsHubViewerRegistry viewers = WatchedSession1();
using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
using IDashboardEventSubscription subscription = broadcaster.Subscribe("session-1");
MxEvent source = BuildEventWithValue();
broadcaster.Publish("session-1", source);
await hubContext.WaitForSendCountAsync(1);
MxEvent received = ReadOne(subscription);
Assert.Null(received.Value);
@@ -153,17 +226,16 @@ public sealed class DashboardEventBroadcasterTests
Assert.NotNull(source.OnAlarmTransition.LimitValue);
}
/// <summary>An in-process subscription opens the viewer gate the same way a hub client does.</summary>
/// <summary>
/// An in-process subscription opens the viewer gate the same way a hub client does,
/// but starts no hub pump: with no remote connection there is nobody to send to.
/// </summary>
[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 DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
using IDashboardEventSubscription subscription = broadcaster.Subscribe("session-1");
@@ -171,8 +243,9 @@ public sealed class DashboardEventBroadcasterTests
broadcaster.Publish("session-1", BuildEventWithValue());
Assert.Equal(1, hubContext.SendCount);
Assert.NotNull(hubContext.LastArgument);
Assert.Equal("session-1", ReadOne(subscription).SessionId);
Assert.Equal(0, broadcaster.ActiveHubPumpCount);
Assert.Equal(0, hubContext.SendCount);
}
/// <summary>Disposing the last in-process subscription restores the no-viewers short-circuit.</summary>
@@ -181,11 +254,10 @@ public sealed class DashboardEventBroadcasterTests
{
CapturingHubContext hubContext = new();
EventsHubViewerRegistry viewers = new();
DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
using 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();
@@ -194,26 +266,28 @@ public sealed class DashboardEventBroadcasterTests
broadcaster.Publish("session-1", BuildEventWithValue());
Assert.Equal(1, hubContext.SendCount);
Assert.Equal(0, hubContext.SendCount);
Assert.False(subscription.Reader.TryRead(out _));
}
/// <summary>Hub viewers and in-process subscribers are audiences of their own session only.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public void Publish_DeliversOnlyToTheSubscribedSessionsAudience()
public async Task Publish_DeliversOnlyToTheSubscribedSessionsAudience()
{
CapturingHubContext hubContext = new();
EventsHubViewerRegistry viewers = new();
viewers.AddViewer("conn-1", "session-1");
DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
using 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);
await hubContext.WaitForSendCountAsync(1);
Assert.False(subscription.Reader.TryRead(out _));
// The in-process-only session reaches its subscriber and starts no hub pump.
broadcaster.Publish("session-2", BuildEventWithValue("session-2"));
Assert.Equal("session-2", ReadOne(subscription).SessionId);
@@ -221,7 +295,8 @@ public sealed class DashboardEventBroadcasterTests
// A session with no audience at all still short-circuits.
broadcaster.Publish("session-3", BuildEventWithValue("session-3"));
Assert.Equal(2, hubContext.SendCount);
Assert.Equal(1, broadcaster.ActiveHubPumpCount);
Assert.Equal(1, hubContext.SendCount);
}
/// <summary>Disposing twice is a no-op and cannot release a sibling subscription's registration.</summary>
@@ -230,7 +305,7 @@ public sealed class DashboardEventBroadcasterTests
{
CapturingHubContext hubContext = new();
EventsHubViewerRegistry viewers = new();
DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
using DashboardEventBroadcaster broadcaster = Create(hubContext, showTagValues: false, viewers);
IDashboardEventSubscription first = broadcaster.Subscribe("session-1");
using IDashboardEventSubscription second = broadcaster.Subscribe("session-1");
@@ -241,7 +316,6 @@ public sealed class DashboardEventBroadcasterTests
broadcaster.Publish("session-1", BuildEventWithValue());
Assert.Equal(1, hubContext.SendCount);
Assert.False(first.Reader.TryRead(out _));
Assert.Equal("session-1", ReadOne(second).SessionId);
}
@@ -258,7 +332,8 @@ public sealed class DashboardEventBroadcasterTests
private static DashboardEventBroadcaster Create(
CapturingHubContext hubContext,
bool showTagValues,
EventsHubViewerRegistry viewers)
EventsHubViewerRegistry viewers,
TimeSpan? hubSendTimeout = null)
{
GatewayOptions gatewayOptions = new()
{
@@ -269,7 +344,21 @@ public sealed class DashboardEventBroadcasterTests
hubContext,
viewers,
Options.Create(gatewayOptions),
NullLogger<DashboardEventBroadcaster>.Instance);
NullLogger<DashboardEventBroadcaster>.Instance,
hubSendTimeout ?? TimeSpan.FromSeconds(30));
}
/// <summary>Builds a value-less event whose item handle carries its publish order.</summary>
/// <param name="sequence">Publish order stamped into <see cref="MxEvent.ItemHandle"/>.</param>
/// <returns>The event.</returns>
private static MxEvent BuildSequencedEvent(int sequence)
{
return new MxEvent
{
Family = MxEventFamily.OnDataChange,
SessionId = "session-1",
ItemHandle = sequence,
};
}
/// <summary>A registry with one hub connection watching <c>session-1</c>.</summary>
@@ -313,81 +402,157 @@ public sealed class DashboardEventBroadcasterTests
/// <summary>Gets the group manager.</summary>
public IGroupManager Groups { get; } = new NoopGroupManager();
/// <summary>Gets the capturing proxy every client lookup resolves to.</summary>
public CapturingClientProxy Proxy => _clients.Proxy;
/// <summary>Gets the first argument of the most recent send call.</summary>
public object? LastArgument => _clients.GroupProxy.LastArgument;
public object? LastArgument => _clients.Proxy.LastArgument;
/// <summary>Gets the number of send calls this fake has observed.</summary>
public int SendCount => _clients.GroupProxy.SendCount;
public int SendCount => _clients.Proxy.SendCount;
/// <summary>Waits until at least <paramref name="count"/> sends have started.</summary>
/// <param name="count">The send count to wait for.</param>
/// <returns>A task that completes once the count is reached.</returns>
public Task WaitForSendCountAsync(int count) =>
WaitUntilAsync(() => SendCount >= count, $"{count} sends (saw {SendCount})");
/// <summary>Waits until a send carrying the given item handle has started.</summary>
/// <param name="itemHandle">Item handle of the event to wait for.</param>
/// <returns>A task that completes once the event was sent.</returns>
public Task WaitForLastItemHandleAsync(int itemHandle) =>
WaitUntilAsync(
() => LastArgument is MxEvent sent && sent.ItemHandle == itemHandle,
$"a send of item handle {itemHandle}");
private static async Task WaitUntilAsync(Func<bool> condition, string description)
{
DateTime deadline = DateTime.UtcNow.AddSeconds(10);
while (!condition())
{
if (DateTime.UtcNow > deadline)
{
Assert.Fail($"Timed out waiting for {description}.");
}
await Task.Delay(5);
}
}
}
private sealed class CapturingHubClients : IHubClients
{
/// <summary>Gets the capturing client proxy shared by this fake.</summary>
public CapturingClientProxy GroupProxy { get; } = new();
public CapturingClientProxy Proxy { get; } = new();
/// <summary>Gets the shared capturing proxy for all connections.</summary>
public IClientProxy All => GroupProxy;
public IClientProxy All => Proxy;
/// <summary>Returns the shared capturing proxy, ignoring exclusions.</summary>
/// <param name="excludedConnectionIds">Ignored.</param>
/// <returns>The shared capturing proxy.</returns>
public IClientProxy AllExcept(IReadOnlyList<string> excludedConnectionIds) => GroupProxy;
public IClientProxy AllExcept(IReadOnlyList<string> excludedConnectionIds) => Proxy;
/// <summary>Returns the shared capturing proxy for any connection.</summary>
/// <param name="connectionId">Ignored.</param>
/// <returns>The shared capturing proxy.</returns>
public IClientProxy Client(string connectionId) => GroupProxy;
public IClientProxy Client(string connectionId) => Proxy;
/// <summary>Returns the shared capturing proxy for any connection set.</summary>
/// <param name="connectionIds">Ignored.</param>
/// <returns>The shared capturing proxy.</returns>
public IClientProxy Clients(IReadOnlyList<string> connectionIds) => GroupProxy;
public IClientProxy Clients(IReadOnlyList<string> connectionIds) => Proxy;
/// <summary>Returns the shared capturing proxy for any group.</summary>
/// <param name="groupName">Ignored.</param>
/// <returns>The shared capturing proxy.</returns>
public IClientProxy Group(string groupName) => GroupProxy;
public IClientProxy Group(string groupName) => Proxy;
/// <summary>Returns the shared capturing proxy for any group, ignoring exclusions.</summary>
/// <param name="groupName">Ignored.</param>
/// <param name="excludedConnectionIds">Ignored.</param>
/// <returns>The shared capturing proxy.</returns>
public IClientProxy GroupExcept(string groupName, IReadOnlyList<string> excludedConnectionIds) => GroupProxy;
public IClientProxy GroupExcept(string groupName, IReadOnlyList<string> excludedConnectionIds) => Proxy;
/// <summary>Returns the shared capturing proxy for any group set.</summary>
/// <param name="groupNames">Ignored.</param>
/// <returns>The shared capturing proxy.</returns>
public IClientProxy Groups(IReadOnlyList<string> groupNames) => GroupProxy;
public IClientProxy Groups(IReadOnlyList<string> groupNames) => Proxy;
/// <summary>Returns the shared capturing proxy for any user.</summary>
/// <param name="userId">Ignored.</param>
/// <returns>The shared capturing proxy.</returns>
public IClientProxy User(string userId) => GroupProxy;
public IClientProxy User(string userId) => Proxy;
/// <summary>Returns the shared capturing proxy for any user set.</summary>
/// <param name="userIds">Ignored.</param>
/// <returns>The shared capturing proxy.</returns>
public IClientProxy Users(IReadOnlyList<string> userIds) => GroupProxy;
public IClientProxy Users(IReadOnlyList<string> userIds) => Proxy;
}
private sealed class CapturingClientProxy : IClientProxy
{
private int _sendCount;
private int _inFlight;
private int _maxConcurrentSends;
private object? _lastArgument;
/// <summary>
/// Gets or sets a gate every send waits on before completing; null completes
/// sends synchronously. Set it before the first publish.
/// </summary>
public TaskCompletionSource? Gate { get; set; }
/// <summary>Gets or sets a value indicating whether a gated send ignores its cancellation token.</summary>
public bool IgnoreCancellation { get; set; }
/// <summary>Gets the first argument of the most recent send call.</summary>
public object? LastArgument { get; private set; }
public object? LastArgument => Volatile.Read(ref _lastArgument);
/// <summary>Gets the number of send calls made through this proxy.</summary>
public int SendCount { get; private set; }
public int SendCount => Volatile.Read(ref _sendCount);
/// <summary>Records the send call arguments and completes synchronously.</summary>
/// <summary>Gets the highest number of sends that were in flight at once.</summary>
public int MaxConcurrentSends => Volatile.Read(ref _maxConcurrentSends);
/// <summary>Records the send call arguments and completes once the gate (if any) opens.</summary>
/// <param name="method">The SignalR method name.</param>
/// <param name="args">The method arguments.</param>
/// <param name="cancellationToken">Token to observe for cancellation.</param>
/// <returns>A completed task.</returns>
/// <returns>A task that completes when the send is released.</returns>
public Task SendCoreAsync(string method, object?[] args, CancellationToken cancellationToken = default)
{
SendCount++;
LastArgument = args.Length > 0 ? args[0] : null;
return Task.CompletedTask;
Volatile.Write(ref _lastArgument, args.Length > 0 ? args[0] : null);
Interlocked.Increment(ref _sendCount);
TaskCompletionSource? gate = Gate;
if (gate is null || gate.Task.IsCompleted)
{
return Task.CompletedTask;
}
return TrackAsync(gate.Task, IgnoreCancellation ? CancellationToken.None : cancellationToken);
}
private async Task TrackAsync(Task gate, CancellationToken cancellationToken)
{
int inFlight = Interlocked.Increment(ref _inFlight);
int observed;
do
{
observed = Volatile.Read(ref _maxConcurrentSends);
}
while (inFlight > observed
&& Interlocked.CompareExchange(ref _maxConcurrentSends, inFlight, observed) != observed);
try
{
await gate.WaitAsync(cancellationToken);
}
finally
{
Interlocked.Decrement(ref _inFlight);
}
}
}
src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/EventsHubViewerRegistryTests.cs
+44
View File
@@ -164,4 +164,48 @@ public sealed class EventsHubViewerRegistryTests
Assert.False(registry.HasViewers("session-1"));
}
/// <summary>In-process subscribers count as viewers but not as hub viewers.</summary>
[Fact]
public void HasHubViewers_CountsOnlyRemoteConnections()
{
EventsHubViewerRegistry registry = new();
registry.AddViewer(EventsHubViewerRegistry.InProcessConnectionIdPrefix + "a", "session-1");
Assert.True(registry.HasViewers("session-1"));
Assert.False(registry.HasHubViewers("session-1"));
registry.AddViewer("conn-1", "session-1");
Assert.True(registry.HasHubViewers("session-1"));
}
/// <summary>
/// The released event fires once, when the last hub viewer leaves — whether it
/// unsubscribes or disconnects — and never for in-process subscribers.
/// </summary>
[Fact]
public void HubViewersReleased_FiresWhenTheLastHubViewerLeaves()
{
EventsHubViewerRegistry registry = new();
List<string> released = [];
registry.HubViewersReleased += released.Add;
registry.AddViewer("conn-1", "session-1");
registry.AddViewer("conn-2", "session-1");
registry.AddViewer(EventsHubViewerRegistry.InProcessConnectionIdPrefix + "a", "session-1");
registry.RemoveViewer("conn-1", "session-1");
Assert.Empty(released);
registry.ReleaseConnection("conn-2");
Assert.Equal(["session-1"], released);
Assert.False(registry.HasHubViewers("session-1"));
// The in-process subscriber still keeps the session watched, and leaving raises nothing.
Assert.True(registry.HasViewers("session-1"));
registry.RemoveViewer(EventsHubViewerRegistry.InProcessConnectionIdPrefix + "a", "session-1");
Assert.Equal(["session-1"], released);
}
}