voice-cat/dotnet/src/VoiceCat.Audio/AudioEngine.cs

using System.Diagnostics;
using VoiceCat.Protocol;
using Voicecat.V1;

namespace VoiceCat.Audio;

public sealed class AudioEngine : IDisposable
{
    private readonly object gate = new();
    private readonly EncodedVoiceSender sender;
    private readonly int[] mixed = new int[1920];
    private readonly short[] output = new short[1920];
    private Routes routes = new([], [], 0);
    private readonly List<(IDisposable Stream, long Epoch)> retired = [];
    private long completedEpoch;
    private readonly CancellationTokenSource stop = new();
    private readonly Task maintenance;
    private Thread? worker;
    private int disposed;
    public uint SampleClock { get; private set; }
    public event MixedPcmHandler? MixedPcm;
    public event PcmStreamHandler? StreamPcm;
    public volatile float InputGain = 1, OutputGain = 1, VadThreshold = 0.02f;
    public volatile bool InputNoiseReduction, MicMuted, Deafened, PushToTalk;
    public volatile AudioInputMode InputMode = AudioInputMode.VoiceActivation;
    public Exception? Failure { get; private set; }

    public AudioEngine(EncodedVoiceSender sender, bool startWorker = true)
    {
        this.sender = sender;
        maintenance = MaintainAsync();
        if (startWorker)
        {
            worker = new Thread(Work) { IsBackground = true, Name = "VoiceCat managed audio" };
            worker.Start();
        }
    }

    public void AddLocalStream(StreamInfo info, int captureChannels = 1)
    {
        if (captureChannels is not (1 or 2)) throw new ArgumentOutOfRangeException(nameof(captureChannels));
        lock (gate)
        {
            ObjectDisposedException.ThrowIf(disposed != 0, this);
            var stream = new LocalStream(info, captureChannels);
            Routes previous = routes;
            var locals = previous.Local.Where(s => s.Info.StreamId != info.StreamId).Append(stream).ToArray();
            Publish(locals, previous.Remote);
        }
    }
    public void RemoveLocalStream(uint streamId)
    {
        lock (gate) Publish(routes.Local.Where(s => s.Info.StreamId != streamId).ToArray(), routes.Remote);
    }
    public void SetCaptureChannels(uint streamId, int channels)
    {
        if (channels is not (1 or 2)) throw new ArgumentOutOfRangeException(nameof(channels));
        lock (gate)
        {
            var current = routes.Local.FirstOrDefault(s => s.Info.StreamId == streamId) ?? throw new ArgumentException("Stream not found.");
            if (current.CaptureChannels != channels) Publish(routes.Local.Select(s => s == current ? new LocalStream(s.Info, channels) : s).ToArray(), routes.Remote);
        }
    }

    public void SetRemoteStreams(IReadOnlyList<User> users, uint selfId, uint channelId)
    {
        lock (gate)
        {
            var next = new List<ReceiveStream>();
            foreach (User user in users.Where(u => u.Id != selfId && u.ChannelId == channelId))
                foreach (StreamInfo info in user.Streams)
                {
                    var previous = routes.Remote.FirstOrDefault(s => s.UserId == user.Id && s.Info.Equals(info));
                    next.Add(previous ?? new ReceiveStream(user.Id, info));
                }
            Publish(routes.Local, next.ToArray());
        }
    }

