rustriff_lib/infrastructure/

audio_handler.rs

use cpal::traits::{DeviceTrait, StreamTrait};
use cpal::{Device, Stream, StreamConfig};
use derive_getters::Getters;
use mockall::automock;
use ringbuf::consumer::Consumer;
use ringbuf::producer::Producer;
use ringbuf::traits::Split;
use ringbuf::{HeapCons, HeapProd, HeapRb};
use tracing::error;

/// A thin wrapper around a CPAL [`Stream`] that allows it to be started
/// through a trait object, enabling mocking in tests.
pub trait PlayableStream: Send {
    /// Starts playback/capture on the underlying stream.
    ///
    /// # Panics
    ///
    /// Panics if the underlying CPAL stream returns an error when playing.
    fn play(&self);
}

impl PlayableStream for Stream {
    fn play(&self) {
        StreamTrait::play(self).unwrap();
    }
}

/// Abstraction over the audio I/O handler used by [`AudioService`].
///
/// Using the [`MockAudioHandlerTrait`] generated by `mockall` allows the audio pipeline to be tested without real hardware.
///
/// [`AudioService`]: crate::services::audio_service::AudioService
#[automock]
pub trait AudioHandlerTrait: Send + Sync {
    /// Builds and returns a started-ready input stream that pushes captured
    /// samples into `prod` (producer).
    ///
    /// # Arguments
    ///
    /// * `prod` - The ring-buffer producer that receives raw `f32` audio samples.
    fn build_input_stream(&self, prod: HeapProd<f32>) -> Box<dyn PlayableStream>;

    /// Builds and returns a started-ready output stream that drains samples
    /// from `cons` (consumer) and sends them to the output device.
    ///
    /// Slots in the output buffer that have no corresponding sample are
    /// filled with silence (`0.0`).
    ///
    /// # Arguments
    ///
    /// * `cons` - The ring-buffer consumer that supplies processed `f32` audio samples.
    fn build_output_stream(&self, cons: HeapCons<f32>) -> Box<dyn PlayableStream>;

    /// Returns a reference to the CPAL input device used by this handler.
    fn input_device(&self) -> &Device;

    /// Returns a reference to the CPAL output device used by this handler.
    fn output_device(&self) -> &Device;

    /// Returns the [`StreamConfig`] used for the input stream.
    fn input_config(&self) -> &StreamConfig;

    /// Returns the [`StreamConfig`] used for the output stream.
    fn output_config(&self) -> &StreamConfig;

    /// Returns the configured input sample rate in Hz.
    fn input_sample_rate(&self) -> u32;

    /// Returns the configured output sample rate in Hz.
    fn output_sample_rate(&self) -> u32;
}

/// Concrete implementation of [`AudioHandlerTrait`] backed by real CPAL devices.
///
/// `AudioHandler` owns the input device, output device, and stream configuration
/// required to build CPAL streams. It is cheaply cloneable so that it can be
/// shared across threads via [`Arc`].
///
/// [`Arc`]: std::sync::Arc
#[derive(Clone, Getters)]
pub struct AudioHandler {
    input_device: Device,
    output_device: Device,
    input_config: StreamConfig,
    output_config: StreamConfig,
    input_sample_rate: u32,
    output_sample_rate: u32,
}

impl AudioHandler {
    /// Creates a new `AudioHandler` with the given CPAL devices and stream config.
    ///
    /// # Arguments
    ///
    /// * `input_device` - The CPAL device used to capture audio.
    /// * `output_device` - The CPAL device used to play back audio.
    /// * `input_config` - The [`StreamConfig`] used for the input stream.
    /// * `output_config` - The [`StreamConfig`] used for the output stream.
    pub fn new(
        input_device: Device,
        output_device: Device,
        input_config: StreamConfig,
        output_config: StreamConfig,
    ) -> Self {
        let input_sample_rate = input_config.sample_rate;
        let output_sample_rate = output_config.sample_rate;

        Self {
            input_device,
            output_device,
            input_config,
            output_config,
            input_sample_rate,
            output_sample_rate,
        }
    }

    /// Creates a lock-free ring buffer of `f32` samples with the given capacity.
    ///
    /// Returns a `(producer, consumer)` pair that can be moved into separate
    /// threads for wait-free, single-producer/single-consumer audio transfer.
    ///
    /// # Arguments
    ///
    /// * `size` - The number of `f32` samples the ring buffer can hold.
    pub fn create_ringbuffer(size: usize) -> (HeapProd<f32>, HeapCons<f32>) {
        let rb = HeapRb::<f32>::new(size);
        rb.split()
    }

