rickub
nandi/jolt-nativepublic⑂ Fork 0
⑂ fa4ecdfed6a830e6099d5fab9be24ef72ea1923b
Branches
main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/jolt-native.git
git clone ssh://git@rickub.com/nandi/jolt-native.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

stream.rs · 588 lines · 20.9 KBRust Blame HistoryRaw
Lift freeq's AV media plane out of sleek 90f8b89 nandi 20d ago1use std::{
2 sync::{
3 atomic::{AtomicU64, Ordering},
4 Arc,
5 },
6 thread::JoinHandle,
7 time::Instant,
8};
9
10use crate::{
11 host::fill_with_equilibrium, traits::StreamTrait, BackendSpecificError, InputCallbackInfo,
12 OutputCallbackInfo, SampleFormat, StreamConfig, StreamError, StreamInstant,
13};
14use pipewire::{
15 self as pw,
16 context::ContextRc,
17 main_loop::MainLoopRc,
18 spa::{
19 param::{
20 format::{MediaSubtype, MediaType},
21 format_utils,
22 },
23 pod::Pod,
24 },
25 stream::{StreamListener, StreamRc, StreamState},
26};
27
28use crate::Data;
29
30#[derive(Debug, Clone, Copy)]
31pub enum StreamCommand {
32 Toggle(bool),
33 Stop,
34}
35
36pub struct Stream {
37 pub(crate) handle: Option<JoinHandle<()>>,
38 pub(crate) controller: pw::channel::Sender<StreamCommand>,
39 pub(crate) last_quantum: Arc<AtomicU64>,
40}
41
42impl Drop for Stream {
43 fn drop(&mut self) {
44 let _ = self.controller.send(StreamCommand::Stop);
45 let _ = self.handle.take().map(|handle| handle.join());
46 }
47}
48
49impl StreamTrait for Stream {
50 fn play(&self) -> Result<(), crate::PlayStreamError> {
51 self.controller
52 .send(StreamCommand::Toggle(true))
53 .map_err(|_| crate::PlayStreamError::BackendSpecific {
54 err: BackendSpecificError {
55 description: "Cannot send message".to_owned(),
56 },
57 })?;
58 Ok(())
59 }
60 fn pause(&self) -> Result<(), crate::PauseStreamError> {
61 self.controller
62 .send(StreamCommand::Toggle(false))
63 .map_err(|_| crate::PauseStreamError::BackendSpecific {
64 err: BackendSpecificError {
65 description: "Cannot send message".to_owned(),
66 },
67 })?;
68 Ok(())
69 }
70
71 fn buffer_size(&self) -> Option<crate::FrameCount> {
72 match self.last_quantum.load(Ordering::Relaxed) {
73 0 => None,
74 n => Some(n as _),
75 }
76 }
77}
78
79pub(crate) const SUPPORTED_FORMATS: &[SampleFormat] = &[
80 SampleFormat::I8,
81 SampleFormat::U8,
82 SampleFormat::I16,
83 SampleFormat::U16,
84 SampleFormat::I24,
85 SampleFormat::U24,
86 SampleFormat::I32,
87 SampleFormat::U32,
88 SampleFormat::I64,
89 SampleFormat::U64,
90 SampleFormat::F32,
91 SampleFormat::F64,
92];
93
94impl From<SampleFormat> for pw::spa::param::audio::AudioFormat {
95 fn from(value: SampleFormat) -> Self {
96 match value {
97 SampleFormat::I8 => Self::S8,
98 SampleFormat::U8 => Self::U8,
99
100 #[cfg(target_endian = "little")]
101 SampleFormat::I16 => Self::S16LE,
102 #[cfg(target_endian = "big")]
103 SampleFormat::I16 => Self::S16BE,
104 #[cfg(target_endian = "little")]
105 SampleFormat::U16 => Self::U16LE,
106 #[cfg(target_endian = "big")]
107 SampleFormat::U16 => Self::U16BE,
108
109 #[cfg(target_endian = "little")]
110 SampleFormat::I24 => Self::S24LE,
111 #[cfg(target_endian = "big")]
112 SampleFormat::I24 => Self::S24BE,
113 #[cfg(target_endian = "little")]
114 SampleFormat::U24 => Self::U24LE,
115 #[cfg(target_endian = "big")]
116 SampleFormat::U24 => Self::U24BE,
117 #[cfg(target_endian = "little")]
118 SampleFormat::I32 => Self::S32LE,
119 #[cfg(target_endian = "big")]
120 SampleFormat::I32 => Self::S32BE,
121 #[cfg(target_endian = "little")]
122 SampleFormat::U32 => Self::U32LE,
123 #[cfg(target_endian = "big")]
124 SampleFormat::U32 => Self::U32BE,
125 #[cfg(target_endian = "little")]
126 SampleFormat::F32 => Self::F32LE,
127 #[cfg(target_endian = "big")]
128 SampleFormat::F32 => Self::F32BE,
129 #[cfg(target_endian = "little")]
130 SampleFormat::F64 => Self::F64LE,
131 #[cfg(target_endian = "big")]
132 SampleFormat::F64 => Self::F64BE,
133 // NOTE: Seems PipeWire does support U64 and I64, but libspa doesn't yet.
134 // TODO: Maybe add the support in the future
135 _ => Self::Unknown,
136 }
137 }
138}
139
140pub struct UserData<D, E> {
141 data_callback: D,
142 error_callback: E,
143 sample_format: SampleFormat,
144 format: pw::spa::param::audio::AudioInfoRaw,
145 created_instance: Instant,
146 last_quantum: Arc<AtomicU64>,
147}
148impl<D, E> UserData<D, E>
149where
150 E: FnMut(StreamError) + Send + 'static,
151{
152 fn state_changed(&mut self, new: StreamState) {
153 match new {
154 pipewire::stream::StreamState::Error(e) => {
155 (self.error_callback)(StreamError::BackendSpecific {
156 err: BackendSpecificError { description: e },
157 })
158 }
159 // TODO: maybe we need to log information when every new state comes?
160 pipewire::stream::StreamState::Paused => {}
161 pipewire::stream::StreamState::Streaming => {}
162 pipewire::stream::StreamState::Connecting => {}
163 pipewire::stream::StreamState::Unconnected => {}
164 }
165 }
166}
167
168/// Hardware timestamp from a PipeWire graph cycle.
169struct PwTime {
170 /// CLOCK_MONOTONIC nanoseconds, stamped at the start of the graph cycle.
171 now_ns: i64,
172 /// Pipeline delay converted to nanoseconds.
173 /// For output: how far ahead of the driver our next sample will be played.
174 /// For input: how long ago the data in the buffer was captured.
175 delay_ns: i64,
176}
177
178/// Returns a hardware timestamp for the current graph cycle, or `None` if
179/// the driver has not started yet or the rate is unavailable.
180fn pw_stream_time(stream: &pw::stream::Stream) -> Option<PwTime> {
181 let mut t: pw::sys::pw_time = unsafe { std::mem::zeroed() };
182 let rc = unsafe {
183 pw::sys::pw_stream_get_time_n(
184 stream.as_raw_ptr(),
185 &mut t,
186 std::mem::size_of::<pw::sys::pw_time>(),
187 )
188 };
189 if rc != 0 || t.now == 0 || t.rate.denom == 0 {
190 return None;
191 }
192 debug_assert_eq!(t.rate.num, 1, "unexpected pw_time rate.num");
193 let delay_ns = t.delay * 1_000_000_000i64 / t.rate.denom as i64;
194 Some(PwTime {
195 now_ns: t.now,
196 delay_ns,
197 })
198}
199
200impl<D, E> UserData<D, E>
201where
202 D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
203 E: FnMut(StreamError) + Send + 'static,
204{
205 fn publish_data_in(
206 &mut self,
207 stream: &pw::stream::Stream,
208 frames: usize,
209 data: &Data,
210 ) -> Result<(), BackendSpecificError> {
211 self.last_quantum.store(frames as u64, Ordering::Relaxed);
212 let (callback, capture) = match pw_stream_time(stream) {
213 Some(PwTime { now_ns, delay_ns }) => (
214 StreamInstant::from_nanos(now_ns),
215 StreamInstant::from_nanos(now_ns - delay_ns),
216 ),
217 None => {
218 let cb = stream_timestamp_fallback(self.created_instance)?;
219 let pl = cb
220 .sub(frames_to_duration(frames, self.format.rate()))
221 .ok_or_else(|| BackendSpecificError {
222 description:
223 "`capture` occurs beyond representation supported by `StreamInstant`"
224 .to_string(),
225 })?;
226 (cb, pl)
227 }
228 };
229 let timestamp = crate::InputStreamTimestamp { callback, capture };
230 let info = InputCallbackInfo { timestamp };
231 (self.data_callback)(data, &info);
232 Ok(())
233 }
234}
235impl<D, E> UserData<D, E>
236where
237 D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
238 E: FnMut(StreamError) + Send + 'static,
239{
240 fn publish_data_out(
241 &mut self,
242 stream: &pw::stream::Stream,
243 frames: usize,
244 data: &mut Data,
245 ) -> Result<(), BackendSpecificError> {
246 self.last_quantum.store(frames as u64, Ordering::Relaxed);
247 let (callback, playback) = match pw_stream_time(stream) {
248 Some(PwTime { now_ns, delay_ns }) => (
249 StreamInstant::from_nanos(now_ns),
250 StreamInstant::from_nanos(now_ns + delay_ns),
251 ),
252 None => {
253 let cb = stream_timestamp_fallback(self.created_instance)?;
254 let pl = cb
255 .add(frames_to_duration(frames, self.format.rate()))
256 .ok_or_else(|| BackendSpecificError {
257 description:
258 "`playback` occurs beyond representation supported by `StreamInstant`"
259 .to_string(),
260 })?;
261 (cb, pl)
262 }
263 };
264 let timestamp = crate::OutputStreamTimestamp { callback, playback };
265 let info = OutputCallbackInfo { timestamp };
266 (self.data_callback)(data, &info);
267 Ok(())
268 }
269}
270pub struct StreamData<D, E> {
271 pub mainloop: MainLoopRc,
272 pub listener: StreamListener<UserData<D, E>>,
273 pub stream: StreamRc,
274 pub context: ContextRc,
275}
276
277// Use elapsed duration since stream creation as fallback when hardware timestamps are unavailable.
278//
279// This ensures positive values that are compatible with our `StreamInstant` representation.
280#[inline]
281fn stream_timestamp_fallback(
282 creation: std::time::Instant,
283) -> Result<StreamInstant, BackendSpecificError> {
284 let now = std::time::Instant::now();
285 let duration = now.duration_since(creation);
286 StreamInstant::from_nanos_i128(duration.as_nanos() as i128).ok_or(BackendSpecificError {
287 description: "stream duration has exceeded `StreamInstant` representation".to_string(),
288 })
289}
290
291// Convert the given duration in frames at the given sample rate to a `std::time::Duration`.
292#[inline]
293fn frames_to_duration(frames: usize, rate: crate::SampleRate) -> std::time::Duration {
294 let secsf = frames as f64 / rate as f64;
295 let secs = secsf as u64;
296 let nanos = ((secsf - secs as f64) * 1_000_000_000.0) as u32;
297 std::time::Duration::new(secs, nanos)
298}
299
300pub fn connect_output<D, E>(
301 config: StreamConfig,
302 properties: pw::properties::PropertiesBox,
303 sample_format: SampleFormat,
304 data_callback: D,
305 error_callback: E,
306 last_quantum: Arc<AtomicU64>,
307) -> Result<StreamData<D, E>, pw::Error>
308where
309 D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
310 E: FnMut(StreamError) + Send + 'static,
311{
312 pw::init();
313 let mainloop = pw::main_loop::MainLoopRc::new(None)?;
314 let context = pw::context::ContextRc::new(&mainloop, None)?;
315 let core = context.connect_rc(None)?;
316
317 let data = UserData {
318 data_callback,
319 error_callback,
320 sample_format,
321 format: Default::default(),
322 created_instance: Instant::now(),
323 last_quantum,
324 };
325 let channels = config.channels as _;
326 let rate = config.sample_rate as _;
327 let stream = pw::stream::StreamRc::new(core, "cpal-playback", properties)?;
328 let listener = stream
329 .add_local_listener_with_user_data(data)
330 .param_changed(move|stream, user_data, id, param| {
331 let Some(param) = param else {
332 return;
333 };
334 if id != pw::spa::param::ParamType::Format.as_raw() {
335 return;
336 }
337
338 let (media_type, media_subtype) = match format_utils::parse_format(param) {
339 Ok(v) => v,
340 Err(_) => return,
341 };
342
343 // only accept raw audio
344 if media_type != MediaType::Audio || media_subtype != MediaSubtype::Raw {
345 return;
346 }
347 // call a helper function to parse the format for us.
348 // When the format update, we check the format first, in case it does not fit what we
349 // set
350 if user_data.format.parse(param).is_ok() {
351 let current_channels = user_data.format.channels();
352 let current_rate = user_data.format.rate();
353 if current_channels != channels || rate != current_rate {
354 (user_data.error_callback)(StreamError::BackendSpecific {
355 err: BackendSpecificError {
356 description: format!("channels or rate is not fit, current channels: {current_channels}, current rate: {current_rate}"),
357 },
358 });
359 // if the channels and rate do not match, we stop the stream
360 if let Err(e) = stream.set_active(false) {
361 (user_data.error_callback)(StreamError::BackendSpecific {
362 err: BackendSpecificError {
363 description: format!("failed to stop the stream, reason: {e}"),
364 },
365 });
366 }
367 }
368
369 }
370 })
371 .state_changed(|_stream, user_data, _old, new| {
372 user_data.state_changed(new);
373 })
374 .process(|stream, user_data| match stream.dequeue_buffer() {
375 None => (user_data.error_callback)(StreamError::BufferUnderrun),
376 Some(mut buffer) => {
377 // Read the requested frame count before mutably borrowing datas_mut().
378 let requested = buffer.requested() as usize;
379 let datas = buffer.datas_mut();
380 if datas.is_empty() {
381 return;
382 }
383 let buf_data = &mut datas[0];
384 let n_channels = user_data.format.channels();
385
386 let stride = user_data.sample_format.sample_size() * n_channels as usize;
387 // frames = samples / channels or frames = data_len / stride
388 // Honor the frame count PipeWire requests this cycle, capped by the
389 // mapped buffer capacity to guard against any mismatch.
390 let frames = requested.min(buf_data.as_raw().maxsize as usize / stride);
391 let Some(samples) = buf_data.data() else {
392 return;
393 };
394
395 // samples = frames * channels or samples = data_len / sample_size
396 let n_samples = frames * n_channels as usize;
397
398 // Pre-fill only the active region with silence before handing it to the callback.
399 let active = &mut samples[..frames * stride];
400 fill_with_equilibrium(active, user_data.sample_format);
401
402 let data = active.as_mut_ptr() as *mut ();
403 let mut data =
404 unsafe { Data::from_parts(data, n_samples, user_data.sample_format) };
405 if let Err(err) = user_data.publish_data_out(stream, frames, &mut data) {
406 (user_data.error_callback)(StreamError::BackendSpecific { err });
407 }
408 let chunk = buf_data.chunk_mut();
409 *chunk.offset_mut() = 0;
410 *chunk.stride_mut() = stride as i32;
411 *chunk.size_mut() = (frames * stride) as u32;
412 }
413 })
414 .register()?;
415 let mut audio_info = pw::spa::param::audio::AudioInfoRaw::new();
416 audio_info.set_format(sample_format.into());
417 audio_info.set_rate(rate);
418 audio_info.set_channels(channels);
419
420 let obj = pw::spa::pod::Object {
421 type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
422 id: pw::spa::param::ParamType::EnumFormat.as_raw(),
423 properties: audio_info.into(),
424 };
425 let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
426 std::io::Cursor::new(Vec::new()),
427 &pw::spa::pod::Value::Object(obj),
428 )
429 .unwrap()
430 .0
431 .into_inner();
432
433 let mut params = [Pod::from_bytes(&values).unwrap()];
434
435 // TODO: what about RT_PROCESS?
436 /* Now connect this stream. We ask that our process function is
437 * called in a realtime thread. */
438 stream.connect(
439 pw::spa::utils::Direction::Output,
440 None,
441 pw::stream::StreamFlags::AUTOCONNECT | pw::stream::StreamFlags::MAP_BUFFERS,
442 &mut params,
443 )?;
444
445 Ok(StreamData {
446 mainloop,
447 listener,
448 stream,
449 context,
450 })
451}
452pub fn connect_input<D, E>(
453 config: StreamConfig,
454 properties: pw::properties::PropertiesBox,
455 sample_format: SampleFormat,
456 data_callback: D,
457 error_callback: E,
458 last_quantum: Arc<AtomicU64>,
459) -> Result<StreamData<D, E>, pw::Error>
460where
461 D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
462 E: FnMut(StreamError) + Send + 'static,
463{
464 pw::init();
465 let mainloop = pw::main_loop::MainLoopRc::new(None)?;
466 let context = pw::context::ContextRc::new(&mainloop, None)?;
467 let core = context.connect_rc(None)?;
468
469 let data = UserData {
470 data_callback,
471 error_callback,
472 sample_format,
473 format: Default::default(),
474 created_instance: Instant::now(),
475 last_quantum,
476 };
477
478 let channels = config.channels as _;
479 let rate = config.sample_rate as _;
480
481 let stream = pw::stream::StreamRc::new(core, "cpal-capture", properties)?;
482 let listener = stream
483 .add_local_listener_with_user_data(data)
484 .param_changed(move |stream, user_data, id, param| {
485 let Some(param) = param else {
486 return;
487 };
488 if id != pw::spa::param::ParamType::Format.as_raw() {
489 return;
490 }
491
492 let (media_type, media_subtype) = match format_utils::parse_format(param) {
493 Ok(v) => v,
494 Err(_) => return,
495 };
496
497 // only accept raw audio
498 if media_type != MediaType::Audio || media_subtype != MediaSubtype::Raw {
499 return;
500 }
501
502 // call a helper function to parse the format for us.
503 // When the format update, we check the format first, in case it does not fit what we
504 // set
505 if user_data.format.parse(param).is_ok() {
506 let current_channels = user_data.format.channels();
507 let current_rate = user_data.format.rate();
508 if current_channels != channels || rate != current_rate {
509 (user_data.error_callback)(StreamError::BackendSpecific {
510 err: BackendSpecificError {
511 description: format!("channels or rate is not fit, current channels: {current_channels}, current rate: {current_rate}"),
512 },
513 });
514 // if the channels and rate do not match, we stop the stream
515 if let Err(e) = stream.set_active(false) {
516 (user_data.error_callback)(StreamError::BackendSpecific {
517 err: BackendSpecificError {
518 description: format!("failed to stop the stream, reason: {e}"),
519 },
520 });
521 }
522 }
523 }
524 })
525 .state_changed(|_stream, user_data, _old, new| {
526 user_data.state_changed(new);
527 })
528 .process(|stream, user_data| match stream.dequeue_buffer() {
529 None => (user_data.error_callback)(StreamError::BufferUnderrun),
530 Some(mut buffer) => {
531 let datas = buffer.datas_mut();
532 if datas.is_empty() {
533 return;
534 }
535 let data = &mut datas[0];
536 let n_channels = user_data.format.channels();
537 let n_samples = data.chunk().size() / user_data.sample_format.sample_size() as u32;
538 let frames = n_samples / n_channels;
539
540 let Some(samples) = data.data() else {
541 return;
542 };
543 let data = samples.as_mut_ptr() as *mut ();
544 let data =
545 unsafe { Data::from_parts(data, n_samples as usize, user_data.sample_format) };
546 if let Err(err) = user_data.publish_data_in(stream, frames as usize, &data) {
547 (user_data.error_callback)(StreamError::BackendSpecific { err });
548 }
549 }
550 })
551 .register()?;
552 let mut audio_info = pw::spa::param::audio::AudioInfoRaw::new();
553 audio_info.set_format(sample_format.into());
554 audio_info.set_rate(rate);
555 audio_info.set_channels(channels);
556
557 let obj = pw::spa::pod::Object {
558 type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
559 id: pw::spa::param::ParamType::EnumFormat.as_raw(),
560 properties: audio_info.into(),
561 };
562 let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
563 std::io::Cursor::new(Vec::new()),
564 &pw::spa::pod::Value::Object(obj),
565 )
566 .unwrap()
567 .0
568 .into_inner();
569
570 let mut params = [Pod::from_bytes(&values).unwrap()];
571
572 // TODO: what about RT_PROCESS?
573 /* Now connect this stream. We ask that our process function is
574 * called in a realtime thread. */
575 stream.connect(
576 pw::spa::utils::Direction::Input,
577 None,
578 pw::stream::StreamFlags::AUTOCONNECT | pw::stream::StreamFlags::MAP_BUFFERS,
579 &mut params,
580 )?;
581
582 Ok(StreamData {
583 mainloop,
584 listener,
585 stream,
586 context,
587 })
588}