PlatformProxy for P
+where
+ <<::Device as Device>::StreamHandle as stream::StreamHandle<
+ DynCallback,
+ >>::Error: Sync,
+{
+ fn name(&self) -> Cow<'static, str> {
+ Cow::Borrowed(P::NAME)
+ }
+
+ fn list_devices(&self) -> Result> {
+ Ok(Vec::from_iter(
+ Platform::list_devices(self)?
+ .into_iter()
+ .map(|dev| Rc::new(dev) as DynDevice),
+ ))
+ }
+
+ fn list_devices_matching(&self, device_type: DeviceType) -> Result> {
+ Ok(Vec::from_iter(
+ Platform::list_devices(self)?.into_iter().filter_map(|dev| {
+ Device::device_type(&dev)
+ .contains(device_type)
+ .then(|| Rc::new(dev) as DynDevice)
+ }),
+ ))
+ }
+
+ fn default_device(&self, device_type: DeviceType) -> Result {
+ let device = Platform::default_device(self, device_type)?;
+ Ok(Rc::new(device) as DynDevice)
+ }
+}
+
+/// Trait for device proxies that describes audio device capabilities.
+pub trait DeviceProxy: ExtensionProvider {
+ fn name(&self) -> Cow<'_, str>;
+ fn device_type(&self) -> DeviceType;
+ fn default_config(&self) -> Result;
+ fn is_config_supported(&self, config: &StreamConfig) -> bool;
+ fn buffer_size_range(&self) -> Result<(Option, Option)>;
+ fn create_stream_raw(
+ &self,
+ config: StreamConfig,
+ callback: DynCallback,
+ ) -> Result;
+
+ fn create_default_stream_raw(
+ &self,
+ requested_type: DeviceType,
+ callback: DynCallback,
+ ) -> Result;
+}
+
+impl DeviceProxy for D
+where
+ as stream::StreamHandle>::Error: Sync,
+{
+ #[inline]
+ fn name(&self) -> Cow<'_, str> {
+ Device::name(self)
+ }
+
+ fn device_type(&self) -> DeviceType {
+ Device::device_type(self)
+ }
+
+ fn default_config(&self) -> Result {
+ Ok(Device::default_config(self)?)
+ }
+
+ fn is_config_supported(&self, config: &StreamConfig) -> bool {
+ Device::is_config_supported(self, config)
+ }
+
+ fn buffer_size_range(&self) -> Result<(Option, Option)> {
+ Ok(Device::buffer_size_range(self)?)
+ }
+
+ fn create_stream_raw(
+ &self,
+ config: StreamConfig,
+ callback: DynCallback,
+ ) -> Result {
+ let handle = Device::create_stream(self, config, callback).context("Cannot open stream")?;
+ Ok(RawStreamHandle::from_handle(handle))
+ }
+
+ fn create_default_stream_raw(
+ &self,
+ requested_type: DeviceType,
+ callback: DynCallback,
+ ) -> Result {
+ let handle =
+ Device::default_stream(self, requested_type, callback).context("Cannot open stream")?;
+ Ok(RawStreamHandle::from_handle(handle))
+ }
+}
+
+pub struct DynCallback {
+ type_id: TypeId,
+ handle: NonNull<()>,
+ prepare: unsafe fn(NonNull<()>, CallbackContext),
+ process_audio: unsafe fn(NonNull<()>, CallbackContext, &AudioInput, &mut AudioOutput),
+}
+
+unsafe impl Send for DynCallback {}
+
+impl DynCallback {
+ pub fn from_callback(callback: Callback) -> Self {
+ Self {
+ type_id: TypeId::of::(),
+ handle: unsafe { create_nonnull(callback).cast() },
+ prepare: |handle, context| {
+ unsafe { handle.cast::().as_mut() }.prepare(context)
+ },
+ process_audio: |handle, context, input, output| {
+ unsafe { handle.cast::().as_mut() }.process_audio(context, input, output)
+ },
+ }
+ }
+
+ pub fn into_inner(self) -> Option> {
+ if self.type_id != TypeId::of::() {
+ return None;
+ }
+ Some(unsafe { consume_nonnull(self.handle.cast::()) })
+ }
+}
+
+impl stream::Callback for DynCallback {
+ fn prepare(&mut self, context: CallbackContext) {
+ unsafe { (self.prepare)(self.handle, context) }
+ }
+
+ fn process_audio(
+ &mut self,
+ context: CallbackContext,
+ input: &AudioInput,
+ output: &mut AudioOutput,
+ ) {
+ unsafe { (self.process_audio)(self.handle, context, input, output) }
+ }
+}
+
+pub struct RawStreamHandle {
+ handle: Option>,
+ eject: unsafe fn(NonNull<()>) -> Result,
+ drop: unsafe fn(NonNull<()>),
+}
+
+impl RawStreamHandle {
+ pub fn from_handle>(
+ handle: Handle,
+ ) -> Self {
+ let handle = Box::into_raw(Box::new(handle));
+ let handle = Some(unsafe { NonNull::new_unchecked(handle).cast() });
+ Self {
+ handle,
+ eject: |ptr| {
+ let handle = unsafe { Box::from_raw(ptr.cast::().as_ptr()) };
+ Ok(handle.eject()?)
+ },
+ drop: |ptr| {
+ let _ = unsafe { Box::from_raw(ptr.cast::().as_ptr()) };
+ },
+ }
+ }
+
+ pub fn eject(mut self) -> Result {
+ let Some(handle) = self.handle.take() else {
+ anyhow::bail!("Stream already ejected");
+ };
+ Ok(unsafe { (self.eject)(handle)? })
+ }
+}
+
+impl Drop for RawStreamHandle {
+ fn drop(&mut self) {
+ let Some(handle) = self.handle.take() else {
+ return;
+ };
+ unsafe { (self.drop)(handle) };
+ }
+}
+
+pub trait CreateStreamExt: DeviceProxy {
+ fn create_stream(
+ &self,
+ config: StreamConfig,
+ callback: Callback,
+ ) -> Result> {
+ let callback = DynCallback::from_callback(callback);
+ let handle = self.create_stream_raw(config, callback)?;
+ Ok(unsafe { StreamHandle::from_raw(handle) })
+ }
+
+ fn default_stream(
+ &self,
+ requested_type: DeviceType,
+ callback: Callback,
+ ) -> Result> {
+ let callback = DynCallback::from_callback(callback);
+ let handle = self.create_default_stream_raw(requested_type, callback)?;
+ Ok(unsafe { StreamHandle::from_raw(handle) })
+ }
+}
+
+impl CreateStreamExt for C {}
+
+#[derive(Debug, thiserror::Error)]
+#[error(transparent)]
+pub struct AnyError(anyhow::Error);
+
+pub struct StreamHandle {
+ __callback: PhantomData,
+ raw: RawStreamHandle,
+}
+
+impl StreamHandle {
+ unsafe fn from_raw(raw: RawStreamHandle) -> Self {
+ Self {
+ __callback: PhantomData,
+ raw,
+ }
+ }
+}
+
+impl stream::StreamHandle>
+ for StreamHandle
+{
+ type Error = AnyError;
+
+ fn eject(self) -> std::result::Result, Self::Error> {
+ (|| -> Result> {
+ let raw_callback = self.raw.eject()?;
+ Ok(raw_callback.into_inner().unwrap())
+ })()
+ .map_err(AnyError)
+ }
+}
+
+unsafe fn create_nonnull(value: T) -> NonNull {
+ NonNull::new_unchecked(Box::into_raw(Box::new(value)))
+}
+
+unsafe fn consume_nonnull(handle: NonNull) -> Box {
+ Box::from_raw(handle.as_ptr())
+}
diff --git a/crates/interflow-core/src/stream.rs b/crates/interflow-core/src/stream.rs
new file mode 100644
index 0000000..727cb59
--- /dev/null
+++ b/crates/interflow-core/src/stream.rs
@@ -0,0 +1,111 @@
+use crate::buffer::{AudioMut, AudioRef};
+use crate::device::ResolvedStreamConfig;
+use crate::stream;
+use crate::timing::Timestamp;
+use crate::traits::ExtensionProvider;
+use bitflags::bitflags;
+
+pub trait StreamProxy: Send + Sync + ExtensionProvider {}
+
+bitflags! {
+ pub struct ChannelFlags: u32 {
+ const INACTIVE = 0x0000_0001;
+ const UNDERFLOW = 0x0000_0002;
+ const OVERFLOW = 0x0000_0004;
+ }
+}
+
+pub trait StreamLatency {
+ fn input_latency(&self, channel: usize) -> Option;
+ fn output_latency(&self, channel: usize) -> Option;
+}
+
+#[duplicate::duplicate_item(
+ name bufty derive;
+ [AudioInput] [AudioRef < 'a, T >] [derive(Copy, Clone)];
+ [AudioOutput] [AudioMut < 'a, T >] [derive()];
+)]
+/// Plain-old-data object holding references to the audio buffer and the associated time-keeping
+/// [`Timestamp`]. This timestamp is associated with the stream, and in the cases where the
+/// driver provides timing information, it is used instead of relying on sample-counting.
+#[derive]
+pub struct name<'a, T> {
+ /// Associated time stamp for this callback. The time represents the duration for which the
+ /// stream has been opened, and is either provided by the driver if available, or is kept up
+ /// manually by the library.
+ pub timestamp: Timestamp,
+ /// Audio buffer data.
+ pub buffer: bufty,
+ pub channel_flags: &'a [ChannelFlags],
+}
+
+/// Plain-old-data object holding the passed-in stream configuration, as well as a general
+/// callback timestamp, which can be different from the input and output streams in case of
+/// cross-stream latencies; differences in timing can indicate desync.
+#[derive(Copy, Clone)]
+pub struct CallbackContext<'a> {
+ /// Passed-in stream configuration. Values have been updated where necessary to correspond to
+ /// the actual stream properties.
+ pub stream_config: &'a ResolvedStreamConfig,
+ /// Callback-wide timestamp.
+ pub timestamp: Timestamp,
+ pub stream_proxy: &'a dyn StreamProxy,
+}
+
+/// Trait for types which handles an audio stream (input or output).
+pub trait StreamHandle {
+ /// Type of errors which have caused the stream to fail.
+ type Error: Send + std::error::Error;
+
+ /// Eject the stream, returning ownership of the callback.
+ ///
+ /// An error can occur when an irrecoverable error has occured and ownership has been lost
+ /// already.
+ fn eject(self) -> Result;
+}
+
+/// Trait of types which process audio data. This is the trait that users will want to
+/// implement when processing audio from a device.
+pub trait Callback: Send {
+ /// Prepare the audio callback to process audio. This function is *not* real-time safe (i.e., allocations can be
+ /// performed), in preparation for processing the stream with [`Self::process_audio`].
+ fn prepare(&mut self, context: CallbackContext);
+
+ /// Callback called when audio data can be processed.
+ fn process_audio(
+ &mut self,
+ context: CallbackContext,
+ input: &AudioInput,
+ output: &mut AudioOutput,
+ );
+}
+
+impl Callback for Box {
+ fn prepare(&mut self, context: CallbackContext) {
+ Callback::prepare(&mut **self, context)
+ }
+
+ fn process_audio(
+ &mut self,
+ context: CallbackContext,
+ input: &AudioInput,
+ output: &mut AudioOutput,
+ ) {
+ Callback::process_audio(&mut **self, context, input, output)
+ }
+}
+
+impl, &mut AudioOutput)> Callback
+ for &mut F
+{
+ fn prepare(&mut self, _: CallbackContext) {}
+
+ fn process_audio(
+ &mut self,
+ context: CallbackContext,
+ input: &AudioInput,
+ output: &mut AudioOutput,
+ ) {
+ (self)(context, input, output);
+ }
+}
diff --git a/src/timestamp.rs b/crates/interflow-core/src/timing.rs
similarity index 68%
rename from src/timestamp.rs
rename to crates/interflow-core/src/timing.rs
index 3831dc0..7e378de 100644
--- a/src/timestamp.rs
+++ b/crates/interflow-core/src/timing.rs
@@ -1,31 +1,6 @@
-//! Module for handling timestamp and duration calculations in audio processing.
-//!
-//! This module provides the [`Timestamp`] type, which manages time-related operations
-//! for audio streams by tracking sample counts and their corresponding durations based
-//! on a specified sample rate. It supports basic arithmetic operations for sample
-//! counting and duration calculations, making it useful for audio stream synchronization
-//! and timing operations.
-//!
-//! # Examples
-//!
-//! ```rust
-//! use std::time::Duration;
-//! use interflow::timestamp::Timestamp;
-//!
-//! // Create a timestamp for 48 kHz audio
-//! let mut ts = Timestamp::new(48000.);
-//!
-//! // Add 48 samples (1ms at 48kHz)
-//! ts += 48;
-//! assert_eq!(ts.as_duration(), Duration::from_millis(1));
-//!
-//! // Convert a duration to samples
-//! let ts2 = Timestamp::from_duration(48000., Duration::from_millis(100));
-//! assert_eq!(ts2.counter, 4800);
-//! ```
-
use std::ops;
use std::ops::AddAssign;
+use std::sync::atomic::AtomicU64;
use std::time::Duration;
/// Timestamp value, which computes duration information from a provided samplerate and a running
@@ -140,13 +115,68 @@ impl Timestamp {
}
}
+ /// Compute the number of seconds represented in this [`Timestamp`].
+ pub fn as_seconds(&self) -> f64 {
+ self.counter as f64 / self.samplerate
+ }
+
/// Compute the duration represented by this [`Timestamp`].
pub fn as_duration(&self) -> Duration {
Duration::from_secs_f64(self.as_seconds())
}
+}
- /// Compute the number of seconds represented in this [`Timestamp`].
- pub fn as_seconds(&self) -> f64 {
- self.counter as f64 / self.samplerate
+/// Atomic version of [`Timestamp`] to be shared between threads. Mainly used by the [`crate::duplex`] module, but
+/// may be useful in user code as well.
+pub struct AtomicTimestamp {
+ samplerate: AtomicU64,
+ counter: AtomicU64,
+}
+
+impl AtomicTimestamp {
+ /// Create a new [`AtomicTimestamp`] with zeroed values.
+ pub fn zeroed() -> Self {
+ Self {
+ samplerate: AtomicU64::new(0),
+ counter: AtomicU64::new(0),
+ }
+ }
+ /// Update the contents with the provided [`Timestamp`].
+ pub fn update(&self, ts: Timestamp) {
+ self.samplerate.store(
+ ts.samplerate.to_bits(),
+ std::sync::atomic::Ordering::Relaxed,
+ );
+ self.counter
+ .store(ts.counter, std::sync::atomic::Ordering::Relaxed);
+ }
+
+ /// Load values and return them as a [`Timestamp`].
+ pub fn as_timestamp(&self) -> Timestamp {
+ Timestamp {
+ samplerate: f64::from_bits(self.samplerate.load(std::sync::atomic::Ordering::Relaxed)),
+ counter: self.counter.load(std::sync::atomic::Ordering::Relaxed),
+ }
+ }
+
+ /// Add the provided number of frames to this.
+ pub fn add_frames(&self, frames: u64) {
+ self.counter
+ .fetch_add(frames, std::sync::atomic::Ordering::Relaxed);
+ }
+}
+
+impl From for AtomicTimestamp {
+ fn from(value: Timestamp) -> Self {
+ Self {
+ samplerate: AtomicU64::new(value.samplerate.to_bits()),
+ counter: AtomicU64::new(value.counter),
+ }
+ }
+}
+
+impl From for Timestamp {
+ fn from(value: AtomicTimestamp) -> Self {
+ value.as_timestamp()
}
}
diff --git a/crates/interflow-core/src/traits.rs b/crates/interflow-core/src/traits.rs
new file mode 100644
index 0000000..7b866c3
--- /dev/null
+++ b/crates/interflow-core/src/traits.rs
@@ -0,0 +1,206 @@
+use crate::dyn_compatible;
+use std::any::TypeId;
+use std::marker::PhantomData;
+use std::mem::MaybeUninit;
+
+/// A fully type-erased pointer that can work with both thin and fat pointers.
+/// Copied from .
+#[derive(Copy, Clone)]
+struct ErasedPtr {
+ value: MaybeUninit<[usize; 2]>,
+}
+
+impl ErasedPtr {
+ /// Erase `ptr`.
+ fn new(ptr: *const T) -> Self {
+ let mut res = ErasedPtr {
+ value: MaybeUninit::zeroed(),
+ };
+
+ let len = size_of::<*const T>();
+ assert!(len <= size_of::<[usize; 2]>());
+
+ let ptr_val = (&ptr) as *const *const T as *const u8;
+ let target = res.value.as_mut_ptr() as *mut u8;
+ // SAFETY: The target is valid for at least `len` bytes, and has no
+ // requirements on the value.
+ unsafe {
+ core::ptr::copy_nonoverlapping(ptr_val, target, len);
+ }
+
+ res
+ }
+
+ /// Erase `ptr`.
+ fn new_mut(ptr: *mut T) -> Self {
+ let mut res = ErasedPtr {
+ value: MaybeUninit::zeroed(),
+ };
+
+ let len = size_of::<*const T>();
+ assert!(len <= size_of::<[usize; 2]>());
+
+ let ptr_val = (&ptr) as *const *mut T as *mut u8;
+ let target = res.value.as_mut_ptr() as *mut u8;
+ // SAFETY: The target is valid for at least `len` bytes, and has no
+ // requirements on the value.
+ unsafe {
+ core::ptr::copy_nonoverlapping(ptr_val, target, len);
+ }
+
+ res
+ }
+
+ /// Convert the type erased pointer back into a pointer.
+ ///
+ /// # Safety
+ ///
+ /// The type `T` must be the same type as the one used with `new` or `new_mut`.
+ unsafe fn as_ptr(&self) -> *const T {
+ // SAFETY: The constructor ensures that the first `size_of::()`
+ // bytes of `&self.value` are a valid `*const T` pointer.
+ unsafe { core::mem::transmute_copy(&self.value) }
+ }
+
+ /// Convert the type erased pointer back into a pointer.
+ ///
+ /// # Safety
+ ///
+ /// This [`ErasedPtr`] must have been constructed from [`Selector::new_mut`], using the same `T` type used in
+ /// this function.
+ unsafe fn as_mut_ptr(&self) -> *mut T {
+ // SAFETY: The constructor ensures that the first `size_of::()`
+ // bytes of `&self.value` are a valid `*const T` pointer.
+ // The caller ensures that the original pointer was mutable.
+ unsafe { core::mem::transmute_copy(&self.value) }
+ }
+}
+
+/// Type that can dynamically retrieve a type registered by an [`ExtensionProvider`] object.
+/// Types are queried on-demand, every time an extension is requested.
+///
+/// Consumers of [`ExtensionProvider`] should instead use [`ExtensionExt::lookup`].
+pub struct Selector<'a> {
+ __lifetime: PhantomData<&'a ()>,
+ target: TypeId,
+ found: Option,
+ found_mut: Option,
+}
+
+impl<'a> Selector<'a> {
+ pub(crate) const fn new() -> Self {
+ Self {
+ __lifetime: PhantomData,
+ target: TypeId::of::(),
+ found: None,
+ found_mut: None,
+ }
+ }
+
+ pub fn register(&mut self, value: &T) -> &mut Self {
+ if self.target == TypeId::of::() {
+ self.found = Some(ErasedPtr::new(value));
+ }
+ self
+ }
+
+ pub fn register_mut(&mut self, value: &mut T) -> &mut Self {
+ if self.target == TypeId::of::() {
+ self.found_mut = Some(ErasedPtr::new_mut(value));
+ }
+ self
+ }
+
+ pub(crate) fn finish(self) -> Option<&'a I> {
+ assert_eq!(self.target, TypeId::of::());
+ let target = self.found.or(self.found_mut)?;
+ Some(unsafe { &*target.as_ptr() })
+ }
+
+ pub(crate) fn finish_mut(self) -> Option<&'a mut I> {
+ assert_eq!(self.target, TypeId::of::());
+ Some(unsafe { &mut *self.found_mut?.as_mut_ptr() })
+ }
+}
+
+/// Trait for types that have optional data available on-demand.
+pub trait ExtensionProvider: 'static {
+ /// Register on-demand types. Note that the implementation details around registering extensions mean that this
+ /// function will be called for every request. Runtime checks are expected, but this function should remain as fast
+ /// as possible.
+ fn register<'a, 'sel>(&'a self, selector: &'sel mut Selector<'a>) -> &'sel mut Selector<'a>;
+ fn register_mut<'a, 'sel>(
+ &'a mut self,
+ selector: &'sel mut Selector<'a>,
+ ) -> &'sel mut Selector<'a> {
+ self.register(selector)
+ }
+}
+
+dyn_compatible!(ExtensionProvider);
+
+pub trait ExtensionProviderExt: ExtensionProvider {
+ /// Look up [`T`] from the extension if it is registered. Returns an immutable reference.
+ fn lookup(&self) -> Option<&T>;
+
+ /// Look up [`T`] from the extension if it is registered. Returns as mutable reference.
+ fn lookup_mut(&mut self) -> Option<&mut T>;
+}
+
+impl ExtensionProviderExt for E {
+ fn lookup(&self) -> Option<&T> {
+ let mut selector = Selector::new::();
+ selector.register::(self);
+ self.register(&mut selector);
+ selector.finish::()
+ }
+
+ fn lookup_mut(&mut self) -> Option<&mut T> {
+ let mut selector = Selector::new::();
+ selector.register_mut::(self);
+ self.register_mut(&mut selector);
+ selector.finish_mut::()
+ }
+}
+
+pub trait Enumerator {
+ type Item;
+ fn len(&self) -> usize;
+ fn at(&self, index: usize) -> Option;
+}
+
+impl<'a, T> Enumerator for &'a dyn Enumerator- {
+ type Item = T;
+
+ fn len(&self) -> usize {
+ (*self).len()
+ }
+
+ fn at(&self, index: usize) -> Option {
+ (*self).at(index)
+ }
+}
+
+pub struct Enumerate {
+ enumerator: E,
+ i: usize,
+}
+
+impl Iterator for Enumerate {
+ type Item = E::Item;
+
+ fn next(&mut self) -> Option {
+ let result = self.enumerator.at(self.i);
+ self.i += 1;
+ result
+ }
+
+ fn size_hint(&self) -> (usize, Option) {
+ let len = self.enumerator.len();
+ (len, Some(len))
+ }
+}
+
+pub fn enumerate(enumerator: &dyn Enumerator
- ) -> impl use<'_, T> + Iterator
- {
+ Enumerate { enumerator, i: 0 }
+}
diff --git a/crates/interflow-coreaudio/Cargo.toml b/crates/interflow-coreaudio/Cargo.toml
new file mode 100644
index 0000000..125a0d7
--- /dev/null
+++ b/crates/interflow-coreaudio/Cargo.toml
@@ -0,0 +1,19 @@
+[package]
+name = "interflow-coreaudio"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+
+[dependencies]
+interflow-core.workspace = true
+
+coreaudio-rs = "0.14.1"
+coreaudio-sys = "0.2.17"
+log.workspace = true
+thiserror.workspace = true
+oneshot = { version = "0.2.1", features = ["std"] }
+scattered-collect = { workspace = true, optional = true }
+
+[features]
+collect = ["interflow-core/collect", "dep:scattered-collect"]
diff --git a/crates/interflow-coreaudio/src/device.rs b/crates/interflow-coreaudio/src/device.rs
new file mode 100644
index 0000000..5cf71ee
--- /dev/null
+++ b/crates/interflow-coreaudio/src/device.rs
@@ -0,0 +1,316 @@
+//! CoreAudio device implementation
+use crate::stream::Handle;
+use crate::{utils, Error};
+use coreaudio::audio_unit::macos_helpers::{
+ get_audio_device_supports_scope, get_default_device_id, get_device_name,
+};
+use coreaudio::audio_unit::{AudioUnit, Element, IOType, Scope};
+use coreaudio_sys::{
+ kAudioDevicePropertyBufferFrameSizeRange, kAudioHardwarePropertyDefaultInputDevice,
+ kAudioHardwarePropertyDefaultOutputDevice, kAudioHardwarePropertyDefaultSystemOutputDevice,
+ kAudioObjectPropertyElementMaster, kAudioObjectPropertyScopeGlobal,
+ kAudioObjectPropertyScopeInput, kAudioObjectPropertyScopeOutput, kAudioObjectSystemObject,
+ kAudioOutputUnitProperty_CurrentDevice, kAudioOutputUnitProperty_EnableIO, AudioDeviceID,
+ AudioObjectGetPropertyData, AudioObjectID, AudioObjectPropertyAddress, AudioValueRange,
+};
+use interflow_core::device::StreamConfig;
+use interflow_core::traits::{ExtensionProvider, Selector};
+use interflow_core::{device, stream, DeviceType};
+use std::borrow::Cow;
+use std::cell::OnceCell;
+use std::ffi::c_uint;
+use std::ptr::null;
+
+/// Audio device request type. CoreAudio has separate code paths for getting the default device, and getting a
+/// specific ID. Furthermore, "default device"s are automatically re-routed when the user changes the default device.
+#[derive(Debug, Default, Copy, Clone)]
+pub enum DeviceRequest {
+ /// Automatically connects to the default output device, generic stream.
+ #[default]
+ DefaultOutput,
+ /// Automatically connects to the default output device, notification stream.
+ SystemOutput,
+ /// Automatically connects to the default input device.
+ DefaultInput,
+ /// Automatically connects to the default voice input device, using the system's voice enhancement processing if
+ /// enabled.
+ VoiceInput,
+ /// Connect to this specific output.
+ Specific(AudioDeviceID),
+}
+
+impl DeviceRequest {
+ pub(crate) fn to_audio_unit(&self) -> Result {
+ match self {
+ Self::DefaultOutput => Ok(AudioUnit::new(IOType::DefaultOutput)?),
+ Self::SystemOutput => Ok(AudioUnit::new(IOType::SystemOutput)?),
+ Self::VoiceInput => Ok(AudioUnit::new(IOType::VoiceProcessingIO)?),
+ Self::DefaultInput => {
+ let mut unit = AudioUnit::new_uninitialized(IOType::HalOutput)?;
+ unit.set_property(
+ kAudioOutputUnitProperty_EnableIO,
+ Scope::Input,
+ Element::Input,
+ Some(&1u32),
+ )?;
+ unit.set_property(
+ kAudioOutputUnitProperty_EnableIO,
+ Scope::Output,
+ Element::Output,
+ Some(&0u32),
+ )?;
+ let input_device_id = get_default_device_id(true).unwrap_or(0);
+ unit.set_property(
+ kAudioOutputUnitProperty_CurrentDevice,
+ Scope::Global,
+ Element::Output,
+ Some(&input_device_id),
+ )?;
+ unit.initialize()?;
+ Ok(unit)
+ }
+ &Self::Specific(id) => {
+ let mut unit = AudioUnit::new_uninitialized(IOType::HalOutput)?;
+ let device_type = self.device_type();
+ let value = if device_type.is_input() { 1u32 } else { 0 };
+ unit.set_property(
+ kAudioOutputUnitProperty_EnableIO,
+ Scope::Input,
+ Element::Input,
+ Some(&value),
+ )?;
+ let value = if device_type.is_output() { 1u32 } else { 0 };
+ unit.set_property(
+ kAudioOutputUnitProperty_EnableIO,
+ Scope::Output,
+ Element::Output,
+ Some(&value),
+ )?;
+ unit.set_property(
+ kAudioOutputUnitProperty_CurrentDevice,
+ Scope::Global,
+ Element::Output,
+ Some(&id),
+ )?;
+ unit.initialize()?;
+ Ok(unit)
+ }
+ }
+ }
+
+ /// Return the name associated with this [`DeviceRequest`]. This is a descriptive name, and does not uniquely
+ /// identify the device.
+ pub fn name(&self) -> Cow<'_, str> {
+ let (selector, fallback) = match self {
+ Self::DefaultOutput => (kAudioHardwarePropertyDefaultOutputDevice, "Default Output"),
+ Self::DefaultInput => (kAudioHardwarePropertyDefaultInputDevice, "Default Input"),
+ Self::SystemOutput => (
+ kAudioHardwarePropertyDefaultSystemOutputDevice,
+ "Notifications output",
+ ),
+ Self::VoiceInput => (c_uint::MAX, "Voice Input"),
+ &Self::Specific(id) => {
+ return match get_device_name(id) {
+ Ok(s) => Cow::Owned(s),
+ Err(err) => {
+ log::error!("Cannot get device name for ID: {}: {err}", id);
+ Cow::Borrowed("")
+ }
+ }
+ }
+ };
+
+ (|| {
+ let property_address = AudioObjectPropertyAddress {
+ mSelector: selector,
+ mScope: kAudioObjectPropertyScopeGlobal,
+ mElement: kAudioObjectPropertyElementMaster,
+ };
+
+ let mut audio_device_id: AudioDeviceID = 0;
+ let mut data_size = size_of::() as u32;
+ let status = unsafe {
+ AudioObjectGetPropertyData(
+ kAudioObjectSystemObject as AudioObjectID,
+ &property_address,
+ 0,
+ null(),
+ &mut data_size,
+ (&raw mut audio_device_id).cast(),
+ )
+ };
+ coreaudio::Error::from_os_status(status)?;
+ let name = get_device_name(audio_device_id)?;
+ let name = Cow::Owned(format!("{fallback} ({name})"));
+ Ok(name)
+ })()
+ .unwrap_or_else(|err: Error| {
+ log::error!("Cannot get device name: {err}");
+ Cow::Borrowed(fallback)
+ })
+ }
+
+ /// Returns the [`DeviceType`] that best describes this [`DeviceRequest`].
+ pub fn device_type(&self) -> DeviceType {
+ let log_error = |result: Result| {
+ result
+ .inspect_err(|err| {
+ log::error!("Cannot get device type for {:?}: {err}", self.name());
+ })
+ .unwrap_or(false)
+ };
+ let supports_scope = |scope| match self {
+ Self::DefaultOutput | Self::SystemOutput if matches!(scope, Scope::Output) => true,
+ Self::DefaultInput | Self::VoiceInput if matches!(scope, Scope::Input) => true,
+ &Self::Specific(id) => log_error(get_audio_device_supports_scope(id, scope)),
+ _ => false,
+ };
+ let is_default = match self {
+ Self::DefaultOutput | Self::SystemOutput | Self::DefaultInput | Self::VoiceInput => {
+ true
+ }
+ &Self::Specific(id) => {
+ get_default_device_id(true) == Some(id) || get_default_device_id(false) == Some(id)
+ }
+ };
+
+ let mut type_ = DeviceType::PHYSICAL;
+ type_.set(DeviceType::INPUT, supports_scope(Scope::Input));
+ type_.set(DeviceType::OUTPUT, supports_scope(Scope::Output));
+ type_.set(DeviceType::DEFAULT, is_default);
+ type_
+ }
+
+ /// Queries the accepted range of buffer sizes that the device accepts.
+ pub fn buffer_size_range(&self) -> Result<(Option, Option), Error> {
+ match self {
+ Self::DefaultOutput | Self::SystemOutput | Self::DefaultInput | Self::VoiceInput => {
+ Ok((None, None))
+ }
+ &Self::Specific(id) => {
+ let address = AudioObjectPropertyAddress {
+ mSelector: kAudioDevicePropertyBufferFrameSizeRange,
+ mScope: if self.device_type().is_input() {
+ kAudioObjectPropertyScopeInput
+ } else {
+ kAudioObjectPropertyScopeOutput
+ },
+ mElement: kAudioObjectPropertyElementMaster,
+ };
+ let range = utils::get_device_property::(id, address)?;
+ Ok((Some(range.mMinimum as usize), Some(range.mMaximum as usize)))
+ }
+ }
+ }
+}
+
+/// CoreAudio device type. Contains the requested device (including default output/notification stream) and
+/// associated audio unit.
+pub struct Device {
+ pub(crate) request: DeviceRequest,
+ audio_unit: OnceCell,
+}
+
+impl Device {
+ /// Create a new device from a [`DeviceRequest`].
+ pub fn new(request: DeviceRequest) -> Self {
+ Self {
+ request,
+ audio_unit: OnceCell::new(),
+ }
+ }
+
+ /// Create a device from a specific device ID.
+ pub fn from_id(id: AudioDeviceID) -> Self {
+ Self::new(DeviceRequest::Specific(id))
+ }
+
+ /// Create a device following the default output.
+ pub fn default_output() -> Self {
+ Self::new(DeviceRequest::DefaultOutput)
+ }
+
+ /// Create a device following the default input.
+ pub fn default_input() -> Self {
+ Self::new(DeviceRequest::DefaultInput)
+ }
+
+ /// Consumes this device object, returning the associated Audio Unit.
+ pub fn into_audio_unit(mut self) -> Result {
+ match self.audio_unit.take() {
+ Some(unit) => Ok(unit),
+ None => self.request.to_audio_unit(),
+ }
+ }
+}
+
+impl ExtensionProvider for Device {
+ fn register<'a, 'sel>(&'a self, selector: &'sel mut Selector<'a>) -> &'sel mut Selector<'a> {
+ selector.register::(self)
+ }
+}
+
+impl device::Device for Device {
+ type Error = Error;
+
+ type StreamHandle = Handle;
+
+ fn name(&self) -> Cow<'_, str> {
+ self.request.name()
+ }
+
+ fn device_type(&self) -> DeviceType {
+ self.request.device_type()
+ }
+
+ fn default_config(&self) -> Result {
+ let au = self.get_audio_unit()?;
+ let input_channels = au
+ .input_stream_format()
+ .map(|fmt| fmt.channels as usize)
+ .unwrap_or(0);
+ let output_channels = au
+ .output_stream_format()
+ .map(|fmt| fmt.channels as usize)
+ .unwrap_or(0);
+ let buffer_size_range = self.request.buffer_size_range()?;
+ Ok(StreamConfig {
+ sample_rate: au.sample_rate()?,
+ input_channels,
+ output_channels,
+ buffer_size_range,
+ exclusive: false,
+ })
+ }
+
+ fn is_config_supported(&self, _config: &StreamConfig) -> bool {
+ true
+ }
+
+ fn create_stream(
+ &self,
+ stream_config: StreamConfig,
+ callback: Callback,
+ ) -> Result, Self::Error> {
+ Handle::new(self, stream_config, callback)
+ }
+}
+
+/// Extension trait for [`Device`]. Either use directly, or lookup the trait with `device.lookup::()`.
+pub trait CoreAudioDeviceExt {
+ /// Return the audio unit associated with this device. Creates it at most once; subsequent calls return the same
+ /// object.
+ fn get_audio_unit(&self) -> Result<&AudioUnit, Error>;
+}
+
+impl CoreAudioDeviceExt for Device {
+ fn get_audio_unit(&self) -> Result<&AudioUnit, Error> {
+ if let Some(unit) = self.audio_unit.get() {
+ return Ok(unit);
+ }
+ let unit = self.request.to_audio_unit()?;
+ let result = self.audio_unit.set(unit);
+ assert!(result.is_ok());
+ Ok(self.audio_unit.get().unwrap())
+ }
+}
diff --git a/crates/interflow-coreaudio/src/lib.rs b/crates/interflow-coreaudio/src/lib.rs
new file mode 100644
index 0000000..53dac73
--- /dev/null
+++ b/crates/interflow-coreaudio/src/lib.rs
@@ -0,0 +1,43 @@
+//! # Interflow CoreAudio
+//!
+//! Interflow backend using CoreAudio for macOS and iOS applications
+#![warn(missing_docs)]
+use coreaudio::audio_unit;
+use interflow_core::DeviceType;
+use std::convert::Infallible;
+
+pub mod device;
+pub mod platform;
+pub mod stream;
+mod utils;
+
+/// Prelude module. Import all with `interflow_coreaudio::prelude::*`.
+pub mod prelude {
+ pub use crate::device::{Device, DeviceRequest};
+ pub use crate::platform::Platform;
+ pub use crate::stream::Handle;
+}
+
+/// Type of errors from the CoreAudio backend
+#[derive(Debug, thiserror::Error)]
+#[error("CoreAudio error: ")]
+pub enum Error {
+ /// Error originating from CoreAudio
+ #[error(transparent)]
+ Backend(#[from] coreaudio::Error),
+ /// The scope given to an audio device is invalid.
+ #[error("Invalid scope {0:?}")]
+ InvalidScope(audio_unit::Scope),
+ /// No matching devices for the given type.
+ #[error("No matching devices for type: {0:?}")]
+ NoMatchingDevices(DeviceType),
+ /// User has requested a duplex (both input and output) stream, which is not supported by CoreAudio.
+ #[error("Duplex devices are not supported")]
+ DuplexUnavailable,
+}
+
+impl From for Error {
+ fn from(_: Infallible) -> Self {
+ unreachable!()
+ }
+}
diff --git a/crates/interflow-coreaudio/src/platform.rs b/crates/interflow-coreaudio/src/platform.rs
new file mode 100644
index 0000000..28e960d
--- /dev/null
+++ b/crates/interflow-coreaudio/src/platform.rs
@@ -0,0 +1,45 @@
+//! CoreAudio platform implementation
+use crate::device::Device;
+use crate::Error;
+use coreaudio::audio_unit::macos_helpers::{get_audio_device_ids, get_default_device_id};
+use interflow_core::traits::{ExtensionProvider, Selector};
+use interflow_core::{collect, platform, DeviceType};
+use std::rc::Rc;
+
+/// The CoreAudio driver.
+pub struct Platform;
+
+impl ExtensionProvider for Platform {
+ fn register<'a, 'sel>(&'a self, selector: &'sel mut Selector<'a>) -> &'sel mut Selector<'a> {
+ selector
+ }
+}
+
+impl platform::Platform for Platform {
+ type Error = Error;
+
+ type Device = Device;
+
+ const NAME: &'static str = "CoreAudio";
+
+ fn default_device(&self, device_type: DeviceType) -> Result {
+ if device_type.is_output() || !device_type.is_input() {
+ Ok(Device::default_output())
+ } else {
+ Ok(Device::default_input())
+ }
+ }
+
+ fn list_devices(&self) -> Result, Self::Error> {
+ Ok([Device::default_output(), Device::default_input()]
+ .into_iter()
+ .chain(get_audio_device_ids()?.into_iter().map(Device::from_id)))
+ }
+}
+
+#[cfg(feature = "collect")]
+#[scattered_collect::scatter(collect::REGISTRAR)]
+static COREAUDIO_PLATFORM_REGISTRATION: collect::Registration = collect::Registration {
+ constructor: || Some(Rc::new(Platform)),
+ priority: collect::DEFAULT_PLATFORM_PRIORITY,
+};
diff --git a/crates/interflow-coreaudio/src/stream.rs b/crates/interflow-coreaudio/src/stream.rs
new file mode 100644
index 0000000..07b3e80
--- /dev/null
+++ b/crates/interflow-coreaudio/src/stream.rs
@@ -0,0 +1,270 @@
+//! CoreAudio stream implementation.
+use crate::device::Device;
+use crate::Error;
+use coreaudio::audio_unit::audio_format::LinearPcmFlags;
+use coreaudio::audio_unit::render_callback::{data, Args};
+use coreaudio::audio_unit::{AudioUnit, Element, SampleFormat, Scope, StreamFormat};
+use coreaudio_sys::{kAudioDevicePropertyBufferFrameSize, kAudioUnitProperty_StreamFormat};
+use interflow_core::buffer::AudioBuffer;
+use interflow_core::device::{Device as _, ResolvedStreamConfig, StreamConfig};
+use interflow_core::stream;
+use interflow_core::stream::{AudioInput, AudioOutput, CallbackContext, StreamProxy};
+use interflow_core::timing::Timestamp;
+use interflow_core::traits::{ExtensionProvider, Selector};
+use std::convert::Infallible;
+use std::num::NonZeroUsize;
+
+/// Stream type created by opening up a stream on a [`Device`].
+pub struct Handle {
+ audio_unit: AudioUnit,
+ callback_retrieve: oneshot::Sender>,
+}
+
+impl stream::StreamHandle for Handle {
+ type Error = Infallible;
+
+ fn eject(mut self) -> Result {
+ let (tx, rx) = oneshot::channel();
+ self.callback_retrieve
+ .send(tx)
+ .expect("Callback receiver cannot have been dropped yet");
+ let callback = rx.recv().expect("Oneshot receiver must be used");
+ self.audio_unit.free_input_callback();
+ self.audio_unit.free_render_callback();
+ Ok(callback)
+ }
+}
+
+fn input_stream_format(sample_rate: f64, channel_count: usize) -> StreamFormat {
+ StreamFormat {
+ sample_rate,
+ sample_format: SampleFormat::I16,
+ flags: LinearPcmFlags::IS_SIGNED_INTEGER,
+ channels: channel_count as _,
+ }
+}
+
+fn output_stream_format(sample_rate: f64, channel_count: usize) -> StreamFormat {
+ StreamFormat {
+ sample_rate,
+ sample_format: SampleFormat::F32,
+ flags: LinearPcmFlags::IS_NON_INTERLEAVED | LinearPcmFlags::IS_FLOAT,
+ channels: channel_count as _,
+ }
+}
+
+struct DummyStreamProxy;
+
+impl ExtensionProvider for DummyStreamProxy {
+ fn register<'a, 'sel>(&'a self, selector: &'sel mut Selector<'a>) -> &'sel mut Selector<'a> {
+ selector
+ }
+}
+
+impl StreamProxy for DummyStreamProxy {}
+
+static STREAM_PROXY: DummyStreamProxy = DummyStreamProxy;
+
+impl Handle {
+ pub(crate) fn new(
+ device: &Device,
+ stream_config: StreamConfig,
+ callback: Callback,
+ ) -> Result {
+ let requested_type = stream_config.requested_device_type();
+ if requested_type.is_duplex() {
+ return Err(Error::DuplexUnavailable);
+ }
+
+ let unsupported = device.device_type() & !requested_type;
+ if !unsupported.is_empty() {
+ log::warn!(
+ "Cannot request {unsupported:?} for {device}, ignoring",
+ device = device.name()
+ );
+ }
+ let unit = device.request.to_audio_unit()?;
+ if requested_type.is_input() {
+ Self::new_input(unit, stream_config, callback)
+ } else {
+ Self::new_output(unit, stream_config, callback)
+ }
+ }
+
+ fn new_input(
+ mut audio_unit: AudioUnit,
+ stream_config: StreamConfig,
+ mut callback: Callback,
+ ) -> Result {
+ let asbd =
+ input_stream_format(stream_config.sample_rate, stream_config.input_channels).to_asbd();
+ audio_unit.set_property(
+ kAudioUnitProperty_StreamFormat,
+ Scope::Output,
+ Element::Input,
+ Some(&asbd),
+ )?;
+ let frame_count: u32 = audio_unit.get_property(
+ kAudioDevicePropertyBufferFrameSize,
+ Scope::Global,
+ Element::Output,
+ )?;
+ let resolved_config = ResolvedStreamConfig {
+ sample_rate: asbd.mSampleRate,
+ input_channels: asbd.mChannelsPerFrame as _,
+ output_channels: 0,
+ max_frame_count: frame_count as _,
+ };
+ let mut buffer = AudioBuffer::zeroed(
+ NonZeroUsize::new(frame_count as _).unwrap(),
+ asbd.mChannelsPerFrame as _,
+ );
+
+ let (tx, rx) = oneshot::channel::>();
+ callback.prepare(CallbackContext {
+ stream_config: &resolved_config,
+ timestamp: Timestamp::new(asbd.mSampleRate),
+ stream_proxy: &STREAM_PROXY,
+ });
+ let mut callback = Some(callback);
+
+ audio_unit.set_input_callback(move |args: Args>| {
+ if let Ok(sender) = rx.try_recv() {
+ sender.send(callback.take().unwrap()).unwrap();
+ return Err(());
+ }
+ let num_frames = args.num_frames;
+ let channels = asbd.mChannelsPerFrame as usize;
+ let input_data = args.data.buffer;
+
+ for frame_idx in 0..num_frames {
+ for ch in 0..channels {
+ let sample = input_data[frame_idx * channels + ch];
+ buffer
+ .frame_mut(frame_idx)
+ .set(ch, (sample as f32) / (i16::MAX as f32));
+ }
+ }
+
+ let timestamp = Timestamp::from_count(
+ resolved_config.sample_rate,
+ args.time_stamp.mSampleTime as _,
+ );
+
+ let input = AudioInput {
+ buffer: buffer.as_ref(),
+ timestamp,
+ channel_flags: &[],
+ };
+
+ let mut dummy_buf = AudioBuffer::zeroed(NonZeroUsize::new(1).unwrap(), channels as _);
+ let mut dummy_output = AudioOutput {
+ buffer: dummy_buf.as_mut(),
+ timestamp: Timestamp::new(asbd.mSampleRate),
+ channel_flags: &[],
+ };
+
+ if let Some(callback) = &mut callback {
+ callback.process_audio(
+ CallbackContext {
+ stream_config: &resolved_config,
+ timestamp,
+ stream_proxy: &STREAM_PROXY,
+ },
+ &input,
+ &mut dummy_output,
+ );
+ }
+ Ok(())
+ })?;
+ audio_unit.start()?;
+ Ok(Self {
+ audio_unit,
+ callback_retrieve: tx,
+ })
+ }
+
+ fn new_output(
+ mut audio_unit: AudioUnit,
+ stream_config: StreamConfig,
+ mut callback: Callback,
+ ) -> Result {
+ let asbd = output_stream_format(stream_config.sample_rate, stream_config.output_channels)
+ .to_asbd();
+ audio_unit.set_property(
+ kAudioUnitProperty_StreamFormat,
+ Scope::Input,
+ Element::Output,
+ Some(&asbd),
+ )?;
+ let frame_size: u32 = audio_unit.get_property(
+ kAudioDevicePropertyBufferFrameSize,
+ Scope::Output,
+ Element::Output,
+ )?;
+ let resolved_config = ResolvedStreamConfig {
+ sample_rate: asbd.mSampleRate,
+ input_channels: 0,
+ output_channels: asbd.mChannelsPerFrame as _,
+ max_frame_count: frame_size as _,
+ };
+ let mut buffer = AudioBuffer::zeroed(
+ NonZeroUsize::new(frame_size as _).unwrap(),
+ asbd.mChannelsPerFrame as _,
+ );
+
+ let (tx, rx) = oneshot::channel::>();
+ callback.prepare(CallbackContext {
+ stream_config: &resolved_config,
+ timestamp: Timestamp::new(resolved_config.sample_rate),
+ stream_proxy: &STREAM_PROXY,
+ });
+ let mut callback = Some(callback);
+
+ audio_unit.set_render_callback(move |mut args: Args>| {
+ if let Ok(sender) = rx.try_recv() {
+ sender.send(callback.take().unwrap()).unwrap();
+ return Err(());
+ }
+ let timestamp = Timestamp::from_count(
+ resolved_config.sample_rate,
+ args.time_stamp.mSampleTime as _,
+ );
+
+ let dummy_buf =
+ AudioBuffer::zeroed(NonZeroUsize::new(1).unwrap(), asbd.mChannelsPerFrame as _);
+ let dummy_input = AudioInput {
+ buffer: dummy_buf.as_ref(),
+ timestamp: Timestamp::new(resolved_config.sample_rate),
+ channel_flags: &[],
+ };
+
+ let mut output = AudioOutput {
+ buffer: buffer.as_mut(),
+ timestamp,
+ channel_flags: &[],
+ };
+
+ if let Some(callback) = &mut callback {
+ callback.process_audio(
+ CallbackContext {
+ stream_config: &resolved_config,
+ timestamp,
+ stream_proxy: &STREAM_PROXY,
+ },
+ &dummy_input,
+ &mut output,
+ );
+ for (out_ch, in_ch) in args.data.channels_mut().zip(buffer.iter_channels()) {
+ out_ch.copy_from_slice(in_ch);
+ }
+ }
+ Ok(())
+ })?;
+ audio_unit.start()?;
+ Ok(Self {
+ audio_unit,
+ callback_retrieve: tx,
+ })
+ }
+}
diff --git a/crates/interflow-coreaudio/src/utils.rs b/crates/interflow-coreaudio/src/utils.rs
new file mode 100644
index 0000000..7eaccae
--- /dev/null
+++ b/crates/interflow-coreaudio/src/utils.rs
@@ -0,0 +1,41 @@
+use coreaudio_sys::{AudioDeviceID, AudioObjectGetPropertyData, AudioObjectPropertyAddress};
+
+pub(crate) fn get_device_property(
+ device_id: AudioDeviceID,
+ address: AudioObjectPropertyAddress,
+) -> Result {
+ let mut data = std::mem::MaybeUninit::::uninit();
+ let mut size = size_of::() as u32;
+ let status = unsafe {
+ AudioObjectGetPropertyData(
+ device_id,
+ &address,
+ 0,
+ std::ptr::null(),
+ &mut size,
+ data.as_mut_ptr() as *mut _,
+ )
+ };
+ coreaudio::Error::from_os_status(status)?;
+ Ok(unsafe { data.assume_init() })
+}
+
+#[expect(unused)]
+pub(crate) fn set_device_property(
+ device_id: AudioDeviceID,
+ address: AudioObjectPropertyAddress,
+ data: &T,
+) -> Result<(), coreaudio::Error> {
+ let size = size_of::() as u32;
+ let status = unsafe {
+ coreaudio_sys::AudioObjectSetPropertyData(
+ device_id,
+ &address,
+ 0,
+ std::ptr::null(),
+ size,
+ data as *const T as *const _,
+ )
+ };
+ coreaudio::Error::from_os_status(status)
+}
diff --git a/crates/interflow-wasapi/Cargo.toml b/crates/interflow-wasapi/Cargo.toml
new file mode 100644
index 0000000..c289fa2
--- /dev/null
+++ b/crates/interflow-wasapi/Cargo.toml
@@ -0,0 +1,31 @@
+[package]
+name = "interflow-wasapi"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+
+[features]
+collect = ["interflow-core/collect", "dep:scattered-collect"]
+
+[dependencies]
+bitflags.workspace = true
+duplicate.workspace = true
+interflow-core.workspace = true
+scattered-collect = { workspace = true, optional = true }
+thiserror.workspace = true
+log.workspace = true
+windows = { version = "0.62.2", features = [
+ "Win32_Media_Audio",
+ "Win32_Foundation",
+ "Win32_Devices_Properties",
+ "Win32_Media_KernelStreaming",
+ "Win32_System_Com_StructuredStorage",
+ "Win32_System_Threading",
+ "Win32_Security",
+ "Win32_System_SystemServices",
+ "Win32_System_Variant",
+ "Win32_Media_Multimedia",
+ "Win32_UI_Shell_PropertiesSystem",
+] }
+zerocopy = { version = "0.8.39", features = ["std"] }
diff --git a/crates/interflow-wasapi/src/device.rs b/crates/interflow-wasapi/src/device.rs
new file mode 100644
index 0000000..fcdbc6b
--- /dev/null
+++ b/crates/interflow-wasapi/src/device.rs
@@ -0,0 +1,205 @@
+use crate::util::MMDevice;
+use crate::Error;
+use interflow_core::device::StreamConfig;
+use interflow_core::traits::{ExtensionProvider, Selector};
+use interflow_core::{device, stream, DeviceType};
+use std::borrow::Cow;
+use std::ptr::NonNull;
+use windows::Win32::Media::Audio;
+use windows::Win32::Media::Audio::{IAudioClient, IAudioClient3};
+use windows::Win32::Media::Multimedia;
+
+#[derive(Debug, Clone)]
+pub struct Device {
+ pub(crate) handle: MMDevice,
+ pub(crate) device_type: DeviceType,
+}
+
+impl ExtensionProvider for Device {
+ fn register<'a, 'sel>(&'a self, selector: &'sel mut Selector<'a>) -> &'sel mut Selector<'a> {
+ selector
+ }
+}
+
+impl device::Device for Device {
+ type Error = Error;
+ type StreamHandle = crate::stream::Handle;
+
+ fn name(&self) -> Cow<'_, str> {
+ Cow::Owned(self.handle.name())
+ }
+
+ fn device_type(&self) -> DeviceType {
+ self.device_type
+ }
+
+ fn default_config(&self) -> Result {
+ self.get_mix_format_iac3()
+ .or_else(|_| self.get_mix_format())
+ }
+
+ fn is_config_supported(&self, config: &StreamConfig) -> bool {
+ self.check_format(config).unwrap_or(false)
+ }
+
+ fn create_stream(
+ &self,
+ stream_config: StreamConfig,
+ callback: Callback,
+ ) -> Result, Self::Error> {
+ crate::stream::Handle::new(self, stream_config, callback)
+ }
+}
+
+impl Device {
+ pub(crate) fn activate_audio_client(&self) -> Result {
+ self.handle.activate::()
+ }
+
+ fn get_mix_format(&self) -> Result {
+ let client = self.activate_audio_client()?;
+ let mix_format = unsafe { client.GetMixFormat() }?;
+ let format = unsafe { mix_format.read_unaligned() };
+ let channels = format.nChannels as usize;
+ let input_channels = if self.device_type.is_input() {
+ channels
+ } else {
+ 0
+ };
+ let output_channels = if self.device_type.is_output() {
+ channels
+ } else {
+ 0
+ };
+ Ok(StreamConfig {
+ sample_rate: format.nSamplesPerSec as _,
+ input_channels,
+ output_channels,
+ buffer_size_range: (None, None),
+ exclusive: false,
+ })
+ }
+
+ fn get_mix_format_iac3(&self) -> Result {
+ let client = self.handle.activate::()?;
+ let mut period_default = 0u32;
+ let mut period_min = 0u32;
+ let mut period_max = 0u32;
+ let format = unsafe { client.GetMixFormat() }?;
+ unsafe {
+ let mut _fundamental_period = 0u32;
+ client.GetSharedModeEnginePeriod(
+ format.cast_const(),
+ &mut period_default,
+ &mut _fundamental_period,
+ &mut period_min,
+ &mut period_max,
+ )?;
+ }
+ let format = unsafe { format.read_unaligned() };
+ let channels = format.nChannels as usize;
+ let input_channels = if self.device_type.is_input() {
+ channels
+ } else {
+ 0
+ };
+ let output_channels = if self.device_type.is_output() {
+ channels
+ } else {
+ 0
+ };
+ Ok(StreamConfig {
+ sample_rate: format.nSamplesPerSec as _,
+ input_channels,
+ output_channels,
+ buffer_size_range: (Some(period_min as usize), Some(period_max as usize)),
+ exclusive: false,
+ })
+ }
+
+ fn check_format(&self, config: &StreamConfig) -> Result {
+ unsafe {
+ let audio_client = self.activate_audio_client()?;
+ let sharemode = if config.exclusive {
+ Audio::AUDCLNT_SHAREMODE_EXCLUSIVE
+ } else {
+ Audio::AUDCLNT_SHAREMODE_SHARED
+ };
+ let format = Self::build_format(config);
+ if config.exclusive {
+ audio_client
+ .IsFormatSupported(sharemode, &format, None)
+ .ok()?;
+ return Ok(true);
+ }
+ let mut closest_ptr: *mut Audio::WAVEFORMATEX = std::ptr::null_mut();
+ let result = audio_client
+ .IsFormatSupported(sharemode, &format, Some(&mut closest_ptr));
+ let hr = result.0;
+ if hr == 0 {
+ return Ok(true);
+ }
+ if hr > 0 && !closest_ptr.is_null() {
+ let closest =
+ crate::util::CoTask::new(NonNull::new_unchecked(closest_ptr));
+ let closest_format = closest.as_ptr().read_unaligned();
+ return Ok(closest_format.nSamplesPerSec == config.sample_rate as u32
+ && closest_format.nChannels as usize
+ == config.output_channels.max(config.input_channels));
+ }
+ Ok(false)
+ }
+ }
+
+ pub(crate) fn build_format(config: &StreamConfig) -> Audio::WAVEFORMATEX {
+ let channels = (config.input_channels.max(config.output_channels)) as u16;
+ let sample_rate = config.sample_rate as u32;
+ let sample_bytes = size_of::() as u16;
+ let avg_bytes_per_sec = channels as u32 * sample_rate * sample_bytes as u32;
+ let block_align = channels * sample_bytes;
+ Audio::WAVEFORMATEX {
+ wFormatTag: Multimedia::WAVE_FORMAT_IEEE_FLOAT as u16,
+ nChannels: channels,
+ nSamplesPerSec: sample_rate,
+ nAvgBytesPerSec: avg_bytes_per_sec,
+ nBlockAlign: block_align,
+ wBitsPerSample: 8 * sample_bytes,
+ cbSize: 0,
+ }
+ }
+}
+
+pub(crate) struct DeviceList {
+ pub(crate) collection: Audio::IMMDeviceCollection,
+ pub(crate) total_count: u32,
+ pub(crate) next_item: u32,
+ pub(crate) device_type: DeviceType,
+}
+
+unsafe impl Send for DeviceList {}
+unsafe impl Sync for DeviceList {}
+
+impl Iterator for DeviceList {
+ type Item = Device;
+
+ fn next(&mut self) -> Option {
+ if self.next_item >= self.total_count {
+ return None;
+ }
+ unsafe {
+ let device = self.collection.Item(self.next_item).ok()?;
+ self.next_item += 1;
+ Some(Device {
+ handle: MMDevice::new(device),
+ device_type: self.device_type,
+ })
+ }
+ }
+
+ fn size_hint(&self) -> (usize, Option) {
+ let rest = (self.total_count - self.next_item) as usize;
+ (rest, Some(rest))
+ }
+}
+
+impl ExactSizeIterator for DeviceList {}
diff --git a/crates/interflow-wasapi/src/lib.rs b/crates/interflow-wasapi/src/lib.rs
new file mode 100644
index 0000000..b176865
--- /dev/null
+++ b/crates/interflow-wasapi/src/lib.rs
@@ -0,0 +1,30 @@
+pub mod device;
+pub mod platform;
+pub mod stream;
+mod util;
+
+pub mod prelude {
+ pub use crate::device::Device;
+ pub use crate::platform::Platform;
+ pub use crate::stream::Handle;
+}
+
+use std::convert::Infallible;
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+ #[error("{} (code {})", .0.message(), .0.code())]
+ BackendError(#[from] windows::core::Error),
+ #[error("Configuration not available")]
+ ConfigurationNotAvailable,
+ #[error("Win32 error: {0}")]
+ FoundationError(String),
+ #[error("Unsupported duplex stream requested")]
+ DuplexStreamRequested,
+}
+
+impl From for Error {
+ fn from(_: Infallible) -> Self {
+ unreachable!()
+ }
+}
diff --git a/crates/interflow-wasapi/src/platform.rs b/crates/interflow-wasapi/src/platform.rs
new file mode 100644
index 0000000..0a907a7
--- /dev/null
+++ b/crates/interflow-wasapi/src/platform.rs
@@ -0,0 +1,151 @@
+use crate::device::{Device, DeviceList};
+use crate::util::MMDevice;
+use crate::Error;
+use bitflags::bitflags_match;
+#[cfg(feature = "collect")]
+use interflow_core::collect;
+use interflow_core::platform;
+use interflow_core::traits::{ExtensionProvider, Selector};
+use interflow_core::DeviceType;
+#[cfg(feature = "collect")]
+use std::rc::Rc;
+use std::sync::OnceLock;
+use windows::Win32::Media::Audio;
+use windows::Win32::Media::Audio::{EDataFlow, ERole};
+use windows::Win32::System::Com;
+use interflow_core::proxies::PlatformProxy;
+
+pub struct Platform;
+
+impl ExtensionProvider for Platform {
+ fn register<'a, 'sel>(&'a self, selector: &'sel mut Selector<'a>) -> &'sel mut Selector<'a> {
+ selector
+ .register::(self)
+ }
+}
+
+impl platform::Platform for Platform {
+ type Error = Error;
+ type Device = Device;
+ const NAME: &'static str = "WASAPI";
+
+ fn default_device(&self, device_type: DeviceType) -> Result {
+ let Some(device) = audio_device_enumerator().get_default_device(device_type)? else {
+ return Err(Error::ConfigurationNotAvailable);
+ };
+ Ok(device)
+ }
+
+ fn list_devices(&self) -> Result, Self::Error> {
+ audio_device_enumerator().get_device_list()
+ }
+}
+
+pub trait DefaultByRole {
+ fn default_by_role(&self, flow: Audio::EDataFlow, role: Audio::ERole) -> Result;
+}
+
+impl DefaultByRole for Platform {
+ fn default_by_role(&self, flow: Audio::EDataFlow, role: Audio::ERole) -> Result {
+ audio_device_enumerator().get_default_device_with_role(flow, role)
+ }
+}
+
+fn audio_device_enumerator() -> &'static AudioDeviceEnumerator {
+ static ENUMERATOR: OnceLock = OnceLock::new();
+ ENUMERATOR.get_or_init(|| {
+ let com = crate::util::com().unwrap();
+ unsafe {
+ let enumerator = com
+ .create_instance::<_, Audio::IMMDeviceEnumerator>(
+ &Audio::MMDeviceEnumerator,
+ None,
+ Com::CLSCTX_ALL,
+ )
+ .unwrap();
+ AudioDeviceEnumerator(enumerator)
+ }
+ })
+}
+
+struct AudioDeviceEnumerator(Audio::IMMDeviceEnumerator);
+
+unsafe impl Send for AudioDeviceEnumerator {}
+unsafe impl Sync for AudioDeviceEnumerator {}
+
+impl AudioDeviceEnumerator {
+ fn get_default_device(&self, device_type: DeviceType) -> Result, Error> {
+ let Some(flow) = bitflags_match!(device_type, {
+ DeviceType::INPUT => Some(Audio::eCapture),
+ DeviceType::OUTPUT => Some(Audio::eRender),
+ _ => None,
+ }) else {
+ return Ok(None);
+ };
+ self.get_default_device_with_role(flow, Audio::eConsole)
+ .map(Some)
+ }
+
+ fn get_default_device_with_role(
+ &self,
+ flow: Audio::EDataFlow,
+ role: Audio::ERole,
+ ) -> Result {
+ unsafe {
+ let device = self.0.GetDefaultAudioEndpoint(flow, role)?;
+ let device_type = match flow {
+ Audio::eRender => DeviceType::OUTPUT,
+ _ => DeviceType::INPUT,
+ };
+ Ok(Device {
+ handle: MMDevice::new(device),
+ device_type: DeviceType::PHYSICAL | device_type,
+ })
+ }
+ }
+
+ fn get_device_list(&self) -> Result, Error> {
+ unsafe {
+ let output_collection = self
+ .0
+ .EnumAudioEndpoints(Audio::eRender, Audio::DEVICE_STATE_ACTIVE)?;
+ let count = output_collection.GetCount()?;
+ let output_device_list = DeviceList {
+ collection: output_collection,
+ total_count: count,
+ next_item: 0,
+ device_type: DeviceType::OUTPUT,
+ };
+
+ let input_collection = self
+ .0
+ .EnumAudioEndpoints(Audio::eCapture, Audio::DEVICE_STATE_ACTIVE)?;
+ let count = input_collection.GetCount()?;
+ let input_device_list = DeviceList {
+ collection: input_collection,
+ total_count: count,
+ next_item: 0,
+ device_type: DeviceType::INPUT,
+ };
+
+ Ok(output_device_list.chain(input_device_list))
+ }
+ }
+}
+
+pub trait DefaultForRole: PlatformProxy {
+ fn default_for_role(&self, flow: Audio::EDataFlow, role: Audio::ERole) -> Result;
+}
+
+impl DefaultForRole for Platform {
+ fn default_for_role(&self, flow: EDataFlow, role: ERole) -> Result {
+ Ok(audio_device_enumerator().get_default_device_with_role(flow, role)?)
+ }
+}
+
+#[cfg(feature = "collect")]
+#[scattered_collect::scatter(collect::REGISTRAR)]
+static WASAPI_PLATFORM_REGISTRATION: collect::Registration = collect::Registration {
+ constructor: || Some(Rc::new(Platform)),
+ priority: collect::DEFAULT_PLATFORM_PRIORITY,
+};
diff --git a/crates/interflow-wasapi/src/stream.rs b/crates/interflow-wasapi/src/stream.rs
new file mode 100644
index 0000000..7dd3967
--- /dev/null
+++ b/crates/interflow-wasapi/src/stream.rs
@@ -0,0 +1,366 @@
+use crate::device::Device;
+use crate::Error;
+use interflow_core::buffer::AudioBuffer;
+use interflow_core::device::{ResolvedStreamConfig, StreamConfig};
+use interflow_core::stream;
+use interflow_core::stream::{AudioInput, AudioOutput, CallbackContext, StreamProxy};
+use interflow_core::timing::Timestamp;
+use interflow_core::traits::{ExtensionProvider, Selector};
+use std::num::NonZeroUsize;
+use std::ptr;
+use std::slice;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::thread::JoinHandle;
+use std::time::Duration;
+use windows::Win32::Foundation::{CloseHandle, HANDLE, WAIT_FAILED};
+use windows::Win32::Media::Audio;
+use windows::Win32::System::Threading;
+
+type EjectSignal = Arc;
+
+pub struct Handle {
+ join_handle: JoinHandle>,
+ eject_signal: EjectSignal,
+}
+
+impl stream::StreamHandle for Handle {
+ type Error = Error;
+
+ fn eject(self) -> Result {
+ self.eject_signal.store(true, Ordering::Relaxed);
+ self.join_handle.join().expect("Audio thread panicked")
+ }
+}
+
+impl Handle {
+ pub(crate) fn new(
+ device: &Device,
+ stream_config: StreamConfig,
+ callback: Callback,
+ ) -> Result