Extract out rubato-audio-source

Cargo.toml

@@ -1,3 +1,3 @@
 [workspace]
-members = ["insanity-core", "insanity-native-tui-app", "insanity-tui-adapter"]
+members = ["insanity-core", "insanity-native-tui-app", "insanity-tui-adapter", "rubato-audio-source"]
 resolver = "2"

insanity-native-tui-app/Cargo.toml

@@ -26,6 +26,7 @@ serde_json = "1.0.114"
 rubato = "0.10"
 insanity-core = { path = "../insanity-core" }
 insanity-tui-adapter = { path = "../insanity-tui-adapter" }
+rubato-audio-source = { path = "../rubato-audio-source" }
 sha2 = "0.10.8"
 whoami = "1.5.1"
 sled = "0.34.7"

insanity-native-tui-app/src/clerver.rs

@@ -6,6 +6,7 @@ use cpal::traits::{HostTrait, StreamTrait};
 use insanity_core::audio_source::AudioSource;
 use insanity_tui_adapter::AppEvent;
 use opus::{Application, Channels, Decoder, Encoder};
+use rubato_audio_source::ResampledAudioSource;
 use serde::{Deserialize, Serialize};
 use tokio::sync::{broadcast, mpsc};
 use veq::veq::VeqSessionAlias;
@@ -14,7 +15,6 @@ use crate::{
     client::{get_output_config, setup_output_stream},
     processor::{AudioChunk, AudioFormat, AudioProcessor, AUDIO_CHUNK_SIZE},
     protocol::ProtocolMessage,
-    resampler::ResampledAudioSource,
     server::make_audio_receiver,
 };
 
@@ -33,7 +33,7 @@ async fn run_audio_sender<R: AudioSource + Send + Sync + 'static>(
     let channels_count = audio_receiver.channels();
     let channels = u16_to_channels(channels_count);
 
-    let mut audio_receiver = ResampledAudioSource::new(audio_receiver, 48000);
+    let mut audio_receiver = ResampledAudioSource::new(audio_receiver, 48000, AUDIO_CHUNK_SIZE);
     let mut encoder = Encoder::new(48000, channels, Application::Audio).unwrap();
     let mut sequence_number = 0;
 

insanity-native-tui-app/src/lib.rs

@@ -5,6 +5,5 @@ pub mod managed_peer;
 pub mod processor;
 pub mod protocol;
 pub mod realtime_buffer;
-pub mod resampler;
 pub mod room_handler;
 pub mod server;

insanity-native-tui-app/src/processor.rs

@@ -8,10 +8,10 @@ use std::sync::{Arc, Mutex};
 use cpal::{Sample, SampleRate};
 use insanity_core::audio_source::SyncAudioSource;
 use nnnoiseless::DenoiseState;
+use rubato_audio_source::ResampledAudioSource;
 use serde::{Deserialize, Serialize};
 
 use crate::realtime_buffer::RealTimeBuffer;
-use crate::resampler::ResampledAudioSource;
 use crate::server::RealtimeAudioSource;
 
 pub const AUDIO_CHUNK_SIZE: usize = 480;
