feat: make Stream implement Send on all hosts (#1021)

* feat: add compile-time Send assertions for Stream types in all hosts
* feat: impl Send for Stream on CoreAudio, WASAPI, WASM
* docs: note removal of Clone from Stream
* refactor: move error callback handling into helper function

Consolidate repeated error callback logic into invoke_error_callback for
better readability and maintainability. Update usages in device.rs and
disconnect manager. Add documentation for the helper.

Author: roderickvd

CHANGELOG.md

@@ -22,14 +22,18 @@
 - CoreAudio: Update `mach2` to 0.5.
 - CoreAudio: Configure device buffer to ensure predictable callback buffer sizes.
 - CoreAudio: Fix timestamp accuracy.
+- CoreAudio: Make `Stream` implement `Send`.
+- CoreAudio: Remove `Clone` impl from `Stream`.
 - Emscripten: Add `BufferSize::Fixed` validation against supported range.
 - iOS: Fix example by properly activating audio session.
 - iOS: Add complete AVAudioSession integration for device enumeration and buffer size control.
 - JACK: Add `BufferSize::Fixed` validation to reject requests that don't match server buffer size.
 - WASAPI: Expose `IMMDevice` from WASAPI host Device.
 - WASAPI: Add `I24` and `U24` sample format support (24-bit samples stored in 4 bytes).
 - WASAPI: Update `windows` to >= 0.58, <= 0.62.
+- WASAPI: Make `Stream` implement `Send`.
 - Wasm: Removed optional `wee-alloc` feature for security reasons.
+- Wasm: Make `Stream` implement `Send`.
 - WebAudio: Add `BufferSize::Fixed` validation against supported range.
 
 # Version 0.16.0 (2025-06-07)

src/host/aaudio/mod.rs

@@ -49,6 +49,9 @@ pub enum Stream {
 // TODO: Is this still in-progress? https://github.com/rust-mobile/ndk/pull/497
 unsafe impl Send for Stream {}
 
+// Compile-time assertion that Stream is Send
+crate::assert_stream_send!(Stream);
+
 pub type SupportedInputConfigs = VecIntoIter<SupportedStreamConfigRange>;
 pub type SupportedOutputConfigs = VecIntoIter<SupportedStreamConfigRange>;
 pub type Devices = VecIntoIter<Device>;

src/host/alsa/mod.rs

@@ -624,6 +624,9 @@ pub struct Stream {
     trigger: TriggerSender,
 }
 
+// Compile-time assertion that Stream is Send
+crate::assert_stream_send!(Stream);
+
 struct StreamWorkerContext {
     descriptors: Box<[libc::pollfd]>,
     transfer_buffer: Box<[u8]>,

src/host/asio/stream.rs

@@ -22,6 +22,9 @@ pub struct Stream {
     callback_id: sys::CallbackId,
 }
 
+// Compile-time assertion that Stream is Send
+crate::assert_stream_send!(Stream);
+
 impl Stream {
     pub fn play(&self) -> Result<(), PlayStreamError> {
         self.playing.store(true, Ordering::SeqCst);

src/host/coreaudio/ios/mod.rs

@@ -1,6 +1,6 @@
 //! CoreAudio implementation for iOS using AVAudioSession and RemoteIO Audio Units.
 
-use std::cell::RefCell;
+use std::sync::Mutex;
 
 use coreaudio::audio_unit::render_callback::data;
 use coreaudio::audio_unit::{render_callback, AudioUnit, Element, Scope};
@@ -212,20 +212,27 @@ impl DeviceTrait for Device {
 }
 
 pub struct Stream {
-    inner: RefCell<StreamInner>,
+    inner: Mutex<StreamInner>,
 }
 
 impl Stream {
     fn new(inner: StreamInner) -> Self {
         Self {
-            inner: RefCell::new(inner),
+            inner: Mutex::new(inner),
         }
     }
 }
 
 impl StreamTrait for Stream {
     fn play(&self) -> Result<(), PlayStreamError> {
-        let mut stream = self.inner.borrow_mut();
+        let mut stream = self
+            .inner
+            .lock()
+            .map_err(|_| PlayStreamError::BackendSpecific {
+                err: BackendSpecificError {
+                    description: "A cpal stream operation panicked while holding the lock - this is a bug, please report it".to_string(),
+                },
+            })?;
 
         if !stream.playing {
             if let Err(e) = stream.audio_unit.start() {
@@ -239,7 +246,14 @@ impl StreamTrait for Stream {
     }
 
     fn pause(&self) -> Result<(), PauseStreamError> {
-        let mut stream = self.inner.borrow_mut();
+        let mut stream = self
+            .inner
+            .lock()
+            .map_err(|_| PauseStreamError::BackendSpecific {
+                err: BackendSpecificError {
+                    description: "A cpal stream operation panicked while holding the lock - this is a bug, please report it".to_string(),
+                },
+            })?;
 
         if stream.playing {
             if let Err(e) = stream.audio_unit.stop() {

src/host/coreaudio/macos/device.rs

@@ -18,13 +18,12 @@ use objc2_audio_toolbox::{
 };
 use objc2_core_audio::{
     kAudioDevicePropertyAvailableNominalSampleRates, kAudioDevicePropertyBufferFrameSize,
-    kAudioDevicePropertyBufferFrameSizeRange, kAudioDevicePropertyDeviceIsAlive,
-    kAudioDevicePropertyNominalSampleRate, kAudioDevicePropertyStreamConfiguration,
-    kAudioDevicePropertyStreamFormat, kAudioObjectPropertyElementMaster,
-    kAudioObjectPropertyScopeGlobal, kAudioObjectPropertyScopeInput,
-    kAudioObjectPropertyScopeOutput, AudioDeviceID, AudioObjectGetPropertyData,
-    AudioObjectGetPropertyDataSize, AudioObjectID, AudioObjectPropertyAddress,
-    AudioObjectPropertyScope, AudioObjectSetPropertyData,
+    kAudioDevicePropertyBufferFrameSizeRange, kAudioDevicePropertyNominalSampleRate,
+    kAudioDevicePropertyStreamConfiguration, kAudioDevicePropertyStreamFormat,
+    kAudioObjectPropertyElementMaster, kAudioObjectPropertyScopeGlobal,
+    kAudioObjectPropertyScopeInput, kAudioObjectPropertyScopeOutput, AudioDeviceID,
+    AudioObjectGetPropertyData, AudioObjectGetPropertyDataSize, AudioObjectID,
+    AudioObjectPropertyAddress, AudioObjectPropertyScope, AudioObjectSetPropertyData,
 };
 use objc2_core_audio_types::{
     AudioBuffer, AudioBufferList, AudioStreamBasicDescription, AudioValueRange,
@@ -36,12 +35,12 @@ pub use super::enumerate::{
 use std::fmt;
 use std::mem::{self};
 use std::ptr::{null, NonNull};
-use std::rc::Rc;
 use std::slice;
 use std::sync::mpsc::{channel, RecvTimeoutError};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
+use super::invoke_error_callback;
 use super::property_listener::AudioObjectPropertyListener;
 use coreaudio::audio_unit::macos_helpers::get_device_name;
 /// Attempt to set the device sample rate to the provided rate.
@@ -253,45 +252,6 @@ fn get_io_buffer_frame_size_range(
     })
 }
 
-/// Register the on-disconnect callback.
-/// This will both stop the stream and call the error callback with DeviceNotAvailable.
-/// This function should only be called once per stream.
-fn add_disconnect_listener<E>(
-    stream: &Stream,
-    error_callback: Arc<Mutex<E>>,
-) -> Result<(), BuildStreamError>
-where
-    E: FnMut(StreamError) + Send + 'static,
-{
-    let stream_inner_weak = Rc::downgrade(&stream.inner);
-    let mut stream_inner = stream.inner.borrow_mut();
-    stream_inner._disconnect_listener = Some(AudioObjectPropertyListener::new(
-        stream_inner.device_id,
-        AudioObjectPropertyAddress {
-            mSelector: kAudioDevicePropertyDeviceIsAlive,
-            mScope: kAudioObjectPropertyScopeGlobal,
-            mElement: kAudioObjectPropertyElementMaster,
-        },
-        move || {
-            if let Some(stream_inner_strong) = stream_inner_weak.upgrade() {
-                match stream_inner_strong.try_borrow_mut() {
-                    Ok(mut stream_inner) => {
-                        let _ = stream_inner.pause();
-                    }
-                    Err(_) => {
-                        // Could not acquire mutable borrow. This can occur if there are
-                        // overlapping borrows, if the stream is already in use, or if a panic
-                        // occurred during a previous borrow. Still notify about device
-                        // disconnection even if we can't pause.
-                    }
-                }
-                (error_callback.lock().unwrap())(StreamError::DeviceNotAvailable);
-            }
-        },
-    )?);
-    Ok(())
-}
-
 impl DeviceTrait for Device {
     type SupportedInputConfigs = SupportedInputConfigs;
     type SupportedOutputConfigs = SupportedOutputConfigs;
@@ -710,24 +670,22 @@ impl Device {
 
         type Args = render_callback::Args<data::Raw>;
         audio_unit.set_input_callback(move |args: Args| unsafe {
-            let ptr = (*args.data.data).mBuffers.as_ptr();
-            let len = (*args.data.data).mNumberBuffers as usize;
-            let buffers: &[AudioBuffer] = slice::from_raw_parts(ptr, len);
-
-            // TODO: Perhaps loop over all buffers instead?
+            // SAFETY: We configure the stream format as interleaved (via asbd_from_config which
+            // does not set kAudioFormatFlagIsNonInterleaved). Interleaved format always has
+            // exactly one buffer containing all channels, so mBuffers[0] is always valid.
             let AudioBuffer {
                 mNumberChannels: channels,
                 mDataByteSize: data_byte_size,
                 mData: data,
-            } = buffers[0];
+            } = (*args.data.data).mBuffers[0];
 
             let data = data as *mut ();
             let len = data_byte_size as usize / bytes_per_channel;
             let data = Data::from_parts(data, len, sample_format);
 
             let callback = match host_time_to_stream_instant(args.time_stamp.mHostTime) {
                 Err(err) => {
-                    (error_callback.lock().unwrap())(err.into());
+                    invoke_error_callback(&error_callback, err.into());
                     return Err(());
                 }
                 Ok(cb) => cb,
@@ -750,21 +708,36 @@ impl Device {
             Ok(())
         })?;
 
-        let stream = Stream::new(StreamInner {
-            playing: true,
-            _disconnect_listener: None,
-            audio_unit,
-            device_id: self.audio_device_id,
-            _loopback_device: loopback_aggregate,
-        });
-
-        // If we didn't request the default device, stop the stream if the
-        // device disconnects.
-        if !is_default_device(self) {
-            add_disconnect_listener(&stream, error_callback_disconnect)?;
-        }
+        // Create error callback for stream - either dummy or real based on device type
+        let error_callback_for_stream: super::ErrorCallback = if is_default_device(self) {
+            Box::new(|_: StreamError| {})
+        } else {
+            let error_callback_clone = error_callback_disconnect.clone();
+            Box::new(move |err: StreamError| {
+                invoke_error_callback(&error_callback_clone, err);
+            })
+        };
+
+        let stream = Stream::new(
+            StreamInner {
+                playing: true,
+                audio_unit,
+                device_id: self.audio_device_id,
+                _loopback_device: loopback_aggregate,
+            },
+            error_callback_for_stream,
+        )?;
 
-        stream.inner.borrow_mut().audio_unit.start()?;
+        stream
+            .inner
+            .lock()
+            .map_err(|_| BuildStreamError::BackendSpecific {
+                err: BackendSpecificError {
+                    description: "A cpal stream operation panicked while holding the lock - this is a bug, please report it".to_string(),
+                },
+            })?
+            .audio_unit
+            .start()?;
 
         Ok(stream)
     }
@@ -800,9 +773,9 @@ impl Device {
 
         type Args = render_callback::Args<data::Raw>;
         audio_unit.set_render_callback(move |args: Args| unsafe {
-            // If `run()` is currently running, then a callback will be available from this list.
-            // Otherwise, we just fill the buffer with zeroes and return.
-
+            // SAFETY: We configure the stream format as interleaved (via asbd_from_config which
+            // does not set kAudioFormatFlagIsNonInterleaved). Interleaved format always has
+            // exactly one buffer containing all channels, so mBuffers[0] is always valid.
             let AudioBuffer {
                 mNumberChannels: channels,
                 mDataByteSize: data_byte_size,
@@ -815,7 +788,7 @@ impl Device {
 
             let callback = match host_time_to_stream_instant(args.time_stamp.mHostTime) {
                 Err(err) => {
-                    (error_callback.lock().unwrap())(err.into());
+                    invoke_error_callback(&error_callback, err.into());
                     return Err(());
                 }
                 Ok(cb) => cb,
@@ -838,21 +811,36 @@ impl Device {
             Ok(())
         })?;
 
-        let stream = Stream::new(StreamInner {
-            playing: true,
-            _disconnect_listener: None,
-            audio_unit,
-            device_id: self.audio_device_id,
-            _loopback_device: None,
-        });
-
-        // If we didn't request the default device, stop the stream if the
-        // device disconnects.
-        if !is_default_device(self) {
-            add_disconnect_listener(&stream, error_callback_disconnect)?;
-        }
+        // Create error callback for stream - either dummy or real based on device type
+        let error_callback_for_stream: super::ErrorCallback = if is_default_device(self) {
+            Box::new(|_: StreamError| {})
+        } else {
+            let error_callback_clone = error_callback_disconnect.clone();
+            Box::new(move |err: StreamError| {
+                invoke_error_callback(&error_callback_clone, err);
+            })
+        };
+
+        let stream = Stream::new(
+            StreamInner {
+                playing: true,
+                audio_unit,
+                device_id: self.audio_device_id,
+                _loopback_device: None,
+            },
+            error_callback_for_stream,
+        )?;
 
-        stream.inner.borrow_mut().audio_unit.start()?;
+        stream
+            .inner
+            .lock()
+            .map_err(|_| BuildStreamError::BackendSpecific {
+                err: BackendSpecificError {
+                    description: "A cpal stream operation panicked while holding the lock - this is a bug, please report it".to_string(),
+                },
+            })?
+            .audio_unit
+            .start()?;
 
         Ok(stream)
     }

src/host/coreaudio/macos/enumerate.rs

@@ -73,9 +73,6 @@ impl Devices {
     }
 }
 
-unsafe impl Send for Devices {}
-unsafe impl Sync for Devices {}
-
 impl Iterator for Devices {
     type Item = Device;
     fn next(&mut self) -> Option<Device> {