lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs

using System.Buffers;
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) 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 int _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 (Volatile.Read(ref _disposed) != 0) throw new ObjectDisposedException(nameof(ModbusRtuOverTcpTransport));
        if (_life is null && _testStream 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.Stream is not null && _life.ShouldReconnectForIdle())
                await _life.TearDownAsync().ConfigureAwait(false);

            // Self-heal (real socket only): a socket torn down by an earlier failure is re-dialled here,
            // once, throttled by the backoff window — same contract as ModbusTcpTransport.
            var freshlyConnected = _life is not null && _life.Stream is null;
            if (freshlyConnected)
                await _life!.EnsureConnectedAsync(ct).ConfigureAwait(false);

            try
            {
                var result = await SendOnceAsync(unitId, pdu, ct).ConfigureAwait(false);
                _life?.MarkSuccess();
                return result;
            }
            catch (Exception ex) when (!freshlyConnected && _life is not null && _life.AutoReconnect
                                       && !ct.IsCancellationRequested
                                       && ModbusSocketLifecycle.IsSocketLevelFailure(ex))
            {
                // Mid-transaction drop: tear down the dead socket, make ONE throttled reconnect attempt
                // (never a sleep inside the gate), 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.EnsureConnectedAsync(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. Built in a pooled buffer (R07:L5); the same bytes
        // ModbusRtuFraming.BuildAdu produces (pinned by a test).
        var aduLength = ModbusRtuFraming.AduLength(pdu.Length);
        var adu = ArrayPool<byte>.Shared.Rent(aduLength);
        ModbusRtuFraming.WriteAdu(unitId, pdu, adu.AsSpan(0, aduLength));

        using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
        cts.CancelAfter(_timeout);
        try
        {
            await stream.WriteAsync(adu.AsMemory(0, aduLength), 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");
        }
        finally
        {
            // The write has completed, or was cancelled and the stream torn down, so nothing still reads it.
            ArrayPool<byte>.Shared.Return(adu);
        }
    }

    /// <summary>Asynchronously disposes the transport and underlying socket resources.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async ValueTask DisposeAsync()
    {
        if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
        if (_life is not null) await _life.DisposeAsync().ConfigureAwait(false);
        // _gate is deliberately NOT disposed: a SendAsync still in flight at teardown releases it in its finally.
    }
}