diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs index 0b9fa38..ac426a4 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs @@ -764,11 +764,24 @@ public sealed class GatewaySession // The distributor's single event source. Drains the worker event stream once (the // distributor guarantees a single consumer) and maps each frame to the public MxEvent, // preserving worker order. Mirrors the former ProduceEventsAsync mapping exactly. + // + // This deliberately duplicates the three lines of ReadEventsAsync rather than enumerating + // it: every worker event crosses this source, and routing it through a second pure + // pass-through iterator cost two extra MoveNextAsync state-machine hops per event for no + // semantic value. ReadEventsAsync stays for ISessionManager.ReadEventsAsync; keep the two + // bodies in step. Only one of them may run per attach — WorkerClient.ReadEventsAsync + // single-reader-claims the event channel and throws on a second consumer — and on the + // distributor path that one consumer is this method. private async IAsyncEnumerable MapWorkerEventsAsync( [EnumeratorCancellation] CancellationToken cancellationToken) { MxAccessGrpcMapper mapper = _eventStreaming.Mapper; - await foreach (WorkerEvent workerEvent in ReadEventsAsync(cancellationToken) + IWorkerClient workerClient = await GetReadyWorkerClientAsync(cancellationToken).ConfigureAwait(false); + TouchClientActivity(_eventStreaming.TimeProvider.GetUtcNow()); + + await foreach (WorkerEvent workerEvent in workerClient + .ReadEventsAsync(cancellationToken) + .WithCancellation(cancellationToken) .ConfigureAwait(false)) { yield return mapper.MapEvent(workerEvent); @@ -1513,6 +1526,13 @@ public sealed class GatewaySession /// Reads events from the worker as an asynchronous enumerable stream. /// /// Token to cancel the asynchronous operation. + /// + /// Backs ISessionManager.ReadEventsAsync. The distributor does not come + /// through here — MapWorkerEventsAsync inlines this body to save a per-event + /// iterator hop, so changes made here belong there too. The two are mutually exclusive + /// per attach: claims the worker event + /// channel for a single reader and throws on the second consumer. + /// /// An asynchronous stream of worker events. public async IAsyncEnumerable ReadEventsAsync( [EnumeratorCancellation] CancellationToken cancellationToken)