riojoy/src/RioJoy.Core/Feedback/FeedbackPipeServer.cs

using System.IO.Pipes;

namespace RioJoy.Core.Feedback;

/// <summary>
/// Named-pipe listener for the inbound feedback protocol: serves
/// <c>\\.\pipe\&lt;name&gt;</c>, reassembles lines
/// (<see cref="FeedbackLineBuffer"/>), and hands each to the owner — parsing
/// and routing live in <see cref="FeedbackService"/>, so this class is pure
/// transport. Modeled on vRIO's <c>VRioPipeService</c> (dedicated background
/// threads — net40 has no <c>WaitForConnectionAsync</c>; throwaway poke-connect
/// on stop because a pending <c>WaitForConnection</c> can survive Dispose on
/// net48), with two deliberate differences: the pipe is
/// <see cref="PipeDirection.In"/> — the server never writes, so the 0-byte
/// pipe-buffer write deadlock class cannot occur and no reply path exists — and
/// up to <see cref="MaxClients"/> clients may stay connected at once (a sim
/// export script and a SimHub plugin both live here). Clients reconnect
/// forever; a malformed or overlong line never costs a client its connection.
/// </summary>
public sealed class FeedbackPipeServer : IDisposable
{
    /// <summary>Concurrent client cap (pipe instances of the served name).</summary>
    public const int MaxClients = 4;

    private readonly string _pipeName;
    private readonly Action<string> _onLine;
    private readonly Action<string>? _log;
    private readonly SemaphoreSlim _slots = new(MaxClients, MaxClients);
    private readonly object _stateGate = new();
    private readonly List<NamedPipeServerStream> _open = new();
    private readonly List<Thread> _readers = new();
    private Thread? _accept;
    private volatile bool _running;

    public FeedbackPipeServer(string pipeName, Action<string> onLine, Action<string>? log = null)
    {
        if (string.IsNullOrWhiteSpace(pipeName))
            throw new ArgumentException("Pipe name is required.", nameof(pipeName));
        _pipeName = pipeName;
        _onLine = onLine ?? throw new ArgumentNullException(nameof(onLine));
        _log = log;
    }

    /// <summary>The served pipe name (without the <c>\\.\pipe\</c> prefix).</summary>
    public string PipeName => _pipeName;

    /// <summary>Start listening (idempotent). Clients may come and go forever.</summary>
    public void Start()
    {
        if (_running)
            return;
        _running = true;

        _accept = new Thread(AcceptLoop)
        {
            IsBackground = true,
            Name = $"RIOJoy feedback pipe ({_pipeName})",
        };
        _accept.Start();
        _log?.Invoke($@"feedback: listening on \\.\pipe\{_pipeName}");
    }

    public void Dispose()
    {
        if (!_running)
            return;
        _running = false;

        // A WaitForConnection pending on a disposed stream can survive the
        // Dispose on net48; a throwaway client connect releases it either way.
        try
        {
            using var poke = new NamedPipeClientStream(".", _pipeName, PipeDirection.Out);
            poke.Connect(100);
        }
        catch (Exception ex) when (ex is IOException or TimeoutException or UnauthorizedAccessException) { }

        NamedPipeServerStream[] open;
        Thread[] readers;
        lock (_stateGate)
        {
            open = _open.ToArray();
            _open.Clear();
            readers = _readers.ToArray();
            _readers.Clear();
        }
        foreach (NamedPipeServerStream pipe in open)
        {
            try { pipe.Dispose(); }
            catch (IOException) { }
        }

        _accept?.Join(1000);
        _accept = null;
        foreach (Thread reader in readers)
            reader.Join(1000);
    }

    private void AcceptLoop()
    {
        bool busyLogged = false; // log a name collision once, not per retry

        while (_running)
        {
            // At capacity, park until a reader frees its slot (timed, so
            // shutdown can't wedge on a missed release).
            if (!_slots.Wait(200))
                continue;
            if (!_running)
            {
                _slots.Release();
                return;
            }

            NamedPipeServerStream pipe;
            try
            {
                pipe = new NamedPipeServerStream(_pipeName, PipeDirection.In, MaxClients,
                    PipeTransmissionMode.Byte, PipeOptions.Asynchronous);
            }
            catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
            {
                // Name already served — most likely a second RIOJoy instance.
                _slots.Release();
                if (!busyLogged)
                {
                    busyLogged = true;
                    _log?.Invoke($@"feedback: \\.\pipe\{_pipeName} is busy ({ex.Message.TrimEnd('.')}) — retrying");
                }
                for (int i = 0; i < 20 && _running; i++)
                    Thread.Sleep(100);
                continue;
            }
            busyLogged = false;
            lock (_stateGate)
                _open.Add(pipe);

            try
            {
                pipe.WaitForConnection();
            }
            catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException)
            {
                Drop(pipe);
                continue; // disposed by Dispose(), or the client vanished mid-connect
            }

            if (!_running)
            {
                Drop(pipe);
                return;
            }

            var reader = new Thread(() => ReadUntilDisconnect(pipe))
            {
                IsBackground = true,
                Name = $"RIOJoy feedback pipe reader ({_pipeName})",
            };
            lock (_stateGate)
                _readers.Add(reader);
            reader.Start(); // the reader owns the slot + stream from here
        }
    }

    private void ReadUntilDisconnect(NamedPipeServerStream pipe)
    {
        var buffer = new byte[512];
        var lines = new FeedbackLineBuffer();
        try
        {
            while (_running)
            {
                int n;
                try
                {
                    n = pipe.Read(buffer, 0, buffer.Length);
                }
                catch (Exception ex) when (
                    ex is IOException or ObjectDisposedException or InvalidOperationException)
                {
                    return; // client gone or shutdown
                }

                if (n == 0)
                    return; // client closed its end

                foreach (string line in lines.Feed(buffer, n))
                    Handle(line);
            }
        }
        finally
        {
            Drop(pipe);
            lock (_stateGate)
                _readers.Remove(Thread.CurrentThread);
        }
    }

    private void Handle(string line)
    {
        try
        {
            _onLine(line);
        }
        catch (Exception ex)
        {
            // The line sink must never kill a reader; log and keep serving.
            _log?.Invoke($"feedback: line handler failed: {ex.Message}");
        }
    }

    private void Drop(NamedPipeServerStream pipe)
    {
        lock (_stateGate)
            _open.Remove(pipe);
        try { pipe.Dispose(); }
        catch (IOException) { }
        _slots.Release();
    }
}