separating audio from tcp; separate resampler for every input stream

client/src/audio_streams.rs

@@ -10,18 +10,21 @@ use ringbuf::{
     traits::{Consumer, Observer, Producer, Split},
     wrap::caching::Caching,
 };
-use ropus::Encoder;
+use ropus::{Decoder, Encoder};
-use rubato::{Async, FixedAsync, PolynomialDegree, Resampler, audioadapter::Adapter};
+use rubato::{Fft, FixedSync, Resampler, audioadapter::Adapter};
 use shared::AudioStreamInfo;
-use std::sync::{Arc, RwLock};
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex, RwLock},
+};
 use tokio::{
     net::UdpSocket,
     sync::broadcast::{Receiver, Sender},
 };
 
 use crate::models::{
-    ClientEvent, ClientInfo, InputAudioStreamsConsumers, InputAudioStreamsEncoders,
+    ClientEvent, ClientInfo, InputAudioStreamsConsumers, InputAudioStreamsDecoders,
-    InputAudioStreamsInfo, InputAudioStreamsProducers,
+    InputAudioStreamsInfo, InputAudioStreamsProducers, InputAudioStreamsResamplers,
 };
 
 fn cpal_build_output_stream<T>(
@@ -58,62 +61,65 @@ where
 pub async fn audio_stream_receiver(
     output_device: Device,
     output_device_config: SupportedStreamConfig,
-    active_input_audio_streams_producers: InputAudioStreamsProducers,
-    active_input_audio_streams_consumers: InputAudioStreamsConsumers,
-    active_input_audio_streams_encoders: InputAudioStreamsEncoders,
     active_input_audio_streams_info: InputAudioStreamsInfo,
     udp_socket: Arc<UdpSocket>,
 ) {
+    let mut active_input_audio_streams_producers: InputAudioStreamsProducers = HashMap::new();
+    let active_input_audio_streams_consumers: InputAudioStreamsConsumers =
+        Arc::new(Mutex::new(HashMap::new()));
+    let mut active_input_audio_streams_decoders: InputAudioStreamsDecoders = HashMap::new();
+    let mut active_input_audio_streams_resamplers: InputAudioStreamsResamplers = HashMap::new();
+
     let output_stream = match output_device_config.sample_format() {
         SampleFormat::F64 => cpal_build_output_stream::<f64>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::I64 => cpal_build_output_stream::<i64>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::U64 => cpal_build_output_stream::<u64>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::F32 => cpal_build_output_stream::<f32>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::I32 => cpal_build_output_stream::<i32>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::U32 => cpal_build_output_stream::<u32>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::I16 => cpal_build_output_stream::<i16>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::U16 => cpal_build_output_stream::<u16>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::I8 => cpal_build_output_stream::<i8>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         SampleFormat::U8 => cpal_build_output_stream::<u8>(
             &output_device,
             output_device_config.into(),
-            active_input_audio_streams_consumers,
+            active_input_audio_streams_consumers.clone(),
         ),
         _ => {
             eprintln!("unsupported output stream format :(");
@@ -125,8 +131,15 @@ pub async fn audio_stream_receiver(
 
     let mut receive_buffer = vec![0u8; 4096];
 
-    let mut decoded_samples = [0.0f32; 2048];
+    let max_input_raw_samples = ((20.0 / 1000.0) * 2.0 * SAMPLE_RATE_48K as f32) as usize; // 20ms, 2channels, 48khz
-    let mut decoded_samples_converted = [0.0f32; 2048];
+    let mut decoded_samples = vec![0.0f32; max_input_raw_samples];
+
+    let mut resampled_samples = vec![0.0f32; max_input_raw_samples];
+
+    let output_raw_samples = ((20.0 / 1000.0)
+        * output_device_config.channels() as f32
+        * output_device_config.sample_rate() as f32) as usize;
+    let mut decoded_samples_converted = vec![0.0f32; output_raw_samples];
 
     loop {
         match udp_socket.recv_from(&mut receive_buffer).await {
@@ -155,15 +168,56 @@ pub async fn audio_stream_receiver(
                     }
                 };
 
-                let decoders_map = active_input_audio_streams_encoders.read().unwrap();
+                let input_frames = ((20.0 / 1000.0) * stream_info.sample_rate as f32) as usize;
+                let resampler_output_frames =
+                    ((20.0 / 1000.0) * output_device_config.sample_rate() as f32) as usize;
 
-                let decoder_mutex = decoders_map
+                if !active_input_audio_streams_decoders.contains_key(&(client_id, stream_id)) {
-                    .get(&(client_id, stream_id))
+                    let channels: ropus::Channels = {
-                    .expect("failed to get encoder mutex");
+                        if stream_info.channels == 2 {
+                            ropus::Channels::Stereo
+                        } else {
+                            ropus::Channels::Mono
+                        }
+                    };
 
-                let mut decoder = decoder_mutex.lock().expect("failed to get encoder");
+                    let decoder = Decoder::new(stream_info.sample_rate, channels)
+                        .expect("failed to create decoder");
+                    active_input_audio_streams_decoders
+                        .insert((stream_info.client_id, stream_info.stream_id), decoder);
+                }
 
-                let decoded_len = decoder
+                if !active_input_audio_streams_resamplers.contains_key(&(client_id, stream_id)) {
+                    let resampler = Fft::<f32>::new(
+                        stream_info.sample_rate as usize,
+                        output_device_config.sample_rate() as usize,
+                        input_frames,
+                        stream_info.channels as usize,
+                        FixedSync::Both,
+                    )
+                    .unwrap();
+                    active_input_audio_streams_resamplers.insert((client_id, stream_id), resampler);
+                }
+
+                if !active_input_audio_streams_producers.contains_key(&(client_id, stream_id)) {
+                    let ring_buffer_frames = (100.0 / 1000.0) * stream_info.sample_rate as f32;
+                    let ring_buffer_samples =
+                        ring_buffer_frames as usize * stream_info.channels as usize;
+                    let ring_buffer = HeapRb::<f32>::new(ring_buffer_samples);
+                    let (producer, consumer) = ring_buffer.split();
+
+                    active_input_audio_streams_producers.insert((client_id, stream_id), producer);
+                    active_input_audio_streams_consumers
+                        .lock()
+                        .unwrap()
+                        .insert((client_id, stream_id), consumer);
+                }
+
+                let decoder = active_input_audio_streams_decoders
+                    .get_mut(&(client_id, stream_id))
+                    .expect("failed to get decoder");
+
+                let _ = decoder
                     .decode_float(
                         audio_packet,
                         &mut decoded_samples,
@@ -171,41 +225,50 @@ pub async fn audio_stream_receiver(
                     )
                     .expect("failed to decode opus");
 
-                let mut active_streams_map = active_input_audio_streams_producers.lock().unwrap();
+                let decoded_samples_adapter = InterleavedSlice::new(
+                    &decoded_samples,
+                    stream_info.channels as usize,
+                    input_frames,
+                )
+                .unwrap();
 
-                if let Some(producer) = active_streams_map.get_mut(&(client_id, stream_id)) {
+                let mut resampled_samples_adapter = InterleavedSlice::new_mut(
-                    if stream_info.channels == output_device_config.channels() {
+                    &mut resampled_samples,
-                        producer.push_slice(
+                    stream_info.channels as usize,
-                            &decoded_samples[..decoded_len * stream_info.channels as usize],
+                    resampler_output_frames,
-                        );
+                )
-                    } else if stream_info.channels > output_device_config.channels() {
+                .unwrap();
-                        for (i, frame) in decoded_samples
+
-                            [..decoded_len * stream_info.channels as usize]
+                let (_, frames_out) = active_input_audio_streams_resamplers
-                            .chunks_exact(2)
+                    .get_mut(&(client_id, stream_id))
-                            .enumerate()
+                    .expect("failed to get resampler")
-                        {
+                    .process_into_buffer(
-                            decoded_samples_converted[i] = (frame[0] + frame[1]) / 2.0;
+                        &decoded_samples_adapter,
+                        &mut resampled_samples_adapter,
+                        None,
+                    )
+                    .expect("error during resampling");
+
+                if let Some(producer) =
+                    active_input_audio_streams_producers.get_mut(&(client_id, stream_id))
+                {
+                    for (&resampled_sample, output_frame) in resampled_samples
+                        .iter()
+                        .step_by(stream_info.channels as usize)
+                        .zip(
+                            decoded_samples_converted
+                                .chunks_exact_mut(output_device_config.channels() as usize),
+                        )
+                    {
+                        for output_sample in output_frame.iter_mut() {
+                            *output_sample = resampled_sample;
                         }
-
-                        producer.push_slice(
-                            &decoded_samples_converted
-                                [..decoded_len * stream_info.channels as usize],
-                        );
-                    } else {
-                        for (i, frame) in decoded_samples_converted
-                            [..decoded_len * output_device_config.channels() as usize]
-                            .chunks_exact_mut(2)
-                            .enumerate()
-                        {
-                            frame[0] = decoded_samples[i];
-                            frame[1] = decoded_samples[i];
-                        }
-
-                        producer.push_slice(
-                            &decoded_samples_converted
-                                [..decoded_len * output_device_config.channels() as usize],
-                        );
                     }
+
+                    producer.push_slice(
+                        &decoded_samples_converted
+                            [..frames_out * output_device_config.channels() as usize],
+                    );
                 }
             }
             Err(e) => eprintln!("failed to receive udp packet: {}", e),
@@ -329,13 +392,12 @@ pub async fn audio_stream_spawner(
 
     let mut resampler_input_buffer = vec![0.0f32; frames_needed * force_channels as usize];
 
-    let mut resampler = Async::<f32>::new_poly(
+    let mut resampler = Fft::<f32>::new(
-        48000.0 / 44100.0,
+        input_device_config.sample_rate() as usize,
-        1.1,
+        SAMPLE_RATE_48K as usize,
-        PolynomialDegree::Cubic,
         frames_needed,
         force_channels as usize,
-        FixedAsync::Output,
+        FixedSync::Both,
     )
     .unwrap();
 

client/src/main.rs

@@ -7,17 +7,14 @@ use clap::Parser;
 use cpal::traits::{DeviceTrait, HostTrait};
 use std::{
     collections::HashMap,
-    sync::{Arc, Mutex, RwLock},
+    sync::{Arc, RwLock},
 };
 use tokio::net::UdpSocket;
 use tokio::sync::broadcast;
 
 use crate::{
     audio_streams::{audio_stream_receiver, audio_stream_spawner},
-    models::{
+    models::{AudioControl, CliArgs, ClientEvent, ClientInfo, InputAudioStreamsInfo},
-        AudioControl, CliArgs, ClientEvent, ClientInfo, InputAudioStreamsConsumers,
-        InputAudioStreamsEncoders, InputAudioStreamsInfo, InputAudioStreamsProducers,
-    },
     tcp_connection::tcp_connection,
     tui::Tui,
 };
@@ -28,23 +25,12 @@ async fn main() {
 
     let (global_event_tx, _) = broadcast::channel::<ClientEvent>(16);
 
-    let active_input_audio_streams_producers: InputAudioStreamsProducers =
-        Arc::new(Mutex::new(HashMap::new()));
-    let active_input_audio_streams_consumers: InputAudioStreamsConsumers =
-        Arc::new(Mutex::new(HashMap::new()));
-
-    let active_input_audio_streams_encoders: InputAudioStreamsEncoders =
-        Arc::new(RwLock::new(HashMap::new()));
-
     let active_input_audio_streams_info: InputAudioStreamsInfo =
         Arc::new(RwLock::new(HashMap::new()));
 
     let udp_socket = Arc::new(UdpSocket::bind("0.0.0.0:0").await.unwrap());
 
     let audio_control: Arc<AudioControl> = Arc::new(AudioControl {
-        active_input_audio_streams_producers: active_input_audio_streams_producers.clone(),
-        active_input_audio_streams_consumers: active_input_audio_streams_consumers.clone(),
-        active_input_audio_streams_encoders: active_input_audio_streams_encoders.clone(),
         active_input_audio_streams_info: active_input_audio_streams_info.clone(),
         udp_socket: udp_socket.clone(),
     });
@@ -87,9 +73,6 @@ async fn main() {
         tokio::spawn(audio_stream_receiver(
             output_device,
             output_device_config.into(),
-            active_input_audio_streams_producers.clone(),
-            active_input_audio_streams_consumers.clone(),
-            active_input_audio_streams_encoders.clone(),
             active_input_audio_streams_info.clone(),
             udp_socket.clone(),
         ));

client/src/models.rs

@@ -1,6 +1,7 @@
 use clap::Parser;
 use ringbuf::{HeapRb, wrap::caching::Caching};
 use ropus::Decoder;
+use rubato::Fft;
 use shared::AudioStreamInfo;
 use std::{
     collections::HashMap,
@@ -39,21 +40,19 @@ pub struct ClientInfo {
     pub connected: bool,
 }
 
-type AudioStreamKey = (u32, u8);
+pub type AudioStreamKey = (u32, u8);
 
 pub type InputAudioStreamsProducers =
-    Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, true, false>>>>;
+    HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, true, false>>;
 pub type InputAudioStreamsConsumers =
     Arc<Mutex<HashMap<AudioStreamKey, Caching<Arc<HeapRb<f32>>, false, true>>>>;
 
-pub type InputAudioStreamsEncoders = Arc<RwLock<HashMap<AudioStreamKey, Mutex<Decoder>>>>;
+pub type InputAudioStreamsDecoders = HashMap<AudioStreamKey, Decoder>;
+pub type InputAudioStreamsResamplers = HashMap<AudioStreamKey, Fft<f32>>;
 
 pub type InputAudioStreamsInfo = Arc<RwLock<HashMap<AudioStreamKey, AudioStreamInfo>>>;
 
 pub struct AudioControl {
-    pub active_input_audio_streams_producers: InputAudioStreamsProducers,
-    pub active_input_audio_streams_consumers: InputAudioStreamsConsumers,
-    pub active_input_audio_streams_encoders: InputAudioStreamsEncoders,
     pub active_input_audio_streams_info: InputAudioStreamsInfo,
     pub udp_socket: Arc<UdpSocket>,
 }

client/src/tcp_connection.rs

@@ -1,8 +1,6 @@
 use futures::{SinkExt, StreamExt};
-use ringbuf::{HeapRb, traits::Split};
-use ropus::Decoder;
 use shared::{ResponseStatus, TcpRequest, TcpResponse, TcpResponseData};
-use std::sync::{Arc, Mutex, RwLock};
+use std::sync::{Arc, RwLock};
 use tokio::{
     net::TcpStream,
     sync::broadcast::{Receiver, Sender},
@@ -57,49 +55,10 @@ pub async fn tcp_connection(
                                         println!("user disconnected: {}", user_id);
                                     }
                                     TcpResponseData::AudioStreamsFullUpdate { audio_stream_info } => {
-                                        let mut producer_map = audio_control.active_input_audio_streams_producers.lock().unwrap();
-                                        producer_map.clear();
-                                        let mut consumer_map = audio_control.active_input_audio_streams_consumers.lock().unwrap();
-                                        consumer_map.clear();
-
-                                        let mut encoder_map = audio_control.active_input_audio_streams_encoders.write().unwrap();
-                                        encoder_map.clear();
-
                                         let mut info_map = audio_control.active_input_audio_streams_info.write().unwrap();
                                         info_map.clear();
 
-
                                         for audio_stream in audio_stream_info {
-                                            let ring_buffer_ms = 150.0;
-                                            let ring_buffer_frames = (ring_buffer_ms / 1000.0) * audio_stream.sample_rate as f32;
-                                            let ring_buffer_samples = ring_buffer_frames as usize * audio_stream.channels as usize;
-                                            let ring_buffer_size = ring_buffer_samples;
-                                            let ring_buffer = HeapRb::<f32>::new(ring_buffer_size);
-                                            let (producer, consumer) = ring_buffer.split();
-
-                                            producer_map.insert(
-                                                (audio_stream.client_id, audio_stream.stream_id),
-                                                producer,
-                                            );
-                                            consumer_map.insert(
-                                                (audio_stream.client_id, audio_stream.stream_id),
-                                                consumer,
-                                            );
-
-                                            let channels: ropus::Channels = {
-                                                if audio_stream.channels == 1 {
-                                                    ropus::Channels::Mono
-                                                } else {
-                                                    ropus::Channels::Stereo
-                                                }
-                                            };
-
-                                            let decoder = Decoder::new(audio_stream.sample_rate, channels).expect("failed to creat decoder");
-                                            encoder_map.insert(
-                                                (audio_stream.client_id, audio_stream.stream_id),
-                                                Mutex::new(decoder),
-                                            );
-
                                             info_map.insert(
                                                 (audio_stream.client_id, audio_stream.stream_id),
                                                 audio_stream,

client/src/tui.rs

@@ -345,15 +345,6 @@ impl DeviceSelectScreen {
             tokio::spawn(audio_stream_receiver(
                 output_device.clone(),
                 output_config.clone(),
-                self.audio_control
-                    .active_input_audio_streams_producers
-                    .clone(),
-                self.audio_control
-                    .active_input_audio_streams_consumers
-                    .clone(),
-                self.audio_control
-                    .active_input_audio_streams_encoders
-                    .clone(),
                 self.audio_control.active_input_audio_streams_info.clone(),
                 self.audio_control.udp_socket.clone(),
             ));