in progress add buffer processor

crates/livekit_client/Cargo.toml

@@ -41,7 +41,8 @@ tokio-tungstenite.workspace = true
 util.workspace = true
 workspace-hack.workspace = true
 
-rodio = { workspace = true, features = ["wav_output"] }
+rodio = { workspace = true, features = ["wav_output", "recording"] }
+dasp_sample = "0.11"
 
 [target.'cfg(not(any(all(target_os = "windows", target_env = "gnu"), target_os = "freebsd")))'.dependencies]
 libwebrtc = { rev = "5f04705ac3f356350ae31534ffbc476abc9ea83d", git = "https://github.com/zed-industries/livekit-rust-sdks" }

crates/livekit_client/src/lib.rs

@@ -9,19 +9,19 @@ use rodio::DeviceTrait as _;
 mod record;
 pub use record::CaptureInput;
 
-#[cfg(not(any(
-    test,
-    feature = "test-support",
-    all(target_os = "windows", target_env = "gnu"),
-    target_os = "freebsd"
-)))]
+// #[cfg(not(any(
+//     test,
+//     feature = "test-support",
+//     all(target_os = "windows", target_env = "gnu"),
+//     target_os = "freebsd"
+// )))]
 mod livekit_client;
-#[cfg(not(any(
-    test,
-    feature = "test-support",
-    all(target_os = "windows", target_env = "gnu"),
-    target_os = "freebsd"
-)))]
+// #[cfg(not(any(
+//     test,
+//     feature = "test-support",
+//     all(target_os = "windows", target_env = "gnu"),
+//     target_os = "freebsd"
+// )))]
 pub use livekit_client::*;
 
 // If you need proper LSP in livekit_client you've got to comment
@@ -29,27 +29,27 @@ pub use livekit_client::*;
 // - the mods: mock_client & test and their conditional blocks
 // - the pub use mock_client::* and their conditional blocks
 
