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

feat(sitestream): add additive SubscribeSite alarm-only gRPC stream + server handler (plan #10 T2)

Browse Source 
Adds an additive, site-wide, alarm-only gRPC stream backing the aggregated
Alarm Summary. Proto: new SiteStreamRequest { correlation_id } message + rpc
SubscribeSite(SiteStreamRequest) returns (stream SiteStreamEvent) on
SiteStreamService — purely additive, no field renumbering. Regenerated the
checked-in SiteStreamGrpc/*.cs.

Server: SubscribeInstance and the new SubscribeSite now delegate to a shared
RunSubscriptionStreamAsync helper (readiness/shutdown guards, correlation-id
validation, duplicate replacement, concurrency cap, bounded DropOldest channel,
relay actor, SiteConnectionOpened/Closed telemetry, guaranteed cleanup). The
only variation is the subscribe delegate: SubscribeSite calls
ISiteStreamSubscriber.SubscribeSiteAlarms (no per-instance filter). Added
SubscribeSiteAlarms to the ISiteStreamSubscriber contract (SiteStreamManager
already implements it from T1). StreamRelayActor reused unchanged — it already
drops IsConfiguredPlaceholder rows and maps the enriched AlarmStateUpdate.

Tests: SubscribeSite subscribes site alarms + removes on cancel, rejects unsafe
correlation ids, and relays a domain AlarmStateChanged as a proto
AlarmStateUpdate on the stream.

Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj
This commit is contained in:
Joseph Doherty
2026-07-10 11:46:36 -04:00
parent b14a4dbc6b
commit 9c631b47e1
6 changed files with 463 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs
+58 -17
View File
@@ -210,10 +210,52 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
internal TimeSpan MaxStreamLifetime => _maxStreamLifetime;
/// <inheritdoc />
public override async Task SubscribeInstance(
public override Task SubscribeInstance(
InstanceStreamRequest request,
IServerStreamWriter<SiteStreamEvent> responseStream,
ServerCallContext context)
=> RunSubscriptionStreamAsync(
request.CorrelationId,
responseStream,
context,
relay => _streamSubscriber.Subscribe(request.InstanceUniqueName, relay),
request.InstanceUniqueName);
/// <inheritdoc />
public override Task SubscribeSite(
SiteStreamRequest request,
IServerStreamWriter<SiteStreamEvent> responseStream,
ServerCallContext context)
=> RunSubscriptionStreamAsync(
request.CorrelationId,
responseStream,
context,
// Site-wide, alarm-only: no per-instance filter. StreamRelayActor
// already drops IsConfiguredPlaceholder rows and maps only the
// enriched AlarmStateUpdate, so it is reused unchanged.
_streamSubscriber.SubscribeSiteAlarms,
"site-wide alarms");
/// <summary>
/// Shared streaming pipeline behind <see cref="SubscribeInstance"/> and
/// <see cref="SubscribeSite"/>: readiness/shutdown guards, correlation-id
/// validation, duplicate replacement, the concurrency cap, the bounded
/// <c>DropOldest</c> channel, the relay actor, connection telemetry, and the
/// guaranteed cleanup. The only variation between the two RPCs is how the
/// relay actor is subscribed to the site broadcast hub — supplied by
/// <paramref name="subscribe"/> — and the log/telemetry description.
/// </summary>
/// <param name="correlationId">The client-supplied correlation id (also the actor-name/dedup key).</param>
/// <param name="responseStream">The gRPC response stream to pump events into.</param>
/// <param name="context">The server call context (carries the client cancellation token).</param>
/// <param name="subscribe">Subscribes the relay actor to the hub, returning a subscription id.</param>
/// <param name="description">Human-readable subscription description for logging.</param>
private async Task RunSubscriptionStreamAsync(
string correlationId,
IServerStreamWriter<SiteStreamEvent> responseStream,
ServerCallContext context,
Func<IActorRef, string> subscribe,
string description)
{
if (!_ready)
throw new RpcException(new GrpcStatus(StatusCode.Unavailable, "Server not ready"));
@@ -228,15 +270,15 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
// have a restricted character set — a id containing '/', whitespace, or other
// disallowed characters would make ActorOf throw InvalidActorNameException,
// escaping as an unhandled RPC fault. Reject unsafe ids cleanly up front.
if (string.IsNullOrEmpty(request.CorrelationId) ||
!ActorPath.IsValidPathElement(request.CorrelationId))
if (string.IsNullOrEmpty(correlationId) ||
!ActorPath.IsValidPathElement(correlationId))
{
throw new RpcException(new GrpcStatus(
StatusCode.InvalidArgument, "correlation_id is missing or not a valid identifier"));
}
// Duplicate prevention -- cancel existing stream for this correlationId
if (_activeStreams.TryRemove(request.CorrelationId, out var existingEntry))
if (_activeStreams.TryRemove(correlationId, out var existingEntry))
{
existingEntry.Cts.Cancel();
existingEntry.Cts.Dispose();
@@ -256,10 +298,9 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
if (_maxStreamLifetime > TimeSpan.Zero && _maxStreamLifetime != Timeout.InfiniteTimeSpan)
streamCts.CancelAfter(_maxStreamLifetime);
var entry = new StreamEntry(streamCts);
_activeStreams[request.CorrelationId] = entry;
_activeStreams[correlationId] = entry;
long dropped = 0;
var correlationId = request.CorrelationId;
var channel = Channel.CreateBounded<SiteStreamEvent>(
new BoundedChannelOptions(1000) { FullMode = BoundedChannelFullMode.DropOldest },
_ =>
@@ -275,8 +316,8 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
var actorSeq = Interlocked.Increment(ref _actorCounter);
var relayActor = _actorSystem!.ActorOf(
Props.Create(typeof(Actors.StreamRelayActor), request.CorrelationId, channel.Writer),
$"stream-relay-{request.CorrelationId}-{actorSeq}");
Props.Create(typeof(Actors.StreamRelayActor), correlationId, channel.Writer),
$"stream-relay-{correlationId}-{actorSeq}");
// The previous code called _streamSubscriber.Subscribe
// OUTSIDE the try block that owns relay-actor cleanup. If Subscribe threw
@@ -289,23 +330,23 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
string subscriptionId;
try
{
subscriptionId = _streamSubscriber.Subscribe(request.InstanceUniqueName, relayActor);
subscriptionId = subscribe(relayActor);
}
catch (Exception ex)
{
_logger.LogWarning(ex,
"Subscribe failed for {Instance} (correlation {CorrelationId}); cleaning up relay actor.",
request.InstanceUniqueName, request.CorrelationId);
"Subscribe failed for {Description} (correlation {CorrelationId}); cleaning up relay actor.",
description, correlationId);
_actorSystem!.Stop(relayActor);
channel.Writer.TryComplete();
_activeStreams.TryRemove(
new KeyValuePair<string, StreamEntry>(request.CorrelationId, entry));
new KeyValuePair<string, StreamEntry>(correlationId, entry));
throw;
}
_logger.LogInformation(
"Stream {CorrelationId} started for {Instance} (subscription {SubscriptionId})",
request.CorrelationId, request.InstanceUniqueName, subscriptionId);
"Stream {CorrelationId} started for {Description} (subscription {SubscriptionId})",
correlationId, description, subscriptionId);
// Telemetry follow-on: the connection is now fully established (Subscribe
// succeeded, so no leak via the catch above). Count it up here and balance
@@ -335,11 +376,11 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
// Only remove our own entry -- a replacement stream may have already taken the slot
_activeStreams.TryRemove(
new KeyValuePair<string, StreamEntry>(request.CorrelationId, entry));
new KeyValuePair<string, StreamEntry>(correlationId, entry));
_logger.LogInformation(
"Stream {CorrelationId} for {Instance} ended",
request.CorrelationId, request.InstanceUniqueName);
"Stream {CorrelationId} for {Description} ended",
correlationId, description);
}
}