diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs
new file mode 100644
index 0000000..424bd66
--- /dev/null
+++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs
@@ -0,0 +1,642 @@
+using Google.Protobuf.WellKnownTypes;
+using Grpc.Core;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.MxGateway.Contracts;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Server.Configuration;
+using ZB.MOM.WW.MxGateway.Server.Grpc;
+using ZB.MOM.WW.MxGateway.Server.Metrics;
+using ZB.MOM.WW.MxGateway.Server.Security.Authentication;
+using ZB.MOM.WW.MxGateway.Server.Security.Authorization;
+using ZB.MOM.WW.MxGateway.Server.Sessions;
+using ZB.MOM.WW.MxGateway.Server.Workers;
+using ZB.MOM.WW.MxGateway.Tests.Gateway.Workers.Fakes;
+using ZB.MOM.WW.MxGateway.Tests.TestSupport;
+
+namespace ZB.MOM.WW.MxGateway.Tests.Gateway;
+
+///
+/// End-to-end reconnect/replay tests through the real gRPC StreamEvents path via the
+/// fake worker harness (TST-01, server half). A single subscriber detaches (cancel + await
+/// the stream task so its lease is fully disposed), the session is retained by detach-grace
+/// while the distributor pump keeps appending worker events to the replay ring, and a second
+/// stream reconnects with . Covers both
+/// the no-gap resume (cursor inside the retained window) and the ReplayGap sentinel
+/// (cursor predates the oldest retained event after capacity eviction).
+///
+///
+/// These tests run in single-subscriber mode (AllowMultipleEventSubscribers=false).
+/// Reconnect works because the first stream is FULLY detached before the reconnect attaches:
+/// awaiting the first stream task runs EventStreamService's finally block, which
+/// disposes the subscriber lease and drops the session's active-subscriber count back to
+/// zero, so the reconnect's AttachEventSubscriberWithReplay sees an empty slot rather
+/// than an "already active" rejection. The distributor (and its replay ring) is created once
+/// per session and survives detach, so events emitted while no subscriber is attached are
+/// retained for the reconnecting stream.
+///
+public sealed class GatewayEndToEndReconnectReplayTests
+{
+ private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10);
+ private const int ServerHandle = 3001;
+ private const int ItemHandle = 4002;
+
+ ///
+ /// Reconnecting inside the retained window replays exactly the events newer than the
+ /// resume cursor — the retained tail of the first batch plus the events emitted while
+ /// detached — in strictly ascending order, with no duplicates and no ReplayGap
+ /// sentinel.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task StreamEvents_ReconnectInsideRetainedWindow_ReplaysTailNoGap()
+ {
+ const int firstBatch = 4;
+ const int secondBatch = 3;
+
+ GatedEventFakeWorkerProcessLauncher launcher = new();
+ // Capacity 16 retains every event emitted here (7 total), so nothing is evicted and a
+ // resume from a middle cursor never produces a gap.
+ await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: 16);
+
+ string sessionId = await OpenSessionAsync(fixture, "reconnect-no-gap");
+
+ // ---- first connection: receive the first batch, capture its real sequences ----
+ using CancellationTokenSource writer1Cts = new();
+ RecordingServerStreamWriter writer1 = new();
+ Task stream1Task = Task.Run(async () =>
+ await fixture.Service.StreamEvents(
+ new StreamEventsRequest { SessionId = sessionId },
+ writer1,
+ new TestServerCallContext(cancellationToken: writer1Cts.Token)));
+
+ await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout);
+ await WireUpAdviseAsync(fixture, sessionId);
+
+ for (int i = 0; i < firstBatch; i++)
+ {
+ launcher.AllowNextEvent();
+ }
+
+ IReadOnlyList batch1 = await writer1.WaitForMessageCountAsync(firstBatch, TestTimeout);
+ ulong[] batch1Sequences = batch1.Select(e => e.WorkerSequence).ToArray();
+
+ // Resume cursor: a middle event of the first batch. Everything strictly newer than this
+ // must be redelivered to the reconnecting stream.
+ ulong cursor = batch1Sequences[1];
+ int retainedTailFromBatch1 = batch1Sequences.Count(s => s > cursor);
+ Assert.Equal(2, retainedTailFromBatch1); // sanity: events at index 2 and 3
+
+ // ---- fully detach writer1 (cancel + await so its lease is disposed) ----
+ await DetachAsync(writer1Cts, stream1Task);
+
+ // ---- emit more events while detached; the ring must retain them for the reconnect ----
+ for (int i = 0; i < secondBatch; i++)
+ {
+ launcher.AllowNextEvent();
+ }
+
+ // ---- reconnect with the middle cursor ----
+ RecordingServerStreamWriter writer2 = new();
+ Task stream2Task = Task.Run(async () =>
+ await fixture.Service.StreamEvents(
+ new StreamEventsRequest { SessionId = sessionId, AfterWorkerSequence = cursor },
+ writer2,
+ new TestServerCallContext()));
+
+ await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout);
+
+ int expected = retainedTailFromBatch1 + secondBatch;
+ IReadOnlyList resumed = await writer2.WaitForMessageCountAsync(expected, TestTimeout);
+
+ // ---- tear down before asserting so a hang can't wedge the run ----
+ launcher.StopEmitting();
+ await CloseAndDrainAsync(fixture, sessionId, stream2Task, launcher);
+
+ // ---- assertions ----
+ Assert.Equal(expected, resumed.Count);
+
+ // No sentinel: every message is a real event, none carries a ReplayGap.
+ Assert.All(resumed, e => Assert.Null(e.ReplayGap));
+
+ // Every replayed/live event is strictly newer than the cursor.
+ Assert.All(resumed, e => Assert.True(
+ e.WorkerSequence > cursor,
+ $"Event sequence {e.WorkerSequence} is not newer than cursor {cursor}."));
+
+ // Strictly ascending, no duplicates.
+ for (int i = 1; i < resumed.Count; i++)
+ {
+ Assert.True(
+ resumed[i].WorkerSequence > resumed[i - 1].WorkerSequence,
+ $"Sequences must be strictly ascending: {resumed[i - 1].WorkerSequence} then {resumed[i].WorkerSequence}.");
+ }
+
+ Assert.Equal(resumed.Count, resumed.Select(e => e.WorkerSequence).Distinct().Count());
+
+ // The retained tail of the first batch (the two events newer than the cursor) is
+ // replayed first, before the events emitted while detached.
+ Assert.Equal(batch1Sequences[2], resumed[0].WorkerSequence);
+ Assert.Equal(batch1Sequences[3], resumed[1].WorkerSequence);
+ }
+
+ ///
+ /// Reconnecting with a cursor that predates the oldest retained event (after capacity
+ /// eviction) yields the ReplayGap sentinel FIRST — family unspecified, no body, no
+ /// per-item fields, correct requested/oldest sequences — followed by exactly the retained
+ /// tail, in ascending order.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task StreamEvents_ReconnectWithStaleCursor_EmitsReplayGapSentinelFirst()
+ {
+ const int capacity = 3;
+ const int totalEvents = 6;
+
+ GatedEventFakeWorkerProcessLauncher launcher = new();
+ // Small capacity forces eviction: with 6 events emitted and a 3-slot ring, the first 3
+ // are evicted and only the newest 3 remain replayable.
+ await using ReconnectReplayGatewayServiceFixture fixture = new(launcher, replayBufferCapacity: capacity);
+
+ string sessionId = await OpenSessionAsync(fixture, "reconnect-gap");
+
+ using CancellationTokenSource writer1Cts = new();
+ RecordingServerStreamWriter writer1 = new();
+ Task stream1Task = Task.Run(async () =>
+ await fixture.Service.StreamEvents(
+ new StreamEventsRequest { SessionId = sessionId },
+ writer1,
+ new TestServerCallContext(cancellationToken: writer1Cts.Token)));
+
+ await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout);
+ await WireUpAdviseAsync(fixture, sessionId);
+
+ // Emit more events than the ring can hold. Waiting for writer1 to receive all of them
+ // proves the pump has appended (and evicted) every event, so the ring is settled to its
+ // final newest-`capacity` contents before we reconnect.
+ for (int i = 0; i < totalEvents; i++)
+ {
+ launcher.AllowNextEvent();
+ }
+
+ IReadOnlyList all = await writer1.WaitForMessageCountAsync(totalEvents, TestTimeout);
+ ulong[] sequences = all.Select(e => e.WorkerSequence).ToArray();
+
+ // With capacity C and N total events, the newest C are retained; the oldest still
+ // available is the (N-C+1)th event — index N-C in emission (ascending) order.
+ ulong oldestAvailable = sequences[totalEvents - capacity];
+
+ await DetachAsync(writer1Cts, stream1Task);
+
+ // Resume after sequence 1: the real event sequences are envelope-numbered and start well
+ // above 1 (startup + command-reply envelopes consume the low numbers), so cursor 1
+ // predates every event and, with the oldest three already evicted, forces a gap.
+ const ulong staleCursor = 1;
+ RecordingServerStreamWriter writer2 = new();
+ Task stream2Task = Task.Run(async () =>
+ await fixture.Service.StreamEvents(
+ new StreamEventsRequest { SessionId = sessionId, AfterWorkerSequence = staleCursor },
+ writer2,
+ new TestServerCallContext()));
+
+ await fixture.WaitForSubscriberCountAsync(sessionId, n: 1, TestTimeout);
+
+ // sentinel + the retained tail (capacity events, all newer than the stale cursor).
+ int expected = 1 + capacity;
+ IReadOnlyList resumed = await writer2.WaitForMessageCountAsync(expected, TestTimeout);
+
+ launcher.StopEmitting();
+ await CloseAndDrainAsync(fixture, sessionId, stream2Task, launcher);
+
+ // ---- the first message is the ReplayGap sentinel ----
+ Assert.Equal(expected, resumed.Count);
+
+ MxEvent sentinel = resumed[0];
+ Assert.NotNull(sentinel.ReplayGap);
+ Assert.Equal(MxEventFamily.Unspecified, sentinel.Family);
+ Assert.Equal(MxEvent.BodyOneofCase.None, sentinel.BodyCase);
+ Assert.Equal(sessionId, sentinel.SessionId);
+ Assert.Equal(staleCursor, sentinel.ReplayGap.RequestedAfterSequence);
+ Assert.Equal(oldestAvailable, sentinel.ReplayGap.OldestAvailableSequence);
+
+ // ---- the sentinel is followed by the retained tail, ascending, no further sentinels ----
+ IReadOnlyList tail = resumed.Skip(1).ToArray();
+ Assert.Equal(capacity, tail.Count);
+ Assert.All(tail, e => Assert.Null(e.ReplayGap));
+ Assert.All(tail, e => Assert.True(
+ e.WorkerSequence >= oldestAvailable,
+ $"Retained event {e.WorkerSequence} is older than the oldest available {oldestAvailable}."));
+
+ ulong[] expectedTail = sequences.Skip(totalEvents - capacity).ToArray();
+ Assert.Equal(expectedTail, tail.Select(e => e.WorkerSequence).ToArray());
+ }
+
+ // ---- shared flow helpers ----
+
+ private static async Task OpenSessionAsync(
+ ReconnectReplayGatewayServiceFixture fixture,
+ string name)
+ {
+ OpenSessionReply openReply = await fixture.Service.OpenSession(
+ new OpenSessionRequest
+ {
+ ClientSessionName = name,
+ ClientCorrelationId = $"open-{name}",
+ CommandTimeout = Duration.FromTimeSpan(TestTimeout),
+ },
+ new TestServerCallContext()).ConfigureAwait(false);
+
+ Assert.Equal(ProtocolStatusCode.Ok, openReply.ProtocolStatus.Code);
+ return openReply.SessionId;
+ }
+
+ private static async Task WireUpAdviseAsync(
+ ReconnectReplayGatewayServiceFixture fixture,
+ string sessionId)
+ {
+ MxCommandReply registerReply = await fixture.Service.Invoke(
+ CreateRegisterRequest(sessionId),
+ new TestServerCallContext()).ConfigureAwait(false);
+ Assert.Equal(ProtocolStatusCode.Ok, registerReply.ProtocolStatus.Code);
+
+ MxCommandReply addItemReply = await fixture.Service.Invoke(
+ CreateAddItemRequest(sessionId, registerReply.Register.ServerHandle),
+ new TestServerCallContext()).ConfigureAwait(false);
+ Assert.Equal(ProtocolStatusCode.Ok, addItemReply.ProtocolStatus.Code);
+
+ MxCommandReply adviseReply = await fixture.Service.Invoke(
+ CreateAdviseRequest(sessionId, registerReply.Register.ServerHandle, addItemReply.AddItem.ItemHandle),
+ new TestServerCallContext()).ConfigureAwait(false);
+ Assert.Equal(ProtocolStatusCode.Ok, adviseReply.ProtocolStatus.Code);
+ }
+
+ // Cancels the stream's token and awaits the stream task. Awaiting is load-bearing: it lets
+ // EventStreamService's finally block dispose the subscriber lease (dropping the session's
+ // active-subscriber count to zero) BEFORE the reconnect attaches, so the reconnect never
+ // races a still-registered subscriber.
+ private static async Task DetachAsync(CancellationTokenSource cts, Task streamTask)
+ {
+ await cts.CancelAsync().ConfigureAwait(false);
+ try
+ {
+ await streamTask.WaitAsync(TestTimeout).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException)
+ {
+ // Expected: the iterator surfaces the cancellation.
+ }
+ catch (RpcException rpc) when (rpc.StatusCode == StatusCode.Cancelled)
+ {
+ // Also acceptable depending on gRPC exception wrapping.
+ }
+ }
+
+ private static async Task CloseAndDrainAsync(
+ ReconnectReplayGatewayServiceFixture fixture,
+ string sessionId,
+ Task streamTask,
+ GatedEventFakeWorkerProcessLauncher launcher)
+ {
+ await fixture.Service.CloseSession(
+ new CloseSessionRequest { SessionId = sessionId, ClientCorrelationId = "close-reconnect" },
+ new TestServerCallContext()).ConfigureAwait(false);
+
+ try
+ {
+ await streamTask.WaitAsync(TestTimeout).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException)
+ {
+ }
+
+ await launcher.WorkerTask.WaitAsync(TestTimeout).ConfigureAwait(false);
+ }
+
+ // ---- request builders ----
+
+ private static MxCommandRequest CreateRegisterRequest(string sessionId) =>
+ new()
+ {
+ SessionId = sessionId,
+ ClientCorrelationId = "register-rr",
+ Command = new MxCommand
+ {
+ Kind = MxCommandKind.Register,
+ Register = new RegisterCommand { ClientName = "reconnect-replay-e2e-client" },
+ },
+ };
+
+ private static MxCommandRequest CreateAddItemRequest(string sessionId, int serverHandle) =>
+ new()
+ {
+ SessionId = sessionId,
+ ClientCorrelationId = "add-item-rr",
+ Command = new MxCommand
+ {
+ Kind = MxCommandKind.AddItem,
+ AddItem = new AddItemCommand
+ {
+ ServerHandle = serverHandle,
+ ItemDefinition = "Galaxy.Tag.Value",
+ },
+ },
+ };
+
+ private static MxCommandRequest CreateAdviseRequest(
+ string sessionId,
+ int serverHandle,
+ int itemHandle) =>
+ new()
+ {
+ SessionId = sessionId,
+ ClientCorrelationId = "advise-rr",
+ Command = new MxCommand
+ {
+ Kind = MxCommandKind.Advise,
+ Advise = new AdviseCommand { ServerHandle = serverHandle, ItemHandle = itemHandle },
+ },
+ };
+
+ private static void ConfigureCommandReply(MxCommandReply reply, MxCommandKind kind)
+ {
+ switch (kind)
+ {
+ case MxCommandKind.Register:
+ reply.Register = new RegisterReply { ServerHandle = ServerHandle };
+ break;
+ case MxCommandKind.AddItem:
+ reply.AddItem = new AddItemReply { ItemHandle = ItemHandle };
+ break;
+ }
+ }
+
+ // ---- fixture ----
+
+ ///
+ /// Gateway service fixture in single-subscriber mode with a configurable replay ring
+ /// capacity, so each test can retain the whole event history (no gap) or force capacity
+ /// eviction (gap).
+ ///
+ private sealed class ReconnectReplayGatewayServiceFixture : IAsyncDisposable
+ {
+ private readonly GatewayMetrics _metrics = new();
+ private readonly SessionRegistry _registry = new();
+
+ /// Initializes a new instance of the class.
+ /// Fake worker process launcher backing the session manager.
+ /// Replay ring capacity for the session's event distributor.
+ public ReconnectReplayGatewayServiceFixture(
+ IWorkerProcessLauncher launcher,
+ int replayBufferCapacity)
+ {
+ IOptions options = Options.Create(CreateOptions(replayBufferCapacity));
+ SessionWorkerClientFactory workerClientFactory = new(
+ launcher,
+ options,
+ _metrics,
+ NullLoggerFactory.Instance);
+ SessionManager sessionManager = new(
+ _registry,
+ workerClientFactory,
+ options,
+ _metrics,
+ logger: NullLogger.Instance,
+ dashboardEventBroadcaster: NullDashboardEventBroadcaster.Instance);
+ MxAccessGrpcMapper mapper = new();
+ EventStreamService eventStreamService = new(
+ sessionManager,
+ options,
+ _metrics);
+
+ Service = new MxAccessGatewayService(
+ sessionManager,
+ new GatewayRequestIdentityAccessor(),
+ new AllowAllConstraintEnforcer(),
+ new MxAccessGrpcRequestValidator(),
+ mapper,
+ eventStreamService,
+ _metrics,
+ NullLogger.Instance,
+ new FakeGatewayAlarmService());
+ }
+
+ /// Gets the gateway service under test.
+ public MxAccessGatewayService Service { get; }
+
+ ///
+ /// Polls for
+ /// until it reaches , bounded by
+ /// . Fails the test on timeout. This is the deterministic
+ /// gate that proves the production code has (re)registered a subscriber before the
+ /// test drives events or reconnects.
+ ///
+ /// Identifier of the session to poll.
+ /// Target subscriber count to wait for.
+ /// Maximum time to wait before failing the test.
+ /// A task that represents the asynchronous operation.
+ public async Task WaitForSubscriberCountAsync(string sessionId, int n, TimeSpan timeout)
+ {
+ using CancellationTokenSource deadlineCts = new(timeout);
+
+ while (true)
+ {
+ if (_registry.TryGet(sessionId, out GatewaySession? session)
+ && session.ActiveEventSubscriberCount >= n)
+ {
+ return;
+ }
+
+ if (deadlineCts.IsCancellationRequested)
+ {
+ int actual = _registry.TryGet(sessionId, out GatewaySession? s)
+ ? s.ActiveEventSubscriberCount
+ : -1;
+ Assert.Fail(
+ $"Timed out waiting for {n} event subscriber(s) on session {sessionId}. "
+ + $"Actual count after {timeout.TotalSeconds:0.#}s: {actual}.");
+ }
+
+ await Task.Delay(millisecondsDelay: 5, deadlineCts.Token).ConfigureAwait(false);
+ }
+ }
+
+ /// Disposes every session in the registry and releases the fixture's metrics.
+ /// A task that represents the asynchronous operation.
+ public async ValueTask DisposeAsync()
+ {
+ foreach (GatewaySession session in _registry.Snapshot())
+ {
+ await session.DisposeAsync().ConfigureAwait(false);
+ }
+
+ _metrics.Dispose();
+ }
+
+ private static GatewayOptions CreateOptions(int replayBufferCapacity) =>
+ new()
+ {
+ Worker = new WorkerOptions
+ {
+ StartupTimeoutSeconds = 5,
+ ShutdownTimeoutSeconds = 5,
+ HeartbeatIntervalSeconds = 30,
+ HeartbeatGraceSeconds = 30,
+ MaxMessageBytes = WorkerFrameProtocolOptions.DefaultMaxMessageBytes,
+ },
+ Sessions = new SessionOptions
+ {
+ DefaultCommandTimeoutSeconds = 5,
+ MaxSessions = 4,
+
+ // Single-subscriber mode: a fully-detached subscriber (stream task awaited)
+ // frees the sole slot, so the reconnect attaches cleanly without needing
+ // multi-subscriber fan-out. Detach-grace keeps the session Ready across the
+ // detach and the fixture runs no lease-reaper, so the session survives to be
+ // reconnected.
+ AllowMultipleEventSubscribers = false,
+ MaxEventSubscribersPerSession = 8,
+ },
+ Events = new EventOptions
+ {
+ QueueCapacity = 32,
+ ReplayBufferCapacity = replayBufferCapacity,
+
+ // Keep age-eviction effectively off for the duration of a fast test so
+ // capacity is the only eviction axis under test.
+ ReplayRetentionSeconds = 300,
+ },
+ };
+ }
+
+ // ---- fake worker launcher ----
+
+ ///
+ /// Fake worker that emits events one at a time, gated by , so
+ /// the test drives event timing deterministically. Modeled on the multi-subscriber E2E
+ /// tests' gated launcher. The worker loop is independent of subscribers, so events emitted
+ /// while no gRPC stream is attached still flow through the distributor pump into the
+ /// session's replay ring — exactly the condition the reconnect/replay tests exercise. Call
+ /// before closing the session so the loop exits cleanly and can
+ /// process the shutdown envelope.
+ ///
+ private sealed class GatedEventFakeWorkerProcessLauncher : IWorkerProcessLauncher
+ {
+ public const int ProcessId = 7730;
+
+ private readonly FakeWorkerProcess _process = new(ProcessId);
+
+ // Capacity 64 so AllowNextEvent can be called ahead of time without blocking.
+ private readonly SemaphoreSlim _emitGate = new(0, 64);
+ private volatile bool _stopEmitting;
+
+ /// Gets the task representing the fake worker's running background loop.
+ public Task WorkerTask { get; private set; } = Task.CompletedTask;
+
+ /// Releases the gate so the worker emits one event.
+ public void AllowNextEvent() => _emitGate.Release();
+
+ ///
+ /// Signals the worker to stop waiting for the emit gate and process the shutdown
+ /// envelope. Must be called before CloseSession.
+ ///
+ public void StopEmitting()
+ {
+ _stopEmitting = true;
+ _emitGate.Release(); // unblock a pending gate wait if any
+ }
+
+ ///
+ public Task LaunchAsync(
+ WorkerProcessLaunchRequest request,
+ CancellationToken cancellationToken = default)
+ {
+ WorkerTask = RunWorkerAsync(request, cancellationToken);
+
+ return Task.FromResult(new WorkerProcessHandle(
+ _process,
+ new WorkerProcessCommandLine("reconnect-replay-fake-worker.exe", []),
+ DateTimeOffset.UtcNow));
+ }
+
+ private async Task RunWorkerAsync(
+ WorkerProcessLaunchRequest request,
+ CancellationToken cancellationToken)
+ {
+ await using FakeWorkerHarness harness = await FakeWorkerHarness.ConnectToGatewayPipeAsync(
+ request.SessionId,
+ request.Nonce,
+ request.PipeName,
+ request.ProtocolVersion,
+ cancellationToken: cancellationToken).ConfigureAwait(false);
+ await harness.CompleteStartupAsync(ProcessId, cancellationToken: cancellationToken).ConfigureAwait(false);
+
+ int advisedServerHandle = 0;
+ int advisedItemHandle = 0;
+ int emittedCount = 0;
+
+ while (!cancellationToken.IsCancellationRequested)
+ {
+ // While subscribed and not stopped, emit gated events using a non-blocking peek
+ // at the gate so incoming envelopes (including shutdown) are never starved.
+ while (advisedServerHandle != 0
+ && !_stopEmitting
+ && await _emitGate.WaitAsync(millisecondsTimeout: 0).ConfigureAwait(false))
+ {
+ int index = ++emittedCount;
+ await harness.EmitEventAsync(
+ MxEventFamily.OnDataChange,
+ cancellationToken,
+ mxEvent =>
+ {
+ mxEvent.ServerHandle = advisedServerHandle;
+ mxEvent.ItemHandle = advisedItemHandle;
+ mxEvent.Quality = 192;
+ mxEvent.Value = new MxValue
+ {
+ DataType = MxDataType.String,
+ StringValue = $"reconnect-value-{index}",
+ };
+ mxEvent.OnDataChange = new OnDataChangeEvent();
+ }).ConfigureAwait(false);
+ }
+
+ WorkerEnvelope? envelope;
+ try
+ {
+ using CancellationTokenSource readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+ readCts.CancelAfter(TimeSpan.FromMilliseconds(50));
+ envelope = await harness.ReadGatewayEnvelopeAsync(readCts.Token).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
+ {
+ // Timed out waiting for an envelope — loop back to check the gate / emit.
+ continue;
+ }
+
+ if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerShutdown)
+ {
+ await harness.SendShutdownAckAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+ _process.MarkExited(0);
+ return;
+ }
+
+ if (envelope.BodyCase != WorkerEnvelope.BodyOneofCase.WorkerCommand)
+ {
+ throw new InvalidOperationException($"Unexpected envelope {envelope.BodyCase}.");
+ }
+
+ MxCommand command = envelope.WorkerCommand.Command;
+ await harness.ReplyToCommandAsync(
+ envelope,
+ configureReply: reply => ConfigureCommandReply(reply, command.Kind),
+ cancellationToken: cancellationToken).ConfigureAwait(false);
+
+ if (command.Kind == MxCommandKind.Advise)
+ {
+ advisedServerHandle = command.Advise.ServerHandle;
+ advisedItemHandle = command.Advise.ItemHandle;
+ }
+ }
+ }
+ }
+}