@@ -203,7 +203,8 @@ impl AudioProcessor<'_> {
     ) -> Self {
         let chunk_buffer = Arc::new(Mutex::new(RealTimeBuffer::new(10)));
         let audio_receiver = RealtimeAudioSource::new(chunk_buffer.clone(), 48000, 2);
-        let audio_receiver = ResampledAudioSource::new(audio_receiver, output_sample_rate.0);
+        let audio_receiver =
+            ResampledAudioSource::new(audio_receiver, output_sample_rate.0, AUDIO_CHUNK_SIZE);
 
         AudioProcessor {
             enable_denoise,

rubato-audio-source/Cargo.toml

@@ -0,0 +1,9 @@
+[package]
+name = "rubato-audio-source"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+insanity-core = { path = "../insanity-core" }
+log = "0.4.22"
+rubato = "0.10"

insanity-native-tui-app/src/resampler.rs → rubato-audio-source/src/lib.rs

@@ -1,21 +1,20 @@
-use std::{collections::VecDeque, sync::Mutex};
+use std::collections::VecDeque;
 
 use insanity_core::audio_source::{AudioSource, SyncAudioSource};
 use log::trace;
 use rubato::{Resampler, SincFixedIn};
 
-use crate::processor::AUDIO_CHUNK_SIZE;
-
 pub struct ResampledAudioSource<R: AudioSource> {
-    resampler: Mutex<SincFixedIn<f32>>,
+    resampler: SincFixedIn<f32>,
     resampled_buffer: VecDeque<f32>,
     original_samples_buffer: VecDeque<f32>,
     delegate: R,
     sample_rate: u32,
+    chunk_size: usize,
 }
 
 impl<R: AudioSource + Send + Sync> ResampledAudioSource<R> {
-    pub fn new(delegate: R, sample_rate: u32) -> ResampledAudioSource<R> {
+    pub fn new(delegate: R, sample_rate: u32, chunk_size: usize) -> ResampledAudioSource<R> {
         let params = rubato::InterpolationParameters {
             sinc_len: 256,
             f_cutoff: 0.95,
@@ -26,15 +25,16 @@ impl<R: AudioSource + Send + Sync> ResampledAudioSource<R> {
         let resampler = SincFixedIn::<f32>::new(
             sample_rate as f64 / delegate.sample_rate() as f64,
             params,
-            AUDIO_CHUNK_SIZE,
+            chunk_size,
             delegate.channels() as usize,
         );
         ResampledAudioSource {
-            resampler: Mutex::new(resampler),
+            resampler,
             resampled_buffer: VecDeque::new(),
             original_samples_buffer: VecDeque::new(),
             delegate,
             sample_rate,
+            chunk_size,
         }
     }
 }
@@ -68,10 +68,10 @@ impl<R: AudioSource + Send> AudioSource for ResampledAudioSource<R> {
         }
         if self.resampled_buffer.is_empty() {
             // First, try to fill the original_samples buffer with enough samples to resample
-            let target_samples_count = AUDIO_CHUNK_SIZE * self.delegate.channels() as usize;
+            let target_samples_count = self.chunk_size * self.delegate.channels() as usize;
             trace!(
                 "Audio chunk size: {}, channels: {}, target samples count: {}",
-                AUDIO_CHUNK_SIZE,
+                self.chunk_size,
                 self.delegate.channels(),
                 target_samples_count
             );
@@ -91,8 +91,7 @@ impl<R: AudioSource + Send> AudioSource for ResampledAudioSource<R> {
             let samples = self.original_samples_buffer.drain(..).collect::<Vec<f32>>();
             let channels = separate_channels(&samples, self.delegate.channels() as usize);
             trace!("Separated into {} channels", channels.len());
-            let mut resampler_guard = self.resampler.lock().unwrap();
-            let resampled_channels = resampler_guard.process(&channels).unwrap();
+            let resampled_channels = self.resampler.process(&channels).unwrap();
             let resampled_samples = interleave_channels(&resampled_channels);
             self.resampled_buffer = resampled_samples.into();
         }
@@ -115,10 +114,10 @@ impl<R: SyncAudioSource + Send> SyncAudioSource for ResampledAudioSource<R> {
         }
         if self.resampled_buffer.is_empty() {
             // First, try to fill the original_samples buffer with enough samples to resample
-            let target_samples_count = AUDIO_CHUNK_SIZE * self.delegate.channels() as usize;
+            let target_samples_count = self.chunk_size * self.delegate.channels() as usize;
             trace!(
                 "Audio chunk size: {}, channels: {}, target samples count: {}",
-                AUDIO_CHUNK_SIZE,
+                self.chunk_size,
                 self.delegate.channels(),
                 target_samples_count
             );
@@ -138,8 +137,7 @@ impl<R: SyncAudioSource + Send> SyncAudioSource for ResampledAudioSource<R> {
             let samples = self.original_samples_buffer.drain(..).collect::<Vec<f32>>();
             let channels = separate_channels(&samples, self.delegate.channels() as usize);
             trace!("Separated into {} channels", channels.len());
-            let mut resampler_guard = self.resampler.lock().unwrap();
-            let resampled_channels = resampler_guard.process(&channels).unwrap();
+            let resampled_channels = self.resampler.process(&channels).unwrap();
             let resampled_samples = interleave_channels(&resampled_channels);
             self.resampled_buffer = resampled_samples.into();
         }