wip: duplex mode API + implementation for ALSA

Author: SolarLiner

examples/duplex.rs

@@ -4,6 +4,21 @@ use interflow::{duplex::DuplexStreamConfig, prelude::*};
 
 mod util;
 
+#[cfg(os_alsa)]
+fn main() -> Result<()> {
+    env_logger::init();
+
+    let device = default_duplex_device();
+    let mut config = device.default_duplex_config().unwrap();
+    config.buffer_size_range = (Some(128), Some(512));
+    let stream = device.create_duplex_stream(config, RingMod::new()).unwrap();
+    println!("Press Enter to stop");
+    std::io::stdin().read_line(&mut String::new())?;
+    stream.eject().unwrap();
+    Ok(())
+}
+
+#[cfg(not(os_alsa))]
 fn main() -> Result<()> {
     let input = default_input_device();
     let output = default_output_device();
@@ -39,8 +54,12 @@ impl AudioDuplexCallback for RingMod {
         input: AudioInput<f32>,
         mut output: AudioOutput<f32>,
     ) {
+        if input.buffer.num_samples() < output.buffer.num_samples() {
+            log::error!("Input underrun");
+        }
         let sr = context.stream_config.samplerate as f32;
-        for i in 0..output.buffer.num_samples() {
+        let num_samples = output.buffer.num_samples().min(input.buffer.num_samples());
+        for i in 0..num_samples {
             let inp = input.buffer.get_frame(i)[0];
             let c = self.carrier.next_sample(sr);
             output.buffer.set_mono(i, inp * c);

examples/enumerate_alsa.rs

@@ -1,3 +1,5 @@
+use crate::util::enumerate::enumerate_duplex_devices;
+
 mod util;
 
 #[cfg(os_alsa)]
@@ -7,7 +9,9 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     env_logger::init();
 
-    enumerate_devices(AlsaDriver)
+    enumerate_devices(AlsaDriver)?;
+    enumerate_duplex_devices(AlsaDriver)?;
+    Ok(())
 }
 
 #[cfg(not(os_alsa))]

examples/loopback.rs

@@ -6,6 +6,23 @@ use std::sync::Arc;
 
 mod util;
 
+#[cfg(os_alsa)]
+fn main() -> Result<()> {
+    env_logger::init();
+
+    let device = default_duplex_device();
+    let mut config = device.default_duplex_config().unwrap();
+    config.buffer_size_range = (Some(128), Some(512));
+    let value = Arc::new(AtomicF32::new(0.0));
+    let stream = device
+        .create_duplex_stream(config, Loopback::new(44100., value.clone()))
+        .unwrap();
+    util::display_peakmeter(value)?;
+    stream.eject().unwrap();
+    Ok(())
+}
+
+#[cfg(not(os_alsa))]
 fn main() -> Result<()> {
     env_logger::init();
 

examples/util/enumerate.rs

@@ -23,3 +23,24 @@ where
     }
     Ok(())
 }
+
+pub fn enumerate_duplex_devices<Driver: AudioDuplexDriver>(
+    driver: Driver,
+) -> Result<(), Box<dyn Error>>
+where
+    <Driver as AudioDriver>::Error: 'static,
+{
+    eprintln!("Driver name   : {}", Driver::DISPLAY_NAME);
+    eprintln!("Driver version: {}", driver.version()?);
+    if let Some(device) = driver.default_duplex_device()? {
+        eprintln!("Default duplex device: {}", device.name());
+    } else {
+        eprintln!("No default duplex device");
+    }
+
+    eprintln!("All duplex devices");
+    for device in driver.list_duplex_devices()? {
+        eprintln!("\t{}", device.name());
+    }
+    Ok(())
+}

src/backends/alsa/device.rs

@@ -1,4 +1,4 @@
-use crate::backends::alsa::stream::AlsaStream;
+use crate::{backends::alsa::stream::AlsaStream, device::AudioDuplexDevice, duplex::AudioDuplexCallback, SendEverywhereButOnWeb};
 use crate::backends::alsa::AlsaError;
 use crate::device::Channel;
 use crate::device::{AudioDevice, AudioInputDevice, AudioOutputDevice, DeviceType};
@@ -16,6 +16,23 @@ pub struct AlsaDevice {
     pub(super) direction: alsa::Direction,
 }
 
+impl AlsaDevice {
+    fn channel_map(&self, requested_direction: alsa::Direction) -> impl Iterator<Item = Channel> {
+        let max_channels = if self.direction == requested_direction {
+            self.pcm
+                .hw_params_current()
+                .and_then(|hwp| hwp.get_channels_max())
+                .unwrap_or(0)
+        } else {
+            0
+        };
+        (0..max_channels as usize).map(|i| Channel {
+            index: i,
+            name: Cow::Owned(format!("Channel {}", i)),
+        })
+    }
+}
+
 impl fmt::Debug for AlsaDevice {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("AlsaDevice")
@@ -111,8 +128,8 @@ impl AlsaDevice {
     }
 
     pub(super) fn new(name: &str, direction: alsa::Direction) -> Result<Self, alsa::Error> {
-        let pcm = PCM::new(name, direction, true)?;
-        let pcm = Rc::new(pcm);
+        log::info!("Opening device: {name}, direction {direction:?}");
+        let pcm = Rc::new(PCM::new(name, direction, true)?);
         Ok(Self {
             name: name.to_string(),
             direction,
@@ -125,11 +142,12 @@ impl AlsaDevice {
         hwp.set_channels(config.channels as _)?;
         hwp.set_rate(config.samplerate as _, alsa::ValueOr::Nearest)?;
         if let Some(min) = config.buffer_size_range.0 {
-            hwp.set_buffer_size_min(min as _)?;
+            hwp.set_buffer_size_min(min as pcm::Frames * 2)?;
         }
         if let Some(max) = config.buffer_size_range.1 {
-            hwp.set_buffer_size_max(max as _)?;
+            hwp.set_buffer_size_max(max as pcm::Frames * 2)?;
         }
+        hwp.set_periods(2, alsa::ValueOr::Nearest)?;
         hwp.set_format(pcm::Format::float())?;
         hwp.set_access(pcm::Access::RWInterleaved)?;
         Ok(hwp)
@@ -147,13 +165,24 @@ impl AlsaDevice {
 
         log::debug!("Apply config: hwp {hwp:#?}");
 
+        swp.set_avail_min(hwp.get_period_size()?)?;
         swp.set_start_threshold(hwp.get_buffer_size()?)?;
         self.pcm.sw_params(&swp)?;
         log::debug!("Apply config: swp {swp:#?}");
 
         Ok((hwp, swp, io))
     }
 
+    pub(super) fn ensure_state(&self, hwp: &pcm::HwParams) -> Result<bool, AlsaError> {
+        match self.pcm.state() {
+            pcm::State::Suspended if hwp.can_resume() => self.pcm.resume()?,
+            pcm::State::Suspended => self.pcm.prepare()?,
+            pcm::State::Paused => return Ok(true),
+            _ => {}
+        }
+        Ok(false)
+    }
+
     fn default_config(&self) -> Result<StreamConfig, AlsaError> {
         let samplerate = 48e3; // Default ALSA sample rate
         let channel_count = 2; // Stereo stream
@@ -166,3 +195,81 @@ impl AlsaDevice {
         })
     }
 }
+
+pub struct AlsaDuplexDevice {
+    pub(super) input: AlsaDevice,
+    pub(super) output: AlsaDevice,
+}
+
+impl AudioDevice for AlsaDuplexDevice {
+    type Error = AlsaError;
+
+    fn name(&self) -> Cow<str> {
+        Cow::Owned(format!("{} / {}", self.input.name(), self.output.name()))
+    }
+
+    fn device_type(&self) -> DeviceType {
+        DeviceType::Duplex
+    }
+
+    fn is_config_supported(&self, config: &StreamConfig) -> bool {
+        let Ok((hwp, _, _)) = self.output.apply_config(config) else {
+            return false;
+        };
+        let Ok(period) = hwp.get_period_size() else {
+            return false;
+        };
+        let period = period as usize;
+        self.input
+            .apply_config(&StreamConfig {
+                buffer_size_range: (Some(period), Some(period)),
+                ..*config
+            })
+            .is_ok()
+    }
+
+    fn enumerate_configurations(&self) -> Option<impl IntoIterator<Item = StreamConfig>> {
+        Some(
+            self.output
+                .enumerate_configurations()?
+                .into_iter()
+                .filter(|config| self.is_config_supported(config)),
+        )
+    }
+}
+
+impl AudioDuplexDevice for AlsaDuplexDevice {
+    type StreamHandle<Callback: AudioDuplexCallback> = AlsaStream<Callback>;
+
+    fn default_duplex_config(&self) -> Result<StreamConfig, Self::Error> {
+        self.output.default_output_config()
+    }
+
+    fn create_duplex_stream<Callback: SendEverywhereButOnWeb + AudioDuplexCallback>(
+        &self,
+        config: StreamConfig,
+        callback: Callback,
+    ) -> Result<<Self as AudioDuplexDevice>::StreamHandle<Callback>, Self::Error> {
+        AlsaStream::new_duplex(
+            config,
+            self.input.name.clone(),
+            self.output.name.clone(),
+            callback,
+        )
+    }
+}
+
+impl AlsaDuplexDevice {
+    /// Create a new duplex device from an input and output device.
+    pub fn new(input: AlsaDevice, output: AlsaDevice) -> Self {
+        Self { input, output }
+    }
+
+    /// Create a full-duplex device from the given name.
+    pub fn full_duplex(name: &str) -> Result<Self, AlsaError> {
+        Ok(Self::new(
+            AlsaDevice::new(name, alsa::Direction::Capture)?,
+            AlsaDevice::new(name, alsa::Direction::Playback)?,
+        ))
+    }
+}