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]