diff --git a/ZB.MOM.WW.OtOpcUa.slnx b/ZB.MOM.WW.OtOpcUa.slnx index a1090eab..b6647bd2 100644 --- a/ZB.MOM.WW.OtOpcUa.slnx +++ b/ZB.MOM.WW.OtOpcUa.slnx @@ -76,6 +76,7 @@ + diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs new file mode 100644 index 00000000..900ee4a2 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs @@ -0,0 +1,167 @@ +using Google.Protobuf.WellKnownTypes; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; + +namespace ZB.MOM.WW.OtOpcUa.Host.Grpc; + +/// +/// Node-side projection of a domain onto its wire +/// envelope (per-cluster mesh Phase 5). The exact reverse of +/// telemetry.proto: it transcribes every domain field onto the matching proto field and +/// selects the envelope's oneof arm from the subtype. +/// +/// +/// +/// DateTime → Timestamp is Kind-defensive. throws +/// unless the source is . Producers +/// use today, but a stray or +/// value must never crash the stream — so a Local value +/// is converted and an Unspecified value is assumed already-UTC (matching the producer intent +/// and preserving the instant on any machine). +/// +/// +/// Nullable presence is preserved. A null nullable-DateTime leaves the proto +/// Timestamp unset (absent == null per the contract); a null optional string leaves the +/// proto field unset so the generated Has… presence distinguishes null from "". +/// +/// +public static class TelemetryProtoMapNode +{ + /// Projects onto a wire . + /// The domain telemetry item to project. + /// Stream correlation id echoed on every envelope. + /// The wire envelope with the matching oneof arm populated. + public static TelemetryEvent ToProto(TelemetryItem item, string correlationId) + { + ArgumentNullException.ThrowIfNull(item); + + return item switch + { + TelemetryItem.Alarm a => new TelemetryEvent + { + CorrelationId = correlationId, + AlarmTransition = MapAlarm(a.E), + }, + TelemetryItem.Script s => new TelemetryEvent + { + CorrelationId = correlationId, + ScriptLog = MapScript(s.E), + }, + TelemetryItem.Health h => new TelemetryEvent + { + CorrelationId = correlationId, + DriverHealth = MapHealth(h.E), + }, + TelemetryItem.Resilience r => new TelemetryEvent + { + CorrelationId = correlationId, + DriverResilience = MapResilience(r.E), + }, + _ => throw new ArgumentOutOfRangeException( + nameof(item), item.GetType().Name, "Unknown TelemetryItem subtype"), + }; + } + + private static AlarmTransition MapAlarm(AlarmTransitionEvent e) + { + var msg = new AlarmTransition + { + AlarmId = e.AlarmId, + EquipmentPath = e.EquipmentPath, + AlarmName = e.AlarmName, + TransitionKind = e.TransitionKind, + Severity = e.Severity, + Message = e.Message, + User = e.User, + TimestampUtc = ToUtcTimestamp(e.TimestampUtc), + AlarmTypeName = e.AlarmTypeName, + }; + + if (e.Comment is not null) + msg.Comment = e.Comment; + if (e.HistorizeToAveva is not null) + msg.HistorizeToAveva = e.HistorizeToAveva.Value; + msg.ReferencingEquipmentPaths.AddRange(e.ReferencingEquipmentPaths ?? Enumerable.Empty()); + + return msg; + } + + private static ScriptLog MapScript(ScriptLogEntry e) + { + var msg = new ScriptLog + { + ScriptId = e.ScriptId, + Level = e.Level, + Message = e.Message, + TimestampUtc = ToUtcTimestamp(e.TimestampUtc), + }; + + if (e.VirtualTagId is not null) + msg.VirtualTagId = e.VirtualTagId; + if (e.AlarmId is not null) + msg.AlarmId = e.AlarmId; + if (e.EquipmentId is not null) + msg.EquipmentId = e.EquipmentId; + + return msg; + } + + private static DriverHealth MapHealth(DriverHealthChanged e) + { + var msg = new DriverHealth + { + ClusterId = e.ClusterId, + DriverInstanceId = e.DriverInstanceId, + State = e.State, + ErrorCount5Min = e.ErrorCount5Min, + PublishedUtc = ToUtcTimestamp(e.PublishedUtc), + }; + + if (e.LastSuccessfulReadUtc is not null) + msg.LastSuccessfulReadUtc = ToUtcTimestamp(e.LastSuccessfulReadUtc.Value); + if (e.LastError is not null) + msg.LastError = e.LastError; + + return msg; + } + + private static DriverResilienceStatus MapResilience(DriverResilienceStatusChanged e) + { + var msg = new DriverResilienceStatus + { + DriverInstanceId = e.DriverInstanceId, + HostName = e.HostName, + BreakerOpen = e.BreakerOpen, + ConsecutiveFailures = e.ConsecutiveFailures, + CurrentInFlight = e.CurrentInFlight, + LastSampledUtc = ToUtcTimestamp(e.LastSampledUtc), + PublishedUtc = ToUtcTimestamp(e.PublishedUtc), + }; + + if (e.LastBreakerOpenUtc is not null) + msg.LastBreakerOpenUtc = ToUtcTimestamp(e.LastBreakerOpenUtc.Value); + + return msg; + } + + /// + /// Converts a domain to a proto without ever + /// throwing on Kind: Utc is used as-is, Local is converted, and Unspecified is assumed + /// already-UTC (the producer contract — — with the wall-clock + /// ticks preserved so the instant round-trips). + /// + private static Timestamp ToUtcTimestamp(DateTime dt) + { + var utc = dt.Kind switch + { + DateTimeKind.Utc => dt, + DateTimeKind.Local => dt.ToUniversalTime(), + _ => DateTime.SpecifyKind(dt, DateTimeKind.Utc), + }; + + return Timestamp.FromDateTime(utc); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryStreamGrpcService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryStreamGrpcService.cs new file mode 100644 index 00000000..20dbc490 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryStreamGrpcService.cs @@ -0,0 +1,114 @@ +using Grpc.Core; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; +// The generated service container is itself named TelemetryStreamService; alias its nested server +// base so this impl can keep the natural name without colliding with the generated type. +using GeneratedServiceBase = + ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1.TelemetryStreamService.TelemetryStreamServiceBase; + +namespace ZB.MOM.WW.OtOpcUa.Host.Grpc; + +/// +/// Node-side live-telemetry streaming server (per-cluster mesh Phase 5). Central dials a driver +/// node and opens ; the node attaches a fresh reader on its +/// and pumps this node's own telemetry +/// (alarm-transitions / script-logs / driver-health / driver-resilience) to the caller as a +/// sequence of envelopes until the client cancels or the +/// max-lifetime cap fires. +/// +/// +/// The hub carries ONLY this node's telemetry, so a per-node stream never double-counts a peer's +/// events. Fan-out is lossy under backpressure by design (the per-subscriber channel is +/// bounded/DropOldest inside the hub). A simple process-wide concurrency cap sheds excess dials +/// with , and every accepted stream is bounded by a +/// linked max-lifetime CTS. +/// +public sealed class TelemetryStreamGrpcService : GeneratedServiceBase +{ + /// Process-wide cap on concurrently-open telemetry streams. + private const int MaxConcurrentStreams = 100; + + /// Upper bound on an accepted correlation id (defends against absurd inputs). + private const int MaxCorrelationIdLength = 256; + + /// Per-subscriber bounded-channel capacity requested from the hub. + private const int SubscriptionCapacity = 1000; + + /// Hard ceiling on a single stream's lifetime; the linked CTS cancels the pump when it fires. + private static readonly TimeSpan MaxStreamLifetime = TimeSpan.FromHours(4); + + /// Process-wide count of currently-open streams; mutated only via . + private static int _activeStreams; + + private readonly ITelemetryLocalHub _hub; + private readonly ILogger _log; + + /// Initializes a new instance of the class. + /// The node-local telemetry fan-out hub this stream drains. + /// Logger. + public TelemetryStreamGrpcService(ITelemetryLocalHub hub, ILogger log) + { + _hub = hub; + _log = log; + } + + /// + public override async Task Subscribe( + TelemetryStreamRequest request, + IServerStreamWriter responseStream, + ServerCallContext context) + { + ArgumentNullException.ThrowIfNull(request); + ArgumentNullException.ThrowIfNull(responseStream); + ArgumentNullException.ThrowIfNull(context); + + var correlationId = request.CorrelationId; + if (string.IsNullOrWhiteSpace(correlationId) || correlationId.Length > MaxCorrelationIdLength) + throw new RpcException(new Status( + StatusCode.InvalidArgument, + "correlation_id must be non-empty and at most 256 characters")); + + // Concurrency cap. The increment is ALWAYS paired: on rejection we decrement here and throw + // before entering the try; on acceptance the finally decrements exactly once. No path + // decrements without a matching successful increment, and no path double-decrements. + var active = Interlocked.Increment(ref _activeStreams); + if (active > MaxConcurrentStreams) + { + Interlocked.Decrement(ref _activeStreams); + _log.LogWarning( + "Telemetry stream rejected (correlationId={CorrelationId}): {Active} open streams exceeds cap {Cap}", + correlationId, active - 1, MaxConcurrentStreams); + throw new RpcException(new Status( + StatusCode.ResourceExhausted, "too many concurrent telemetry streams")); + } + + try + { + using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken); + lifetime.CancelAfter(MaxStreamLifetime); + using var sub = _hub.Subscribe(SubscriptionCapacity); + + _log.LogDebug("Telemetry stream opened (correlationId={CorrelationId})", correlationId); + + try + { + await foreach (var item in sub.Reader.ReadAllAsync(lifetime.Token).ConfigureAwait(false)) + await responseStream + .WriteAsync(TelemetryProtoMapNode.ToProto(item, correlationId), lifetime.Token) + .ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Normal termination: the client disconnected (context token) or the max-lifetime cap + // fired (linked token). Neither is an error — swallow and let the stream close cleanly. + } + + _log.LogDebug("Telemetry stream closed (correlationId={CorrelationId})", correlationId); + } + finally + { + Interlocked.Decrement(ref _activeStreams); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/AssemblyInfo.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/AssemblyInfo.cs new file mode 100644 index 00000000..63dc64f4 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/AssemblyInfo.cs @@ -0,0 +1,6 @@ +using Xunit; + +// The TelemetryStreamGrpcService concurrency counter is process-wide (static). Run this assembly's +// tests serially so the cap test observes a deterministic active-stream count with no cross-test +// interference. This project holds only telemetry tests, so serial execution costs nothing. +[assembly: CollectionBehavior(DisableTestParallelization = true)] diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/TelemetryStreamGrpcServiceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/TelemetryStreamGrpcServiceTests.cs new file mode 100644 index 00000000..f98e9c58 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/TelemetryStreamGrpcServiceTests.cs @@ -0,0 +1,287 @@ +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging; +using ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1; +using ZB.MOM.WW.OtOpcUa.Host.Grpc; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; + +namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Grpc; + +/// +/// Verifies the node-side telemetry streaming service (per-cluster mesh Phase 5): the hub is +/// fanned to a connected caller as envelopes, every domain field is +/// transcribed onto its proto field, DateTime→Timestamp mapping is Kind-defensive, a cancelled +/// context ends the stream cleanly, and the process-wide concurrency cap sheds excess dials with +/// . +/// +public sealed class TelemetryStreamGrpcServiceTests +{ + private static TelemetryStreamGrpcService NewService(ITelemetryLocalHub hub) => + new(hub, NullLogger.Instance); + + private static ServerCallContext Ctx(CancellationToken token) => new FakeServerCallContext(token); + + private static async Task WaitUntilAsync(Func condition, string because) + { + var deadline = DateTime.UtcNow.AddSeconds(5); + while (DateTime.UtcNow < deadline) + { + if (condition()) + return; + await Task.Delay(15); + } + + throw new TimeoutException($"Timed out waiting: {because}"); + } + + [Fact] + public async Task Streams_one_of_each_kind_with_fields_transcribed() + { + var hub = new TelemetryLocalHub(); + var service = NewService(hub); + var writer = new CapturingStreamWriter(); + using var cts = new CancellationTokenSource(); + var request = new TelemetryStreamRequest { CorrelationId = "corr-1" }; + + var publishedHealth = new DateTime(2026, 7, 23, 10, 0, 0, DateTimeKind.Utc); + var health = new DriverHealthChanged( + "cluster-a", "drv-1", "Healthy", LastSuccessfulReadUtc: null, LastError: null, + ErrorCount5Min: 3, PublishedUtc: publishedHealth); + var resilience = new DriverResilienceStatusChanged( + "drv-1", "host-x", BreakerOpen: true, ConsecutiveFailures: 4, CurrentInFlight: 2, + LastBreakerOpenUtc: null, LastSampledUtc: DateTime.UtcNow, PublishedUtc: DateTime.UtcNow); + + // Snapshot-style items are cached, so emitting them BEFORE the service attaches guarantees they + // are replayed to the fresh subscription — a deterministic "the pump is draining" signal. + hub.Emit(new TelemetryItem.Health(health)); + hub.Emit(new TelemetryItem.Resilience(resilience)); + + var pump = service.Subscribe(request, writer, Ctx(cts.Token)); + + await WaitUntilAsync(() => writer.Count >= 2, "cached snapshots to be replayed"); + + // Append-style items are live-only: emit them once the subscription is proven attached. + var alarm = new AlarmTransitionEvent( + "raw/dev/Speed", "Area/Line/Equip", "HighSpeed", "Activated", 700, "over limit", "operator1", + new DateTime(2026, 7, 23, 11, 0, 0, DateTimeKind.Utc), "LimitAlarm", + Comment: "manual ack", HistorizeToAveva: true, + ReferencingEquipmentPaths: ["Area/Line/EquipA", "Area/Line/EquipB"]); + var script = new ScriptLogEntry( + "script-9", "Warning", "threshold near", new DateTime(2026, 7, 23, 12, 0, 0, DateTimeKind.Utc), + VirtualTagId: "vt-1", AlarmId: null, EquipmentId: "equip-1"); + + hub.Emit(new TelemetryItem.Alarm(alarm)); + hub.Emit(new TelemetryItem.Script(script)); + + await WaitUntilAsync(() => writer.Count >= 4, "all four telemetry events to be streamed"); + + cts.Cancel(); + await pump; // cancellation must be swallowed — Subscribe returns cleanly. + + var events = writer.Snapshot(); + events.ShouldAllBe(e => e.CorrelationId == "corr-1"); + + var healthEvt = events.Single(e => e.EventCase == TelemetryEvent.EventOneofCase.DriverHealth); + healthEvt.DriverHealth.ClusterId.ShouldBe("cluster-a"); + healthEvt.DriverHealth.State.ShouldBe("Healthy"); + healthEvt.DriverHealth.ErrorCount5Min.ShouldBe(3); + healthEvt.DriverHealth.PublishedUtc.ToDateTime().ShouldBe(publishedHealth); + healthEvt.DriverHealth.LastSuccessfulReadUtc.ShouldBeNull(); // nullable DateTime absent + healthEvt.DriverHealth.HasLastError.ShouldBeFalse(); // nullable string absent + + var resilienceEvt = events.Single(e => e.EventCase == TelemetryEvent.EventOneofCase.DriverResilience); + resilienceEvt.DriverResilience.DriverInstanceId.ShouldBe("drv-1"); + resilienceEvt.DriverResilience.BreakerOpen.ShouldBeTrue(); + resilienceEvt.DriverResilience.LastBreakerOpenUtc.ShouldBeNull(); + + var alarmEvt = events.Single(e => e.EventCase == TelemetryEvent.EventOneofCase.AlarmTransition); + alarmEvt.AlarmTransition.AlarmId.ShouldBe("raw/dev/Speed"); + alarmEvt.AlarmTransition.TransitionKind.ShouldBe("Activated"); + alarmEvt.AlarmTransition.Severity.ShouldBe(700); + alarmEvt.AlarmTransition.Comment.ShouldBe("manual ack"); + alarmEvt.AlarmTransition.HistorizeToAveva.ShouldBeTrue(); + alarmEvt.AlarmTransition.ReferencingEquipmentPaths.ShouldBe(["Area/Line/EquipA", "Area/Line/EquipB"]); + + var scriptEvt = events.Single(e => e.EventCase == TelemetryEvent.EventOneofCase.ScriptLog); + scriptEvt.ScriptLog.ScriptId.ShouldBe("script-9"); + scriptEvt.ScriptLog.Level.ShouldBe("Warning"); + scriptEvt.ScriptLog.VirtualTagId.ShouldBe("vt-1"); + scriptEvt.ScriptLog.HasAlarmId.ShouldBeFalse(); // nullable string absent + scriptEvt.ScriptLog.EquipmentId.ShouldBe("equip-1"); + } + + [Fact] + public async Task Cancelled_context_ends_stream_cleanly() + { + var hub = new TelemetryLocalHub(); + var service = NewService(hub); + var writer = new CapturingStreamWriter(); + using var cts = new CancellationTokenSource(); + var request = new TelemetryStreamRequest { CorrelationId = "corr-cancel" }; + + var pump = service.Subscribe(request, writer, Ctx(cts.Token)); + + cts.Cancel(); + await Should.NotThrowAsync(() => pump); // client disconnect is normal, not an error. + } + + [Theory] + [InlineData("")] + [InlineData(" ")] + public async Task Empty_correlation_id_is_invalid_argument(string correlationId) + { + var service = NewService(new TelemetryLocalHub()); + using var cts = new CancellationTokenSource(); + + var ex = await Should.ThrowAsync(() => service.Subscribe( + new TelemetryStreamRequest { CorrelationId = correlationId }, + new CapturingStreamWriter(), Ctx(cts.Token))); + + ex.StatusCode.ShouldBe(StatusCode.InvalidArgument); + } + + [Fact] + public async Task Over_long_correlation_id_is_invalid_argument() + { + var service = NewService(new TelemetryLocalHub()); + using var cts = new CancellationTokenSource(); + + var ex = await Should.ThrowAsync(() => service.Subscribe( + new TelemetryStreamRequest { CorrelationId = new string('x', 257) }, + new CapturingStreamWriter(), Ctx(cts.Token))); + + ex.StatusCode.ShouldBe(StatusCode.InvalidArgument); + } + + [Fact] + public async Task Concurrency_cap_sheds_the_over_the_limit_dial() + { + // Serial assembly (see AssemblyInfo) => the static active-stream counter starts at 0 here. + var hub = new TelemetryLocalHub(); + var service = NewService(hub); + using var cts = new CancellationTokenSource(); + + // Open exactly the cap. Each stream blocks in the pump (hub emits nothing and is never completed), + // so all 100 stay open, holding the counter at 100. + var open = new List(); + for (var i = 0; i < 100; i++) + open.Add(service.Subscribe( + new TelemetryStreamRequest { CorrelationId = $"cap-{i}" }, + new CapturingStreamWriter(), Ctx(cts.Token))); + + // The 101st dial must be rejected. + var ex = await Should.ThrowAsync(() => service.Subscribe( + new TelemetryStreamRequest { CorrelationId = "cap-overflow" }, + new CapturingStreamWriter(), Ctx(cts.Token))); + ex.StatusCode.ShouldBe(StatusCode.ResourceExhausted); + + // Drain the 100 open streams; the counter returns to 0 for the next test. + cts.Cancel(); + await Task.WhenAll(open); + + // With the counter back at 0, a fresh single dial is accepted (proves no leak / no double-count). + using var cts2 = new CancellationTokenSource(); + var writer = new CapturingStreamWriter(); + var pump = service.Subscribe( + new TelemetryStreamRequest { CorrelationId = "cap-after" }, writer, Ctx(cts2.Token)); + cts2.Cancel(); + await Should.NotThrowAsync(() => pump); + } + + [Fact] + public void ToProto_health_with_unspecified_kind_datetime_does_not_throw_and_roundtrips() + { + // Producers use UtcNow, but a Kind=Unspecified value must never crash the stream. The mapper + // assumes an Unspecified value is already-UTC, so the instant round-trips on any machine. + var wall = new DateTime(2026, 7, 23, 13, 30, 0, DateTimeKind.Unspecified); + var expectedUtc = DateTime.SpecifyKind(wall, DateTimeKind.Utc); + var health = new DriverHealthChanged( + "cluster-a", "drv-1", "Healthy", LastSuccessfulReadUtc: wall, LastError: null, + ErrorCount5Min: 0, PublishedUtc: wall); + + TelemetryEvent evt = null!; + Should.NotThrow(() => evt = TelemetryProtoMapNode.ToProto(new TelemetryItem.Health(health), "c")); + + evt.DriverHealth.PublishedUtc.ToDateTime().ShouldBe(expectedUtc); + evt.DriverHealth.LastSuccessfulReadUtc.ToDateTime().ShouldBe(expectedUtc); + } + + [Fact] + public void ToProto_alarm_omits_absent_nullables() + { + var alarm = new AlarmTransitionEvent( + "id", "Area/Line/Equip", "Name", "Cleared", 100, "msg", "system", + DateTime.UtcNow, Comment: null, HistorizeToAveva: null, ReferencingEquipmentPaths: null); + + var msg = TelemetryProtoMapNode.ToProto(new TelemetryItem.Alarm(alarm), "c").AlarmTransition; + + msg.HasComment.ShouldBeFalse(); + msg.HasHistorizeToAveva.ShouldBeFalse(); + msg.ReferencingEquipmentPaths.ShouldBeEmpty(); + } + + /// Minimal exposing a caller-controlled cancellation token. + private sealed class FakeServerCallContext(CancellationToken token) : ServerCallContext + { + protected override string MethodCore => "/telemetry.v1.TelemetryStreamService/Subscribe"; + protected override string HostCore => "localhost"; + protected override string PeerCore => "ipv4:127.0.0.1:0"; + protected override DateTime DeadlineCore => DateTime.MaxValue; + protected override Metadata RequestHeadersCore => []; + protected override CancellationToken CancellationTokenCore => token; + protected override Metadata ResponseTrailersCore { get; } = []; + protected override Status StatusCore { get; set; } + protected override WriteOptions? WriteOptionsCore { get; set; } + protected override AuthContext AuthContextCore => new(null, new Dictionary>()); + + protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) => + throw new NotSupportedException(); + + protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask; + } + + /// + /// Captures every streamed . Implements the cancellable + /// WriteAsync overload explicitly — otherwise the interface's default method throws + /// for the pump's cancellable lifetime token. + /// + private sealed class CapturingStreamWriter : IServerStreamWriter + { + private readonly List _items = []; + private readonly Lock _gate = new(); + + public WriteOptions? WriteOptions { get; set; } + + public int Count + { + get + { + lock (_gate) + return _items.Count; + } + } + + public Task WriteAsync(TelemetryEvent message) + { + lock (_gate) + _items.Add(message); + return Task.CompletedTask; + } + + public Task WriteAsync(TelemetryEvent message, CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + return WriteAsync(message); + } + + public List Snapshot() + { + lock (_gate) + return [.. _items]; + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/ZB.MOM.WW.OtOpcUa.Host.Tests.csproj b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/ZB.MOM.WW.OtOpcUa.Host.Tests.csproj new file mode 100644 index 00000000..63747137 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/ZB.MOM.WW.OtOpcUa.Host.Tests.csproj @@ -0,0 +1,26 @@ + + + + false + true + ZB.MOM.WW.OtOpcUa.Host.Tests + true + + + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + +