riojoy/tests/RioJoy.Core.Tests/Serial/GatedTransport.cs

using System.Threading.Channels;
using RioJoy.Core.Serial;

namespace RioJoy.Core.Tests.Serial;

/// <summary>
/// <see cref="IRioTransport"/> whose first write blocks until <see cref="Open"/>
/// — lets a test park one writer mid-sequence while another tries to cut in
/// (write-lock and latest-wins assertions).
/// </summary>
internal sealed class GatedTransport : IRioTransport
{
    private readonly Channel<byte[]> _writes = Channel.CreateUnbounded<byte[]>();
    private readonly SemaphoreSlim _gate = new(0, 1);
    private readonly object _armLock = new();
    private bool _gateArmed = true;

    public string Description => "gated";

    /// <summary>Release the parked first write.</summary>
    public void Open() => _gate.Release();

    public Task<int> ReadAsync(byte[] buffer, CancellationToken cancellationToken) =>
        Task.FromResult(0);

    public async Task WriteAsync(byte[] data, CancellationToken cancellationToken)
    {
        bool wait;
        lock (_armLock)
        {
            wait = _gateArmed;
            _gateArmed = false;
        }
        if (wait)
            await _gate.WaitAsync(cancellationToken);
        _writes.Writer.TryWrite((byte[])data.Clone());
    }

    /// <summary>Read the next write, failing if none arrives in time.</summary>
    public async Task<byte[]> NextWriteAsync(TimeSpan? timeout = null)
    {
        using var cts = new CancellationTokenSource(timeout ?? TimeSpan.FromSeconds(5));
        return await _writes.Reader.ReadAsync(cts.Token);
    }

    /// <summary>True when no further write has arrived.</summary>
    public bool NoMoreWrites => !_writes.Reader.TryPeek(out _);

    public void Dispose() { }
}