    /// Replaces the current output device.
    ///
    /// # Arguments
    ///
    /// * `output_device` - The new CPAL output device.
    pub fn set_output_device(&mut self, output_device: Device) {
        self.output_device = output_device;
    }

    /// Replaces the current input device.
    ///
    /// # Arguments
    ///
    /// * `input_device` - The new CPAL input device.
    pub fn set_input_device(&mut self, input_device: Device) {
        self.input_device = input_device;
    }
}

impl AudioHandlerTrait for AudioHandler {
    /// Builds a CPAL input stream that forwards every captured sample into the
    /// provided ring-buffer producer.
    ///
    /// Samples that cannot be pushed (i.e. the ring buffer is full) are silently
    /// dropped. Input errors are added to logs as error.
    ///
    /// # Panics
    ///
    /// Panics if CPAL fails to build the input stream.
    fn build_input_stream(&self, mut producer: HeapProd<f32>) -> Box<dyn PlayableStream> {
        let stream = self
            .input_device
            .build_input_stream(
                &self.input_config,
                move |data: &[f32], _| {
                    for &s in data {
                        let _ = producer.try_push(s);
                    }
                },
                move |err| error!("Input error: {:?}", err),
                None,
            )
            .unwrap();
        Box::new(stream)
    }

    /// Builds a CPAL output stream that drains samples from the provided
    /// ring-buffer consumer into the hardware output buffer.
    ///
    /// Any output slot that has no corresponding sample is filled with `0.0`
    /// (silence). Output errors are printed to stderr.
    ///
    /// # Panics
    ///
    /// Panics if CPAL fails to build the output stream.
    fn build_output_stream(&self, mut consumer: HeapCons<f32>) -> Box<dyn PlayableStream> {
        let stream = self
            .output_device
            .build_output_stream(
                &self.output_config,
                move |out: &mut [f32], _| {
                    //println!("Output buffer: {:?}", &out[..10.min(out.len())]);
                    for o in out.iter_mut() {
                        *o = consumer.try_pop().unwrap_or(0.0);
                    }
                },
                move |err| eprintln!("Output error: {:?}", err),
                None,
            )
            .unwrap();
        Box::new(stream)
    }

    fn input_device(&self) -> &Device {
        &self.input_device
    }

    fn output_device(&self) -> &Device {
        &self.output_device
    }

    fn input_config(&self) -> &StreamConfig {
        &self.input_config
    }

    fn output_config(&self) -> &StreamConfig {
        &self.output_config
    }

    fn input_sample_rate(&self) -> u32 {
        self.input_sample_rate
    }

    fn output_sample_rate(&self) -> u32 {
        self.output_sample_rate
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[cfg(test)]
    mod success_path {
        use super::*;

        #[test]
        fn test_input_callback_pushes_samples() {
            let (mut prod, mut cons) = AudioHandler::create_ringbuffer(16);

            let input_data = vec![0.1, 0.2, 0.3, 0.4];
            {
                for &s in &input_data {
                    let _ = prod.try_push(s);
                }
            }

            for expected in input_data {
                assert_eq!(cons.try_pop(), Some(expected));
            }
        }
        #[test]
        fn test_output_callback_reads_samples() {
            let (mut prod, mut cons) = AudioHandler::create_ringbuffer(16);
            prod.try_push(10.0).unwrap();
            prod.try_push(20.0).unwrap();
            prod.try_push(30.0).unwrap();

            let mut out = [0.0f32; 5];
            {
                for o in out.iter_mut() {
                    *o = cons.try_pop().unwrap_or(0.0);
                }
            }

            assert_eq!(out, [10.0, 20.0, 30.0, 0.0, 0.0]);
        }
    }

    #[cfg(test)]
    mod failure_path {
        use super::*;

        #[test]
        fn test_input_callback_drops_when_full() {
            let (mut prod, mut cons) = AudioHandler::create_ringbuffer(3);

            let input_data = vec![1.0, 2.0, 3.0, 4.0, 5.0];

            for &s in &input_data {
                let _ = prod.try_push(s);
            }

            assert_eq!(cons.try_pop(), Some(1.0));
            assert_eq!(cons.try_pop(), Some(2.0));
            assert_eq!(cons.try_pop(), Some(3.0));
            assert_eq!(cons.try_pop(), None);
        }

        #[test]
        fn test_output_callback_zero_fills_when_empty() {
            let (_prod, mut cons) = AudioHandler::create_ringbuffer(8);

            let mut out = [1.0f32; 4];

            {
                for o in out.iter_mut() {
                    *o = cons.try_pop().unwrap_or(0.0);
                }
            }

            assert_eq!(out, [0.0, 0.0, 0.0, 0.0]);
        }
    }
}