-#[cfg(any(
-    test,
-    feature = "test-support",
-    all(target_os = "windows", target_env = "gnu"),
-    target_os = "freebsd"
-))]
-mod mock_client;
-#[cfg(any(
-    test,
-    feature = "test-support",
-    all(target_os = "windows", target_env = "gnu"),
-    target_os = "freebsd"
-))]
-pub mod test;
-#[cfg(any(
-    test,
-    feature = "test-support",
-    all(target_os = "windows", target_env = "gnu"),
-    target_os = "freebsd"
-))]
-pub use mock_client::*;
+// #[cfg(any(
+//     test,
+//     feature = "test-support",
+//     all(target_os = "windows", target_env = "gnu"),
+//     target_os = "freebsd"
+// ))]
+// mod mock_client;
+// #[cfg(any(
+//     test,
+//     feature = "test-support",
+//     all(target_os = "windows", target_env = "gnu"),
+//     target_os = "freebsd"
+// ))]
+// pub mod test;
+// #[cfg(any(
+//     test,
+//     feature = "test-support",
+//     all(target_os = "windows", target_env = "gnu"),
+//     target_os = "freebsd"
+// ))]
+// pub use mock_client::*;
 
 #[derive(Debug, Clone)]
 pub enum Participant {

crates/livekit_client/src/livekit_client/playback.rs

@@ -1,6 +1,8 @@
 use anyhow::{Context as _, Result};
 
+use cpal::Sample;
 use cpal::traits::{DeviceTrait, StreamTrait as _};
+use dasp_sample::ToSample;
 use futures::channel::mpsc::UnboundedSender;
 use futures::{Stream, StreamExt as _};
 use gpui::{
@@ -19,7 +21,9 @@ use livekit::webrtc::{
 };
 use parking_lot::Mutex;
 use rodio::Source;
+use rodio::source::{LimitSettings, UniformSourceIterator};
 use std::cell::RefCell;
+use std::num::NonZero;
 use std::sync::Weak;
 use std::sync::atomic::{AtomicBool, AtomicI32, Ordering};
 use std::time::Duration;
@@ -254,6 +258,48 @@ impl AudioStack {
         }
     }
 
+    async fn capture_input_rodio(
+        apm: Arc<Mutex<apm::AudioProcessingModule>>,
+        frame_tx: UnboundedSender<AudioFrame<'static>>,
+        sample_rate: u32,
+        num_channels: u32,
+    ) -> Result<()> {
+        use crate::livekit_client::playback::source::RodioExt;
+        thread::spawn(|| {
+            let stream = rodio::microphone::MicrophoneBuilder::new()
+                .with_default_device()?
+                .with_default_config()?
+                .open_stream()?;
+            let stream = UniformSourceIterator::new(
+                stream,
+                NonZero::new(1).expect("1 is not zero"),
+                NonZero::new(SAMPLE_RATE).expect("constant is not zero"),
+            )
+            .limit(LimitSettings::live_performance())
+            .process_buffer(|buffer| {
+                let mut int_buffer: [i16; _] = buffer.map(|s| s.to_sample());
+                apm.lock()
+                    .process_stream(&mut int_buffer, sample_rate as i32, num_channels as i32)
+                    .unwrap(); // TODO dvdsk fix this
+                for (sample, processed) in buffer.iter().zip(&int_buffer) {
+                    *sample = (*processed).to_sample_();
+                }
+            })
+            .automatic_gain_control(1.0, 4.0, 0.0, 5.0);
+
+            loop {
+                frame_tx.unbounded_send(AudioFrame {
+                    data: Cow::Owned(sampled),
+                    sample_rate,
+                    num_channels,
+                    samples_per_channel: sample_rate / 100,
+                })
+            }
+        });
+
+        Ok(())
+    }
+
     async fn capture_input(
         apm: Arc<Mutex<apm::AudioProcessingModule>>,
         frame_tx: UnboundedSender<AudioFrame<'static>>,

crates/livekit_client/src/livekit_client/playback/source.rs

@@ -1,3 +1,5 @@
+use std::num::NonZero;
+
 use futures::StreamExt;
 use libwebrtc::{audio_stream::native::NativeAudioStream, prelude::AudioFrame};
 use livekit::track::RemoteAudioTrack;
@@ -9,7 +11,11 @@ fn frame_to_samplesbuffer(frame: AudioFrame) -> SamplesBuffer {
     let samples = frame.data.iter().copied();
     let samples = SampleTypeConverter::<_, _>::new(samples);
     let samples: Vec<f32> = samples.collect();
-    SamplesBuffer::new(frame.num_channels as u16, frame.sample_rate, samples)
+    SamplesBuffer::new(
+        NonZero::new(frame.num_channels as u16).expect("audio frame channels is nonzero"),
+        NonZero::new(frame.sample_rate).expect("audio frame sample rate is nonzero"),
+        samples,
+    )
 }
 
 pub struct LiveKitStream {
@@ -65,3 +71,76 @@ impl Source for LiveKitStream {
         self.inner.total_duration()
     }
 }
+
+pub trait RodioExt: Source + Sized {
+    fn process_buffer<F>(
+        self,
+        callback: impl FnMut(&mut [rodio::Sample; 200]),
+    ) -> ProcessBuffer<Self, F>
+    where
+        F: FnMut(&mut [rodio::Sample]);
+}
+
+impl<S: Source> RodioExt for S {
+    fn process_buffer<F>(
+        self,
+        callback: impl FnMut(&mut [rodio::Sample; 200]),
+    ) -> ProcessBuffer<Self, F>
+    where
+        F: FnMut(&mut [rodio::Sample]),
+    {
+        ProcessBuffer {
+            inner: self,
+            callback,
+            in_buffer: [0.0; 200],
+            out_buffer: [0.0; 200],
+        }
+    }
+}
+
+pub struct ProcessBuffer<S, F>
+where
+    S: Source + Sized,
+    F: FnMut(&mut [rodio::Sample; 200]),
+{
+    inner: S,
+    callback: F,
+    in_buffer: [rodio::Sample; 200],
+    out_buffer: std::array::IntoIter<rodio::Sample, N>,
+}
+
+impl<S, F> Iterator for ProcessBuffer<S, F>
+where
+    S: Source + Sized,
+    F: FnMut(&mut [rodio::Sample; 200]),
+{
+    type Item = rodio::Sample;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        for sample in &mut in_buffer {
+            *sample = self.inner.next()?;
+        }
+    }
+}
+
+impl<S, F> Source for ProcessBuffer<S, F>
+where
+    S: Source + Sized,
+    F: FnMut(&mut [rodio::Sample; 200]),
+{
+    fn current_span_len(&self) -> Option<usize> {
+        todo!()
+    }
+
+    fn channels(&self) -> rodio::ChannelCount {
+        todo!()
+    }
+
+    fn sample_rate(&self) -> rodio::SampleRate {
+        todo!()
+    }
+
+    fn total_duration(&self) -> Option<std::time::Duration> {
+        todo!()
+    }
+}

crates/livekit_client/src/record.rs

@@ -1,5 +1,6 @@
 use std::{
     env,
+    num::NonZero,
     path::{Path, PathBuf},
     sync::{Arc, Mutex},
     time::Duration,
@@ -83,7 +84,11 @@ fn write_out(
             .expect("Stream has ended, callback cant hold the lock"),
     );
     let samples: Vec<f32> = SampleTypeConverter::<_, f32>::new(samples.into_iter()).collect();
-    let mut samples = SamplesBuffer::new(config.channels(), config.sample_rate().0, samples);
+    let mut samples = SamplesBuffer::new(
+        NonZero::new(config.channels()).expect("config channel is never zero"),
+        NonZero::new(config.sample_rate().0).expect("config sample_rate is never zero"),
+        samples,
+    );
     match rodio::output_to_wav(&mut samples, path) {
         Ok(_) => Ok(()),
         Err(e) => Err(anyhow::anyhow!("Failed to write wav file: {}", e)),