diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs new file mode 100644 index 00000000..7b010e4f --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs @@ -0,0 +1,190 @@ +using System.Net.Sockets; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus; + +/// +/// Concrete Modbus RTU-over-TCP transport. Composes for the +/// connect / reconnect / keep-alive / idle machinery and for the +/// wire layout — the same lifecycle uses, but with CRC framing +/// instead of MBAP. +/// +/// +/// +/// Delta from : an RTU ADU is [unitId][PDU][CRC] with +/// no MBAP header and no transaction id. Because there is no TxId to correlate +/// interleaved responses, at most one transaction may be on the bus at a time — the +/// single-flight is therefore mandatory, not merely a +/// diagnostics convenience. +/// +/// +/// Survives mid-transaction socket drops the same way the TCP transport does: a socket-level +/// failure ( / / +/// , and the +/// that derives from ) triggers a single reconnect-then-retry; a +/// second failure bubbles up so the driver's health surface reflects the real state. +/// +/// +/// Every transaction runs under a per-op deadline (a linked +/// , design §7 / R2-01) so a +/// frozen gateway can never wedge a poll: the deadline firing is normalised to a +/// and tears the socket down so the next attempt +/// reconnects. A caller cancellation is distinguished from that timeout and propagates +/// as a plain with no teardown. +/// +/// +/// The constructor is connection-free; all socket I/O happens in / +/// . The seam injects a pre-connected +/// (an in-memory duplex fake) so the send/receive orchestration, +/// single-flight, and deadline can be exercised with no socket — in that path +/// is and the idle / reconnect machinery is +/// skipped (a raw injected stream cannot reconnect). +/// +/// +public sealed class ModbusRtuOverTcpTransport : IModbusTransport +{ + private readonly ModbusSocketLifecycle? _life; + private readonly Stream? _testStream; + private readonly TimeSpan _timeout; + private readonly SemaphoreSlim _gate = new(1, 1); + private bool _disposed; + + /// Initializes a new instance of the class. + /// The host address or hostname of the Modbus gateway. + /// The TCP port of the Modbus gateway. + /// The timeout for socket operations and the per-op response deadline. + /// Whether to automatically reconnect on socket failures. + /// Optional keep-alive configuration for the socket. + /// Optional duration after which an idle socket is disconnected. + /// Optional reconnect backoff configuration. + public ModbusRtuOverTcpTransport( + string host, int port, TimeSpan timeout, bool autoReconnect = true, + ModbusKeepAliveOptions? keepAlive = null, + TimeSpan? idleDisconnect = null, + ModbusReconnectOptions? reconnect = null) + { + _timeout = timeout; + _life = new ModbusSocketLifecycle(host, port, timeout, autoReconnect, keepAlive, idleDisconnect, reconnect); + } + + private ModbusRtuOverTcpTransport(Stream testStream, TimeSpan timeout) + { + _timeout = timeout; + _testStream = testStream; + } + + /// + /// Test seam: builds a transport over a pre-connected, in-memory duplex , + /// bypassing and the idle / reconnect machinery so the send/receive + /// orchestration, single-flight, and per-op deadline can be exercised with no real socket. + /// + /// A pre-connected duplex stream standing in for the socket. + /// The per-op response deadline. + /// A transport driving directly. + internal static ModbusRtuOverTcpTransport ForTest(Stream stream, TimeSpan timeout) => new(stream, timeout); + + /// + public Task ConnectAsync(CancellationToken ct) => + // The ForTest seam is handed a pre-connected stream — nothing to dial. + _life is null ? Task.CompletedTask : _life.ConnectAsync(ct); + + /// + public async Task SendAsync(byte unitId, byte[] pdu, CancellationToken ct) + { + if (_disposed) throw new ObjectDisposedException(nameof(ModbusRtuOverTcpTransport)); + if (CurrentStream is null) throw new InvalidOperationException("Transport not connected"); + + // Single-flight: RTU carries no transaction id, so at most one transaction may be on the bus + // at a time — serialise every send/receive under the gate. + await _gate.WaitAsync(ct).ConfigureAwait(false); + try + { + // Proactive idle-disconnect (real socket only): if the socket has been quiet longer than + // the configured threshold, tear it down + reconnect before this PDU lands. Defends + // against silent NAT / firewall reaping. + if (_life is not null && _life.ShouldReconnectForIdle()) + { + await _life.TearDownAsync().ConfigureAwait(false); + await _life.ConnectWithBackoffAsync(ct).ConfigureAwait(false); + } + + try + { + var result = await SendOnceAsync(unitId, pdu, ct).ConfigureAwait(false); + _life?.MarkSuccess(); + return result; + } + catch (Exception ex) when (_life is not null && _life.AutoReconnect && ModbusSocketLifecycle.IsSocketLevelFailure(ex)) + { + // Mid-transaction drop: tear down the dead socket, reconnect (with backoff if + // configured), resend. Single retry — a second failure propagates so health/status + // reflect reality. Never reached in the ForTest seam (a raw injected stream has no + // lifecycle to reconnect). + await _life.TearDownAsync().ConfigureAwait(false); + await _life.ConnectWithBackoffAsync(ct).ConfigureAwait(false); + var result = await SendOnceAsync(unitId, pdu, ct).ConfigureAwait(false); + _life.MarkSuccess(); + return result; + } + } + finally + { + _gate.Release(); + } + } + + /// The live stream: the lifecycle's , or the injected test stream. + private Stream? CurrentStream => _life?.Stream ?? _testStream; + + /// + /// Executes exactly one RTU transaction on the current stream: build the ADU, write + flush, + /// read back one deframed response PDU. See the class remarks for the desync / timeout / + /// caller-cancel handling. + /// + private async Task SendOnceAsync(byte unitId, byte[] pdu, CancellationToken ct) + { + var stream = CurrentStream; + if (stream is null) throw new InvalidOperationException("Transport not connected"); + + // RTU ADU: [unitId] + PDU + CRC — no MBAP header, no TxId. + var adu = ModbusRtuFraming.BuildAdu(unitId, pdu); + + using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); + cts.CancelAfter(_timeout); + try + { + await stream.WriteAsync(adu.AsMemory(), cts.Token).ConfigureAwait(false); + await stream.FlushAsync(cts.Token).ConfigureAwait(false); + + // Framing owns the length-less RTU deframe + CRC + unit-address validation. A CRC-valid + // exception PDU throws ModbusException (socket still coherent — propagates, no teardown); + // truncation / CRC / unit-mismatch throw ModbusTransportDesyncException. + return await ModbusRtuFraming.ReadResponsePduAsync(stream, unitId, cts.Token).ConfigureAwait(false); + } + catch (ModbusTransportDesyncException) + { + // Framing violation: the stream is desynchronized — never reuse it. + if (_life is not null) await _life.TearDownAsync().ConfigureAwait(false); + throw; + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + // Per-op timeout fired (NOT the caller). The response is unknown / may still land later + // and corrupt the next read — tear the socket down and normalise as a desync so the + // reconnect retry / status mapping engages. + if (_life is not null) await _life.TearDownAsync().ConfigureAwait(false); + throw new ModbusTransportDesyncException( + ModbusTransportDesyncException.DesyncReason.Timeout, + $"Modbus per-op response timeout after {_timeout.TotalMilliseconds:0}ms"); + } + } + + /// Asynchronously disposes the transport and underlying socket resources. + /// A task that represents the asynchronous operation. + public async ValueTask DisposeAsync() + { + if (_disposed) return; + _disposed = true; + if (_life is not null) await _life.DisposeAsync().ConfigureAwait(false); + _gate.Dispose(); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusRtuOverTcpTransportTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusRtuOverTcpTransportTests.cs new file mode 100644 index 00000000..2ba1a622 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusRtuOverTcpTransportTests.cs @@ -0,0 +1,110 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests; + +/// +/// Exercises 's send/receive orchestration, single-flight, +/// and per-op deadline against an in-memory duplex fake injected through the ForTest seam — +/// no real socket. The fake records the request ADU the transport writes and replays a canned RTU +/// response on read, or blocks forever (honouring cancellation) when stall is set. +/// +[Trait("Category", "Unit")] +public sealed class ModbusRtuOverTcpTransportTests +{ + [Fact] + public async Task SendAsync_writes_rtu_adu_and_returns_deframed_pdu() + { + // Response the fake gateway will emit: FC03, 1 reg = 0x000A. + var response = ModbusCrc.Append(new byte[] { 0x01, 0x03, 0x02, 0x00, 0x0A }); + var fake = new CapturingDuplexStream(response); + + await using var transport = ModbusRtuOverTcpTransport.ForTest(fake, TimeSpan.FromSeconds(2)); + var pdu = await transport.SendAsync(0x01, new byte[] { 0x03, 0x00, 0x00, 0x00, 0x01 }, + TestContext.Current.CancellationToken); + + pdu.ShouldBe(new byte[] { 0x03, 0x02, 0x00, 0x0A }); + // The request went out as an RTU ADU: unit + pdu + CRC, NO MBAP header. + fake.Written.ShouldBe(new byte[] { 0x01, 0x03, 0x00, 0x00, 0x00, 0x01, 0x84, 0x0A }); + } + + [Fact] + public async Task SendAsync_exception_pdu_surfaces_ModbusException() + { + var response = ModbusCrc.Append(new byte[] { 0x01, 0x83, 0x02 }); + var fake = new CapturingDuplexStream(response); + await using var transport = ModbusRtuOverTcpTransport.ForTest(fake, TimeSpan.FromSeconds(2)); + await Should.ThrowAsync( + transport.SendAsync(0x01, new byte[] { 0x03, 0x00, 0x00, 0x00, 0x01 }, + TestContext.Current.CancellationToken)); + } + + [Fact] + public async Task SendAsync_stalled_gateway_hits_per_op_deadline() + { + var fake = new CapturingDuplexStream(respondBytes: Array.Empty(), stall: true); + await using var transport = ModbusRtuOverTcpTransport.ForTest(fake, TimeSpan.FromMilliseconds(200)); + await Should.ThrowAsync( + transport.SendAsync(0x01, new byte[] { 0x03, 0x00, 0x00, 0x00, 0x01 }, + TestContext.Current.CancellationToken)); + } + + /// + /// In-memory duplex test double: records everything written and replays + /// respondBytes on read. When stall is set the read blocks until its cancellation + /// token fires (simulating a frozen gateway) so the transport's per-op deadline is exercised. + /// + private sealed class CapturingDuplexStream : Stream + { + private readonly byte[] _response; + private readonly bool _stall; + private readonly MemoryStream _written = new(); + private int _readPos; + + public CapturingDuplexStream(byte[] respondBytes, bool stall = false) + { + _response = respondBytes; + _stall = stall; + } + + /// Gets the bytes the transport wrote to this stream (the request ADU). + public byte[] Written => _written.ToArray(); + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => true; + public override long Length => throw new NotSupportedException(); + public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + + public override async ValueTask ReadAsync(Memory buffer, CancellationToken ct = default) + { + if (_stall) + { + // Frozen gateway: never answer. Block until the per-op deadline (or caller) cancels. + var tcs = new TaskCompletionSource(); + await using (ct.Register(() => tcs.TrySetCanceled(ct))) + await tcs.Task.ConfigureAwait(false); + return 0; // unreachable — the await above always throws when cancelled + } + + var remaining = _response.Length - _readPos; + if (remaining <= 0) return 0; + var n = Math.Min(remaining, buffer.Length); + _response.AsSpan(_readPos, n).CopyTo(buffer.Span); + _readPos += n; + return n; + } + + public override async ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken ct = default) + { + await _written.WriteAsync(buffer, ct).ConfigureAwait(false); + } + + public override Task FlushAsync(CancellationToken ct) => Task.CompletedTask; + public override void Flush() { } + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + } +}