LlamaCasty/Services/Audio/NamedPipeAudioWriter.cs

using System.IO.Pipes;
using System.Runtime.InteropServices;

namespace ytLive.Services.Audio;

/// <summary>
/// The mixer's sink for the live loop: interleaved stereo PCM float at 48 kHz
/// written to a named pipe, byte-identical to what ffmpeg expects from
/// <c>-f f32le -ar 48000 -ac 2 -i \\.\pipe\&lt;name&gt;</c>. The encoder side is
/// handled by FfmpegArgs/EncoderOptions; this side just owns the server end.
/// </summary>
public interface IAudioPipeWriter : IDisposable
{
    /// <summary>Creates the named pipe server and waits (async) for the client.</summary>
    void Start(string pipeName);

    /// <summary>True once ffmpeg has connected and the pipe is writable.</summary>
    bool IsConnected { get; }

    /// <summary>Count of <see cref="WriteAsync"/> calls that wrote nothing because
    /// the pipe wasn't connected yet or was already closed. Part of the silent-
    /// recording diagnosis: nonzero means audio was dropped before ffmpeg ever
    /// connected to the pipe.</summary>
    long DroppedWrites { get; }

    /// <summary>Writes interleaved float samples; dropped until the client connects.</summary>
    Task WriteAsync(ReadOnlyMemory<float> samples, CancellationToken cancellationToken = default);

    /// <summary>Closes the server end — ffmpeg sees EOF and ends the audio input.</summary>
    void Stop();
}

public sealed class NamedPipeAudioWriter : IAudioPipeWriter
{
    private readonly object _sync = new();
    private NamedPipeServerStream? _pipe;

    /// <inheritdoc/>
    public long DroppedWrites { get; private set; }

    public void Start(string pipeName)
    {
        lock (_sync)
        {
            if (_pipe != null)
                throw new InvalidOperationException("Already started.");
            _pipe = new NamedPipeServerStream(
                pipeName,
                PipeDirection.Out,
                1,
                PipeTransmissionMode.Byte,
                PipeOptions.Asynchronous);
        }
        _ = Task.Run(WaitForConnectionAsync);
    }

    public bool IsConnected
    {
        get
        {
            lock (_sync)
            {
                return _pipe is { IsConnected: true };
            }
        }
    }

    public async Task WriteAsync(ReadOnlyMemory<float> samples, CancellationToken cancellationToken = default)
    {
        NamedPipeServerStream? pipe;
        lock (_sync)
        {
            pipe = _pipe;
        }
        if (pipe == null || !pipe.IsConnected || samples.IsEmpty)
        {
            if (pipe?.IsConnected == false || pipe == null)
                DroppedWrites++;
            return;
        }

        var bytes = MemoryMarshal.AsBytes(samples.Span).ToArray();
        await pipe.WriteAsync(bytes, cancellationToken).ConfigureAwait(false);
    }

    public void Stop()
    {
        NamedPipeServerStream? pipe;
        lock (_sync)
        {
            pipe = _pipe;
            _pipe = null;
        }
        pipe?.Dispose();
    }

    public void Dispose() => Stop();

    private async Task WaitForConnectionAsync()
    {
        NamedPipeServerStream? pipe;
        lock (_sync)
        {
            pipe = _pipe;
        }
        if (pipe == null)
            return;

        try
        {
            await pipe.WaitForConnectionAsync().ConfigureAwait(false);
        }
        catch
        {
            Stop();
        }
    }
}