diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/ISiteStreamSubscriber.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/ISiteStreamSubscriber.cs index 2995d7d2..398cbf3e 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/ISiteStreamSubscriber.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/ISiteStreamSubscriber.cs @@ -17,6 +17,16 @@ public interface ISiteStreamSubscriber /// A subscription ID that can be used for unsubscription. string Subscribe(string instanceName, IActorRef subscriber); + /// + /// Subscribes an actor to receive ALARM events for ALL instances on the site + /// (no per-instance filter). Only AlarmStateChanged events are + /// forwarded; attribute-value events are dropped. Backs the site-wide + /// SubscribeSite gRPC stream used by the aggregated Alarm Summary. + /// + /// The actor reference that will receive alarm stream event messages. + /// A subscription ID that can be used for unsubscription. + string SubscribeSiteAlarms(IActorRef subscriber); + /// /// Removes all subscriptions for the given actor. /// diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs index a8e2c465..0fb89c4a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs @@ -210,10 +210,52 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase internal TimeSpan MaxStreamLifetime => _maxStreamLifetime; /// - public override async Task SubscribeInstance( + public override Task SubscribeInstance( InstanceStreamRequest request, IServerStreamWriter responseStream, ServerCallContext context) + => RunSubscriptionStreamAsync( + request.CorrelationId, + responseStream, + context, + relay => _streamSubscriber.Subscribe(request.InstanceUniqueName, relay), + request.InstanceUniqueName); + + /// + public override Task SubscribeSite( + SiteStreamRequest request, + IServerStreamWriter 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"); + + /// + /// Shared streaming pipeline behind and + /// : readiness/shutdown guards, correlation-id + /// validation, duplicate replacement, the concurrency cap, the bounded + /// DropOldest 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 + /// — and the log/telemetry description. + /// + /// The client-supplied correlation id (also the actor-name/dedup key). + /// The gRPC response stream to pump events into. + /// The server call context (carries the client cancellation token). + /// Subscribes the relay actor to the hub, returning a subscription id. + /// Human-readable subscription description for logging. + private async Task RunSubscriptionStreamAsync( + string correlationId, + IServerStreamWriter responseStream, + ServerCallContext context, + Func 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( 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(request.CorrelationId, entry)); + new KeyValuePair(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(request.CorrelationId, entry)); + new KeyValuePair(correlationId, entry)); _logger.LogInformation( - "Stream {CorrelationId} for {Instance} ended", - request.CorrelationId, request.InstanceUniqueName); + "Stream {CorrelationId} for {Description} ended", + correlationId, description); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto b/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto index a7c9c478..5876dcbf 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto @@ -7,6 +7,10 @@ import "google/protobuf/wrappers.proto"; // Int32Value service SiteStreamService { rpc SubscribeInstance(InstanceStreamRequest) returns (stream SiteStreamEvent); + // Site-wide, alarm-only live stream (aggregated Alarm Summary): every + // AlarmStateChanged for ALL instances on the site, no per-instance filter. + // Attribute updates are never carried on this stream. + rpc SubscribeSite(SiteStreamRequest) returns (stream SiteStreamEvent); rpc IngestAuditEvents(AuditEventBatch) returns (IngestAck); rpc IngestCachedTelemetry(CachedTelemetryBatch) returns (IngestAck); rpc PullAuditEvents(PullAuditEventsRequest) returns (PullAuditEventsResponse); @@ -18,6 +22,13 @@ message InstanceStreamRequest { string instance_unique_name = 2; } +// Request for the site-wide, alarm-only SubscribeSite stream. Unlike +// InstanceStreamRequest there is NO instance filter — the stream carries alarm +// transitions for every instance on the site. +message SiteStreamRequest { + string correlation_id = 1; +} + message SiteStreamEvent { string correlation_id = 1; oneof event { diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs index ca300fa5..6cd9eb49 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs @@ -28,6 +28,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { "L3Byb3RvYnVmL3RpbWVzdGFtcC5wcm90bxoeZ29vZ2xlL3Byb3RvYnVmL3dy", "YXBwZXJzLnByb3RvIk0KFUluc3RhbmNlU3RyZWFtUmVxdWVzdBIWCg5jb3Jy", "ZWxhdGlvbl9pZBgBIAEoCRIcChRpbnN0YW5jZV91bmlxdWVfbmFtZRgCIAEo", + "CSIrChFTaXRlU3RyZWFtUmVxdWVzdBIWCg5jb3JyZWxhdGlvbl9pZBgBIAEo", "CSKoAQoPU2l0ZVN0cmVhbUV2ZW50EhYKDmNvcnJlbGF0aW9uX2lkGAEgASgJ", "Ej0KEWF0dHJpYnV0ZV9jaGFuZ2VkGAIgASgLMiAuc2l0ZXN0cmVhbS5BdHRy", "aWJ1dGVWYWx1ZVVwZGF0ZUgAEjUKDWFsYXJtX2NoYW5nZWQYAyABKAsyHC5z", @@ -95,22 +96,25 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { "VEFURV9BQ1RJVkUQAiqFAQoOQWxhcm1MZXZlbEVudW0SFAoQQUxBUk1fTEVW", "RUxfTk9ORRAAEhMKD0FMQVJNX0xFVkVMX0xPVxABEhcKE0FMQVJNX0xFVkVM", "X0xPV19MT1cQAhIUChBBTEFSTV9MRVZFTF9ISUdIEAMSGQoVQUxBUk1fTEVW", - "RUxfSElHSF9ISUdIEAQytwMKEVNpdGVTdHJlYW1TZXJ2aWNlElUKEVN1YnNj", + "RUxfSElHSF9ISUdIEAQyhgQKEVNpdGVTdHJlYW1TZXJ2aWNlElUKEVN1YnNj", "cmliZUluc3RhbmNlEiEuc2l0ZXN0cmVhbS5JbnN0YW5jZVN0cmVhbVJlcXVl", - "c3QaGy5zaXRlc3RyZWFtLlNpdGVTdHJlYW1FdmVudDABEkcKEUluZ2VzdEF1", - "ZGl0RXZlbnRzEhsuc2l0ZXN0cmVhbS5BdWRpdEV2ZW50QmF0Y2gaFS5zaXRl", - "c3RyZWFtLkluZ2VzdEFjaxJQChVJbmdlc3RDYWNoZWRUZWxlbWV0cnkSIC5z", - "aXRlc3RyZWFtLkNhY2hlZFRlbGVtZXRyeUJhdGNoGhUuc2l0ZXN0cmVhbS5J", - "bmdlc3RBY2sSWgoPUHVsbEF1ZGl0RXZlbnRzEiIuc2l0ZXN0cmVhbS5QdWxs", - "QXVkaXRFdmVudHNSZXF1ZXN0GiMuc2l0ZXN0cmVhbS5QdWxsQXVkaXRFdmVu", - "dHNSZXNwb25zZRJUCg1QdWxsU2l0ZUNhbGxzEiAuc2l0ZXN0cmVhbS5QdWxs", - "U2l0ZUNhbGxzUmVxdWVzdBohLnNpdGVzdHJlYW0uUHVsbFNpdGVDYWxsc1Jl", - "c3BvbnNlQiuqAihaQi5NT00uV1cuU2NhZGFCcmlkZ2UuQ29tbXVuaWNhdGlv", - "bi5HcnBjYgZwcm90bzM=")); + "c3QaGy5zaXRlc3RyZWFtLlNpdGVTdHJlYW1FdmVudDABEk0KDVN1YnNjcmli", + "ZVNpdGUSHS5zaXRlc3RyZWFtLlNpdGVTdHJlYW1SZXF1ZXN0Ghsuc2l0ZXN0", + "cmVhbS5TaXRlU3RyZWFtRXZlbnQwARJHChFJbmdlc3RBdWRpdEV2ZW50cxIb", + "LnNpdGVzdHJlYW0uQXVkaXRFdmVudEJhdGNoGhUuc2l0ZXN0cmVhbS5Jbmdl", + "c3RBY2sSUAoVSW5nZXN0Q2FjaGVkVGVsZW1ldHJ5EiAuc2l0ZXN0cmVhbS5D", + "YWNoZWRUZWxlbWV0cnlCYXRjaBoVLnNpdGVzdHJlYW0uSW5nZXN0QWNrEloK", + "D1B1bGxBdWRpdEV2ZW50cxIiLnNpdGVzdHJlYW0uUHVsbEF1ZGl0RXZlbnRz", + "UmVxdWVzdBojLnNpdGVzdHJlYW0uUHVsbEF1ZGl0RXZlbnRzUmVzcG9uc2US", + "VAoNUHVsbFNpdGVDYWxscxIgLnNpdGVzdHJlYW0uUHVsbFNpdGVDYWxsc1Jl", + "cXVlc3QaIS5zaXRlc3RyZWFtLlB1bGxTaXRlQ2FsbHNSZXNwb25zZUIrqgIo", + "WkIuTU9NLldXLlNjYWRhQnJpZGdlLkNvbW11bmljYXRpb24uR3JwY2IGcHJv", + "dG8z")); descriptor = pbr::FileDescriptor.FromGeneratedCode(descriptorData, new pbr::FileDescriptor[] { global::Google.Protobuf.WellKnownTypes.TimestampReflection.Descriptor, global::Google.Protobuf.WellKnownTypes.WrappersReflection.Descriptor, }, new pbr::GeneratedClrTypeInfo(new[] {typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.Quality), typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmStateEnum), typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmLevelEnum), }, null, new pbr::GeneratedClrTypeInfo[] { new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.InstanceStreamRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.InstanceStreamRequest.Parser, new[]{ "CorrelationId", "InstanceUniqueName" }, null, null, null, null), + new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamRequest), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamRequest.Parser, new[]{ "CorrelationId" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamEvent), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamEvent.Parser, new[]{ "CorrelationId", "AttributeChanged", "AlarmChanged" }, new[]{ "Event" }, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AttributeValueUpdate), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AttributeValueUpdate.Parser, new[]{ "InstanceUniqueName", "AttributePath", "AttributeName", "Value", "Quality", "Timestamp" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmStateUpdate), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmStateUpdate.Parser, new[]{ "InstanceUniqueName", "AlarmName", "State", "Priority", "Timestamp", "Level", "Message", "Kind", "Active", "Acknowledged", "Confirmed", "ShelveState", "Suppressed", "SourceReference", "AlarmTypeName", "Category", "OperatorUser", "OperatorComment", "OriginalRaiseTime", "CurrentValue", "LimitValue", "NativeSourceCanonicalName", "IsConfiguredPlaceholder" }, null, null, null, null), @@ -394,6 +398,209 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { } + /// + /// Request for the site-wide, alarm-only SubscribeSite stream. Unlike + /// InstanceStreamRequest there is NO instance filter — the stream carries alarm + /// transitions for every instance on the site. + /// + [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")] + public sealed partial class SiteStreamRequest : pb::IMessage + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + , pb::IBufferMessage + #endif + { + private static readonly pb::MessageParser _parser = new pb::MessageParser(() => new SiteStreamRequest()); + private pb::UnknownFieldSet _unknownFields; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public static pb::MessageParser Parser { get { return _parser; } } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public static pbr::MessageDescriptor Descriptor { + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[1]; } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + pbr::MessageDescriptor pb::IMessage.Descriptor { + get { return Descriptor; } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public SiteStreamRequest() { + OnConstruction(); + } + + partial void OnConstruction(); + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public SiteStreamRequest(SiteStreamRequest other) : this() { + correlationId_ = other.correlationId_; + _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public SiteStreamRequest Clone() { + return new SiteStreamRequest(this); + } + + /// Field number for the "correlation_id" field. + public const int CorrelationIdFieldNumber = 1; + private string correlationId_ = ""; + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public string CorrelationId { + get { return correlationId_; } + set { + correlationId_ = pb::ProtoPreconditions.CheckNotNull(value, "value"); + } + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override bool Equals(object other) { + return Equals(other as SiteStreamRequest); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public bool Equals(SiteStreamRequest other) { + if (ReferenceEquals(other, null)) { + return false; + } + if (ReferenceEquals(other, this)) { + return true; + } + if (CorrelationId != other.CorrelationId) return false; + return Equals(_unknownFields, other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override int GetHashCode() { + int hash = 1; + if (CorrelationId.Length != 0) hash ^= CorrelationId.GetHashCode(); + if (_unknownFields != null) { + hash ^= _unknownFields.GetHashCode(); + } + return hash; + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public override string ToString() { + return pb::JsonFormatter.ToDiagnosticString(this); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void WriteTo(pb::CodedOutputStream output) { + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + output.WriteRawMessage(this); + #else + if (CorrelationId.Length != 0) { + output.WriteRawTag(10); + output.WriteString(CorrelationId); + } + if (_unknownFields != null) { + _unknownFields.WriteTo(output); + } + #endif + } + + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + void pb::IBufferMessage.InternalWriteTo(ref pb::WriteContext output) { + if (CorrelationId.Length != 0) { + output.WriteRawTag(10); + output.WriteString(CorrelationId); + } + if (_unknownFields != null) { + _unknownFields.WriteTo(ref output); + } + } + #endif + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public int CalculateSize() { + int size = 0; + if (CorrelationId.Length != 0) { + size += 1 + pb::CodedOutputStream.ComputeStringSize(CorrelationId); + } + if (_unknownFields != null) { + size += _unknownFields.CalculateSize(); + } + return size; + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void MergeFrom(SiteStreamRequest other) { + if (other == null) { + return; + } + if (other.CorrelationId.Length != 0) { + CorrelationId = other.CorrelationId; + } + _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields); + } + + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public void MergeFrom(pb::CodedInputStream input) { + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + input.ReadRawMessage(this); + #else + uint tag; + while ((tag = input.ReadTag()) != 0) { + if ((tag & 7) == 4) { + // Abort on any end group tag. + return; + } + switch(tag) { + default: + _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, input); + break; + case 10: { + CorrelationId = input.ReadString(); + break; + } + } + } + #endif + } + + #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + void pb::IBufferMessage.InternalMergeFrom(ref pb::ParseContext input) { + uint tag; + while ((tag = input.ReadTag()) != 0) { + if ((tag & 7) == 4) { + // Abort on any end group tag. + return; + } + switch(tag) { + default: + _unknownFields = pb::UnknownFieldSet.MergeFieldFrom(_unknownFields, ref input); + break; + case 10: { + CorrelationId = input.ReadString(); + break; + } + } + } + } + #endif + + } + [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")] public sealed partial class SiteStreamEvent : pb::IMessage #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE @@ -409,7 +616,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[1]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[2]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -740,7 +947,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[2]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[3]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -1132,7 +1339,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[3]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[4]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -2201,7 +2408,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[4]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[5]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -3216,7 +3423,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[5]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[6]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -3403,7 +3610,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[6]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[7]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -3596,7 +3803,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[7]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[8]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -4254,7 +4461,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[8]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[9]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -4507,7 +4714,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[9]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[10]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -4702,7 +4909,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[10]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[11]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -4946,7 +5153,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[11]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[12]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -5179,7 +5386,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[12]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[13]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] @@ -5469,7 +5676,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public static pbr::MessageDescriptor Descriptor { - get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[13]; } + get { return global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor.MessageTypes[14]; } } [global::System.Diagnostics.DebuggerNonUserCodeAttribute] diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/SitestreamGrpc.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/SitestreamGrpc.cs index b57de38e..6db5da2a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/SitestreamGrpc.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/SitestreamGrpc.cs @@ -50,6 +50,8 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] static readonly grpc::Marshaller __Marshaller_sitestream_SiteStreamEvent = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamEvent.Parser)); [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + static readonly grpc::Marshaller __Marshaller_sitestream_SiteStreamRequest = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamRequest.Parser)); + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] static readonly grpc::Marshaller __Marshaller_sitestream_AuditEventBatch = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventBatch.Parser)); [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] static readonly grpc::Marshaller __Marshaller_sitestream_IngestAck = grpc::Marshallers.Create(__Helper_SerializeMessage, context => __Helper_DeserializeMessage(context, global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.IngestAck.Parser)); @@ -72,6 +74,14 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { __Marshaller_sitestream_InstanceStreamRequest, __Marshaller_sitestream_SiteStreamEvent); + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + static readonly grpc::Method __Method_SubscribeSite = new grpc::Method( + grpc::MethodType.ServerStreaming, + __ServiceName, + "SubscribeSite", + __Marshaller_sitestream_SiteStreamRequest, + __Marshaller_sitestream_SiteStreamEvent); + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] static readonly grpc::Method __Method_IngestAuditEvents = new grpc::Method( grpc::MethodType.Unary, @@ -120,6 +130,21 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { throw new grpc::RpcException(new grpc::Status(grpc::StatusCode.Unimplemented, "")); } + /// + /// Site-wide, alarm-only live stream (aggregated Alarm Summary): every + /// AlarmStateChanged for ALL instances on the site, no per-instance filter. + /// Attribute updates are never carried on this stream. + /// + /// The request received from the client. + /// Used for sending responses back to the client. + /// The context of the server-side call handler being invoked. + /// A task indicating completion of the handler. + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual global::System.Threading.Tasks.Task SubscribeSite(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamRequest request, grpc::IServerStreamWriter responseStream, grpc::ServerCallContext context) + { + throw new grpc::RpcException(new grpc::Status(grpc::StatusCode.Unimplemented, "")); + } + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] public virtual global::System.Threading.Tasks.Task IngestAuditEvents(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventBatch request, grpc::ServerCallContext context) { @@ -183,6 +208,34 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { { return CallInvoker.AsyncServerStreamingCall(__Method_SubscribeInstance, null, options, request); } + /// + /// Site-wide, alarm-only live stream (aggregated Alarm Summary): every + /// AlarmStateChanged for ALL instances on the site, no per-instance filter. + /// Attribute updates are never carried on this stream. + /// + /// The request to send to the server. + /// The initial metadata to send with the call. This parameter is optional. + /// An optional deadline for the call. The call will be cancelled if deadline is hit. + /// An optional token for canceling the call. + /// The call object. + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual grpc::AsyncServerStreamingCall SubscribeSite(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamRequest request, grpc::Metadata headers = null, global::System.DateTime? deadline = null, global::System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)) + { + return SubscribeSite(request, new grpc::CallOptions(headers, deadline, cancellationToken)); + } + /// + /// Site-wide, alarm-only live stream (aggregated Alarm Summary): every + /// AlarmStateChanged for ALL instances on the site, no per-instance filter. + /// Attribute updates are never carried on this stream. + /// + /// The request to send to the server. + /// The options for the call. + /// The call object. + [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] + public virtual grpc::AsyncServerStreamingCall SubscribeSite(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamRequest request, grpc::CallOptions options) + { + return CallInvoker.AsyncServerStreamingCall(__Method_SubscribeSite, null, options, request); + } [global::System.CodeDom.Compiler.GeneratedCode("grpc_csharp_plugin", null)] public virtual global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.IngestAck IngestAuditEvents(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventBatch request, grpc::Metadata headers = null, global::System.DateTime? deadline = null, global::System.Threading.CancellationToken cancellationToken = default(global::System.Threading.CancellationToken)) { @@ -278,6 +331,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { { return grpc::ServerServiceDefinition.CreateBuilder() .AddMethod(__Method_SubscribeInstance, serviceImpl.SubscribeInstance) + .AddMethod(__Method_SubscribeSite, serviceImpl.SubscribeSite) .AddMethod(__Method_IngestAuditEvents, serviceImpl.IngestAuditEvents) .AddMethod(__Method_IngestCachedTelemetry, serviceImpl.IngestCachedTelemetry) .AddMethod(__Method_PullAuditEvents, serviceImpl.PullAuditEvents) @@ -292,6 +346,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { public static void BindService(grpc::ServiceBinderBase serviceBinder, SiteStreamServiceBase serviceImpl) { serviceBinder.AddMethod(__Method_SubscribeInstance, serviceImpl == null ? null : new grpc::ServerStreamingServerMethod(serviceImpl.SubscribeInstance)); + serviceBinder.AddMethod(__Method_SubscribeSite, serviceImpl == null ? null : new grpc::ServerStreamingServerMethod(serviceImpl.SubscribeSite)); serviceBinder.AddMethod(__Method_IngestAuditEvents, serviceImpl == null ? null : new grpc::UnaryServerMethod(serviceImpl.IngestAuditEvents)); serviceBinder.AddMethod(__Method_IngestCachedTelemetry, serviceImpl == null ? null : new grpc::UnaryServerMethod(serviceImpl.IngestCachedTelemetry)); serviceBinder.AddMethod(__Method_PullAuditEvents, serviceImpl == null ? null : new grpc::UnaryServerMethod(serviceImpl.PullAuditEvents)); diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs index 1e3b01c7..a44e570e 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs @@ -21,6 +21,8 @@ public class SiteStreamGrpcServerTests : TestKit _subscriber = Substitute.For(); _subscriber.Subscribe(Arg.Any(), Arg.Any()) .Returns("sub-1"); + _subscriber.SubscribeSiteAlarms(Arg.Any()) + .Returns("site-sub-1"); _logger = NullLogger.Instance; } @@ -266,6 +268,102 @@ public class SiteStreamGrpcServerTests : TestKit await streamTask; } + // --- SubscribeSite (site-wide, alarm-only aggregated stream, plan #10 T2) --- + + private static SiteStreamRequest MakeSiteRequest(string correlationId = "site-corr-1") + => new() { CorrelationId = correlationId }; + + [Fact] + public async Task SubscribeSite_SubscribesSiteAlarmsAndRemovesOnCancel() + { + var server = CreateServer(); + server.SetReady(Sys); + + var cts = new CancellationTokenSource(); + var context = CreateMockContext(cts.Token); + var writer = Substitute.For>(); + + var streamTask = Task.Run(() => server.SubscribeSite( + MakeSiteRequest("site-corr-sub"), writer, context)); + + await WaitForConditionAsync(() => server.ActiveStreamCount == 1); + + // Site-wide handler must call SubscribeSiteAlarms (no instance filter), + // never the per-instance Subscribe. + _subscriber.Received(1).SubscribeSiteAlarms(Arg.Any()); + _subscriber.DidNotReceive().Subscribe(Arg.Any(), Arg.Any()); + + cts.Cancel(); + await streamTask; + + _subscriber.Received(1).RemoveSubscriber(Arg.Any()); + Assert.Equal(0, server.ActiveStreamCount); + } + + [Fact] + public async Task SubscribeSite_RejectsUnsafeCorrelationId() + { + var server = CreateServer(); + server.SetReady(Sys); + + var writer = Substitute.For>(); + var context = CreateMockContext(); + + var ex = await Assert.ThrowsAsync( + () => server.SubscribeSite(MakeSiteRequest("bad/id"), writer, context)); + + Assert.Equal(StatusCode.InvalidArgument, ex.StatusCode); + Assert.Equal(0, server.ActiveStreamCount); + } + + [Fact] + public async Task SubscribeSite_RelaysAlarmStateChangedAsAlarmStateUpdate() + { + var server = CreateServer(); + server.SetReady(Sys); + + // Capture the relay actor spawned for the site-wide subscription. + IActorRef? capturedActor = null; + _subscriber.SubscribeSiteAlarms(Arg.Any()) + .Returns(ci => + { + capturedActor = ci.Arg(); + return "site-sub-relay"; + }); + + var cts = new CancellationTokenSource(); + var context = CreateMockContext(cts.Token); + var writer = Substitute.For>(); + var writtenEvents = new List(); + writer.WriteAsync(Arg.Any(), Arg.Any()) + .Returns(Task.CompletedTask) + .AndDoes(ci => writtenEvents.Add(ci.Arg())); + + var streamTask = Task.Run(() => server.SubscribeSite( + MakeSiteRequest("site-corr-write"), writer, context)); + + await WaitForConditionAsync(() => capturedActor != null); + + // A real alarm transition for ANY instance must arrive as an AlarmStateUpdate. + capturedActor!.Tell(new Commons.Messages.Streaming.AlarmStateChanged( + "Site1.Pump01", + "HighPressure", + Commons.Types.Enums.AlarmState.Active, + 700, + DateTimeOffset.UtcNow)); + + await WaitForConditionAsync(() => writtenEvents.Count >= 1); + + Assert.Single(writtenEvents); + Assert.Equal("site-corr-write", writtenEvents[0].CorrelationId); + Assert.Equal(SiteStreamEvent.EventOneofCase.AlarmChanged, writtenEvents[0].EventCase); + Assert.Equal("Site1.Pump01", writtenEvents[0].AlarmChanged.InstanceUniqueName); + Assert.Equal("HighPressure", writtenEvents[0].AlarmChanged.AlarmName); + + cts.Cancel(); + await streamTask; + } + [Theory] [InlineData("corr/with/slash")] [InlineData("corr with space")]