riojoy/src/RioJoy.Core/Serial/RioSerialLink.cs

using System.Diagnostics;
using RioJoy.Core.Protocol;

namespace RioJoy.Core.Serial;

/// <summary>
/// Drives the RIO serial link: a receive loop that frames incoming bytes into
/// packets (with ACK/NAK replies), plus the analog poll + reset-recovery timer.
/// The modern equivalent of the legacy overlapped-I/O watch thread
/// (<c>CommWatchProc</c>/<c>ReadCommBlock</c>); see docs/PROTOCOL.md §2 and §4.
///
/// <para>The link drives a supplied <see cref="IRioTransport"/> but does not own
/// it: acquiring/releasing the COM port is the transport's lifecycle (create to
/// acquire, dispose to yield).</para>
/// </summary>
public sealed class RioSerialLink
{
    private readonly IRioTransport _transport;
    private readonly RioSerialLinkOptions _options;
    private readonly PacketParser _parser = new();
    private readonly SemaphoreSlim _writeLock = new(1, 1);

    // Time since the last accepted AnalogReply, for the recovery watchdog.
    private readonly Stopwatch _sinceAnalog = new();

    // NAK retransmit state: the last COMMAND packet sent (control-byte replies
    // are excluded — the board never NAKs those) and how many resends it has
    // consumed. Guarded by _retransmitGate; see RioSerialLinkOptions.NakRetransmitLimit.
    private readonly object _retransmitGate = new();
    private byte[]? _lastCommand;
    private int _lastCommandResends;
    private long _nakRetransmits;

    public RioSerialLink(IRioTransport transport, RioSerialLinkOptions? options = null)
    {
        _transport = transport ?? throw new ArgumentNullException(nameof(transport));
        _options = options ?? new RioSerialLinkOptions();
    }

    /// <summary>Raised for every framed packet (after the ACK/NAK reply is sent).</summary>
    public event Action<RioPacket>? PacketReceived;

    /// <summary>Raised for a decoded, valid <see cref="RioCommand.AnalogReply"/>.</summary>
    public event Action<AnalogReport>? AnalogReceived;

    /// <summary>Raised for a decoded <see cref="RioCommand.VersionReply"/>.</summary>
    public event Action<VersionInfo>? VersionReceived;

    /// <summary>Raised for a decoded <see cref="RioCommand.CheckReply"/>.</summary>
    public event Action<CheckStatus>? CheckReceived;

    /// <summary>Raised for a control byte received outside framing (ACK/NAK/RESTART/IDLE/garbage).</summary>
    public event Action<byte>? ControlReceived;

    /// <summary>Raised when a mid-packet framing error forced a resync.</summary>
    public event Action? FramingError;

    /// <summary>The transport's description, surfaced for status/logging.</summary>
    public string Description => _transport.Description;

    /// <summary>Total NAK-triggered retransmits since the link was created.</summary>
    public long NakRetransmits => Interlocked.Read(ref _nakRetransmits);

    /// <summary>
    /// Run the receive loop and (if enabled) the analog poll loop until
    /// <paramref name="cancellationToken"/> fires or the transport closes.
    /// </summary>
    public async Task RunAsync(CancellationToken cancellationToken)
    {
        _parser.Reset();
        _sinceAnalog.Restart();

        using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);

        var loops = new List<Task> { ReceiveLoopAsync(linked.Token) };
        if (_options.AutoPollAnalog)
            loops.Add(PollLoopAsync(linked.Token));

