Implement duplex mode API for ALSA (#32)

## Description

Implements the duplex mode API and for ALSA.

Needs #31 .

Closes #30.

## Type of Change

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code cleanup or refactor

## How Has This Been Tested?

Ran the `loopback` example.

## Checklist:

- [x] My code follows the style guidelines of this project
- [x] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] Wherever possible, I have added tests that prove my fix is
effective or that my feature works. For changes that need to be
validated manually (i.e. a new audio driver), use examples that can be
run to easily validate them.
- [ ] New and existing unit tests pass locally with my changes
- [x] I have checked my code and corrected any misspellings

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

@@ -19,7 +19,28 @@ where
 
     eprintln!("All devices");
     for device in driver.list_devices()? {
-        eprintln!("\t{} ({:?})", device.name(), device.device_type());
+        eprintln!("\t{}", device.name());
+    }
+    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(())
 }

flake.nix

@@ -26,7 +26,10 @@
         };
         devShells.default = pkgs.clangStdenv.mkDerivation {
           name = "interflow-devshell";
-          buildInputs = buildInputs ++ nativeBuildInputs;
+          buildInputs = buildInputs ++ nativeBuildInputs ++ (with pkgs; [pre-commit]);
+          shellHook = ''
+			  pre-commit install
+          '';
           inherit LIBCLANG_PATH;
         };
       }

src/backends/alsa/device.rs

@@ -1,8 +1,11 @@
-use crate::backends::alsa::stream::AlsaStream;
 use crate::backends::alsa::AlsaError;
 use crate::device::Channel;
 use crate::device::{AudioDevice, AudioInputDevice, AudioOutputDevice, DeviceType};
 use crate::stream::{AudioInputCallback, AudioOutputCallback, StreamConfig};
+use crate::{
+    backends::alsa::stream::AlsaStream, device::AudioDuplexDevice, duplex::AudioDuplexCallback,
+    SendEverywhereButOnWeb,
+};
 use alsa::{pcm, PCM};
 use std::borrow::Cow;
 use std::fmt;
@@ -16,6 +19,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")
@@ -32,13 +52,6 @@ impl AudioDevice for AlsaDevice {
         Cow::Borrowed(self.name.as_str())
     }
 
-    fn device_type(&self) -> DeviceType {
-        match self.direction {
-            alsa::Direction::Playback => DeviceType::Output,
-            alsa::Direction::Capture => DeviceType::Input,
-        }
-    }
-
     fn is_config_supported(&self, config: &StreamConfig) -> bool {
         self.get_hwp(config)
             .inspect_err(|err| {
@@ -110,8 +123,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,
@@ -124,11 +137,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)
@@ -146,13 +160,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
@@ -165,3 +190,77 @@ 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 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)?,
+        ))
+    }
+}