Robovoice/Robovoice.Stt.Tcp/TcpSttSource.cs

using System.Net;
using System.Net.Sockets;
using System.Text;
using Robovoice.Core;

namespace Robovoice.Stt.Tcp;

public sealed class TcpSttSource : ISttSource
{
    private TcpClient? _tcp;
    private NetworkStream? _stream;
    private StreamWriter? _writer;
    private Thread? _recvThread;
    private Thread? _connectThread;
    private volatile bool _running;
    private readonly object _sendLock = new();
    private uint _session;
    private bool _disposed;

    public string Endpoint { get; set; } = "127.0.0.1:6996";

    public Action<string>? Log { get; set; }

    public event TranscriptEventHandler? TranscriptReceived;

    public Task StartAsync(CancellationToken ct = default)
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        if (_running)
            return Task.CompletedTask;

        _running = true;
        _connectThread = new Thread(ConnectLoop) { IsBackground = true, Name = "TcpSttSource-Connect" };
        _connectThread.Start();

        return Task.CompletedTask;
    }

    public Task StopAsync(CancellationToken ct = default)
    {
        _running = false;

        lock (_sendLock)
        {
            _writer?.Dispose();
            _stream?.Dispose();
            _tcp?.Close();
            _writer = null;
            _stream = null;
            _tcp = null;
        }

        // Threads are background — they'll die when the process exits.
        // Closing the socket unblocks any pending Read.
        _connectThread?.Join(1000);
        _recvThread?.Join(1000);

        return Task.CompletedTask;
    }

    private void ConnectLoop()
    {
        while (_running)
        {
            IPEndPoint? endpoint = ParseEndpoint(Endpoint);
            if (endpoint is null)
            {
                Log?.Invoke($"STT: invalid endpoint '{Endpoint}'");
                SleepInterruptible(3000);
                continue;
            }

            try
            {
                var tcp = new TcpClient();
                tcp.Connect(endpoint.Address, endpoint.Port);
                tcp.NoDelay = true;

                lock (_sendLock)
                {
                    _tcp = tcp;
                    _stream = tcp.GetStream();
                    _writer = new StreamWriter(_stream, new UTF8Encoding(false)) { AutoFlush = true };
                }

                Log?.Invoke($"STT: connected to {Endpoint}");

                // Blocking receive loop — runs until disconnected or stopped.
                ReceiveLoop();

                Log?.Invoke("STT: disconnected");
            }
            catch (Exception ex)
            {
                if (_running)
                    Log?.Invoke($"STT: connection failed ({ex.Message}), retrying...");
            }
            finally
            {
                lock (_sendLock)
                {
                    _writer?.Dispose();
                    _stream?.Dispose();
                    _tcp?.Close();
                    _writer = null;
                    _stream = null;
                    _tcp = null;
                }
            }

            if (_running)
                SleepInterruptible(3000);
        }
    }

    private void ReceiveLoop()
    {
        byte[] buffer = new byte[4096];
        StringBuilder lineBuf = new();

        while (_running)
        {
            NetworkStream? stream;
            lock (_sendLock)
            {
                stream = _stream;
            }

            if (stream is null)
                break;

            int bytesRead;
            try
            {
                bytesRead = stream.Read(buffer, 0, buffer.Length);
            }
            catch
            {
                break;
            }

            if (bytesRead == 0)
                break;

            for (int i = 0; i < bytesRead; i++)
            {
                byte b = buffer[i];
                if (b == '\n')
                {
                    string line = lineBuf.ToString().TrimEnd('\r');
                    lineBuf.Clear();

                    TranscriptMessage? message = ParseReply(line);
                    if (message is not null)
                    {
                        TranscriptReceived?.Invoke(this, new TranscriptEventArgs
                        {
                            Message = message,
                        });
                    }
                }
                else
                {
                    lineBuf.Append((char)b);
                }
            }
        }
    }

    private void SleepInterruptible(int ms)
    {
        int slice = 100;
        int waited = 0;
        while (_running && waited < ms)
        {
            int chunk = Math.Min(slice, ms - waited);
            Thread.Sleep(chunk);
            waited += chunk;
        }
    }

    public void SendOn()
    {
        _session++;
        Send($"ON {_session}");
    }

    public void SendOff()
    {
        Send($"OFF {_session}");
    }

    private void Send(string message)
    {
        lock (_sendLock)
        {
            if (_writer is null)
                return;

            try
            {
                _writer.WriteLine(message);
            }
            catch
            {
                Log?.Invoke($"STT: failed to send '{message}' (not connected?)");
            }
        }
    }

    private static IPEndPoint? ParseEndpoint(string endpoint)
    {
        int colon = endpoint.LastIndexOf(':');
        if (colon <= 0)
            return null;

        string host = endpoint[..colon];
        if (!int.TryParse(endpoint[(colon + 1)..], out int port))
            return null;

        if (IPAddress.TryParse(host, out var addr))
            return new IPEndPoint(addr, port);

        try
        {
            var addresses = Dns.GetHostAddresses(host);
            addr = addresses.FirstOrDefault(a => a.AddressFamily == AddressFamily.InterNetwork);
            if (addr is null)
                return null;
            return new IPEndPoint(addr, port);
        }
        catch
        {
            return null;
        }
    }

    private TranscriptMessage? ParseReply(string line)
    {
        if (line.StartsWith("P ", StringComparison.Ordinal))
        {
            string rest = line["P ".Length..];
            int space = rest.IndexOf(' ');
            if (space < 0)
                return null;

            if (!uint.TryParse(rest[..space], out uint session))
                return null;

            if (session != _session)
            {
                Log?.Invoke($"STT: dropping stale reply (session {session} != current {_session})");
                return null;
            }

            return new TranscriptMessage(TranscriptType.Partial, rest[(space + 1)..]);
        }

        if (line.StartsWith("F ", StringComparison.Ordinal))
        {
            string rest = line["F ".Length..];
            int space = rest.IndexOf(' ');

            if (space < 0)
            {
                if (uint.TryParse(rest, out uint session) && session == _session)
                    return new TranscriptMessage(TranscriptType.Final, string.Empty);
                return null;
            }

            if (!uint.TryParse(rest[..space], out uint ses))
                return null;

            if (ses != _session)
            {
                Log?.Invoke($"STT: dropping stale reply (session {ses} != current {_session})");
                return null;
            }

            return new TranscriptMessage(TranscriptType.Final, rest[(space + 1)..]);
        }

        return null;
    }

    public ValueTask DisposeAsync()
    {
        if (_disposed) return ValueTask.CompletedTask;
        _disposed = true;
        StopAsync();
        return ValueTask.CompletedTask;
    }
}