        try
        {
            // If any running loop ends (transport closed / error / cancellation),
            // tear the others down too.
            await Compat.TaskCompat.WhenAny(loops).ConfigureAwait(false);
        }
        finally
        {
            linked.Cancel();
            await Compat.TaskCompat.WhenAll(loops.Select(Swallow)).ConfigureAwait(false);
        }
    }

    /// <summary>Send a pre-built packet (see <see cref="PacketBuilder"/>) to the RIO.</summary>
    public async Task SendAsync(byte[] packet, CancellationToken cancellationToken = default)
    {
        if (packet is null) throw new ArgumentNullException(nameof(packet));

        // Only real command packets participate in NAK retransmit; single
        // control bytes (our ACK/NAK replies) are never NAK'd by the board.
        if (packet.Length > 1 && _options.NakRetransmitLimit > 0)
        {
            lock (_retransmitGate)
            {
                _lastCommand = packet;
                _lastCommandResends = 0;
            }
        }

        await WriteAsync(packet, cancellationToken).ConfigureAwait(false);
    }

    private async Task WriteAsync(byte[] packet, CancellationToken cancellationToken)
    {
        await Compat.TaskCompat.WaitAsync(_writeLock, cancellationToken).ConfigureAwait(false);
        try
        {
            await _transport.WriteAsync(packet, cancellationToken).ConfigureAwait(false);
        }
        finally
        {
            _writeLock.Release();
        }
    }

    // The board NAK'd its most recent inbound packet: resend the last command
    // (idempotent, see RioSerialLinkOptions.NakRetransmitLimit) unless its
    // retry budget is spent.
    private async Task RetransmitOnNakAsync(CancellationToken ct)
    {
        byte[]? packet;
        lock (_retransmitGate)
        {
            if (_lastCommand is null || _lastCommandResends >= _options.NakRetransmitLimit)
                return;
            _lastCommandResends++;
            packet = _lastCommand;
        }

        Interlocked.Increment(ref _nakRetransmits);
        await WriteAsync(packet!, ct).ConfigureAwait(false);
    }

    /// <summary>Request an analog update (<see cref="RioCommand.AnalogRequest"/>).</summary>
    public Task RequestAnalogAsync(CancellationToken cancellationToken = default) =>
        SendAsync(PacketBuilder.AnalogRequest(), cancellationToken);

    /// <summary>Request the RIO firmware version (<see cref="RioCommand.VersionRequest"/>).</summary>
    public Task RequestVersionAsync(CancellationToken cancellationToken = default) =>
        SendAsync(PacketBuilder.VersionRequest(), cancellationToken);

    /// <summary>Request a board/lamp status check (<see cref="RioCommand.CheckRequest"/>).</summary>
    public Task RequestCheckAsync(CancellationToken cancellationToken = default) =>
        SendAsync(PacketBuilder.CheckRequest(), cancellationToken);

    /// <summary>Issue a reset to recalibrate an axis or recover the board.</summary>
    public Task ResetAsync(RioResetTarget target, CancellationToken cancellationToken = default) =>
        SendAsync(PacketBuilder.ResetRequest(target), cancellationToken);

    /// <summary>Set a lighted button's state (<see cref="RioCommand.LampRequest"/>).</summary>
    public Task SetLampAsync(byte lampNumber, byte state, CancellationToken cancellationToken = default) =>
        SendAsync(PacketBuilder.LampRequest(lampNumber, state), cancellationToken);

    private async Task ReceiveLoopAsync(CancellationToken ct)
    {
        var buffer = new byte[_options.ReadBufferSize];
        while (!ct.IsCancellationRequested)
        {
            int n = await _transport.ReadAsync(buffer, ct).ConfigureAwait(false);
            if (n == 0)
                break; // transport closed

            for (int i = 0; i < n; i++)
            {
                if (_parser.Feed(buffer[i], out RioRxEvent ev))
                    await HandleEventAsync(ev, ct).ConfigureAwait(false);
            }
        }
    }

    private async Task HandleEventAsync(RioRxEvent ev, CancellationToken ct)
    {
        switch (ev.Kind)
        {
            case RioRxEventKind.Packet:
                await ReplyAsync(ev, ct).ConfigureAwait(false);

                DispatchTyped(ev.Packet);
                PacketReceived?.Invoke(ev.Packet);
                break;

            case RioRxEventKind.ControlByte:
                if (ev.Byte == (byte)RioControl.Nak)
                    await RetransmitOnNakAsync(ct).ConfigureAwait(false);
                ControlReceived?.Invoke(ev.Byte);
                break;

            case RioRxEventKind.FramingError:
                FramingError?.Invoke();
                break;
        }
    }

    private void DispatchTyped(RioPacket packet)
    {
        switch (packet.Command)
        {
            case RioCommand.AnalogReply:
                if (AnalogReport.TryParse(packet.Payload, out AnalogReport report))
                {
                    _sinceAnalog.Restart();
                    AnalogReceived?.Invoke(report);
                }
                break;

            case RioCommand.VersionReply:
                VersionReceived?.Invoke(VersionInfo.Parse(packet.Payload));
                break;

            case RioCommand.CheckReply:
                CheckReceived?.Invoke(CheckStatus.Parse(packet.Payload));
                break;
        }
    }

    private Task ReplyAsync(RioRxEvent ev, CancellationToken ct)
    {
        // Documented contract: ACK an accepted packet; NAK a button packet whose
        // checksum failed. The legacy path force-accepts (always ACK) unless
        // VerifyInboundChecksum re-enables real verification.
        bool nak = _options.VerifyInboundChecksum
                   && !ev.ChecksumValid
                   && ev.Packet.Command is RioCommand.ButtonPressed or RioCommand.ButtonReleased;

        byte reply = nak ? (byte)RioControl.Nak : (byte)RioControl.Ack;
        return SendAsync(new[] { reply }, ct);
    }

    private async Task PollLoopAsync(CancellationToken ct)
    {
        try
        {
            while (!ct.IsCancellationRequested)
            {
                await Compat.TaskCompat.Delay(_options.AnalogPollInterval, ct).ConfigureAwait(false);

                await RequestAnalogAsync(ct).ConfigureAwait(false);

                if (_sinceAnalog.Elapsed > _options.AnalogRecoveryTimeout)
                {
                    // No analog data for too long — recover with a general reset.
                    await SendAsync(PacketBuilder.ResetRequest(RioResetTarget.All), ct).ConfigureAwait(false);
                    _sinceAnalog.Restart(); // avoid re-issuing every tick
                }
            }
        }
        catch (OperationCanceledException)
        {
            // Normal shutdown.
        }
    }

    private static async Task Swallow(Task task)
    {
        try
        {
            await task.ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            // Expected on teardown.
        }
    }
}