From 39a3cd563bd9ec61c4a774ee2ac520983e4ff6ce Mon Sep 17 00:00:00 2001 From: ShootingStarDragons Date: Tue, 14 Jul 2026 12:31:14 +0900 Subject: [PATCH] feat: duplex support for pipewire still in draft, I need to test more --- src/duplex.rs | 4 +- src/host/mod.rs | 17 ++ src/host/pipewire/device.rs | 258 +++++++++++++++- src/host/pipewire/stream.rs | 567 +++++++++++++++++++++++++++++++++++- src/lib.rs | 54 ++++ 5 files changed, 888 insertions(+), 12 deletions(-) diff --git a/src/duplex.rs b/src/duplex.rs index 3800172ff..e020f8e65 100644 --- a/src/duplex.rs +++ b/src/duplex.rs @@ -8,8 +8,8 @@ use crate::{BufferSize, CallbackInfo, ChannelCount, SampleRate}; /// same invocation. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] pub struct DuplexCallbackInfo { - input: CallbackInfo, - output: CallbackInfo, + pub(crate) input: CallbackInfo, + pub(crate) output: CallbackInfo, } impl DuplexCallbackInfo { diff --git a/src/host/mod.rs b/src/host/mod.rs index 3ebb3380e..8ec87a1ce 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -299,6 +299,23 @@ where } } +#[allow(dead_code)] +pub(crate) fn monotonic_duplex_callback( + mut data_callback: D, +) -> impl FnMut(&crate::Data, &mut crate::Data, &crate::DuplexCallbackInfo) + Send + 'static +where + D: FnMut(&crate::Data, &mut crate::Data, &crate::DuplexCallbackInfo) + Send + 'static, +{ + // FnMut runs on one thread at a time, so the floor needs no synchronization. + let mut floor = 0u64; + move |data_out, data_in, info| { + let mut info = *info; + info.input.timestamp.device = non_decreasing(&mut floor, info.input.timestamp.device); + info.output.timestamp.device = non_decreasing(&mut floor, info.output.timestamp.device); + data_callback(data_out, data_in, &info); + } +} + /// Wraps an output data callback so the `playback` timestamp never regresses across callbacks. #[allow(dead_code)] pub(crate) fn monotonic_output_callback( diff --git a/src/host/pipewire/device.rs b/src/host/pipewire/device.rs index 9402f69ec..f8c27bd07 100644 --- a/src/host/pipewire/device.rs +++ b/src/host/pipewire/device.rs @@ -28,9 +28,9 @@ use pipewire::{ use super::stream::Stream; use crate::{ BufferSize, CallbackInfo, ChannelCount, Data, DeviceDescription, DeviceDescriptionBuilder, - DeviceDirection, DeviceId, DeviceType, Error, ErrorKind, FrameCount, HostId, InterfaceType, - SampleFormat, SampleRate, StreamConfig, SupportedBufferSize, SupportedStreamConfig, - SupportedStreamConfigRange, + DeviceDirection, DeviceId, DeviceType, DuplexStreamConfig, Error, ErrorKind, FrameCount, + HostId, InterfaceType, SampleFormat, SampleRate, StreamConfig, SupportedBufferSize, + SupportedStreamConfig, SupportedStreamConfigRange, host::{ Notify, emit_error, latch::Latch, @@ -152,7 +152,41 @@ impl Device { Class::Node => None, } } + pub(crate) fn pw_properties_duplx( + &self, + direction: DeviceDirection, + config: &DuplexStreamConfig, + ) -> PropertiesBox { + let mut properties = match direction { + DeviceDirection::Output => pw::properties::properties! { + *pw::keys::MEDIA_TYPE => "Audio", + *pw::keys::MEDIA_CATEGORY => "Playback", + }, + DeviceDirection::Input => pw::properties::properties! { + *pw::keys::MEDIA_TYPE => "Audio", + *pw::keys::MEDIA_CATEGORY => "Capture", + }, + _ => unreachable!(), + }; + if matches!(self.role, Role::Sink) && matches!(direction, DeviceDirection::Input) { + properties.insert(*pw::keys::STREAM_CAPTURE_SINK, "true"); + } + if matches!(self.class, Class::Node) { + properties.insert(*pw::keys::TARGET_OBJECT, self.object_serial.to_string()); + } + + // Group input and output nodes so PipeWire schedules them in the same quantum, + // preventing phase drift between simultaneous input/output streams. + properties.insert("node.group", format!("cpal-{}", std::process::id())); + if let BufferSize::Fixed(buffer_size) = config.buffer_size { + properties.insert( + *pw::keys::NODE_LATENCY, + format!("{buffer_size}/{rate}", rate = config.sample_rate), + ); + } + properties + } pub(crate) fn pw_properties( &self, direction: DeviceDirection, @@ -251,6 +285,10 @@ impl DeviceTrait for Device { ) } + fn supports_duplex(&self) -> bool { + matches!(self.direction, DeviceDirection::Duplex) + } + fn supported_input_configs(&self) -> Result { if !self.supports_input() { return Ok(vec![].into_iter()); @@ -355,7 +393,7 @@ impl DeviceTrait for Device { D: FnMut(&Data, &CallbackInfo) + Send + 'static, E: FnMut(Error) + Send + 'static, { - crate::validate_stream_config(&config)?; + config.validate()?; if let BufferSize::Fixed(n) = config.buffer_size { // When max_quantum is 0 the server clock metadata has not been received yet. if self.max_quantum > 0 && !(self.min_quantum..=self.max_quantum).contains(&n) { @@ -547,7 +585,7 @@ impl DeviceTrait for Device { D: FnMut(&mut Data, &CallbackInfo) + Send + 'static, E: FnMut(Error) + Send + 'static, { - crate::validate_stream_config(&config)?; + config.validate()?; if let BufferSize::Fixed(n) = config.buffer_size { // When max_quantum is 0 the server clock metadata has not been received yet. if self.max_quantum > 0 && !(self.min_quantum..=self.max_quantum).contains(&n) { @@ -741,6 +779,216 @@ impl DeviceTrait for Device { stream.signal_ready(); Ok(stream) } + + fn build_duplex_stream_raw( + &self, + config: crate::DuplexStreamConfig, + input_sample_format: SampleFormat, + output_sample_format: SampleFormat, + data_callback: D, + error_callback: E, + timeout: Option, + ) -> Result + where + D: FnMut(&Data, &mut Data, &crate::DuplexCallbackInfo) + Send + 'static, + E: FnMut(Error) + Send + 'static, + { + config.validate()?; + if let BufferSize::Fixed(n) = config.buffer_size { + // When max_quantum is 0 the server clock metadata has not been received yet. + if self.max_quantum > 0 && !(self.min_quantum..=self.max_quantum).contains(&n) { + return Err(Error::with_message( + ErrorKind::UnsupportedConfig, + format!( + "Buffer size {n} is not in the supported quantum range {min}..={max}", + min = self.min_quantum, + max = self.max_quantum + ), + )); + } + } + let (pw_play_tx, pw_play_rx) = pw::channel::channel::(); + + let (init_tx, init_rx) = mpsc::channel::>(); + let mut latch = Latch::new(); + let waiter = latch.waiter(); + let device = self.clone(); + let wait_timeout = timeout.unwrap_or(Duration::from_secs(2)); + let initial_quantum = match config.buffer_size { + BufferSize::Fixed(n) => n as u64, + BufferSize::Default => self.quantum as u64, + }; + let last_quantum = Arc::new(AtomicU64::new(initial_quantum)); + let last_quantum_clone = last_quantum.clone(); + let draining = Arc::new(AtomicBool::new(false)); + let draining_clone = draining.clone(); + + let drained: Arc = Arc::new(Notify::default()); + let drained_clone = drained.clone(); + let drained_cmd = drained.clone(); + // Keep `capture` monotonic: pw_time delay() grows when another client joins + // needing a larger buffer, which can pull `capture` backward. + let data_callback = crate::host::monotonic_duplex_callback(data_callback); + let start = std::time::Instant::now(); + let handle = thread::Builder::new() + .name("pw_out".to_owned()) + .spawn(move || { + let _pw = PwInitGuard::new(); + let properties = device.pw_properties_duplx(DeviceDirection::Output, &config); + + let stream_data = match super::stream::connect_duplex( + super::stream::ConnectDuplexParams { + config, + properties, + sample_format_in: input_sample_format, + sample_format_out: output_sample_format, + last_quantum: last_quantum_clone, + start, + connect_automatically: device.connect_automatically.load(Ordering::Relaxed), + draining: draining_clone, + drained: Some(drained_clone), + is_default_device: matches!( + device.class(), + Class::DefaultSink | Class::DefaultInput | Class::DefaultOutput + ), + }, + data_callback, + error_callback, + ) { + Ok(d) => d, + Err(e) => { + let _ = init_tx.send(Err(Error::with_message( + ErrorKind::UnsupportedConfig, + format!("PipeWire stream connection failed: {e}"), + ))); + return; + } + }; + + let StreamData { + mainloop, + listener, + stream, + context, + core, + core_monitor, + error_callback, + pending_device_changed, + invalidated, + } = stream_data; + + let default_monitor = if let Some(key) = device.default_metadata_key() { + match core.get_registry_rc() { + Ok(registry) => Some(DefaultDeviceMonitor::new( + registry, + key, + error_callback.clone(), + invalidated, + pending_device_changed, + )), + Err(e) => { + let _ = init_tx.send(Err(Error::with_message( + ErrorKind::BackendError, + format!("Could not acquire registry: {e}"), + ))); + return; + } + } + } else { + None + }; + let stream_clone = stream.clone(); + let mainloop_rc1 = mainloop.clone(); + let error_callback_cmd = error_callback.clone(); + let _receiver = pw_play_rx.attach(mainloop.loop_(), move |play| match play { + StreamCommand::Toggle(state) => { + if let Err(e) = stream_clone.set_active(state) { + emit_error( + &error_callback_cmd, + Error::with_message( + ErrorKind::StreamInvalidated, + format!("Failed to set stream active ({state}): {e}"), + ), + ); + } + } + StreamCommand::Drain => { + if let Err(e) = stream_clone.flush(true) { + emit_error( + &error_callback_cmd, + Error::with_message( + ErrorKind::StreamInvalidated, + format!("Stream flush failed: {e}"), + ), + ); + let (mutex, cvar) = drained_cmd.as_ref(); + *mutex.lock().unwrap_or_else(|g| g.into_inner()) = true; + cvar.notify_one(); + } + } + StreamCommand::Stop => { + if let Err(e) = stream_clone.disconnect() { + emit_error( + &error_callback_cmd, + Error::with_message( + ErrorKind::StreamInvalidated, + format!("Stream disconnect failed: {e}"), + ), + ); + } + mainloop_rc1.quit(); + } + }); + + if init_tx.send(Ok(())).is_err() { + return; + } + + // If the Latch is dropped without being released (error path), exit cleanly. + if !waiter.wait() { + return; + } + + mainloop.run(); + drop(listener); + drop(default_monitor); + drop(core_monitor); + drop(core); + drop(context); + }) + .map_err(|e| { + Error::with_message( + ErrorKind::ResourceExhausted, + format!("Failed to create thread: {e}"), + ) + })?; + + let init_result = init_rx.recv_timeout(wait_timeout).unwrap_or_else(|_| { + Err(Error::with_message( + ErrorKind::DeviceNotAvailable, + "PipeWire timed out", + )) + }); + + if let Err(e) = init_result { + drop(latch); + return Err(e); + } + + latch.add_thread(handle.thread().clone()); + let stream = Stream::new( + handle, + pw_play_tx, + last_quantum, + start, + latch, + true, + draining, + Some(drained), + ); + stream.signal_ready(); + Ok(stream) + } } #[derive(Clone, Default)] diff --git a/src/host/pipewire/stream.rs b/src/host/pipewire/stream.rs index 895096129..9a39de804 100644 --- a/src/host/pipewire/stream.rs +++ b/src/host/pipewire/stream.rs @@ -33,8 +33,8 @@ use pipewire::{ }; use crate::{ - CallbackInfo, Data, Error, ErrorKind, FrameCount, SampleFormat, StreamConfig, StreamInstant, - StreamTimestamp, + CallbackInfo, Data, DuplexCallbackInfo, DuplexStreamConfig, Error, ErrorKind, FrameCount, + SampleFormat, StreamConfig, StreamInstant, StreamTimestamp, host::{ ErrorCallbackArc, Notify, emit_error, equilibrium::fill_equilibrium, frames_to_duration, latch::Latch, try_emit_error, @@ -469,9 +469,107 @@ where } } +impl UserData>> +where + D: FnMut(&Data, &mut Data, &crate::DuplexCallbackInfo) + Send + 'static, +{ + fn publish_data_duplex( + &mut self, + stream: &pw::stream::Stream, + frames: usize, + data_in: &Data, + data_out: &mut Data, + xrun: bool, + ) { + #[cfg(feature = "realtime")] + { + let prev = self.last_quantum.swap(frames as u64, Ordering::Relaxed); + if !self.rt_promoted || frames as u64 != prev { + self.promote_realtime(frames as FrameCount); + } + } + + let mut data_callback = self.data_callback.lock().unwrap_or_else(|e| e.into_inner()); + #[cfg(not(feature = "realtime"))] + self.last_quantum.store(frames as u64, Ordering::Relaxed); + + if !self.draining.load(Ordering::Relaxed) { + let (callback, playback) = match pw_stream_time(stream) { + Some(t) => { + // `pw_stream_time` guarantees `now > 0` and `denom != 0`. + let now = t.now() as u64; + let delay_ns = + (t.delay() * 1_000_000_000 / t.rate().denom as i64).max(0) as u64; + ( + StreamInstant::from_nanos(now), + StreamInstant::from_nanos(now.saturating_add(delay_ns)), + ) + } + None => { + let cb = monotonic_stream_instant() + .unwrap_or_else(|| stream_instant_from_start(self.start)); + let pl = cb + frames_to_duration(frames as FrameCount, self.format.rate()); + (cb, pl) + } + }; + let timestamp = StreamTimestamp { + callback, + device: playback, + }; + data_callback( + data_in, + data_out, + &DuplexCallbackInfo::new( + CallbackInfo { timestamp, xrun }, + CallbackInfo { timestamp, xrun }, + ), + ) + //(self.data_callback)(data, &CallbackInfo { timestamp, xrun }); + } + } + + fn publish_data_duplex_in( + &mut self, + stream: &pw::stream::Stream, + frames: usize, + data_in: &Data, + xrun: bool, + ) { + let mut data_out_raw: Vec = vec![]; + let data = data_out_raw.as_mut_ptr() as *mut (); + + let mut data_out = unsafe { Data::from_parts(data, 0, self.sample_format) }; + + self.publish_data_duplex(stream, frames, data_in, &mut data_out, xrun); + } + fn publish_data_duplex_out( + &mut self, + stream: &pw::stream::Stream, + frames: usize, + data_out: &mut Data, + xrun: bool, + ) { + let data_in_raw: Vec = vec![]; + let data = data_in_raw.as_ptr() as *mut (); + + let data_in = unsafe { Data::from_parts(data, 0, self.sample_format) }; + + self.publish_data_duplex(stream, frames, &data_in, data_out, xrun); + } +} + +#[allow(unused)] +pub enum PwListener { + Single(StreamListener>), + Duplex { + input: StreamListener>, + output: StreamListener>, + }, +} + pub struct StreamData { pub mainloop: MainLoopRc, - pub listener: StreamListener>, + pub listener: PwListener, pub stream: StreamRc, pub context: ContextRc, pub core: CoreRc, @@ -632,6 +730,19 @@ pub struct ConnectParams { pub is_default_device: bool, } +pub struct ConnectDuplexParams { + pub config: DuplexStreamConfig, + pub properties: PropertiesBox, + pub sample_format_in: SampleFormat, + pub sample_format_out: SampleFormat, + pub last_quantum: Arc, + pub start: Instant, + pub connect_automatically: bool, + pub draining: Arc, + pub drained: Option>, + pub is_default_device: bool, +} + pub fn connect_output( params: ConnectParams, data_callback: D, @@ -887,7 +998,7 @@ where Ok(StreamData { mainloop, - listener, + listener: PwListener::Single(listener), stream, context, core, @@ -1129,7 +1240,453 @@ where Ok(StreamData { mainloop, - listener, + listener: PwListener::Single(listener), + stream, + context, + core, + core_monitor, + error_callback: error_callback_out, + pending_device_changed, + invalidated, + }) +} + +pub fn connect_duplex( + params: ConnectDuplexParams, + data_callback: D, + error_callback: E, +) -> Result>>, pw::Error> +where + D: FnMut(&Data, &mut Data, &crate::DuplexCallbackInfo) + Send + 'static, + E: FnMut(Error) + Send + 'static, +{ + let ConnectDuplexParams { + config, + properties, + sample_format_in, + sample_format_out, + last_quantum, + start, + connect_automatically, + draining, + drained, + is_default_device, + } = params; + + let mainloop = MainLoopRc::new(None)?; + let context = ContextRc::new(&mainloop, None)?; + let core = context.connect_rc(remote_props())?; + + let error_callback: ErrorCallbackArc = Arc::new(Mutex::new(error_callback)); + let data_callback = Arc::new(Mutex::new(data_callback)); + let invalidated = Arc::new(AtomicBool::new(false)); + + let pending_device_changed = Arc::new(AtomicBool::new(false)); + + let core_monitor = { + let invalidated_core = invalidated.clone(); + let error_callback_core = error_callback.clone(); + core.add_listener_local() + .error(move |id, _seq, _res, message| { + if id == PW_ID_CORE && !invalidated_core.swap(true, Ordering::Relaxed) { + emit_error( + &error_callback_core, + Error::with_message( + ErrorKind::StreamInvalidated, + format!("PipeWire server error: {message}"), + ), + ); + } + }) + .register() + }; + + let error_callback_2 = error_callback.clone(); + let error_callback_out = error_callback.clone(); + let data_callback_2 = data_callback.clone(); + let data_in = UserData { + data_callback, + error_callback, + sample_format: sample_format_in, + format: Default::default(), + last_quantum: last_quantum.clone(), + start, + draining: draining.clone(), + invalidated: invalidated.clone(), + is_default_device, + has_connected: false, + pending_device_changed: pending_device_changed.clone(), + spa_io_clock: std::ptr::null(), + xrun_recovering: false, + #[cfg(feature = "realtime")] + rt_promoted: false, + }; + + let channels = config.input_channels as _; + let rate = config.sample_rate as _; + let stream = StreamRc::new( + core.clone(), + &format!("cpal-capture-{}", std::process::id()), + properties, + )?; + let listener_in = stream + .add_local_listener_with_user_data(data_in) + .param_changed(move |stream, user_data, id, param| { + let Some(param) = param else { + return; + }; + if id != ParamType::Format.as_raw() { + return; + } + + let (media_type, media_subtype) = match format_utils::parse_format(param) { + Ok(v) => v, + Err(_) => return, + }; + + // only accept raw audio + if media_type != MediaType::Audio || media_subtype != MediaSubtype::Raw { + return; + } + + // call a helper function to parse the format for us. + // When the format update, we check the format first, in case it does not fit what we + // set + match user_data.format.parse(param) { + Ok(_) => { + let current_channels = user_data.format.channels(); + let current_rate = user_data.format.rate(); + let expected_fmt = + AudioFormat::from(user_data.sample_format); + let current_fmt = user_data.format.format(); + let mismatch = current_channels != channels + || current_rate != rate + || current_fmt != expected_fmt; + if mismatch && !user_data.invalidated.swap(true, Ordering::Relaxed) { + let fmt_note = if current_fmt != expected_fmt { + "; sample format differs" + } else { + "" + }; + emit_error( + &user_data.error_callback, + Error::with_message( + ErrorKind::UnsupportedConfig, + format!("Negotiated format mismatch: expected {channels} channels at {rate} Hz, got {current_channels} channels at {current_rate} Hz{fmt_note}"), + ), + ); + if let Err(e) = stream.set_active(false) { + emit_error( + &user_data.error_callback, + Error::with_message( + ErrorKind::StreamInvalidated, + format!("Failed to stop stream: {e}"), + ), + ); + } + } + } + Err(e) => { + if !user_data.invalidated.swap(true, Ordering::Relaxed) { + emit_error( + &user_data.error_callback, + Error::with_message( + ErrorKind::StreamInvalidated, + format!("Failed to parse negotiated audio format: {e}"), + ), + ); + if let Err(e) = stream.set_active(false) { + emit_error( + &user_data.error_callback, + Error::with_message( + ErrorKind::StreamInvalidated, + format!("Failed to stop stream: {e}"), + ), + ); + } + } + } + } + }) + .state_changed(|_stream, user_data, _old, new| { + user_data.state_changed(new); + }) + .io_changed(|_stream, user_data, id, area, _size| { + if id == SPA_IO_Clock { + user_data.spa_io_clock = area as *const spa_io_clock; + } + }) + .process(|stream, user_data| { + if user_data.pending_device_changed.load(Ordering::Relaxed) + && try_emit_error( + &user_data.error_callback, + Error::with_message(ErrorKind::DeviceChanged, "Default device changed"), + ) + .is_ok() + { + user_data.pending_device_changed.store(false, Ordering::Relaxed); + } + let xrun = user_data.check_xrun(); + + let n_channels = user_data.format.channels(); + if n_channels == 0 { + return; // format not yet negotiated by param_changed + } + + if let Some(mut buffer) = stream.dequeue_buffer() { + let datas = buffer.datas_mut(); + if datas.is_empty() { + return; + } + let data = &mut datas[0]; + let n_samples = data.chunk().size() / user_data.sample_format.sample_size() as u32; + let frames = n_samples / n_channels; + + let Some(samples) = data.data() else { + return; + }; + let data = samples.as_mut_ptr() as *mut (); + let data = + unsafe { Data::from_parts(data, n_samples as usize, user_data.sample_format) }; + user_data.publish_data_duplex_in(stream, frames as usize, &data, xrun); + } + }) + .register()?; + let data_out = UserData { + data_callback: data_callback_2, + error_callback: error_callback_2, + sample_format: sample_format_out, + format: Default::default(), + last_quantum, + start, + draining, + invalidated: invalidated.clone(), + is_default_device, + has_connected: false, + pending_device_changed: pending_device_changed.clone(), + spa_io_clock: std::ptr::null(), + xrun_recovering: false, + #[cfg(feature = "realtime")] + rt_promoted: false, + }; + let channels = config.output_channels as _; + let rate = config.sample_rate as _; + let listener_out = stream + .add_local_listener_with_user_data(data_out) + .param_changed(move |stream, user_data, id, param| { + let Some(param) = param else { + return; + }; + if id != ParamType::Format.as_raw() { + return; + } + + let (media_type, media_subtype) = match format_utils::parse_format(param) { + Ok(v) => v, + Err(_) => return, + }; + + // only accept raw audio + if media_type != MediaType::Audio || media_subtype != MediaSubtype::Raw { + return; + } + // call a helper function to parse the format for us. + // When the format update, we check the format first, in case it does not fit what we + // set + match user_data.format.parse(param) { + Ok(_) => { + let current_channels = user_data.format.channels(); + let current_rate = user_data.format.rate(); + let expected_fmt = + AudioFormat::from(user_data.sample_format); + let current_fmt = user_data.format.format(); + let mismatch = current_channels != channels + || current_rate != rate + || current_fmt != expected_fmt; + if mismatch && !user_data.invalidated.swap(true, Ordering::Relaxed) { + let fmt_note = if current_fmt != expected_fmt { + "; sample format differs" + } else { + "" + }; + emit_error( + &user_data.error_callback, + Error::with_message( + ErrorKind::UnsupportedConfig, + format!("Negotiated format mismatch: expected {channels} channels at {rate} Hz, got {current_channels} channels at {current_rate} Hz{fmt_note}"), + ), + ); + if let Err(e) = stream.set_active(false) { + emit_error( + &user_data.error_callback, + Error::with_message( + ErrorKind::StreamInvalidated, + format!("Failed to stop stream: {e}"), + ), + ); + } + } + } + Err(e) => { + if !user_data.invalidated.swap(true, Ordering::Relaxed) { + emit_error( + &user_data.error_callback, + Error::with_message( + ErrorKind::StreamInvalidated, + format!("Failed to parse negotiated audio format: {e}"), + ), + ); + if let Err(e) = stream.set_active(false) { + emit_error( + &user_data.error_callback, + Error::with_message( + ErrorKind::StreamInvalidated, + format!("Failed to stop stream: {e}"), + ), + ); + } + } + } + } + }) + .state_changed(|_stream, user_data, _old, new| { + user_data.state_changed(new); + }) + .io_changed(|_stream, user_data, id, area, _size| { + if id == SPA_IO_Clock { + user_data.spa_io_clock = area as *const spa_io_clock; + } + }) + .process(|stream, user_data| { + if user_data.pending_device_changed.load(Ordering::Relaxed) + && try_emit_error( + &user_data.error_callback, + Error::with_message(ErrorKind::DeviceChanged, "Default device changed"), + ) + .is_ok() + { + user_data.pending_device_changed.store(false, Ordering::Relaxed); + } + let xrun = user_data.check_xrun(); + + let n_channels = user_data.format.channels(); + if n_channels == 0 { + return; // format not yet negotiated by param_changed + } + + if let Some(mut buffer) = stream.dequeue_buffer() { + // Read the requested frame count before mutably borrowing datas_mut(). + let requested = buffer.requested() as usize; + let datas = buffer.datas_mut(); + if datas.is_empty() { + return; + } + let buf_data = &mut datas[0]; + + let stride = user_data.sample_format.sample_size() * n_channels as usize; + // frames = samples / channels or frames = data_len / stride + // Honor the frame count PipeWire requests this cycle, capped by the + // mapped buffer capacity to guard against any mismatch. + let frames = requested.min(buf_data.as_raw().maxsize as usize / stride); + let Some(samples) = buf_data.data() else { + return; + }; + + // samples = frames * channels or samples = data_len / sample_size + let n_samples = frames * n_channels as usize; + + // Pre-fill only the active region with equilibrium before handing it to the + // callback. + let active = &mut samples[..frames * stride]; + fill_equilibrium(active, user_data.sample_format); + + let data = active.as_mut_ptr() as *mut (); + let mut data = + unsafe { Data::from_parts(data, n_samples, user_data.sample_format) }; + user_data.publish_data_duplex_out(stream, frames, &mut data, xrun); + let chunk = buf_data.chunk_mut(); + *chunk.offset_mut() = 0; + *chunk.stride_mut() = stride as i32; + *chunk.size_mut() = (frames * stride) as u32; + } + }) + .drained(move |_stream, _user_data| { + if let Some(drained) = &drained { + let (mutex, cvar) = drained.as_ref(); + *mutex.lock().unwrap_or_else(|e| e.into_inner()) = true; + cvar.notify_one(); + } + }) + .register()?; + + let mut audio_info = AudioInfoRaw::new(); + audio_info.set_format(sample_format_in.into()); + audio_info.set_rate(rate); + audio_info.set_channels(channels); + + let obj = Object { + type_: SpaTypes::ObjectParamFormat.as_raw(), + id: ParamType::EnumFormat.as_raw(), + properties: audio_info.into(), + }; + let values: Vec = + PodSerializer::serialize(std::io::Cursor::new(Vec::new()), &Value::Object(obj)) + .unwrap() + .0 + .into_inner(); + + let mut params_in = [Pod::from_bytes(&values).unwrap()]; + + // RT_PROCESS is intentionally absent: with add_local_listener the process callback always + // runs on this mainloop thread, not the separate data-loop thread RT_PROCESS creates. + // That thread promotes itself to RT from the process callback, once when the negotiated + // quantum is known and again whenever PipeWire renegotiates it. + let mut flags = StreamFlags::MAP_BUFFERS | StreamFlags::INACTIVE; + if connect_automatically { + flags |= StreamFlags::AUTOCONNECT; + } + if !is_default_device { + flags |= StreamFlags::DONT_RECONNECT; + } + + stream.connect(Direction::Input, None, flags, &mut params_in)?; + let mut audio_info = AudioInfoRaw::new(); + audio_info.set_format(sample_format_out.into()); + audio_info.set_rate(rate); + audio_info.set_channels(channels); + + let obj = Object { + type_: SpaTypes::ObjectParamFormat.as_raw(), + id: ParamType::EnumFormat.as_raw(), + properties: audio_info.into(), + }; + let values: Vec = + PodSerializer::serialize(std::io::Cursor::new(Vec::new()), &Value::Object(obj)) + .unwrap() + .0 + .into_inner(); + + let mut params_out = [Pod::from_bytes(&values).unwrap()]; + + // RT_PROCESS is intentionally absent: with add_local_listener the process callback always + // runs on this mainloop thread, not the separate data-loop thread RT_PROCESS creates. + // That thread promotes itself to RT from the process callback, once when the negotiated + // quantum is known and again whenever PipeWire renegotiates it. + let mut flags = StreamFlags::MAP_BUFFERS | StreamFlags::INACTIVE; + if connect_automatically { + flags |= StreamFlags::AUTOCONNECT; + } + if !is_default_device { + flags |= StreamFlags::DONT_RECONNECT; + } + stream.connect(Direction::Output, None, flags, &mut params_out)?; + + Ok(StreamData { + mainloop, + listener: PwListener::Duplex { + input: listener_in, + output: listener_out, + }, stream, context, core, diff --git a/src/lib.rs b/src/lib.rs index 469e4e338..61cd15a23 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -895,6 +895,60 @@ impl From for StreamConfig { } } +impl StreamConfig { + #[allow(dead_code)] + pub fn validate(&self) -> Result<(), Error> { + if self.channels == 0 { + return Err(Error::with_message( + ErrorKind::InvalidInput, + "channel count must be at least 1", + )); + } + if self.sample_rate == 0 { + return Err(Error::with_message( + ErrorKind::InvalidInput, + "sample rate must be at least 1 Hz", + )); + } + if self.buffer_size == BufferSize::Fixed(0) { + return Err(Error::with_message( + ErrorKind::InvalidInput, + "buffer size must be greater than 0", + )); + } + Ok(()) + } +} +impl DuplexStreamConfig { + #[allow(dead_code)] + pub fn validate(&self) -> Result<(), Error> { + if self.input_channels == 0 { + return Err(Error::with_message( + ErrorKind::InvalidInput, + "channel count must be at least 1", + )); + } + if self.output_channels == 0 { + return Err(Error::with_message( + ErrorKind::InvalidInput, + "channel count must be at least 1", + )); + } + if self.sample_rate == 0 { + return Err(Error::with_message( + ErrorKind::InvalidInput, + "sample rate must be at least 1 Hz", + )); + } + if self.buffer_size == BufferSize::Fixed(0) { + return Err(Error::with_message( + ErrorKind::InvalidInput, + "buffer size must be greater than 0", + )); + } + Ok(()) + } +} #[allow(dead_code)] pub(crate) fn validate_stream_config(config: &StreamConfig) -> Result<(), Error> { if config.channels == 0 {