From c9846394ac6cddd82442baadcabad653a28fed66 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Gustav=20S=C3=B6rn=C3=A4s?= <gustav@sornas.net>
Date: Wed, 21 Oct 2020 03:37:22 +0200
Subject: refactor audio into input.rs and output.rs

---
 mumd/src/audio/input.rs  | 52 ++++++++++++++++++++++++++++
 mumd/src/audio/output.rs | 90 ++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 142 insertions(+)
 create mode 100644 mumd/src/audio/input.rs
 create mode 100644 mumd/src/audio/output.rs

(limited to 'mumd/src/audio')

diff --git a/mumd/src/audio/input.rs b/mumd/src/audio/input.rs
new file mode 100644
index 0000000..4e95360
--- /dev/null
+++ b/mumd/src/audio/input.rs
@@ -0,0 +1,52 @@
+use bytes::Bytes;
+use cpal::{InputCallbackInfo, Sample};
+use mumble_protocol::voice::VoicePacketPayload;
+use std::collections::VecDeque;
+use std::sync::{Arc, Mutex};
+use tokio::sync::{mpsc, watch};
+
+pub fn callback<T: Sample>(
+    mut opus_encoder: opus::Encoder,
+    mut input_sender: mpsc::Sender<VoicePacketPayload>,
+    sample_rate: u32,
+    input_volume_receiver: watch::Receiver<f32>,
+    opus_frame_size_blocks: u32, // blocks of 2.5ms
+) -> impl FnMut(&[T], &InputCallbackInfo) + Send + 'static {
+    if !(opus_frame_size_blocks == 1
+        || opus_frame_size_blocks == 2
+        || opus_frame_size_blocks == 4
+        || opus_frame_size_blocks == 8)
+    {
+        panic!(
+            "Unsupported amount of opus frame blocks {}",
+            opus_frame_size_blocks
+        );
+    }
+    let opus_frame_size = opus_frame_size_blocks * sample_rate / 400;
+
+    let buf = Arc::new(Mutex::new(VecDeque::new()));
+    move |data: &[T], _info: &InputCallbackInfo| {
+        let mut buf = buf.lock().unwrap();
+        let input_volume = *input_volume_receiver.borrow();
+        let out: Vec<f32> = data.iter().map(|e| e.to_f32())
+                                       .map(|e| e * input_volume)
+                                       .collect();
+        buf.extend(out);
+        while buf.len() >= opus_frame_size as usize {
+            let tail = buf.split_off(opus_frame_size as usize);
+            let mut opus_buf: Vec<u8> = vec![0; opus_frame_size as usize];
+            let result = opus_encoder
+                .encode_float(&Vec::from(buf.clone()), &mut opus_buf)
+                .unwrap();
+            opus_buf.truncate(result);
+            let bytes = Bytes::copy_from_slice(&opus_buf);
+            match input_sender.try_send(VoicePacketPayload::Opus(bytes, false)) {
+                Ok(_) => {}
+                Err(_e) => {
+                    //warn!("Error sending audio packet: {:?}", e);
+                }
+            }
+            *buf = tail;
+        }
+    }
+}
diff --git a/mumd/src/audio/output.rs b/mumd/src/audio/output.rs
new file mode 100644
index 0000000..94e4b21
--- /dev/null
+++ b/mumd/src/audio/output.rs
@@ -0,0 +1,90 @@
+use cpal::{OutputCallbackInfo, Sample};
+use mumble_protocol::voice::VoicePacketPayload;
+use opus::Channels;
+use std::collections::{HashMap, VecDeque};
+use std::ops::AddAssign;
+use std::sync::{Arc, Mutex};
+
+pub struct ClientStream {
+    buffer: VecDeque<f32>, //TODO ring buffer?
+    opus_decoder: opus::Decoder,
+}
+
+impl ClientStream {
+    pub fn new(sample_rate: u32, channels: u16) -> Self {
+        Self {
+            buffer: VecDeque::new(),
+            opus_decoder: opus::Decoder::new(
+                sample_rate,
+                match channels {
+                    1 => Channels::Mono,
+                    2 => Channels::Stereo,
+                    _ => unimplemented!("Only 1 or 2 channels supported, got {}", channels),
+                },
+            )
+            .unwrap(),
+        }
+    }
+
+    pub fn decode_packet(&mut self, payload: VoicePacketPayload, channels: usize) {
+        match payload {
+            VoicePacketPayload::Opus(bytes, _eot) => {
+                let mut out: Vec<f32> = vec![0.0; 720 * channels * 4]; //720 is because that is the max size of packet we can get that we want to decode
+                let parsed = self
+                    .opus_decoder
+                    .decode_float(&bytes, &mut out, false)
+                    .expect("Error decoding");
+                out.truncate(parsed);
+                self.buffer.extend(out);
+            }
+            _ => {
+                unimplemented!("Payload type not supported");
+            }
+        }
+    }
+}
+
+pub trait SaturatingAdd {
+    fn saturating_add(self, rhs: Self) -> Self;
+}
+
+impl SaturatingAdd for f32 {
+    fn saturating_add(self, rhs: Self) -> Self {
+        match self + rhs {
+            a if a < -1.0 => -1.0,
+            a if a > 1.0 => 1.0,
+            a => a,
+        }
+    }
+}
+
+impl SaturatingAdd for i16 {
+    fn saturating_add(self, rhs: Self) -> Self {
+        i16::saturating_add(self, rhs)
+    }
+}
+
+impl SaturatingAdd for u16 {
+    fn saturating_add(self, rhs: Self) -> Self {
+        u16::saturating_add(self, rhs)
+    }
+}
+
+pub fn curry_callback<T: Sample + AddAssign + SaturatingAdd>(
+    buf: Arc<Mutex<HashMap<u32, ClientStream>>>,
+) -> impl FnMut(&mut [T], &OutputCallbackInfo) + Send + 'static {
+    move |data: &mut [T], _info: &OutputCallbackInfo| {
+        for sample in data.iter_mut() {
+            *sample = Sample::from(&0.0);
+        }
+
+        let mut lock = buf.lock().unwrap();
+        for client_stream in lock.values_mut() {
+            for sample in data.iter_mut() {
+                *sample = sample.saturating_add(Sample::from(
+                    &client_stream.buffer.pop_front().unwrap_or(0.0),
+                ));
+            }
+        }
+    }
+}
-- 
cgit v1.2.1