nandi/jolt-native · mod.rs

⑂ cfd3e3677bed92e80cfe447469a249cfdc0b1401

mod.rs · 377 lines · 12.3 KBRust    Blame HistoryRaw
Lift freeq's AV media plane out of sleek   1#![allow(deprecated)]
2use super::{asbd_from_config, check_os_status, frames_to_duration, host_time_to_stream_instant};
4use super::OSStatus;
5use crate::host::coreaudio::macos::loopback::LoopbackDevice;
6use crate::traits::{HostTrait, StreamTrait};
7use crate::{BackendSpecificError, DevicesError, PauseStreamError, PlayStreamError};
8use coreaudio::audio_unit::AudioUnit;
9use objc2_core_audio::AudioDeviceID;
10use std::sync::{mpsc, Arc, Mutex, Weak};
12pub use self::enumerate::{default_input_device, default_output_device, Devices};
14use objc2_core_audio::{
  kAudioDevicePropertyDeviceIsAlive, kAudioObjectPropertyElementMain,
  kAudioObjectPropertyScopeGlobal, AudioObjectPropertyAddress,
18use property_listener::AudioObjectPropertyListener;
20mod device;
21pub mod enumerate;
22mod loopback;
23mod property_listener;
24pub use device::Device;
26/// Coreaudio host, the default host on macOS.
27#[derive(Debug)]
28pub struct Host;
30impl Host {
  pub fn new() -> Result<Self, crate::HostUnavailable> {
      Ok(Host)
36impl HostTrait for Host {
  type Devices = Devices;
  type Device = Device;
  fn is_available() -> bool {
      // Assume coreaudio is always available
      true
  fn devices(&self) -> Result<Self::Devices, DevicesError> {
      Devices::new()
  fn default_input_device(&self) -> Option<Self::Device> {
      default_input_device()
  fn default_output_device(&self) -> Option<Self::Device> {
      default_output_device()
58/// Type alias for the error callback to reduce complexity
59type ErrorCallback = Box<dyn FnMut(crate::StreamError) + Send + 'static>;
61/// Invoke error callback, recovering from poisoned mutex if needed.
62/// Returns true if callback was invoked, false if skipped due to WouldBlock.
63
64fn invoke_error_callback<E>(error_callback: &Arc<Mutex<E>>, err: crate::StreamError) -> bool
  E: FnMut(crate::StreamError) + Send,
  match error_callback.try_lock() {
      Ok(mut cb) => {
          cb(err);
          true
      Err(std::sync::TryLockError::Poisoned(guard)) => {
          // Recover from poisoned lock to still report this error
          guard.into_inner()(err);
          true
      Err(std::sync::TryLockError::WouldBlock) => {
          // Skip if callback is busy
          false
85/// Manages device disconnection listener on a dedicated thread to ensure the
86/// AudioObjectPropertyListener is always created and dropped on the same thread.
87/// This avoids potential threading issues with CoreAudio APIs.
89/// When a device disconnects, this manager:
90/// 1. Attempts to pause the stream to stop audio I/O
91/// 2. Calls the error callback with `StreamError::DeviceNotAvailable`
93/// The dedicated thread architecture ensures `Stream` can implement `Send`.
94struct DisconnectManager {
  _shutdown_tx: mpsc::Sender<()>,
98impl DisconnectManager {
  /// Create a new DisconnectManager that monitors device disconnection on a dedicated thread
  fn new(
      device_id: AudioDeviceID,
      stream_weak: Weak<Mutex<StreamInner>>,
      error_callback: Arc<Mutex<ErrorCallback>>,
  ) -> Result<Self, crate::BuildStreamError> {
      let (shutdown_tx, shutdown_rx) = mpsc::channel();
      let (disconnect_tx, disconnect_rx) = mpsc::channel();
      let (ready_tx, ready_rx) = mpsc::channel();
      // Spawn dedicated thread to own the AudioObjectPropertyListener
      let disconnect_tx_clone = disconnect_tx.clone();
      std::thread::spawn(move || {
          let property_address = AudioObjectPropertyAddress {
              mSelector: kAudioDevicePropertyDeviceIsAlive,
              mScope: kAudioObjectPropertyScopeGlobal,
              mElement: kAudioObjectPropertyElementMain,
          // Create the listener on this dedicated thread
          let disconnect_fn = move || {
              let _ = disconnect_tx_clone.send(());
          match AudioObjectPropertyListener::new(device_id, property_address, disconnect_fn) {
              Ok(_listener) => {
                  let _ = ready_tx.send(Ok(()));
                  // Drop the listener on this thread after receiving a shutdown signal
                  let _ = shutdown_rx.recv();
              Err(e) => {
                  let _ = ready_tx.send(Err(e));
      });
      // Wait for listener creation to complete or fail
      ready_rx
          .recv()
          .map_err(|_| crate::BuildStreamError::BackendSpecific {
              err: BackendSpecificError {
                  description: "Disconnect listener thread terminated unexpectedly".to_string(),
          })??;
      // Handle disconnect events on the main thread pool
      let stream_weak_clone = stream_weak.clone();
      let error_callback_clone = error_callback.clone();
      std::thread::spawn(move || {
          while disconnect_rx.recv().is_ok() {
              // Check if stream still exists
              if let Some(stream_arc) = stream_weak_clone.upgrade() {
                  // First, try to pause the stream to stop playback
                  if let Ok(mut stream_inner) = stream_arc.try_lock() {
                      let _ = stream_inner.pause();
                  // Always try to notify about device disconnection
                  invoke_error_callback(
                      &error_callback_clone,
                      crate::StreamError::DeviceNotAvailable,
              } else {
                  // Stream is gone, exit the handler thread
                  break;
      });
      Ok(DisconnectManager {
          _shutdown_tx: shutdown_tx,
173struct StreamInner {
  playing: bool,
  audio_unit: AudioUnit,
  // Track the device with which the audio unit was spawned.
  // We must do this so that we can avoid changing the device sample rate if there is already
  // a stream associated with the device.
  #[allow(dead_code)]
  device_id: AudioDeviceID,
  /// Manage the lifetime of the aggregate device used
  /// for loopback recording
  _loopback_device: Option<LoopbackDevice>,
187impl StreamInner {
  fn play(&mut self) -> Result<(), PlayStreamError> {
      if !self.playing {
          if let Err(e) = self.audio_unit.start() {
              let description = format!("{e}");
              let err = BackendSpecificError { description };
              return Err(err.into());
          self.playing = true;
      Ok(())
  fn pause(&mut self) -> Result<(), PauseStreamError> {
      if self.playing {
          if let Err(e) = self.audio_unit.stop() {
              let description = format!("{e}");
              let err = BackendSpecificError { description };
              return Err(err.into());
          self.playing = false;
      Ok(())
213pub struct Stream {
  inner: Arc<Mutex<StreamInner>>,
  // Manages the device disconnection listener separately to allow Stream to be Send.
  // The DisconnectManager contains the non-Send AudioObjectPropertyListener.
  _disconnect_manager: DisconnectManager,
220impl Stream {
  fn new(
      inner: StreamInner,
      error_callback: ErrorCallback,
  ) -> Result<Self, crate::BuildStreamError> {
      let device_id = inner.device_id;
      let inner_arc = Arc::new(Mutex::new(inner));
      let weak_inner = Arc::downgrade(&inner_arc);
      let error_callback = Arc::new(Mutex::new(error_callback));
      let disconnect_manager = DisconnectManager::new(device_id, weak_inner, error_callback)?;
      Ok(Self {
          inner: inner_arc,
          _disconnect_manager: disconnect_manager,
239impl StreamTrait for Stream {
  fn play(&self) -> Result<(), PlayStreamError> {
      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(),
          })?;
      stream.play()
  fn pause(&self) -> Result<(), PauseStreamError> {
      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(),
          })?;
      stream.pause()
  fn buffer_size(&self) -> Option<crate::FrameCount> {
      let stream = self.inner.lock().ok()?;
      device::get_device_buffer_frame_size(&stream.audio_unit)
          .ok()
          .map(|size| size as crate::FrameCount)
275#[cfg(test)]
276
  use crate::{
      default_host,
      traits::{DeviceTrait, HostTrait, StreamTrait},
      Sample,
  #[test]
  fn test_play() {
      let host = default_host();
      let device = host.default_output_device().unwrap();
      let mut supported_configs_range = device.supported_output_configs().unwrap();
      let supported_config = supported_configs_range
          .next()
          .unwrap()
          .with_max_sample_rate();
      let config = supported_config.config();
      let stream = device
          .build_output_stream(
              config,
              write_silence::<f32>,
              move |err| println!("Error: {err}"),
              None, // None=blocking, Some(Duration)=timeout
          .unwrap();
      stream.play().unwrap();
      std::thread::sleep(std::time::Duration::from_secs(1));
  #[test]
  fn test_record() {
      let host = default_host();
      let device = host.default_input_device().unwrap();
      println!("Device: {:?}", device.name());
      let mut supported_configs_range = device.supported_input_configs().unwrap();
      println!("Supported configs:");
      for config in supported_configs_range.clone() {
          println!("{:?}", config)
      let supported_config = supported_configs_range
          .next()
          .unwrap()
          .with_max_sample_rate();
      let config = supported_config.config();
      let stream = device
          .build_input_stream(
              config,
              move |data: &[f32], _: &crate::InputCallbackInfo| {
                  // react to stream events and read or write stream data here.
                  println!("Got data: {:?}", &data[..25]);
              move |err| println!("Error: {err}"),
              None, // None=blocking, Some(Duration)=timeout
          .unwrap();
      stream.play().unwrap();
      std::thread::sleep(std::time::Duration::from_secs(1));
  #[test]
  fn test_record_output() {
      if std::env::var("CI").is_ok() {
          println!("Skipping test_record_output in CI environment due to permissions");
          return;
      let host = default_host();
      let device = host.default_output_device().unwrap();
      let mut supported_configs_range = device.supported_output_configs().unwrap();
      let supported_config = supported_configs_range
          .next()
          .unwrap()
          .with_max_sample_rate();
      let config = supported_config.config();
      println!("Building input stream");
      let stream = device
          .build_input_stream(
              config,
              move |data: &[f32], _: &crate::InputCallbackInfo| {
                  // react to stream events and read or write stream data here.
                  println!("Got data: {:?}", &data[..25]);
              move |err| println!("Error: {err}"),
              None, // None=blocking, Some(Duration)=timeout
          .unwrap();
      stream.play().unwrap();
      std::thread::sleep(std::time::Duration::from_secs(1));
  fn write_silence<T: Sample>(data: &mut [T], _: &crate::OutputCallbackInfo) {
      for sample in data.iter_mut() {
          *sample = Sample::EQUILIBRIUM;

Lift freeq's AV media plane out of sleek 90f8b89 nandi 19d ago	1	#![allow(deprecated)]
	2	use super::{asbd_from_config, check_os_status, frames_to_duration, host_time_to_stream_instant};
	3
	4	use super::OSStatus;
	5	use crate::host::coreaudio::macos::loopback::LoopbackDevice;
	6	use crate::traits::{HostTrait, StreamTrait};
	7	use crate::{BackendSpecificError, DevicesError, PauseStreamError, PlayStreamError};
	8	use coreaudio::audio_unit::AudioUnit;
	9	use objc2_core_audio::AudioDeviceID;
	10	use std::sync::{mpsc, Arc, Mutex, Weak};
	11
	12	pub use self::enumerate::{default_input_device, default_output_device, Devices};
	13
	14	use objc2_core_audio::{
	15	kAudioDevicePropertyDeviceIsAlive, kAudioObjectPropertyElementMain,
	16	kAudioObjectPropertyScopeGlobal, AudioObjectPropertyAddress,
	17	};
	18	use property_listener::AudioObjectPropertyListener;
	19
	20	mod device;
	21	pub mod enumerate;
	22	mod loopback;
	23	mod property_listener;
	24	pub use device::Device;
	25
	26	/// Coreaudio host, the default host on macOS.
	27	#[derive(Debug)]
	28	pub struct Host;
	29
	30	impl Host {
	31	pub fn new() -> Result<Self, crate::HostUnavailable> {
	32	Ok(Host)
	33	}
	34	}
	35
	36	impl HostTrait for Host {
	37	type Devices = Devices;
	38	type Device = Device;
	39
	40	fn is_available() -> bool {
	41	// Assume coreaudio is always available
	42	true
	43	}
	44
	45	fn devices(&self) -> Result<Self::Devices, DevicesError> {
	46	Devices::new()
	47	}
	48
	49	fn default_input_device(&self) -> Option<Self::Device> {
	50	default_input_device()
	51	}
	52
	53	fn default_output_device(&self) -> Option<Self::Device> {
	54	default_output_device()
	55	}
	56	}
	57
	58	/// Type alias for the error callback to reduce complexity
	59	type ErrorCallback = Box<dyn FnMut(crate::StreamError) + Send + 'static>;
	60
	61	/// Invoke error callback, recovering from poisoned mutex if needed.
	62	/// Returns true if callback was invoked, false if skipped due to WouldBlock.
	63	#[inline]
	64	fn invoke_error_callback<E>(error_callback: &Arc<Mutex<E>>, err: crate::StreamError) -> bool
65	where
66	E: FnMut(crate::StreamError) + Send,
67	{
68	match error_callback.try_lock() {
69	Ok(mut cb) => {
70	cb(err);
71	true
72	}
73	Err(std::sync::TryLockError::Poisoned(guard)) => {
74	// Recover from poisoned lock to still report this error
75	guard.into_inner()(err);
76	true
77	}
78	Err(std::sync::TryLockError::WouldBlock) => {
79	// Skip if callback is busy
80	false
81	}
82	}
83	}
84
85	/// Manages device disconnection listener on a dedicated thread to ensure the
86	/// AudioObjectPropertyListener is always created and dropped on the same thread.
87	/// This avoids potential threading issues with CoreAudio APIs.
88	///
89	/// When a device disconnects, this manager:
90	/// 1. Attempts to pause the stream to stop audio I/O
91	/// 2. Calls the error callback with `StreamError::DeviceNotAvailable`
92	///
93	/// The dedicated thread architecture ensures `Stream` can implement `Send`.
94	struct DisconnectManager {
95	_shutdown_tx: mpsc::Sender<()>,
96	}
97
98	impl DisconnectManager {
99	/// Create a new DisconnectManager that monitors device disconnection on a dedicated thread
100	fn new(
101	device_id: AudioDeviceID,
102	stream_weak: Weak<Mutex<StreamInner>>,
103	error_callback: Arc<Mutex<ErrorCallback>>,
104	) -> Result<Self, crate::BuildStreamError> {
105	let (shutdown_tx, shutdown_rx) = mpsc::channel();
106	let (disconnect_tx, disconnect_rx) = mpsc::channel();
107	let (ready_tx, ready_rx) = mpsc::channel();
108
109	// Spawn dedicated thread to own the AudioObjectPropertyListener
110	let disconnect_tx_clone = disconnect_tx.clone();
111	std::thread::spawn(move \|\| {
112	let property_address = AudioObjectPropertyAddress {
113	mSelector: kAudioDevicePropertyDeviceIsAlive,
114	mScope: kAudioObjectPropertyScopeGlobal,
115	mElement: kAudioObjectPropertyElementMain,
116	};
117
118	// Create the listener on this dedicated thread
119	let disconnect_fn = move \|\| {
120	let _ = disconnect_tx_clone.send(());
121	};
122	match AudioObjectPropertyListener::new(device_id, property_address, disconnect_fn) {
123	Ok(_listener) => {
124	let _ = ready_tx.send(Ok(()));
125	// Drop the listener on this thread after receiving a shutdown signal
126	let _ = shutdown_rx.recv();
127	}
128	Err(e) => {
129	let _ = ready_tx.send(Err(e));
130	}
131	}
132	});
133
134	// Wait for listener creation to complete or fail
135	ready_rx
136	.recv()
137	.map_err(\|_\| crate::BuildStreamError::BackendSpecific {
138	err: BackendSpecificError {
139	description: "Disconnect listener thread terminated unexpectedly".to_string(),
140	},
141	})??;
142
143	// Handle disconnect events on the main thread pool
144	let stream_weak_clone = stream_weak.clone();
145	let error_callback_clone = error_callback.clone();
146	std::thread::spawn(move \|\| {
147	while disconnect_rx.recv().is_ok() {
148	// Check if stream still exists
149	if let Some(stream_arc) = stream_weak_clone.upgrade() {
150	// First, try to pause the stream to stop playback
151	if let Ok(mut stream_inner) = stream_arc.try_lock() {
152	let _ = stream_inner.pause();
153	}
154
155	// Always try to notify about device disconnection
156	invoke_error_callback(
157	&error_callback_clone,
158	crate::StreamError::DeviceNotAvailable,
159	);
160	} else {
161	// Stream is gone, exit the handler thread
162	break;
163	}
164	}
165	});
166
167	Ok(DisconnectManager {
168	_shutdown_tx: shutdown_tx,
169	})
170	}
171	}
172
173	struct StreamInner {
174	playing: bool,
175	audio_unit: AudioUnit,
176	// Track the device with which the audio unit was spawned.
177	//
178	// We must do this so that we can avoid changing the device sample rate if there is already
179	// a stream associated with the device.
180	#[allow(dead_code)]
181	device_id: AudioDeviceID,
182	/// Manage the lifetime of the aggregate device used
183	/// for loopback recording
184	_loopback_device: Option<LoopbackDevice>,
185	}
186
187	impl StreamInner {
188	fn play(&mut self) -> Result<(), PlayStreamError> {
189	if !self.playing {
190	if let Err(e) = self.audio_unit.start() {
191	let description = format!("{e}");
192	let err = BackendSpecificError { description };
193	return Err(err.into());
194	}
195	self.playing = true;
196	}
197	Ok(())
198	}
199
200	fn pause(&mut self) -> Result<(), PauseStreamError> {
201	if self.playing {
202	if let Err(e) = self.audio_unit.stop() {
203	let description = format!("{e}");
204	let err = BackendSpecificError { description };
205	return Err(err.into());
206	}
207	self.playing = false;
208	}
209	Ok(())
210	}
211	}
212
213	pub struct Stream {
214	inner: Arc<Mutex<StreamInner>>,
215	// Manages the device disconnection listener separately to allow Stream to be Send.
216	// The DisconnectManager contains the non-Send AudioObjectPropertyListener.
217	_disconnect_manager: DisconnectManager,
218	}
219
220	impl Stream {
221	fn new(
222	inner: StreamInner,
223	error_callback: ErrorCallback,
224	) -> Result<Self, crate::BuildStreamError> {
225	let device_id = inner.device_id;
226	let inner_arc = Arc::new(Mutex::new(inner));
227	let weak_inner = Arc::downgrade(&inner_arc);
228
229	let error_callback = Arc::new(Mutex::new(error_callback));
230	let disconnect_manager = DisconnectManager::new(device_id, weak_inner, error_callback)?;
231
232	Ok(Self {
233	inner: inner_arc,
234	_disconnect_manager: disconnect_manager,
235	})
236	}
237	}
238
239	impl StreamTrait for Stream {
240	fn play(&self) -> Result<(), PlayStreamError> {
241	let mut stream = self
242	.inner
243	.lock()
244	.map_err(\|_\| PlayStreamError::BackendSpecific {
245	err: BackendSpecificError {
246	description: "A cpal stream operation panicked while holding the lock - this is a bug, please report it".to_string(),
247	},
248	})?;
249
250	stream.play()
251	}
252
253	fn pause(&self) -> Result<(), PauseStreamError> {
254	let mut stream = self
255	.inner
256	.lock()
257	.map_err(\|_\| PauseStreamError::BackendSpecific {
258	err: BackendSpecificError {
259	description: "A cpal stream operation panicked while holding the lock - this is a bug, please report it".to_string(),
260	},
261	})?;
262
263	stream.pause()
264	}
265
266	fn buffer_size(&self) -> Option<crate::FrameCount> {
267	let stream = self.inner.lock().ok()?;
268
269	device::get_device_buffer_frame_size(&stream.audio_unit)
270	.ok()
271	.map(\|size\| size as crate::FrameCount)
272	}
273	}
274
275	#[cfg(test)]
276	mod test {
277	use crate::{
278	default_host,
279	traits::{DeviceTrait, HostTrait, StreamTrait},
280	Sample,
281	};
282
283	#[test]
284	fn test_play() {
285	let host = default_host();
286	let device = host.default_output_device().unwrap();
287
288	let mut supported_configs_range = device.supported_output_configs().unwrap();
289	let supported_config = supported_configs_range
290	.next()
291	.unwrap()
292	.with_max_sample_rate();
293	let config = supported_config.config();
294
295	let stream = device
296	.build_output_stream(
297	config,
298	write_silence::<f32>,
299	move \|err\| println!("Error: {err}"),
300	None, // None=blocking, Some(Duration)=timeout
301	)
302	.unwrap();
303	stream.play().unwrap();
304	std::thread::sleep(std::time::Duration::from_secs(1));
305	}
306
307	#[test]
308	fn test_record() {
309	let host = default_host();
310	let device = host.default_input_device().unwrap();
311	println!("Device: {:?}", device.name());
312
313	let mut supported_configs_range = device.supported_input_configs().unwrap();
314	println!("Supported configs:");
315	for config in supported_configs_range.clone() {
316	println!("{:?}", config)
317	}
318	let supported_config = supported_configs_range
319	.next()
320	.unwrap()
321	.with_max_sample_rate();
322	let config = supported_config.config();
323
324	let stream = device
325	.build_input_stream(
326	config,
327	move \|data: &[f32], _: &crate::InputCallbackInfo\| {
328	// react to stream events and read or write stream data here.
329	println!("Got data: {:?}", &data[..25]);
330	},
331	move \|err\| println!("Error: {err}"),
332	None, // None=blocking, Some(Duration)=timeout
333	)
334	.unwrap();
335	stream.play().unwrap();
336	std::thread::sleep(std::time::Duration::from_secs(1));
337	}
338
339	#[test]
340	fn test_record_output() {
341	if std::env::var("CI").is_ok() {
342	println!("Skipping test_record_output in CI environment due to permissions");
343	return;
344	}
345
346	let host = default_host();
347	let device = host.default_output_device().unwrap();
348
349	let mut supported_configs_range = device.supported_output_configs().unwrap();
350	let supported_config = supported_configs_range
351	.next()
352	.unwrap()
353	.with_max_sample_rate();
354	let config = supported_config.config();
355
356	println!("Building input stream");
357	let stream = device
358	.build_input_stream(
359	config,
360	move \|data: &[f32], _: &crate::InputCallbackInfo\| {
361	// react to stream events and read or write stream data here.
362	println!("Got data: {:?}", &data[..25]);
363	},
364	move \|err\| println!("Error: {err}"),
365	None, // None=blocking, Some(Duration)=timeout
366	)
367	.unwrap();
368	stream.play().unwrap();
369	std::thread::sleep(std::time::Duration::from_secs(1));
370	}
371
372	fn write_silence<T: Sample>(data: &mut [T], _: &crate::OutputCallbackInfo) {
373	for sample in data.iter_mut() {
374	*sample = Sample::EQUILIBRIUM;
375	}
376	}
377	}