gstreamer/

bus.rs

// Take a look at the license at the top of the repository in the LICENSE file.

use std::mem::transmute;

use glib::{
    ControlFlow,
    ffi::{gboolean, gpointer},
    source::Priority,
    translate::*,
};

use crate::{Bus, BusSyncReply, Message, MessageType, ffi};

#[cfg(feature = "futures")]
pub use crate::bus_futures::BusStream;
#[cfg(feature = "futures")]
use futures_util::{StreamExt, stream::FusedStream};

unsafe extern "C" fn trampoline_watch<F: FnMut(&Bus, &Message) -> ControlFlow + Send + 'static>(
    bus: *mut ffi::GstBus,
    msg: *mut ffi::GstMessage,
    func: gpointer,
) -> gboolean {
    unsafe {
        let func: &mut F = &mut *(func as *mut F);
        func(&from_glib_borrow(bus), &Message::from_glib_borrow(msg)).into_glib()
    }
}

unsafe extern "C" fn destroy_closure_watch<
    F: FnMut(&Bus, &Message) -> ControlFlow + Send + 'static,
>(
    ptr: gpointer,
) {
    unsafe {
        let _ = Box::<F>::from_raw(ptr as *mut _);
    }
}

fn into_raw_watch<F: FnMut(&Bus, &Message) -> ControlFlow + Send + 'static>(func: F) -> gpointer {
    #[allow(clippy::type_complexity)]
    let func: Box<F> = Box::new(func);
    Box::into_raw(func) as gpointer
}

unsafe extern "C" fn trampoline_watch_local<F: FnMut(&Bus, &Message) -> ControlFlow + 'static>(
    bus: *mut ffi::GstBus,
    msg: *mut ffi::GstMessage,
    func: gpointer,
) -> gboolean {
    unsafe {
        let func: &mut glib::thread_guard::ThreadGuard<F> =
            &mut *(func as *mut glib::thread_guard::ThreadGuard<F>);
        (func.get_mut())(&from_glib_borrow(bus), &Message::from_glib_borrow(msg)).into_glib()
    }
}

unsafe extern "C" fn destroy_closure_watch_local<
    F: FnMut(&Bus, &Message) -> ControlFlow + 'static,
>(
    ptr: gpointer,
) {
    unsafe {
        let _ = Box::<glib::thread_guard::ThreadGuard<F>>::from_raw(ptr as *mut _);
    }
}

fn into_raw_watch_local<F: FnMut(&Bus, &Message) -> ControlFlow + 'static>(func: F) -> gpointer {
    #[allow(clippy::type_complexity)]
    let func: Box<glib::thread_guard::ThreadGuard<F>> =
        Box::new(glib::thread_guard::ThreadGuard::new(func));
    Box::into_raw(func) as gpointer
}

unsafe extern "C" fn trampoline_sync<
    F: Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static,
>(
    bus: *mut ffi::GstBus,
    msg: *mut ffi::GstMessage,
    func: gpointer,
) -> ffi::GstBusSyncReply {
    unsafe {
        let f: &F = &*(func as *const F);
        let res = f(&from_glib_borrow(bus), &Message::from_glib_borrow(msg)).into_glib();

        if res == ffi::GST_BUS_DROP {
            ffi::gst_mini_object_unref(msg as *mut _);
        }

        res
    }
}

unsafe extern "C" fn destroy_closure_sync<
    F: Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static,
>(
    ptr: gpointer,
) {
    unsafe {
        let _ = Box::<F>::from_raw(ptr as *mut _);
    }
}

fn into_raw_sync<F: Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static>(
    func: F,
) -> gpointer {
    let func: Box<F> = Box::new(func);
    Box::into_raw(func) as gpointer
}

impl Bus {
    /// Adds a bus signal watch to the default main context with the given `priority`
    /// (e.g. `G_PRIORITY_DEFAULT`). It is also possible to use a non-default main
    /// context set up using [`glib::MainContext::push_thread_default()`][crate::glib::MainContext::push_thread_default()]
    /// (before one had to create a bus watch source and attach it to the desired
    /// main context 'manually').
    ///
    /// After calling this statement, the bus will emit the "message" signal for each
    /// message posted on the bus when the `GMainLoop` is running.
    ///
    /// This function may be called multiple times. To clean up, the caller is
    /// responsible for calling [`remove_signal_watch()`][Self::remove_signal_watch()] as many times as this
    /// function is called.
    ///
    /// There can only be a single bus watch per bus, you must remove any signal
    /// watch before you can set another type of watch.
    /// ## `priority`
    /// The priority of the watch.
    #[doc(alias = "gst_bus_add_signal_watch")]
    #[doc(alias = "gst_bus_add_signal_watch_full")]
    pub fn add_signal_watch_full(&self, priority: Priority) {
        unsafe {
            ffi::gst_bus_add_signal_watch_full(self.to_glib_none().0, priority.into_glib());
        }
    }

