dohertj2/mxaccessgw
1
0
Fork 0
Code Issues Pull Requests 2 Actions 29 Packages Projects Releases Wiki Activity

test(gateway): TST-01 end-to-end reconnect/replay integration test

Browse Source 
Proves the default-on reconnect protocol end to end through the real gRPC
StreamEvents path via the fake worker harness (no live COM). Two facts:

- ReconnectInsideRetainedWindow_ReplaysTailNoGap: capacity 16 retains all
  events; reconnect from a mid-batch after_worker_sequence cursor replays
  exactly the events with WorkerSequence > cursor (retained tail + events
  emitted while detached), strictly ascending, distinct, no ReplayGap.
- ReconnectWithStaleCursor_EmitsReplayGapSentinelFirst: capacity 3 with 6
  events forces eviction; reconnect with AfterWorkerSequence=1 yields the
  ReplayGap sentinel first (Family=Unspecified, no body, RequestedAfterSequence=1,
  OldestAvailableSequence=oldest retained), then the retained tail ascending.

Single-subscriber mode: the first stream is fully detached (cancel + await the
stream task runs EventStreamService's finally, dropping the subscriber count to
0) before reconnect; the distributor + replay ring are created once per session
and survive detach, so events emitted while detached are retained.

macOS note: fake-worker E2E tests need TMPDIR=/tmp (default macOS TMPDIR pushes
the CoreFxPipe Unix-socket path past the 104-byte sun_path limit). Windows CI
is unaffected. Server reconnect/replay behavior matched the contract exactly.

Claude-Session: https://claude.ai/code/session_01DMXXvNuPekkkrTEyPNxEkW
This commit is contained in:
Joseph Doherty
2026-07-09 16:26:40 -04:00
parent 0c6e5b3ace
commit fed0685633
1 changed files with 642 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.MxGateway.Tests/Gateway/GatewayEndToEndReconnectReplayTests.cs
+642
View File
@@ -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;
/// <summary>
/// End-to-end reconnect/replay tests through the real gRPC <c>StreamEvents</c> 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 <see cref="StreamEventsRequest.AfterWorkerSequence"/>. Covers both
/// the no-gap resume (cursor inside the retained window) and the <c>ReplayGap</c> sentinel
/// (cursor predates the oldest retained event after capacity eviction).
/// </summary>
/// <remarks>
/// These tests run in single-subscriber mode (<c>AllowMultipleEventSubscribers=false</c>).
/// Reconnect works because the first stream is FULLY detached before the reconnect attaches:
/// awaiting the first stream task runs <c>EventStreamService</c>'s finally block, which
/// disposes the subscriber lease and drops the session's active-subscriber count back to
/// zero, so the reconnect's <c>AttachEventSubscriberWithReplay</c> 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.
/// </remarks>
public sealed class GatewayEndToEndReconnectReplayTests
{
private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10);
private const int ServerHandle = 3001;
private const int ItemHandle = 4002;
/// <summary>
/// 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 <c>ReplayGap</c>
/// sentinel.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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<MxEvent> 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<MxEvent> 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<MxEvent> 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<MxEvent> 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);
}
/// <summary>
/// Reconnecting with a cursor that predates the oldest retained event (after capacity
/// eviction) yields the <c>ReplayGap</c> sentinel FIRST — family unspecified, no body, no
/// per-item fields, correct requested/oldest sequences — followed by exactly the retained
/// tail, in ascending order.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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<MxEvent> 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<MxEvent> 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<MxEvent> 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<MxEvent> 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<MxEvent> 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<string> 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 ----
/// <summary>
/// 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).
/// </summary>
private sealed class ReconnectReplayGatewayServiceFixture : IAsyncDisposable
{
private readonly GatewayMetrics _metrics = new();
private readonly SessionRegistry _registry = new();
/// <summary>Initializes a new instance of the <see cref="ReconnectReplayGatewayServiceFixture"/> class.</summary>
/// <param name="launcher">Fake worker process launcher backing the session manager.</param>
/// <param name="replayBufferCapacity">Replay ring capacity for the session's event distributor.</param>
public ReconnectReplayGatewayServiceFixture(
IWorkerProcessLauncher launcher,
int replayBufferCapacity)
{
IOptions<GatewayOptions> options = Options.Create(CreateOptions(replayBufferCapacity));
SessionWorkerClientFactory workerClientFactory = new(
launcher,
options,
_metrics,
NullLoggerFactory.Instance);
SessionManager sessionManager = new(
_registry,
workerClientFactory,
options,
_metrics,
logger: NullLogger<SessionManager>.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<MxAccessGatewayService>.Instance,
new FakeGatewayAlarmService());
}
/// <summary>Gets the gateway service under test.</summary>
public MxAccessGatewayService Service { get; }
/// <summary>
/// Polls <see cref="GatewaySession.ActiveEventSubscriberCount"/> for
/// <paramref name="sessionId"/> until it reaches <paramref name="n"/>, bounded by
/// <paramref name="timeout"/>. 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.
/// </summary>
/// <param name="sessionId">Identifier of the session to poll.</param>
/// <param name="n">Target subscriber count to wait for.</param>
/// <param name="timeout">Maximum time to wait before failing the test.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
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);
}
}
/// <summary>Disposes every session in the registry and releases the fixture's metrics.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
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 ----
/// <summary>
/// Fake worker that emits events one at a time, gated by <see cref="AllowNextEvent"/>, 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
/// <see cref="StopEmitting"/> before closing the session so the loop exits cleanly and can
/// process the shutdown envelope.
/// </summary>
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;
/// <summary>Gets the task representing the fake worker's running background loop.</summary>
public Task WorkerTask { get; private set; } = Task.CompletedTask;
/// <summary>Releases the gate so the worker emits one event.</summary>
public void AllowNextEvent() => _emitGate.Release();
/// <summary>
/// Signals the worker to stop waiting for the emit gate and process the shutdown
/// envelope. Must be called before <c>CloseSession</c>.
/// </summary>
public void StopEmitting()
{
_stopEmitting = true;
_emitGate.Release(); // unblock a pending gate wait if any
}
/// <inheritdoc />
public Task<WorkerProcessHandle> 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;
}
}
}
}
}