From 6d5ecc92a56c7d46868a5c5544054b99a010a8cb Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 10 Jul 2026 11:49:49 -0400 Subject: [PATCH] feat(sitestream): add central SubscribeSiteAsync alarm-only client (plan #10 T3) Add SiteStreamGrpcClient.SubscribeSiteAsync mirroring SubscribeAsync but for the site-wide, alarm-only SubscribeSite RPC: builds a SiteStreamRequest, opens the server stream, and delivers each event via a typed Action callback (this stream is alarm-only by contract, so Task 4's per-site cache consumes an alarm delta with no downstream type test). Reuses the shared enrichment mapping via a new internal ConvertToAlarmEvent helper that returns null for any non-alarm event, defensively filtering anything that should never appear on the stream. Factory unchanged - it already caches a client per (site, endpoint). Adds focused unit tests for the alarm-only filter and the test-only-client guard. Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj --- .../Grpc/SiteStreamGrpcClient.cs | 79 +++++++++++++++++++ .../Grpc/SiteStreamGrpcClientTests.cs | 72 +++++++++++++++++ 2 files changed, 151 insertions(+) diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs index 8e57e851..65afa9ee 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs @@ -200,6 +200,74 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable } } + /// + /// Opens the site-wide, alarm-only server-streaming subscription + /// (SubscribeSite) for a whole site rather than a single instance. This is + /// the central per-site feed that backs the aggregated Alarm Summary live cache + /// (plan #10): the site runtime's SubscribeSiteAlarms hub drops the + /// per-instance filter and carries only events for + /// all instances on the site (attributes are deliberately excluded — the + /// summary never shows them and they are far higher-volume). + /// + /// The callback is deliberately typed as + /// rather than the generic Action<object> used by + /// : this stream is alarm-only by contract, so Task 4's + /// per-site cache consumes an alarm delta directly with no downstream type test. + /// The mapping is still shared via ; a non-alarm + /// event (which should never appear on this stream) is defensively ignored rather + /// than delivered or thrown. + /// + /// This is a long-running async method; the caller launches it as a background task. + /// Error handling, cancellation and cleanup mirror . + /// + /// Unique identifier for this subscription. + /// Callback invoked for each alarm delta received from the site-wide stream. + /// Callback invoked when the subscription encounters an error. + /// Cancellation token to stop the subscription. + /// A task that represents the asynchronous operation. + public virtual async Task SubscribeSiteAsync( + string correlationId, + Action onAlarmEvent, + Action onError, + CancellationToken ct) + { + if (_client is null) + throw new InvalidOperationException("Cannot subscribe on a test-only client."); + + var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); + RegisterSubscription(correlationId, cts); + + var request = new SiteStreamRequest + { + CorrelationId = correlationId + }; + + try + { + using var call = _client.SubscribeSite(request, cancellationToken: cts.Token); + + await foreach (var evt in call.ResponseStream.ReadAllAsync(cts.Token)) + { + // Site-wide stream is alarm-only by contract; defensively ignore anything else. + if (ConvertToAlarmEvent(evt) is { } alarm) + onAlarmEvent(alarm); + } + } + catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled) + { + // Normal cancellation — not an error + } + catch (Exception ex) + { + onError(ex); + } + finally + { + // Remove only our own entry -- a racing reconnect may already own the slot. + RemoveSubscription(correlationId, cts); + } + } + /// /// Cancels an active subscription by correlation ID. /// @@ -260,6 +328,17 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable _ => null }; + /// + /// Maps a proto to a domain , + /// returning null for any non-alarm event. Used by + /// to enforce the alarm-only contract of the site-wide stream without duplicating the + /// enrichment mapping in . Internal for testability. + /// + /// The protobuf site stream event to convert. + /// The mapped , or null if the event is not an alarm. + internal static AlarmStateChanged? ConvertToAlarmEvent(SiteStreamEvent evt) => + ConvertToDomainEvent(evt) as AlarmStateChanged; + /// Parses the wire "kind" string back to ; defaults to Computed. /// The wire "kind" string from the gRPC payload; null or unrecognised defaults to . /// The parsed , or when the value is null or unrecognised. diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs index 82d4addb..34b8563b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs @@ -208,6 +208,78 @@ public class SiteStreamGrpcClientTests Assert.True(cts2.IsCancellationRequested); } + // --- Site-wide (SubscribeSite) alarm-only stream (plan #10 T3) --- + + [Fact] + public void ConvertToAlarmEvent_AlarmChanged_ReturnsMappedAlarm() + { + // An alarm event on the site-wide stream is delivered to onAlarmEvent with full enrichment. + var ts = DateTimeOffset.UtcNow; + var evt = new SiteStreamEvent + { + CorrelationId = "site-corr", + AlarmChanged = new AlarmStateUpdate + { + InstanceUniqueName = "Site1.Motor01", + AlarmName = "T01.Hi", + State = AlarmStateEnum.AlarmStateActive, + Priority = 700, + Timestamp = Timestamp.FromDateTimeOffset(ts), + Kind = "NativeOpcUa", + Active = true, + SourceReference = "T01.Hi" + } + }; + + var alarm = SiteStreamGrpcClient.ConvertToAlarmEvent(evt); + + Assert.NotNull(alarm); + Assert.Equal("Site1.Motor01", alarm!.InstanceUniqueName); + Assert.Equal("T01.Hi", alarm.AlarmName); + Assert.Equal(AlarmState.Active, alarm.State); + Assert.Equal(AlarmKind.NativeOpcUa, alarm.Kind); + Assert.Equal("T01.Hi", alarm.SourceReference); + } + + [Fact] + public void ConvertToAlarmEvent_AttributeChanged_ReturnsNull() + { + // Attribute events must never appear on the alarm-only site-wide stream; if one + // somehow arrives it is defensively filtered out rather than delivered or thrown. + var evt = new SiteStreamEvent + { + CorrelationId = "site-corr", + AttributeChanged = new AttributeValueUpdate + { + InstanceUniqueName = "Site1.Pump01", + AttributePath = "Modules.IO", + AttributeName = "Temperature", + Value = "42.5", + Quality = Quality.Good, + Timestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow) + } + }; + + Assert.Null(SiteStreamGrpcClient.ConvertToAlarmEvent(evt)); + } + + [Fact] + public void ConvertToAlarmEvent_UnknownEvent_ReturnsNull() + { + var evt = new SiteStreamEvent { CorrelationId = "site-corr" }; + Assert.Null(SiteStreamGrpcClient.ConvertToAlarmEvent(evt)); + } + + [Fact] + public async Task SubscribeSiteAsync_OnTestOnlyClient_Throws() + { + // Guards against subscribing on a channel-less test double, mirroring SubscribeAsync. + var client = SiteStreamGrpcClient.CreateForTesting(); + + await Assert.ThrowsAsync(() => + client.SubscribeSiteAsync("corr", _ => { }, _ => { }, CancellationToken.None)); + } + // --- Communication-003 regression tests --- [Fact]