    /// Create watch for this bus. The [`glib::Source`][crate::glib::Source] will be dispatched whenever
    /// a message is on the bus. After the GSource is dispatched, the
    /// message is popped off the bus and unreffed.
    ///
    /// As with other watches, there can only be one watch on the bus, including
    /// any signal watch added with `gst_bus_add_signal_watch`.
    ///
    /// # Returns
    ///
    /// a [`glib::Source`][crate::glib::Source] that can be added to a `GMainLoop`.
    #[doc(alias = "gst_bus_create_watch")]
    pub fn create_watch<F>(&self, name: Option<&str>, priority: Priority, func: F) -> glib::Source
    where
        F: FnMut(&Bus, &Message) -> ControlFlow + Send + 'static,
    {
        skip_assert_initialized!();
        unsafe {
            let source = ffi::gst_bus_create_watch(self.to_glib_none().0);
            glib::ffi::g_source_set_callback(
                source,
                Some(transmute::<
                    *mut (),
                    unsafe extern "C" fn(glib::ffi::gpointer) -> i32,
                >(trampoline_watch::<F> as *mut ())),
                into_raw_watch(func),
                Some(destroy_closure_watch::<F>),
            );
            glib::ffi::g_source_set_priority(source, priority.into_glib());

            if let Some(name) = name {
                glib::ffi::g_source_set_name(source, name.to_glib_none().0);
            }

            from_glib_full(source)
        }
    }

    /// Adds a bus watch to the default main context with the default priority
    /// ( `G_PRIORITY_DEFAULT` ). It is also possible to use a non-default main
    /// context set up using [`glib::MainContext::push_thread_default()`][crate::glib::MainContext::push_thread_default()] (before
    /// one had to create a bus watch source and attach it to the desired main
    /// context 'manually').
    ///
    /// This function is used to receive asynchronous messages in the main loop.
    /// There can only be a single bus watch per bus, you must remove it before you
    /// can set a new one.
    ///
    /// The bus watch will only work if a `GMainLoop` is being run.
    ///
    /// The watch can be removed using [`remove_watch()`][Self::remove_watch()] or by returning [`false`]
    /// from `func`. If the watch was added to the default main context it is also
    /// possible to remove the watch using [`glib::Source::remove()`][crate::glib::Source::remove()].
    ///
    /// The bus watch will take its own reference to the `self`, so it is safe to unref
    /// `self` using `gst_object_unref()` after setting the bus watch.
    /// ## `func`
    /// A function to call when a message is received.
    ///
    /// # Returns
    ///
    /// The event source id or 0 if `self` already got an event source.
    #[doc(alias = "gst_bus_add_watch")]
    #[doc(alias = "gst_bus_add_watch_full")]
    pub fn add_watch<F>(&self, func: F) -> Result<BusWatchGuard, glib::BoolError>
    where
        F: FnMut(&Bus, &Message) -> ControlFlow + Send + 'static,
    {
        unsafe {
            let res = ffi::gst_bus_add_watch_full(
                self.to_glib_none().0,
                glib::ffi::G_PRIORITY_DEFAULT,
                Some(trampoline_watch::<F>),
                into_raw_watch(func),
                Some(destroy_closure_watch::<F>),
            );

            if res == 0 {
                Err(glib::bool_error!("Bus already has a watch"))
            } else {
                Ok(BusWatchGuard { bus: self.clone() })
            }
        }
    }

    #[doc(alias = "gst_bus_add_watch")]
    #[doc(alias = "gst_bus_add_watch_full")]
    pub fn add_watch_local<F>(&self, func: F) -> Result<BusWatchGuard, glib::BoolError>
    where
        F: FnMut(&Bus, &Message) -> ControlFlow + 'static,
    {
        unsafe {
            let ctx = glib::MainContext::ref_thread_default();
            let _acquire = ctx
                .acquire()
                .expect("thread default main context already acquired by another thread");

            let res = ffi::gst_bus_add_watch_full(
                self.to_glib_none().0,
                glib::ffi::G_PRIORITY_DEFAULT,
                Some(trampoline_watch_local::<F>),
                into_raw_watch_local(func),
                Some(destroy_closure_watch_local::<F>),
            );

            if res == 0 {
                Err(glib::bool_error!("Bus already has a watch"))
            } else {
                Ok(BusWatchGuard { bus: self.clone() })
            }
        }
    }

