using System.Diagnostics;
using RioJoy.Core.Protocol;
namespace RioJoy.Core.Serial;
///
/// 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
/// (CommWatchProc/ReadCommBlock); see docs/PROTOCOL.md §2 and §4.
///
/// The link drives a supplied but does not own
/// it: acquiring/releasing the COM port is the transport's lifecycle (create to
/// acquire, dispose to yield).
///
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();
// Stop-and-wait command state: one command in flight at a time
// (_commandGate), resolved by the board's ACK/NAK or by AckTimeout.
// Control-byte replies (our ACKs) bypass the gate — the board never
// ACK/NAKs those, and they must not queue behind a pending command.
private readonly SemaphoreSlim _commandGate = new(1, 1);
private readonly object _pendingGate = new();
private PendingCommand? _pending;
private long _retransmits;
// The in-flight command: its command byte gates which inbound events may
// resolve it (a reply only resolves the MATCHING request — an analog reply
// still in USB transit from the previous poll must never "confirm" a lamp
// command that follows it; bench 2026-07-19).
private sealed class PendingCommand
{
public PendingCommand(byte command) => Command = command;
public byte Command { get; }
public TaskCompletionSource Tcs { get; } = new();
}
public RioSerialLink(IRioTransport transport, RioSerialLinkOptions? options = null)
{
_transport = transport ?? throw new ArgumentNullException(nameof(transport));
_options = options ?? new RioSerialLinkOptions();
}
/// Raised for every framed packet (after the ACK/NAK reply is sent).
public event Action? PacketReceived;
/// Raised for a decoded, valid .
public event Action? AnalogReceived;
/// Raised for a decoded .
public event Action? VersionReceived;
/// Raised for a decoded .
public event Action? CheckReceived;
/// Raised for a control byte received outside framing (ACK/NAK/RESTART/IDLE/garbage).
public event Action? ControlReceived;
/// Raised when a mid-packet framing error forced a resync.
public event Action? FramingError;
/// The transport's description, surfaced for status/logging.
public string Description => _transport.Description;
/// Total command retransmits (NAK- or timeout-triggered) since creation.
public long Retransmits => Interlocked.Read(ref _retransmits);
///
/// Run the receive loop and (if enabled) the analog poll loop until
/// fires or the transport closes.
///
public async Task RunAsync(CancellationToken cancellationToken)
{
_parser.Reset();
_sinceAnalog.Restart();
using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
var loops = new List { 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);
}
}
///
/// Send a pre-built packet (see ) to the RIO.
/// Command packets use stop-and-wait: the call completes once the board
/// ACKs, or after
/// retransmits (NAK- or timeout-triggered) go unacknowledged — the command
/// is then dropped (all commands are idempotent state-setters, and the
/// caller's next update supersedes it). Control bytes bypass the wait.
///
public async Task SendAsync(byte[] packet, CancellationToken cancellationToken = default)
{
if (packet is null) throw new ArgumentNullException(nameof(packet));
// Control-byte replies (len 1) are never ACK/NAK'd by the board, and a
// limit of 0 selects the legacy fire-and-forget behavior.
if (packet.Length <= 1 || _options.CommandRetransmitLimit <= 0)
{
await WriteAsync(packet, cancellationToken).ConfigureAwait(false);
return;
}
await Compat.TaskCompat.WaitAsync(_commandGate, cancellationToken).ConfigureAwait(false);
try
{
for (int attempt = 0; ; attempt++)
{
var pending = new PendingCommand(packet[0]);
lock (_pendingGate) _pending = pending;
await WriteAsync(packet, cancellationToken).ConfigureAwait(false);
Task winner = await Compat.TaskCompat.WhenAny(new[]
{
pending.Tcs.Task,
Compat.TaskCompat.Delay(_options.AckTimeout, cancellationToken),
}).ConfigureAwait(false);
if (winner == pending.Tcs.Task && pending.Tcs.Task.Result)
return; // delivered (ACK, or the request's own reply)
// NAK'd or timed out (a shred so complete the board never NAK'd).
if (attempt >= _options.CommandRetransmitLimit)
{
// Budget spent — drop (idempotent; the next update supersedes).
// Brief settle so a late straggler response can't land on the
// NEXT command's wait (strays with nothing pending are ignored).
lock (_pendingGate) _pending = null;
await Compat.TaskCompat.Delay(TimeSpan.FromMilliseconds(10), cancellationToken)
.ConfigureAwait(false);
return;
}
Interlocked.Increment(ref _retransmits);
}
}
finally
{
lock (_pendingGate) _pending = null;
_commandGate.Release();
}
}
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();
}
}
/// Request an analog update ().
public Task RequestAnalogAsync(CancellationToken cancellationToken = default) =>
SendAsync(PacketBuilder.AnalogRequest(), cancellationToken);
/// Request the RIO firmware version ().
public Task RequestVersionAsync(CancellationToken cancellationToken = default) =>
SendAsync(PacketBuilder.VersionRequest(), cancellationToken);
/// Request a board/lamp status check ().
public Task RequestCheckAsync(CancellationToken cancellationToken = default) =>
SendAsync(PacketBuilder.CheckRequest(), cancellationToken);
/// Issue a reset to recalibrate an axis or recover the board.
public Task ResetAsync(RioResetTarget target, CancellationToken cancellationToken = default) =>
SendAsync(PacketBuilder.ResetRequest(target), cancellationToken);
/// Set a lighted button's state ().
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 is (byte)RioControl.Ack or (byte)RioControl.Nak)
{
// Resolve the in-flight command (stop-and-wait). Safe to be
// type-blind here: the board's TX ISR sends pending ACK/NAK
// ahead of reply data, so a command's ACK cannot arrive
// after its successor starts (unlike replies, see
// DispatchTyped).
PendingCommand? pending;
lock (_pendingGate) pending = _pending;
pending?.Tcs.TrySetResult(ev.Byte == (byte)RioControl.Ack);
}
ControlReceived?.Invoke(ev.Byte);
break;
case RioRxEventKind.FramingError:
FramingError?.Invoke();
break;
}
}
private void DispatchTyped(RioPacket packet)
{
// A reply proves the MATCHING request landed, whether or not the board
// also sent an explicit ACK — but only the matching one: a reply still
// in transit from an earlier poll must never confirm an unrelated
// command (that type-blindness was the residual lamp-glitch hole).
RioCommand? resolves = packet.Command switch
{
RioCommand.AnalogReply => RioCommand.AnalogRequest,
RioCommand.VersionReply => RioCommand.VersionRequest,
RioCommand.CheckReply => RioCommand.CheckRequest,
_ => null,
};
if (resolves is not null)
{
PendingCommand? pending;
lock (_pendingGate) pending = _pending;
if (pending is not null && pending.Command == (byte)resolves.Value)
pending.Tcs.TrySetResult(true);
}
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.
}
}
}