pipewire/stream/
mod.rs

1// Copyright The pipewire-rs Contributors.
2// SPDX-License-Identifier: MIT
3
4//! Streams are higher-level objects providing a convenient way to send and receive data streams to/from PipeWire.
5//!
6//! This module contains wrappers for [`pw_stream`](pw_sys::pw_stream) and related itmes.
7
8use crate::buffer::Buffer;
9use crate::{error::Error, properties::Properties};
10use bitflags::bitflags;
11use spa::utils::result::SpaResult;
12use std::{
13    ffi::{self, CStr, CString},
14    fmt::Debug,
15    mem, os,
16    pin::Pin,
17    ptr,
18};
19
20mod box_;
21pub use box_::*;
22mod rc;
23pub use rc::*;
24
25#[derive(Debug, PartialEq)]
26pub enum StreamState {
27    Error(String),
28    Unconnected,
29    Connecting,
30    Paused,
31    Streaming,
32}
33
34impl StreamState {
35    pub(crate) fn from_raw(state: pw_sys::pw_stream_state, error: *const os::raw::c_char) -> Self {
36        match state {
37            pw_sys::pw_stream_state_PW_STREAM_STATE_UNCONNECTED => StreamState::Unconnected,
38            pw_sys::pw_stream_state_PW_STREAM_STATE_CONNECTING => StreamState::Connecting,
39            pw_sys::pw_stream_state_PW_STREAM_STATE_PAUSED => StreamState::Paused,
40            pw_sys::pw_stream_state_PW_STREAM_STATE_STREAMING => StreamState::Streaming,
41            _ => {
42                let error = if error.is_null() {
43                    "".to_string()
44                } else {
45                    unsafe { ffi::CStr::from_ptr(error).to_string_lossy().to_string() }
46                };
47
48                StreamState::Error(error)
49            }
50        }
51    }
52}
53
54/// Stream timing information, updated every graph cycle.
55///
56/// Obtained from [`Stream::time()`].
57/// The [`now`](Self::now) field holds the timestamp of the last update;
58/// compare it with `Stream::nsec()` (requires the `v1_1_0` feature) to
59/// interpolate the current position.
60///
61/// All timing values are relative to the stream's rate.
62///
63/// For a detailed description, see [`pw_time`'s documentation](https://docs.pipewire.org/structpw__time.html#details).
64#[repr(transparent)]
65pub struct Time(pw_sys::pw_time);
66
67impl Clone for Time {
68    fn clone(&self) -> Self {
69        Self(pw_sys::pw_time { ..self.0 })
70    }
71}
72
73impl Time {
74    pub fn as_raw(&self) -> &pw_sys::pw_time {
75        &self.0
76    }
77
78    /// The monotonic timestamp (in nanoseconds) of this report.
79    pub fn now(&self) -> i64 {
80        self.0.now
81    }
82
83    /// The rate of `ticks` and `delay`, usually expressed as 1/samplerate.
84    pub fn rate(&self) -> spa::utils::Fraction {
85        self.0.rate
86    }
87
88    /// Monotonically increasing stream position in ticks.
89    pub fn ticks(&self) -> u64 {
90        self.0.ticks
91    }
92
93    /// Delay to device in ticks, including all filters on the path. Can be
94    /// negative.
95    ///
96    /// Convert to seconds using `delay * rate.num / rate.denom`.
97    pub fn delay(&self) -> i64 {
98        self.0.delay
99    }
100
101    /// Total bytes queued in the stream.
102    pub fn queued(&self) -> u64 {
103        self.0.queued
104    }
105
106    /// Extra frames buffered in the resampler. Since PipeWire 0.3.50.
107    #[cfg(feature = "v0_3_50")]
108    pub fn buffered(&self) -> u64 {
109        self.0.buffered
110    }
111
112    /// Number of buffers currently queued. Since PipeWire 0.3.50.
113    #[cfg(feature = "v0_3_50")]
114    pub fn queued_buffers(&self) -> u32 {
115        self.0.queued_buffers
116    }
117
118    /// Number of buffers available to dequeue. Since PipeWire 0.3.50.
119    #[cfg(feature = "v0_3_50")]
120    pub fn avail_buffers(&self) -> u32 {
121        self.0.avail_buffers
122    }
123}
124
125impl Debug for Time {
126    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
127        let mut s = f.debug_struct("Time");
128        s.field("now", &self.now())
129            .field("rate", &self.rate())
130            .field("ticks", &self.ticks())
131            .field("delay", &self.delay())
132            .field("queued", &self.queued());
133
134        #[cfg(feature = "v0_3_50")]
135        s.field("buffered", &self.buffered())
136            .field("queued_buffers", &self.queued_buffers())
137            .field("avail_buffers", &self.avail_buffers());
138
139        s.finish()
140    }
141}
142
143/// Transparent wrapper around a [stream](self).
144///
145/// This does not own the underlying object and is usually seen behind a `&` reference.
146///
147/// For owning wrappers that can construct streams, see [`StreamBox`] and [`StreamRc`].
148///
149/// For an explanation of these, see [Smart pointers to PipeWire
150/// objects](crate#smart-pointers-to-pipewire-objects).
151#[repr(transparent)]
152pub struct Stream(pw_sys::pw_stream);
153
154impl Stream {
155    pub fn as_raw(&self) -> &pw_sys::pw_stream {
156        &self.0
157    }
158
159    pub fn as_raw_ptr(&self) -> *mut pw_sys::pw_stream {
160        ptr::addr_of!(self.0).cast_mut()
161    }
162
163    /// Add a local listener builder
164    #[must_use = "Use the builder to register event callbacks"]
165    pub fn add_local_listener_with_user_data<D>(
166        &self,
167        user_data: D,
168    ) -> ListenerLocalBuilder<'_, D> {
169        let mut callbacks = ListenerLocalCallbacks::with_user_data(user_data);
170        callbacks.stream =
171            Some(ptr::NonNull::new(self.as_raw_ptr()).expect("Pointer should be nonnull"));
172        ListenerLocalBuilder {
173            stream: self,
174            callbacks,
175        }
176    }
177
178    /// Add a local listener builder. User data is initialized with its default value
179    #[must_use = "Use the builder to register event callbacks"]
180    pub fn add_local_listener<D: Default>(&self) -> ListenerLocalBuilder<'_, D> {
181        self.add_local_listener_with_user_data(Default::default())
182    }
183
184    /// Connect the stream
185    ///
186    /// Tries to connect to the node `id` in the given `direction`. If no node
187    /// is provided then any suitable node will be used.
188    // FIXME: high-level API for params
189    pub fn connect(
190        &self,
191        direction: spa::utils::Direction,
192        id: Option<u32>,
193        flags: StreamFlags,
194        params: &mut [&spa::pod::Pod],
195    ) -> Result<(), Error> {
196        let r = unsafe {
197            pw_sys::pw_stream_connect(
198                self.as_raw_ptr(),
199                direction.as_raw(),
200                id.unwrap_or(crate::constants::ID_ANY),
201                flags.bits(),
202                // We cast from *mut [&spa::pod::Pod] to *mut [*const spa_sys::spa_pod] here,
203                // which is valid because spa::pod::Pod is a transparent wrapper around spa_sys::spa_pod
204                params.as_mut_ptr().cast(),
205                params.len() as u32,
206            )
207        };
208
209        SpaResult::from_c(r).into_sync_result()?;
210        Ok(())
211    }
212
213    /// Update Parameters
214    ///
215    /// Call from the `param_changed` callback to negotiate a new set of
216    /// parameters for the stream.
217    // FIXME: high-level API for params
218    pub fn update_params(&self, params: &mut [&spa::pod::Pod]) -> Result<(), Error> {
219        let r = unsafe {
220            pw_sys::pw_stream_update_params(
221                self.as_raw_ptr(),
222                params.as_mut_ptr().cast(),
223                params.len() as u32,
224            )
225        };
226
227        SpaResult::from_c(r).into_sync_result()?;
228        Ok(())
229    }
230
231    /// Activate or deactivate the stream
232    pub fn set_active(&self, active: bool) -> Result<(), Error> {
233        let r = unsafe { pw_sys::pw_stream_set_active(self.as_raw_ptr(), active) };
234
235        SpaResult::from_c(r).into_sync_result()?;
236        Ok(())
237    }
238
239    /// Take a Buffer from the Stream
240    ///
241    /// Removes a buffer from the stream. If this is an input stream the buffer
242    /// will contain data ready to process. If this is an output stream it can
243    /// be filled.
244    ///
245    /// # Safety
246    ///
247    /// The pointer returned could be NULL if no buffer is available. The buffer
248    /// should be returned to the stream once processing is complete.
249    pub unsafe fn dequeue_raw_buffer(&self) -> *mut pw_sys::pw_buffer {
250        pw_sys::pw_stream_dequeue_buffer(self.as_raw_ptr())
251    }
252
253    pub fn dequeue_buffer(&self) -> Option<Buffer<'_>> {
254        unsafe { Buffer::from_raw(self.dequeue_raw_buffer(), self) }
255    }
256
257    /// Return a Buffer to the Stream
258    ///
259    /// Give back a buffer once processing is complete. Use this to queue up a
260    /// frame for an output stream, or return the buffer to the pool ready to
261    /// receive new data for an input stream.
262    ///
263    /// # Safety
264    ///
265    /// The buffer pointer should be one obtained from this stream instance by
266    /// a call to [Self::dequeue_raw_buffer()].
267    pub unsafe fn queue_raw_buffer(&self, buffer: *mut pw_sys::pw_buffer) {
268        pw_sys::pw_stream_queue_buffer(self.as_raw_ptr(), buffer);
269    }
270
271    /// Disconnect the stream
272    pub fn disconnect(&self) -> Result<(), Error> {
273        let r = unsafe { pw_sys::pw_stream_disconnect(self.as_raw_ptr()) };
274
275        SpaResult::from_c(r).into_sync_result()?;
276        Ok(())
277    }
278
279    /// Set the stream in error state
280    ///
281    /// # Panics
282    /// Will panic if `error` contains a 0 byte.
283    ///
284    pub fn set_error(&mut self, res: i32, error: &str) {
285        let error = CString::new(error).expect("failed to convert error to CString");
286        let error_cstr = error.as_c_str();
287        Stream::set_error_cstr(self, res, error_cstr)
288    }
289
290    /// Set the stream in error state with CStr
291    ///
292    /// # Panics
293    /// Will panic if `error` contains a 0 byte.
294    ///
295    pub fn set_error_cstr(&mut self, res: i32, error: &CStr) {
296        unsafe {
297            pw_sys::pw_stream_set_error(self.as_raw_ptr(), res, error.as_ptr());
298        }
299    }
300
301    /// Flush the stream. When  `drain` is `true`, the `drained` callback will
302    /// be called when all data is played or recorded.
303    pub fn flush(&self, drain: bool) -> Result<(), Error> {
304        let r = unsafe { pw_sys::pw_stream_flush(self.as_raw_ptr(), drain) };
305
306        SpaResult::from_c(r).into_sync_result()?;
307        Ok(())
308    }
309
310    pub fn set_control(&self, id: u32, values: &[f32]) -> Result<(), Error> {
311        let r = unsafe {
312            pw_sys::pw_stream_set_control(
313                self.as_raw_ptr(),
314                id,
315                values.len() as u32,
316                values.as_ptr() as *mut f32,
317            )
318        };
319        SpaResult::from_c(r).into_sync_result()?;
320        Ok(())
321    }
322
323    // getters
324
325    /// Get the name of the stream.
326    pub fn name(&self) -> String {
327        let name = unsafe {
328            let name = pw_sys::pw_stream_get_name(self.as_raw_ptr());
329            CStr::from_ptr(name)
330        };
331
332        name.to_string_lossy().to_string()
333    }
334
335    /// Get the current state of the stream.
336    pub fn state(&self) -> StreamState {
337        let mut error: *const std::os::raw::c_char = ptr::null();
338        let state = unsafe {
339            pw_sys::pw_stream_get_state(self.as_raw_ptr(), (&mut error) as *mut *const _)
340        };
341        StreamState::from_raw(state, error)
342    }
343
344    /// Get the properties of the stream.
345    pub fn properties(&self) -> &Properties {
346        unsafe {
347            let props = pw_sys::pw_stream_get_properties(self.as_raw_ptr());
348            let props = ptr::NonNull::new(props.cast_mut()).expect("stream properties is NULL");
349            props.cast().as_ref()
350        }
351    }
352
353    /// Get the node ID of the stream.
354    pub fn node_id(&self) -> u32 {
355        unsafe { pw_sys::pw_stream_get_node_id(self.as_raw_ptr()) }
356    }
357
358    #[cfg(feature = "v0_3_34")]
359    pub fn is_driving(&self) -> bool {
360        unsafe { pw_sys::pw_stream_is_driving(self.as_raw_ptr()) }
361    }
362
363    #[cfg(feature = "v0_3_34")]
364    pub fn trigger_process(&self) -> Result<(), Error> {
365        let r = unsafe { pw_sys::pw_stream_trigger_process(self.as_raw_ptr()) };
366
367        SpaResult::from_c(r).into_result()?;
368        Ok(())
369    }
370
371    /// Query the time on the stream.
372    ///
373    /// The time reported by this function is updated every graph cycle, usually
374    /// from the process callback. Values such as `ticks` and `delay` are
375    /// meaningful when used together with `now`: use `pw_stream_get_nsec()` to
376    /// get the current timestamp and interpolate.
377    ///
378    /// With the `v0_3_50` feature enabled, this uses `pw_stream_get_time_n()`
379    /// which also populates the `buffered`, `queued_buffers`, and
380    /// `avail_buffers` fields.
381    ///
382    /// This function is RT-safe.
383    pub fn time(&self) -> Result<Time, Error> {
384        unsafe {
385            let mut time = mem::MaybeUninit::<pw_sys::pw_time>::zeroed();
386
387            #[cfg(feature = "v0_3_50")]
388            let r = pw_sys::pw_stream_get_time_n(
389                self.as_raw_ptr(),
390                time.as_mut_ptr(),
391                mem::size_of::<pw_sys::pw_time>(),
392            );
393
394            #[cfg(not(feature = "v0_3_50"))]
395            let r = pw_sys::pw_stream_get_time(self.as_raw_ptr(), time.as_mut_ptr());
396
397            SpaResult::from_c(r).into_result()?;
398            Ok(Time(time.assume_init()))
399        }
400    }
401
402    /// Get the current time in nanoseconds.
403    ///
404    /// This value can be compared with the [`Time::now`] value to calculate
405    /// the elapsed time since the last time report and interpolate updated
406    /// `ticks` and `delay` values, as described in [`Stream::time`].
407    ///
408    /// This function is RT-safe.
409    #[cfg(feature = "v1_1_0")]
410    pub fn nsec(&self) -> u64 {
411        unsafe { pw_sys::pw_stream_get_nsec(self.as_raw_ptr()) }
412    }
413
414    // TODO: pw_stream_get_core()
415    // TODO: pw_stream_get_data_loop() (since PipeWire 1.1.0, needs v1_1_0 feature)
416}
417
418type ParamChangedCB<D> = dyn FnMut(&Stream, &mut D, u32, Option<&spa::pod::Pod>);
419type ProcessCB<D> = dyn FnMut(&Stream, &mut D);
420
421#[allow(clippy::type_complexity)]
422pub struct ListenerLocalCallbacks<D> {
423    pub state_changed: Option<Box<dyn FnMut(&Stream, &mut D, StreamState, StreamState)>>,
424    pub control_info:
425        Option<Box<dyn FnMut(&Stream, &mut D, u32, *const pw_sys::pw_stream_control)>>,
426    pub io_changed: Option<Box<dyn FnMut(&Stream, &mut D, u32, *mut os::raw::c_void, u32)>>,
427    pub param_changed: Option<Box<ParamChangedCB<D>>>,
428    pub add_buffer: Option<Box<dyn FnMut(&Stream, &mut D, *mut pw_sys::pw_buffer)>>,
429    pub remove_buffer: Option<Box<dyn FnMut(&Stream, &mut D, *mut pw_sys::pw_buffer)>>,
430    pub process: Option<Box<ProcessCB<D>>>,
431    pub drained: Option<Box<dyn FnMut(&Stream, &mut D)>>,
432    #[cfg(feature = "v0_3_39")]
433    pub command: Option<Box<dyn FnMut(&Stream, &mut D, *const spa_sys::spa_command)>>,
434    #[cfg(feature = "v0_3_40")]
435    pub trigger_done: Option<Box<dyn FnMut(&Stream, &mut D)>>,
436    pub user_data: D,
437    stream: Option<ptr::NonNull<pw_sys::pw_stream>>,
438}
439
440unsafe fn unwrap_stream_ptr<'a>(stream: Option<ptr::NonNull<pw_sys::pw_stream>>) -> &'a Stream {
441    stream
442        .map(|ptr| ptr.cast::<Stream>().as_ref())
443        .expect("stream cannot be null")
444}
445
446impl<D> ListenerLocalCallbacks<D> {
447    fn with_user_data(user_data: D) -> Self {
448        ListenerLocalCallbacks {
449            process: Default::default(),
450            stream: Default::default(),
451            drained: Default::default(),
452            add_buffer: Default::default(),
453            control_info: Default::default(),
454            io_changed: Default::default(),
455            param_changed: Default::default(),
456            remove_buffer: Default::default(),
457            state_changed: Default::default(),
458            #[cfg(feature = "v0_3_39")]
459            command: Default::default(),
460            #[cfg(feature = "v0_3_40")]
461            trigger_done: Default::default(),
462            user_data,
463        }
464    }
465
466    pub(crate) fn into_raw(
467        self,
468    ) -> (
469        Pin<Box<pw_sys::pw_stream_events>>,
470        Box<ListenerLocalCallbacks<D>>,
471    ) {
472        let callbacks = Box::new(self);
473
474        unsafe extern "C" fn on_state_changed<D>(
475            data: *mut os::raw::c_void,
476            old: pw_sys::pw_stream_state,
477            new: pw_sys::pw_stream_state,
478            error: *const os::raw::c_char,
479        ) {
480            if let Some(state) = (data as *mut ListenerLocalCallbacks<D>).as_mut() {
481                if let Some(cb) = &mut state.state_changed {
482                    let stream = unwrap_stream_ptr(state.stream);
483                    let old = StreamState::from_raw(old, error);
484                    let new = StreamState::from_raw(new, error);
485                    cb(stream, &mut state.user_data, old, new)
486                };
487            }
488        }
489
490        unsafe extern "C" fn on_control_info<D>(
491            data: *mut os::raw::c_void,
492            id: u32,
493            control: *const pw_sys::pw_stream_control,
494        ) {
495            if let Some(state) = (data as *mut ListenerLocalCallbacks<D>).as_mut() {
496                if let Some(cb) = &mut state.control_info {
497                    let stream = unwrap_stream_ptr(state.stream);
498                    cb(stream, &mut state.user_data, id, control);
499                }
500            }
501        }
502
503        unsafe extern "C" fn on_io_changed<D>(
504            data: *mut os::raw::c_void,
505            id: u32,
506            area: *mut os::raw::c_void,
507            size: u32,
508        ) {
509            if let Some(state) = (data as *mut ListenerLocalCallbacks<D>).as_mut() {
510                if let Some(cb) = &mut state.io_changed {
511                    let stream = unwrap_stream_ptr(state.stream);
512                    cb(stream, &mut state.user_data, id, area, size);
513                }
514            }
515        }
516
517        unsafe extern "C" fn on_param_changed<D>(
518            data: *mut os::raw::c_void,
519            id: u32,
520            param: *const spa_sys::spa_pod,
521        ) {
522            if let Some(state) = (data as *mut ListenerLocalCallbacks<D>).as_mut() {
523                if let Some(cb) = &mut state.param_changed {
524                    let stream = unwrap_stream_ptr(state.stream);
525                    let param = if !param.is_null() {
526                        Some(spa::pod::Pod::from_raw(param))
527                    } else {
528                        None
529                    };
530
531                    cb(stream, &mut state.user_data, id, param);
532                }
533            }
534        }
535
536        unsafe extern "C" fn on_add_buffer<D>(
537            data: *mut ::std::os::raw::c_void,
538            buffer: *mut pw_sys::pw_buffer,
539        ) {
540            if let Some(state) = (data as *mut ListenerLocalCallbacks<D>).as_mut() {
541                if let Some(cb) = &mut state.add_buffer {
542                    let stream = unwrap_stream_ptr(state.stream);
543                    cb(stream, &mut state.user_data, buffer);
544                }
545            }
546        }
547
548        unsafe extern "C" fn on_remove_buffer<D>(
549            data: *mut ::std::os::raw::c_void,
550            buffer: *mut pw_sys::pw_buffer,
551        ) {
552            if let Some(state) = (data as *mut ListenerLocalCallbacks<D>).as_mut() {
553                if let Some(cb) = &mut state.remove_buffer {
554                    let stream = unwrap_stream_ptr(state.stream);
555                    cb(stream, &mut state.user_data, buffer);
556                }
557            }
558        }
559
560        unsafe extern "C" fn on_process<D>(data: *mut ::std::os::raw::c_void) {
561            if let Some(state) = (data as *mut ListenerLocalCallbacks<D>).as_mut() {
562                if let Some(cb) = &mut state.process {
563                    let stream = unwrap_stream_ptr(state.stream);
564                    cb(stream, &mut state.user_data);
565                }
566            }
567        }
568
569        unsafe extern "C" fn on_drained<D>(data: *mut ::std::os::raw::c_void) {
570            if let Some(state) = (data as *mut ListenerLocalCallbacks<D>).as_mut() {
571                if let Some(cb) = &mut state.drained {
572                    let stream = unwrap_stream_ptr(state.stream);
573                    cb(stream, &mut state.user_data);
574                }
575            }
576        }
577
578        #[cfg(feature = "v0_3_39")]
579        unsafe extern "C" fn on_command<D>(
580            data: *mut ::std::os::raw::c_void,
581            command: *const spa_sys::spa_command,
582        ) {
583            if let Some(state) = (data as *mut ListenerLocalCallbacks<D>).as_mut() {
584                if let Some(cb) = &mut state.command {
585                    let stream = unwrap_stream_ptr(state.stream);
586                    cb(stream, &mut state.user_data, command);
587                }
588            }
589        }
590
591        #[cfg(feature = "v0_3_40")]
592        unsafe extern "C" fn on_trigger_done<D>(data: *mut ::std::os::raw::c_void) {
593            if let Some(state) = (data as *mut ListenerLocalCallbacks<D>).as_mut() {
594                if let Some(cb) = &mut state.trigger_done {
595                    let stream = unwrap_stream_ptr(state.stream);
596                    cb(stream, &mut state.user_data);
597                }
598            }
599        }
600
601        let events = unsafe {
602            let mut events: Pin<Box<pw_sys::pw_stream_events>> = Box::pin(mem::zeroed());
603            events.version = pw_sys::PW_VERSION_STREAM_EVENTS;
604
605            if callbacks.state_changed.is_some() {
606                events.state_changed = Some(on_state_changed::<D>);
607            }
608            if callbacks.control_info.is_some() {
609                events.control_info = Some(on_control_info::<D>);
610            }
611            if callbacks.io_changed.is_some() {
612                events.io_changed = Some(on_io_changed::<D>);
613            }
614            if callbacks.param_changed.is_some() {
615                events.param_changed = Some(on_param_changed::<D>);
616            }
617            if callbacks.add_buffer.is_some() {
618                events.add_buffer = Some(on_add_buffer::<D>);
619            }
620            if callbacks.remove_buffer.is_some() {
621                events.remove_buffer = Some(on_remove_buffer::<D>);
622            }
623            if callbacks.process.is_some() {
624                events.process = Some(on_process::<D>);
625            }
626            if callbacks.drained.is_some() {
627                events.drained = Some(on_drained::<D>);
628            }
629            #[cfg(feature = "v0_3_39")]
630            if callbacks.command.is_some() {
631                events.command = Some(on_command::<D>);
632            }
633            #[cfg(feature = "v0_3_40")]
634            if callbacks.trigger_done.is_some() {
635                events.trigger_done = Some(on_trigger_done::<D>);
636            }
637
638            events
639        };
640
641        (events, callbacks)
642    }
643}
644
645/// A builder for registering stream event callbacks.
646///
647/// Use [`Stream::add_local_listener`] or [`Stream::add_local_listener_with_user_data`] to create this and register callbacks that will be called when events of interest occur.
648/// After adding callbacks, use [`register`](Self::register) to get back a [`StreamListener`].
649///
650/// # Examples
651/// ```
652/// # use pipewire::stream::Stream;
653/// # use pipewire::spa::pod::Pod;
654/// # fn example(stream: Stream) {
655/// let stream_listener = stream.add_local_listener::<()>()
656///     .state_changed(|_stream, _user_data, old, new| println!("Stream state changed from {old:?} to {new:?}"))
657///     .control_info(|_stream, _user_data, id, control| println!("Stream control info: id {id}, control {control:?}"))
658///     .io_changed(|_stream, _user_data, id, area, size| println!("Stream IO change: IO type {id}, area {area:?}, size {size}"))
659///     .param_changed(|_stream, _user_data, id, param| println!("Stream param change: id {id}, param {:?}",
660///         param.map(Pod::as_bytes)))
661///     .add_buffer(|_stream, _user_data, buffer| println!("Stream buffer added {buffer:?}"))
662///     .remove_buffer(|_stream, _user_data, buffer| println!("Stream buffer removed {buffer:?}"))
663///     .process(|stream, _user_data| {
664///         println!("Stream can be processed");
665///         let buf = stream.dequeue_buffer();
666///         // Produce or consume data using the buffer
667///         // The buffer is enqueued back to the stream when it's dropped
668///     })
669///     .drained(|_stream, _user_data| println!("Stream is drained"))
670///     .register();
671/// # }
672/// ```
673pub struct ListenerLocalBuilder<'a, D> {
674    stream: &'a Stream,
675    callbacks: ListenerLocalCallbacks<D>,
676}
677
678impl<'a, D> ListenerLocalBuilder<'a, D> {
679    /// Set the stream `state_changed` event callback of the listener.
680    ///
681    /// This event is emitted when the stream state changes.
682    ///
683    /// # Callback parameters
684    /// `stream`: The stream  
685    /// `data`: User data  
686    /// `old`: Old stream state  
687    /// `new`: New stream state
688    ///
689    /// # Examples
690    /// ```
691    /// # use pipewire::stream::Stream;
692    /// # fn example(stream: Stream) {
693    /// let stream_listener = stream.add_local_listener::<()>()
694    ///     .state_changed(|_stream, _user_data, old, new| println!("Stream state changed from {old:?} to {new:?}"))
695    ///     .register();
696    /// # }
697    /// ```
698    #[must_use = "Call `.register()` to start receiving events"]
699    pub fn state_changed<F>(mut self, callback: F) -> Self
700    where
701        F: FnMut(&Stream, &mut D, StreamState, StreamState) + 'static,
702    {
703        self.callbacks.state_changed = Some(Box::new(callback));
704        self
705    }
706
707    /// Set the stream `control_info` event callback of the listener.
708    ///
709    /// This event is emitted when there is information about a control.
710    ///
711    /// # Callback parameters
712    /// `stream`: The stream  
713    /// `user_data`: User data  
714    /// `id`: Type of the control  
715    /// `control`: The control
716    ///
717    /// # Examples
718    /// ```
719    /// # use pipewire::stream::Stream;
720    /// # fn example(stream: Stream) {
721    /// let stream_listener = stream.add_local_listener::<()>()
722    ///     .control_info(|_stream, _user_data, id, _control| println!("Stream control info {id}"))
723    ///     .register();
724    /// # }
725    /// ```
726    #[must_use = "Call `.register()` to start receiving events"]
727    pub fn control_info<F>(mut self, callback: F) -> Self
728    where
729        F: FnMut(&Stream, &mut D, u32, *const pw_sys::pw_stream_control) + 'static,
730    {
731        self.callbacks.control_info = Some(Box::new(callback));
732        self
733    }
734
735    /// Set the stream `io_changed` event callback of the listener.
736    ///
737    /// This event is emitted when IO is changed on the stream.
738    ///
739    /// # Callback parameters
740    /// `stream`: The stream  
741    /// `user_data`: User data  
742    /// `id`: Type of the IO area  
743    /// `area`: The IO area  
744    /// `size`: The IO area size
745    ///
746    /// # Examples
747    /// ```
748    /// # use pipewire::stream::Stream;
749    /// # fn example(stream: Stream) {
750    /// let stream_listener = stream.add_local_listener::<()>()
751    ///     .io_changed(|_stream, _user_data, id, _area, size| println!("Stream IO change: IO type {id}"))
752    ///     .register();
753    /// # }
754    /// ```
755    #[must_use = "Call `.register()` to start receiving events"]
756    pub fn io_changed<F>(mut self, callback: F) -> Self
757    where
758        F: FnMut(&Stream, &mut D, u32, *mut os::raw::c_void, u32) + 'static,
759    {
760        self.callbacks.io_changed = Some(Box::new(callback));
761        self
762    }
763
764    /// Set the stream `param_changed` event callback of the listener.
765    ///
766    /// This event is emitted when a param is changed.
767    ///
768    /// # Callback parameters
769    /// `stream`: The stream  
770    /// `user_data`: User data  
771    /// `id`: Type of the param  
772    /// `param`: The param
773    ///
774    /// # Examples
775    /// ```
776    /// # use pipewire::stream::Stream;
777    /// # use pipewire::spa::pod::Pod;
778    /// # fn example(stream: Stream) {
779    /// let stream_listener = stream.add_local_listener::<()>()
780    ///     .param_changed(|_stream, _user_data, id, param| println!("Stream param change: id {id}, param {:?}",
781    ///         param.map(Pod::as_bytes)))
782    ///     .register();
783    /// # }
784    /// ```
785    #[must_use = "Call `.register()` to start receiving events"]
786    pub fn param_changed<F>(mut self, callback: F) -> Self
787    where
788        F: FnMut(&Stream, &mut D, u32, Option<&spa::pod::Pod>) + 'static,
789    {
790        self.callbacks.param_changed = Some(Box::new(callback));
791        self
792    }
793
794    /// Set the stream `add_buffer` event callback of the listener.
795    ///
796    /// This event is emitted when a buffer was added for this stream.
797    ///
798    /// # Callback parameters
799    /// `stream`: The stream  
800    /// `user_data`: User data  
801    /// `buffer`: The buffer
802    ///
803    /// # Examples
804    /// ```
805    /// # use pipewire::stream::Stream;
806    /// # fn example(stream: Stream) {
807    /// let stream_listener = stream.add_local_listener::<()>()
808    ///     .add_buffer(|_stream, _user_data, buffer| println!("Stream buffer added {buffer:?}"))
809    ///     .register();
810    /// # }
811    /// ```
812    #[must_use = "Call `.register()` to start receiving events"]
813    pub fn add_buffer<F>(mut self, callback: F) -> Self
814    where
815        F: FnMut(&Stream, &mut D, *mut pw_sys::pw_buffer) + 'static,
816    {
817        self.callbacks.add_buffer = Some(Box::new(callback));
818        self
819    }
820
821    /// Set the stream `remove_buffer` event callback of the listener.
822    ///
823    /// This event is emitted when a buffer was removed for this stream.
824    ///
825    /// # Callback parameters
826    /// `stream`: The stream  
827    /// `user_data`: User data  
828    /// `buffer`: The buffer
829    ///
830    /// # Examples
831    /// ```
832    /// # use pipewire::stream::Stream;
833    /// # fn example(stream: Stream) {
834    /// let stream_listener = stream.add_local_listener::<()>()
835    ///     .remove_buffer(|_stream, _user_data, buffer| println!("Stream buffer removed {buffer:?}"))
836    ///     .register();
837    /// # }
838    /// ```
839    #[must_use = "Call `.register()` to start receiving events"]
840    pub fn remove_buffer<F>(mut self, callback: F) -> Self
841    where
842        F: FnMut(&Stream, &mut D, *mut pw_sys::pw_buffer) + 'static,
843    {
844        self.callbacks.remove_buffer = Some(Box::new(callback));
845        self
846    }
847
848    /// Set the stream `process` event callback of the listener.
849    ///
850    /// This event is emitted when a buffer can be queued (for playback streams) or dequeued (for capture streams).
851    ///
852    /// This is normally called from the mainloop but can also be called directly from the realtime data thread if the user is prepared to deal with this.
853    ///
854    /// # Callback parameters
855    /// `stream`: The stream  
856    /// `user_data`: User data
857    ///
858    /// # Examples
859    /// ```
860    /// # use pipewire::stream::Stream;
861    /// # fn example(stream: Stream) {
862    /// let stream_listener = stream.add_local_listener::<()>()
863    ///     .process(|stream, _user_data| {
864    ///         println!("Stream can be processed");
865    ///         let buf = stream.dequeue_buffer();
866    ///         // Produce or consume data using the buffer
867    ///         // The buffer is enqueued back to the stream when it's dropped
868    ///     })
869    ///     .register();
870    /// # }
871    /// ```
872    #[must_use = "Call `.register()` to start receiving events"]
873    pub fn process<F>(mut self, callback: F) -> Self
874    where
875        F: FnMut(&Stream, &mut D) + 'static,
876    {
877        self.callbacks.process = Some(Box::new(callback));
878        self
879    }
880
881    /// Set the stream `drained` event callback of the listener.
882    ///
883    /// This event is emitted when the stream is drained.
884    ///
885    /// # Callback parameters
886    /// `stream`: The stream  
887    /// `user_data`: User data
888    ///
889    /// # Examples
890    /// ```
891    /// # use pipewire::stream::Stream;
892    /// # fn example(stream: Stream) {
893    /// let stream_listener = stream.add_local_listener::<()>()
894    ///     .drained(|_stream, _user_data| println!("Stream is drained"))
895    ///     .register();
896    /// # }
897    /// ```
898    #[must_use = "Call `.register()` to start receiving events"]
899    pub fn drained<F>(mut self, callback: F) -> Self
900    where
901        F: FnMut(&Stream, &mut D) + 'static,
902    {
903        self.callbacks.drained = Some(Box::new(callback));
904        self
905    }
906
907    /// Subscribe to events and register any provided callbacks.
908    pub fn register(self) -> Result<StreamListener<D>, Error> {
909        let (events, data) = self.callbacks.into_raw();
910        let (listener, data) = unsafe {
911            let listener: Box<spa_sys::spa_hook> = Box::new(mem::zeroed());
912            let raw_listener = Box::into_raw(listener);
913            let raw_data = Box::into_raw(data);
914            pw_sys::pw_stream_add_listener(
915                self.stream.as_raw_ptr(),
916                raw_listener,
917                events.as_ref().get_ref(),
918                raw_data as *mut _,
919            );
920            (Box::from_raw(raw_listener), Box::from_raw(raw_data))
921        };
922        Ok(StreamListener {
923            listener,
924            _events: events,
925            _data: data,
926        })
927    }
928}
929
930/// An owned listener for stream events.
931///
932/// This is created by [`stream::ListenerLocalBuilder`][ListenerLocalBuilder] and will receive events as long as it is alive.
933/// When this gets dropped, the listener gets unregistered and no events will be received by it.
934#[must_use = "Listeners unregister themselves when dropped. Keep the listener alive in order to receive events."]
935pub struct StreamListener<D> {
936    listener: Box<spa_sys::spa_hook>,
937    // Need to stay allocated while the listener is registered
938    _events: Pin<Box<pw_sys::pw_stream_events>>,
939    _data: Box<ListenerLocalCallbacks<D>>,
940}
941
942impl<D> StreamListener<D> {
943    /// Stop the listener from receiving any events
944    ///
945    /// Removes the listener registration and cleans up allocated resources.
946    pub fn unregister(self) {
947        // do nothing, drop will clean up.
948    }
949}
950
951impl<D> std::ops::Drop for StreamListener<D> {
952    fn drop(&mut self) {
953        spa::utils::hook::remove(*self.listener);
954    }
955}
956
957bitflags! {
958    /// Extra flags that can be used in [`Stream::connect()`]
959    #[derive(Debug, PartialEq, Eq, Clone, Copy)]
960    pub struct StreamFlags: pw_sys::pw_stream_flags {
961        const AUTOCONNECT = pw_sys::pw_stream_flags_PW_STREAM_FLAG_AUTOCONNECT;
962        const INACTIVE = pw_sys::pw_stream_flags_PW_STREAM_FLAG_INACTIVE;
963        const MAP_BUFFERS = pw_sys::pw_stream_flags_PW_STREAM_FLAG_MAP_BUFFERS;
964        const DRIVER = pw_sys::pw_stream_flags_PW_STREAM_FLAG_DRIVER;
965        const RT_PROCESS = pw_sys::pw_stream_flags_PW_STREAM_FLAG_RT_PROCESS;
966        const NO_CONVERT = pw_sys::pw_stream_flags_PW_STREAM_FLAG_NO_CONVERT;
967        const EXCLUSIVE = pw_sys::pw_stream_flags_PW_STREAM_FLAG_EXCLUSIVE;
968        const DONT_RECONNECT = pw_sys::pw_stream_flags_PW_STREAM_FLAG_DONT_RECONNECT;
969        const ALLOC_BUFFERS = pw_sys::pw_stream_flags_PW_STREAM_FLAG_ALLOC_BUFFERS;
970        #[cfg(feature = "v0_3_41")]
971        const TRIGGER = pw_sys::pw_stream_flags_PW_STREAM_FLAG_TRIGGER;
972    }
973}
pipewire/stream/mod.rs

pipewire/stream/
mod.rs
