Port managed client audio and Windows application

dotnet/src/VoiceCat.Audio/AudioEngine.cs

@@ -0,0 +1,170 @@
+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 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);
+}