    /// Sets the synchronous handler on the bus. The function will be called
    /// every time a new message is posted on the bus. Note that the function
    /// will be called in the same thread context as the posting object. This
    /// function is usually only called by the creator of the bus. Applications
    /// should handle messages asynchronously using the gst_bus watch and poll
    /// functions.
    ///
    /// Before 1.16.3 it was not possible to replace an existing handler and
    /// clearing an existing handler with [`None`] was not thread-safe.
    /// ## `func`
    /// The handler function to install
    /// ## `notify`
    /// called when `user_data` becomes unused
    #[doc(alias = "gst_bus_set_sync_handler")]
    pub fn set_sync_handler<F>(&self, func: F)
    where
        F: Fn(&Bus, &Message) -> BusSyncReply + Send + Sync + 'static,
    {
        unsafe {
            let bus = self.to_glib_none().0;

            #[allow(clippy::manual_dangling_ptr)]
            #[cfg(not(feature = "v1_18"))]
            {
                static SET_ONCE_QUARK: std::sync::OnceLock<glib::Quark> =
                    std::sync::OnceLock::new();

                let set_once_quark = SET_ONCE_QUARK
                    .get_or_init(|| glib::Quark::from_str("gstreamer-rs-sync-handler"));

                // This is not thread-safe before 1.16.3, see
                // https://gitlab.freedesktop.org/gstreamer/gstreamer-rs/merge_requests/416
                if crate::version() < (1, 16, 3, 0) {
                    if !glib::gobject_ffi::g_object_get_qdata(
                        bus as *mut _,
                        set_once_quark.into_glib(),
                    )
                    .is_null()
                    {
                        panic!("Bus sync handler can only be set once");
                    }

                    glib::gobject_ffi::g_object_set_qdata(
                        bus as *mut _,
                        set_once_quark.into_glib(),
                        1 as *mut _,
                    );
                }
            }

            ffi::gst_bus_set_sync_handler(
                bus,
                Some(trampoline_sync::<F>),
                into_raw_sync(func),
                Some(destroy_closure_sync::<F>),
            )
        }
    }

    pub fn unset_sync_handler(&self) {
        #[cfg(not(feature = "v1_18"))]
        {
            // This is not thread-safe before 1.16.3, see
            // https://gitlab.freedesktop.org/gstreamer/gstreamer-rs/merge_requests/416
            if crate::version() < (1, 16, 3, 0) {
                return;
            }
        }

        unsafe {
            use std::ptr;

            ffi::gst_bus_set_sync_handler(self.to_glib_none().0, None, ptr::null_mut(), None)
        }
    }

