diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs index fb2d734d..42c32655 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs @@ -67,6 +67,27 @@ public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection public async Task ConnectAsync(IDictionary connectionDetails, CancellationToken cancellationToken = default) { var cfg = MxGatewayEndpointConfigSerializer.FromFlatDict(connectionDetails); + + // Reconnect reuses this adapter: cancel the previous event loop and dispose + // the previous client so a flapping gateway cannot accumulate orphaned + // streams/sockets, and the old loop's fault path cannot signal Disconnected + // against the new session after the guard resets below. Keep this BEFORE the + // _disconnectFired reset so a fault raised by the dying loop during teardown + // is absorbed by the old guard cycle. Dispose is best-effort fire-and-forget. + if (_eventLoopCts is { } previousLoopCts) + { + previousLoopCts.Cancel(); + previousLoopCts.Dispose(); + _eventLoopCts = null; + } + if (_client is { } previousClient) + { + _ = previousClient.DisposeAsync().AsTask().ContinueWith( + t => _logger.LogWarning(t.Exception?.GetBaseException(), + "Disposing previous MxGateway client failed"), + TaskContinuationOptions.OnlyOnFaulted); + } + Interlocked.Exchange(ref _disconnectFired, 0); // reset guard on (re)connect, like OPC UA _client = _clientFactory.Create(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionReconnectTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionReconnectTests.cs new file mode 100644 index 00000000..cb032d61 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/MxGatewayDataConnectionReconnectTests.cs @@ -0,0 +1,56 @@ +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters; + +/// +/// DataConnectionLayer-S1 (MxGateway half): the adapter instance is reused across +/// reconnects. Before the fix, ConnectAsync overwrote _client and +/// _eventLoopCts without cancelling the previous event loop or disposing the +/// previous client — every gateway flap accumulated an orphaned stream/socket, and +/// the dying old loop's fault path could raise Disconnected against the new +/// session after the guard was reset. +/// +[Collection("DataConnectionManagerActor")] +public class MxGatewayDataConnectionReconnectTests +{ + [Fact] + public async Task Reconnect_DisposesPreviousClient_AndCancelsPreviousEventLoop() + { + var clients = new List(); + var loopTokens = new List(); + var factory = Substitute.For(); + factory.Create().Returns(_ => + { + var c = Substitute.For(); + c.RunEventLoopAsync(Arg.Any>(), Arg.Do(t => loopTokens.Add(t))) + .Returns(Task.Delay(Timeout.Infinite)); + clients.Add(c); + return c; + }); + var adapter = new MxGatewayDataConnection(factory, NullLogger.Instance); + var details = new Dictionary { ["Endpoint"] = "http://gw:5000", ["ApiKey"] = "k" }; + + // The event loop is pumped fire-and-forget (ConnectAsync → Task.Run → + // RunEventLoopAsync), so the token is recorded on a thread-pool continuation. + // Barrier on each recorded token before proceeding. + await adapter.ConnectAsync(details); + await WaitUntilAsync(() => loopTokens.Count == 1); + + await adapter.ConnectAsync(details); // reconnect + await WaitUntilAsync(() => loopTokens.Count == 2); + + Assert.Equal(2, clients.Count); + await clients[0].Received(1).DisposeAsync(); + Assert.True(loopTokens[0].IsCancellationRequested); // old loop cancelled + Assert.False(loopTokens[1].IsCancellationRequested); // new loop live + } + + private static async Task WaitUntilAsync(Func condition) + { + for (var i = 0; i < 100 && !condition(); i++) + await Task.Delay(50); + Assert.True(condition(), "condition not met within timeout"); + } +}