    public bool FeedPcm(uint streamId, ReadOnlySpan<short> pcm, int channels)
    {
        foreach (LocalStream stream in Volatile.Read(ref routes).Local)
            if (stream.Info.StreamId == streamId) return stream.Feed(pcm, channels);
        return false;
    }
    public void Receive(VoiceFrameHeader header, ReadOnlySpan<byte> packet)
    {
        foreach (ReceiveStream stream in Volatile.Read(ref routes).Remote) if (stream.Info.Ssrc == header.Ssrc) { stream.Enqueue(header, packet); return; }
    }
    public (float Level, bool Talking) GetLocalLevel(uint streamId)
    {
        foreach (LocalStream stream in Volatile.Read(ref routes).Local) if (stream.Info.StreamId == streamId) return (stream.Level, stream.Talking);
        return default;
    }
    public void SetLocalGain(uint streamId, float gain)
    {
        if (!float.IsFinite(gain) || gain < 0 || gain > 4) throw new ArgumentOutOfRangeException(nameof(gain));
        foreach (LocalStream stream in Volatile.Read(ref routes).Local)
            if (stream.Info.StreamId == streamId) { stream.Gain = gain; return; }
    }
    public void SetRemotePlayback(uint userId, uint streamId, float gain, bool muted, bool noiseReduction)
    {
        if (!float.IsFinite(gain) || gain < 0 || gain > 4) throw new ArgumentOutOfRangeException(nameof(gain));
        foreach (var stream in Volatile.Read(ref routes).Remote)
            if (stream.UserId == userId && stream.Info.StreamId == streamId) { stream.Gain = gain; stream.Muted = muted; stream.NoiseReduction = noiseReduction; return; }
    }
    public (float Gain, bool Muted, bool NoiseReduction)? GetRemotePlayback(uint userId, uint streamId)
    {
        foreach (var stream in Volatile.Read(ref routes).Remote) if (stream.UserId == userId && stream.Info.StreamId == streamId) return (stream.Gain, stream.Muted, stream.NoiseReduction);
        return null;
    }

    private void Publish(LocalStream[] local, ReceiveStream[] remote)
    {
        Routes previous = routes; var next = new Routes(local, remote, previous.Epoch + 1);
        foreach (var stream in previous.Local) if (!local.Contains(stream)) retired.Add((stream, next.Epoch));
        foreach (var stream in previous.Remote) if (!remote.Contains(stream)) retired.Add((stream, next.Epoch));
        Volatile.Write(ref routes, next);
    }

    // One audio owner calls this. No allocation, waiting, lock, registry mutation or disposal
    // occurs inside a mix cycle. Callbacks receive borrowed spans and must follow that rule.
    internal void ProcessCycle()
    {
        Routes current = Volatile.Read(ref routes);
        mixed.AsSpan().Clear();
        foreach (var stream in current.Local) stream.Process(this, sender);
        foreach (var stream in current.Remote) stream.Mix(mixed, Deafened, StreamPcm);
        float gain = Deafened ? 0 : OutputGain;
        for (int i = 0; i < output.Length; i++) output[i] = (short)Math.Clamp((int)(mixed[i] * gain), short.MinValue, short.MaxValue);
        MixedPcm?.Invoke(output);
        SampleClock = unchecked(SampleClock + 960);
        Volatile.Write(ref completedEpoch, current.Epoch);
    }

    private void Work()
    {
        long deadline = Stopwatch.GetTimestamp();
        try
        {
            while (!stop.IsCancellationRequested)
            {
                ProcessCycle();
                deadline += Stopwatch.Frequency / 50;
                double remaining = (deadline - Stopwatch.GetTimestamp()) * 1000.0 / Stopwatch.Frequency;
                if (remaining > 0) Thread.Sleep((int)Math.Ceiling(remaining));
                else if (remaining < -100) deadline = Stopwatch.GetTimestamp();
            }
        }
        catch (Exception exception) { Failure = exception; stop.Cancel(); }
    }
    private async Task MaintainAsync()
    {
        try
        {
            using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(20));
            while (await timer.WaitForNextTickAsync(stop.Token).ConfigureAwait(false))
                lock (gate)
                    for (int i = retired.Count - 1; i >= 0; i--)
                        if (retired[i].Epoch <= Volatile.Read(ref completedEpoch)) { retired[i].Stream.Dispose(); retired.RemoveAt(i); }
        }
        catch (OperationCanceledException) when (stop.IsCancellationRequested) { }
    }
    public void Dispose()
    {
        if (Interlocked.Exchange(ref disposed, 1) != 0) return;
        stop.Cancel(); worker?.Join(); maintenance.GetAwaiter().GetResult();
        lock (gate)
        {
            foreach (var stream in routes.Local) stream.Dispose(); foreach (var stream in routes.Remote) stream.Dispose();
            foreach (var stream in retired) stream.Stream.Dispose(); retired.Clear();
            routes = new([], [], routes.Epoch + 1);
        }
    }
    private sealed record Routes(LocalStream[] Local, ReceiveStream[] Remote, long Epoch);
}