    #[doc(alias = "gst_bus_pop")]
    pub fn iter(&self) -> Iter<'_> {
        self.iter_timed(Some(crate::ClockTime::ZERO))
    }

    #[doc(alias = "gst_bus_timed_pop")]
    pub fn iter_timed(&self, timeout: impl Into<Option<crate::ClockTime>>) -> Iter<'_> {
        Iter {
            bus: self,
            timeout: timeout.into(),
        }
    }

    #[doc(alias = "gst_bus_pop_filtered")]
    pub fn iter_filtered<'a>(
        &'a self,
        msg_types: &'a [MessageType],
    ) -> impl Iterator<Item = Message> + 'a {
        self.iter_timed_filtered(Some(crate::ClockTime::ZERO), msg_types)
    }

    #[doc(alias = "gst_bus_timed_pop_filtered")]
    pub fn iter_timed_filtered<'a>(
        &'a self,
        timeout: impl Into<Option<crate::ClockTime>>,
        msg_types: &'a [MessageType],
    ) -> impl Iterator<Item = Message> + 'a {
        self.iter_timed(timeout)
            .filter(move |msg| msg_types.contains(&msg.type_()))
    }

    /// Gets a message from the bus whose type matches the message type mask `types`,
    /// waiting up to the specified timeout (and discarding any messages that do not
    /// match the mask provided).
    ///
    /// If `timeout` is 0, this function behaves like [`pop_filtered()`][Self::pop_filtered()]. If
    /// `timeout` is `GST_CLOCK_TIME_NONE`, this function will block forever until a
    /// matching message was posted on the bus.
    /// ## `timeout`
    /// a timeout in nanoseconds, or `GST_CLOCK_TIME_NONE` to wait forever
    /// ## `types`
    /// message types to take into account, `GST_MESSAGE_ANY` for any type
    ///
    /// # Returns
    ///
    /// a [`Message`][crate::Message] matching the
    ///  filter in `types`, or [`None`] if no matching message was found on
    ///  the bus until the timeout expired.
    #[doc(alias = "gst_bus_timed_pop_filtered")]
    pub fn timed_pop_filtered(
        &self,
        timeout: impl Into<Option<crate::ClockTime>>,
        msg_types: &[MessageType],
    ) -> Option<Message> {
        // Infinite wait: just loop forever
        let Some(timeout) = timeout.into() else {
            loop {
                let msg = self.timed_pop(None)?;
                if msg_types.contains(&msg.type_()) {
                    return Some(msg);
                }
            }
        };

        // Finite timeout
        let total = timeout;
        let start = std::time::Instant::now();

        loop {
            let elapsed = crate::ClockTime::from_nseconds(start.elapsed().as_nanos() as u64);

            // If timeout budget is exhausted, return None
            let remaining = total.checked_sub(elapsed)?;

            let msg = self.timed_pop(Some(remaining))?;

            if msg_types.contains(&msg.type_()) {
                return Some(msg);
            }

            // Discard non-matching messages without restarting the timeout
        }
    }

    /// Gets a message matching `type_` from the bus. Will discard all messages on
    /// the bus that do not match `type_` and that have been posted before the first
    /// message that does match `type_`. If there is no message matching `type_` on
    /// the bus, all messages will be discarded. It is not possible to use message
    /// enums beyond `GST_MESSAGE_EXTENDED` in the `events` mask.
    /// ## `types`
    /// message types to take into account
    ///
    /// # Returns
    ///
    /// the next [`Message`][crate::Message] matching
    ///  `type_` that is on the bus, or [`None`] if the bus is empty or there
    ///  is no message matching `type_`.
    #[doc(alias = "gst_bus_pop_filtered")]
    pub fn pop_filtered(&self, msg_types: &[MessageType]) -> Option<Message> {
        loop {
            let msg = self.pop()?;
            if msg_types.contains(&msg.type_()) {
                return Some(msg);
            }
        }
    }

    #[cfg(feature = "futures")]
    pub fn stream(&self) -> BusStream {
        BusStream::new(self)
    }

    #[cfg(feature = "futures")]
    pub fn stream_filtered<'a>(
        &self,
        message_types: &'a [MessageType],
    ) -> impl FusedStream<Item = Message> + Unpin + Send + 'a + use<'a> {
        self.stream().filter(move |message| {
            let message_type = message.type_();

            std::future::ready(message_types.contains(&message_type))
        })
    }
}

#[must_use = "iterators are lazy and do nothing unless consumed"]
#[derive(Debug)]
pub struct Iter<'a> {
    bus: &'a Bus,
    timeout: Option<crate::ClockTime>,
}

impl Iterator for Iter<'_> {
    type Item = Message;

    fn next(&mut self) -> Option<Message> {
        self.bus.timed_pop(self.timeout)
    }
}

// rustdoc-stripper-ignore-next
/// Manages ownership of the bus watch added to a bus with [`Bus::add_watch`] or [`Bus::add_watch_local`]
///
/// When dropped the bus watch is removed from the bus.
#[derive(Debug)]
#[must_use = "if unused the bus watch will immediately be removed"]
pub struct BusWatchGuard {
    bus: Bus,
}

impl Drop for BusWatchGuard {
    fn drop(&mut self) {
        let _ = self.bus.remove_watch();
    }
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};

    use super::*;

    #[test]
    fn test_sync_handler() {
        crate::init().unwrap();

        let bus = Bus::new();
        let msgs = Arc::new(Mutex::new(Vec::new()));
        let msgs_clone = msgs.clone();
        bus.set_sync_handler(move |_, msg| {
            msgs_clone.lock().unwrap().push(msg.clone());
            BusSyncReply::Pass
        });

        bus.post(crate::message::Eos::new()).unwrap();

        let msgs = msgs.lock().unwrap();
        assert_eq!(msgs.len(), 1);
        match msgs[0].view() {
            crate::MessageView::Eos(_) => (),
            _ => unreachable!(),
        }
    }
}