dohertj2/lmxopcua
1
0
Fork 0
Code Issues 12 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(modbus-rtu): add ModbusRtuOverTcpTransport (RTU framing over socket lifecycle)

Browse Source 
Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 13:56:19 -04:00
parent 2f38b5b285
commit 47e9cd56ef
2 changed files with 300 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs
+190
View File
@@ -0,0 +1,190 @@
using System.Net.Sockets;
namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus;
/// <summary>
/// Concrete Modbus RTU-over-TCP transport. Composes <see cref="ModbusSocketLifecycle"/> for the
/// connect / reconnect / keep-alive / idle machinery and <see cref="ModbusRtuFraming"/> for the
/// wire layout — the same lifecycle <see cref="ModbusTcpTransport"/> uses, but with CRC framing
/// instead of MBAP.
/// </summary>
/// <remarks>
/// <para>
/// Delta from <see cref="ModbusTcpTransport"/>: an RTU ADU is <c>[unitId][PDU][CRC]</c> with
/// <b>no MBAP header and no transaction id</b>. Because there is no TxId to correlate
/// interleaved responses, at most one transaction may be on the bus at a time — the
/// <see cref="_gate"/> single-flight is therefore <b>mandatory</b>, not merely a
/// diagnostics convenience.
/// </para>
/// <para>
/// Survives mid-transaction socket drops the same way the TCP transport does: a socket-level
/// failure (<see cref="IOException"/> / <see cref="SocketException"/> /
/// <see cref="EndOfStreamException"/>, and the <see cref="ModbusTransportDesyncException"/>
/// that derives from <see cref="IOException"/>) triggers a single reconnect-then-retry; a
/// second failure bubbles up so the driver's health surface reflects the real state.
/// </para>
/// <para>
/// Every transaction runs under a per-op deadline (a linked
/// <see cref="CancellationTokenSource.CancelAfter(TimeSpan)"/>, design §7 / R2-01) so a
/// frozen gateway can never wedge a poll: the deadline firing is normalised to a
/// <see cref="ModbusTransportDesyncException"/> and tears the socket down so the next attempt
/// reconnects. A <b>caller</b> cancellation is distinguished from that timeout and propagates
/// as a plain <see cref="OperationCanceledException"/> with no teardown.
/// </para>
/// <para>
/// The constructor is connection-free; all socket I/O happens in <see cref="ConnectAsync"/> /
/// <see cref="SendAsync"/>. The <see cref="ForTest"/> seam injects a pre-connected
/// <see cref="Stream"/> (an in-memory duplex fake) so the send/receive orchestration,
/// single-flight, and deadline can be exercised with no socket — in that path
/// <see cref="_life"/> is <see langword="null"/> and the idle / reconnect machinery is
/// skipped (a raw injected stream cannot reconnect).
/// </para>
/// </remarks>
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;
/// <summary>Initializes a new instance of the <see cref="ModbusRtuOverTcpTransport"/> class.</summary>
/// <param name="host">The host address or hostname of the Modbus gateway.</param>
/// <param name="port">The TCP port of the Modbus gateway.</param>
/// <param name="timeout">The timeout for socket operations and the per-op response deadline.</param>
/// <param name="autoReconnect">Whether to automatically reconnect on socket failures.</param>
/// <param name="keepAlive">Optional keep-alive configuration for the socket.</param>
/// <param name="idleDisconnect">Optional duration after which an idle socket is disconnected.</param>
/// <param name="reconnect">Optional reconnect backoff configuration.</param>
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;
}
/// <summary>
/// Test seam: builds a transport over a pre-connected, in-memory duplex <paramref name="stream"/>,
/// bypassing <see cref="ConnectAsync"/> and the idle / reconnect machinery so the send/receive
/// orchestration, single-flight, and per-op deadline can be exercised with no real socket.
/// </summary>
/// <param name="stream">A pre-connected duplex stream standing in for the socket.</param>
/// <param name="timeout">The per-op response deadline.</param>
/// <returns>A transport driving <paramref name="stream"/> directly.</returns>
internal static ModbusRtuOverTcpTransport ForTest(Stream stream, TimeSpan timeout) => new(stream, timeout);
/// <inheritdoc />
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);
/// <inheritdoc />
public async Task<byte[]> 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();
}
}
/// <summary>The live stream: the lifecycle's <see cref="NetworkStream"/>, or the injected test stream.</summary>
private Stream? CurrentStream => _life?.Stream ?? _testStream;
/// <summary>
/// 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.
/// </summary>
private async Task<byte[]> 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");
}
}
/// <summary>Asynchronously disposes the transport and underlying socket resources.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
public async ValueTask DisposeAsync()
{
if (_disposed) return;
_disposed = true;
if (_life is not null) await _life.DisposeAsync().ConfigureAwait(false);
_gate.Dispose();
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusRtuOverTcpTransportTests.cs
+110
View File
@@ -0,0 +1,110 @@
using Shouldly;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests;
/// <summary>
/// Exercises <see cref="ModbusRtuOverTcpTransport"/>'s send/receive orchestration, single-flight,
/// and per-op deadline against an in-memory duplex fake injected through the <c>ForTest</c> 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 <c>stall</c> is set.
/// </summary>
[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<ModbusException>(
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<byte>(), stall: true);
await using var transport = ModbusRtuOverTcpTransport.ForTest(fake, TimeSpan.FromMilliseconds(200));
await Should.ThrowAsync<ModbusTransportDesyncException>(
transport.SendAsync(0x01, new byte[] { 0x03, 0x00, 0x00, 0x00, 0x01 },
TestContext.Current.CancellationToken));
}
/// <summary>
/// In-memory duplex <see cref="Stream"/> test double: records everything written and replays
/// <c>respondBytes</c> on read. When <c>stall</c> is set the read blocks until its cancellation
/// token fires (simulating a frozen gateway) so the transport's per-op deadline is exercised.
/// </summary>
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;
}
/// <summary>Gets the bytes the transport wrote to this stream (the request ADU).</summary>
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<int> ReadAsync(Memory<byte> 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<byte> 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();
}
}