gstreamer_app/

app_src.rs

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

use std::{mem, panic, ptr};

#[cfg(not(panic = "abort"))]
use std::sync::atomic::{AtomicBool, Ordering};

use glib::{
    ffi::{gboolean, gpointer},
    prelude::*,
    translate::*,
};

use crate::{AppSrc, ffi};

#[cfg(feature = "futures")]
pub use crate::app_src_futures::AppSrcSink;

#[allow(clippy::type_complexity)]
pub struct AppSrcCallbacks {
    need_data: Option<Box<dyn FnMut(&AppSrc, u32) + Send + 'static>>,
    enough_data: Option<Box<dyn Fn(&AppSrc) + Send + Sync + 'static>>,
    seek_data: Option<Box<dyn Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>>,
    #[cfg(not(panic = "abort"))]
    panicked: AtomicBool,
    callbacks: ffi::GstAppSrcCallbacks,
}

unsafe impl Send for AppSrcCallbacks {}
unsafe impl Sync for AppSrcCallbacks {}

impl AppSrcCallbacks {
    pub fn builder() -> AppSrcCallbacksBuilder {
        skip_assert_initialized!();

        AppSrcCallbacksBuilder {
            need_data: None,
            enough_data: None,
            seek_data: None,
        }
    }
}

#[allow(clippy::type_complexity)]
#[must_use = "The builder must be built to be used"]
pub struct AppSrcCallbacksBuilder {
    need_data: Option<Box<dyn FnMut(&AppSrc, u32) + Send + 'static>>,
    enough_data: Option<Box<dyn Fn(&AppSrc) + Send + Sync + 'static>>,
    seek_data: Option<Box<dyn Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>>,
}

impl AppSrcCallbacksBuilder {
    pub fn need_data<F: FnMut(&AppSrc, u32) + Send + 'static>(self, need_data: F) -> Self {
        Self {
            need_data: Some(Box::new(need_data)),
            ..self
        }
    }

    pub fn need_data_if<F: FnMut(&AppSrc, u32) + Send + 'static>(
        self,
        need_data: F,
        predicate: bool,
    ) -> Self {
        if predicate {
            self.need_data(need_data)
        } else {
            self
        }
    }

    pub fn need_data_if_some<F: FnMut(&AppSrc, u32) + Send + 'static>(
        self,
        need_data: Option<F>,
    ) -> Self {
        if let Some(need_data) = need_data {
            self.need_data(need_data)
        } else {
            self
        }
    }

    pub fn enough_data<F: Fn(&AppSrc) + Send + Sync + 'static>(self, enough_data: F) -> Self {
        Self {
            enough_data: Some(Box::new(enough_data)),
            ..self
        }
    }

    pub fn enough_data_if<F: Fn(&AppSrc) + Send + Sync + 'static>(
        self,
        enough_data: F,
        predicate: bool,
    ) -> Self {
        if predicate {
            self.enough_data(enough_data)
        } else {
            self
        }
    }

    pub fn enough_data_if_some<F: Fn(&AppSrc) + Send + Sync + 'static>(
        self,
        enough_data: Option<F>,
    ) -> Self {
        if let Some(enough_data) = enough_data {
            self.enough_data(enough_data)
        } else {
            self
        }
    }

    pub fn seek_data<F: Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>(
        self,
        seek_data: F,
    ) -> Self {
        Self {
            seek_data: Some(Box::new(seek_data)),
            ..self
        }
    }

    pub fn seek_data_if<F: Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>(
        self,
        seek_data: F,
        predicate: bool,
    ) -> Self {
        if predicate {
            self.seek_data(seek_data)
        } else {
            self
        }
    }

    pub fn seek_data_if_some<F: Fn(&AppSrc, u64) -> bool + Send + Sync + 'static>(
        self,
        seek_data: Option<F>,
    ) -> Self {
        if let Some(seek_data) = seek_data {
            self.seek_data(seek_data)
        } else {
            self
        }
    }

    #[must_use = "Building the callbacks without using them has no effect"]
    pub fn build(self) -> AppSrcCallbacks {
        let have_need_data = self.need_data.is_some();
        let have_enough_data = self.enough_data.is_some();
        let have_seek_data = self.seek_data.is_some();

        AppSrcCallbacks {
            need_data: self.need_data,
            enough_data: self.enough_data,
            seek_data: self.seek_data,
            #[cfg(not(panic = "abort"))]
            panicked: AtomicBool::new(false),
            callbacks: ffi::GstAppSrcCallbacks {
                need_data: if have_need_data {
                    Some(trampoline_need_data)
                } else {
                    None
                },
                enough_data: if have_enough_data {
                    Some(trampoline_enough_data)
                } else {
                    None
                },
                seek_data: if have_seek_data {
                    Some(trampoline_seek_data)
                } else {
                    None
                },
                _gst_reserved: [
                    ptr::null_mut(),
                    ptr::null_mut(),
                    ptr::null_mut(),
                    ptr::null_mut(),
                ],
            },
        }
    }
}

unsafe extern "C" fn trampoline_need_data(
    appsrc: *mut ffi::GstAppSrc,
    length: u32,
    callbacks: gpointer,
) {
    unsafe {
        let callbacks = callbacks as *mut AppSrcCallbacks;
        let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);

        #[cfg(not(panic = "abort"))]
        if (*callbacks).panicked.load(Ordering::Relaxed) {
            let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
            gst::subclass::post_panic_error_message(
                element.upcast_ref(),
                element.upcast_ref(),
                None,
            );
            return;
        }

        if let Some(ref mut need_data) = (*callbacks).need_data {
            let result =
                panic::catch_unwind(panic::AssertUnwindSafe(|| need_data(&element, length)));
            match result {
                Ok(result) => result,
                Err(err) => {
                    #[cfg(panic = "abort")]
                    {
                        unreachable!("{err:?}");
                    }
                    #[cfg(not(panic = "abort"))]
                    {
                        (*callbacks).panicked.store(true, Ordering::Relaxed);
                        gst::subclass::post_panic_error_message(
                            element.upcast_ref(),
                            element.upcast_ref(),
                            Some(err),
                        );
                    }
                }
            }
        }
    }
}

unsafe extern "C" fn trampoline_enough_data(appsrc: *mut ffi::GstAppSrc, callbacks: gpointer) {
    unsafe {
        let callbacks = callbacks as *const AppSrcCallbacks;
        let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);

        #[cfg(not(panic = "abort"))]
        if (*callbacks).panicked.load(Ordering::Relaxed) {
            let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
            gst::subclass::post_panic_error_message(
                element.upcast_ref(),
                element.upcast_ref(),
                None,
            );
            return;
        }

        if let Some(ref enough_data) = (*callbacks).enough_data {
            let result = panic::catch_unwind(panic::AssertUnwindSafe(|| enough_data(&element)));
            match result {
                Ok(result) => result,
                Err(err) => {
                    #[cfg(panic = "abort")]
                    {
                        unreachable!("{err:?}");
                    }
                    #[cfg(not(panic = "abort"))]
                    {
                        (*callbacks).panicked.store(true, Ordering::Relaxed);
                        gst::subclass::post_panic_error_message(
                            element.upcast_ref(),
                            element.upcast_ref(),
                            Some(err),
                        );
                    }
                }
            }
        }
    }
}

unsafe extern "C" fn trampoline_seek_data(
    appsrc: *mut ffi::GstAppSrc,
    offset: u64,
    callbacks: gpointer,
) -> gboolean {
    unsafe {
        let callbacks = callbacks as *const AppSrcCallbacks;
        let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);

        #[cfg(not(panic = "abort"))]
        if (*callbacks).panicked.load(Ordering::Relaxed) {
            let element: Borrowed<AppSrc> = from_glib_borrow(appsrc);
            gst::subclass::post_panic_error_message(
                element.upcast_ref(),
                element.upcast_ref(),
                None,
            );
            return false.into_glib();
        }

        let ret = if let Some(ref seek_data) = (*callbacks).seek_data {
            let result =
                panic::catch_unwind(panic::AssertUnwindSafe(|| seek_data(&element, offset)));
            match result {
                Ok(result) => result,
                Err(err) => {
                    #[cfg(panic = "abort")]
                    {
                        unreachable!("{err:?}");
                    }
                    #[cfg(not(panic = "abort"))]
                    {
                        (*callbacks).panicked.store(true, Ordering::Relaxed);
                        gst::subclass::post_panic_error_message(
                            element.upcast_ref(),
                            element.upcast_ref(),
                            Some(err),
                        );

                        false
                    }
                }
            }
        } else {
            false
        };

        ret.into_glib()
    }
}

unsafe extern "C" fn destroy_callbacks(ptr: gpointer) {
    unsafe {
        let _ = Box::<AppSrcCallbacks>::from_raw(ptr as *mut _);
    }
}

impl AppSrc {
    // rustdoc-stripper-ignore-next
    /// Creates a new builder-pattern struct instance to construct [`AppSrc`] objects.
    ///
    /// This method returns an instance of [`AppSrcBuilder`](crate::builders::AppSrcBuilder) which can be used to create [`AppSrc`] objects.
    pub fn builder<'a>() -> AppSrcBuilder<'a> {
        assert_initialized_main_thread!();
        AppSrcBuilder {
            builder: gst::Object::builder(),
            callbacks: None,
            automatic_eos: None,
        }
    }

    /// Set callbacks which will be executed when data is needed, enough data has
    /// been collected or when a seek should be performed.
    /// This is an alternative to using the signals, it has lower overhead and is thus
    /// less expensive, but also less flexible.
    ///
    /// If callbacks are installed, no signals will be emitted for performance
    /// reasons.
    ///
    /// Before 1.16.3 it was not possible to change the callbacks in a thread-safe
    /// way.
    ///
    /// Since 1.28.3 it is allowed to set the `callbacks` to [`None`] to unset them.
    ///
    /// Note that `gst_app_src_set_callbacks()` and
    /// `gst_app_src_set_simple_callbacks()` are mutually exclusive and setting one
    /// will unset the other.
    /// ## `callbacks`
    /// the callbacks
    /// ## `notify`
    /// a destroy notify function
    #[doc(alias = "gst_app_src_set_callbacks")]
    pub fn set_callbacks(&self, callbacks: AppSrcCallbacks) {
        unsafe {
            let src = 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-app-src-callbacks"));

                // This is not thread-safe before 1.16.3, see
                // https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/merge_requests/570
                if gst::version() < (1, 16, 3, 0) {
                    if !glib::gobject_ffi::g_object_get_qdata(
                        src as *mut _,
                        set_once_quark.into_glib(),
                    )
                    .is_null()
                    {
                        panic!("AppSrc callbacks can only be set once");
                    }

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

            ffi::gst_app_src_set_callbacks(
                src,
                mut_override(&callbacks.callbacks),
                Box::into_raw(Box::new(callbacks)) as *mut _,
                Some(destroy_callbacks),
            );
        }
    }

    /// Configure the `min` and `max` latency in `src`. If `min` is set to -1, the
    /// default latency calculations for pseudo-live sources will be used.
    /// ## `min`
    /// the min latency
    /// ## `max`
    /// the max latency
    #[doc(alias = "gst_app_src_set_latency")]
    pub fn set_latency(
        &self,
        min: impl Into<Option<gst::ClockTime>>,
        max: impl Into<Option<gst::ClockTime>>,
    ) {
        unsafe {
            ffi::gst_app_src_set_latency(
                self.to_glib_none().0,
                min.into().into_glib(),
                max.into().into_glib(),
            );
        }
    }

    /// Retrieve the min and max latencies in `min` and `max` respectively.
    ///
    /// # Returns
    ///
    ///
    /// ## `min`
    /// the min latency
    ///
    /// ## `max`
    /// the max latency
    #[doc(alias = "get_latency")]
    #[doc(alias = "gst_app_src_get_latency")]
    pub fn latency(&self) -> (Option<gst::ClockTime>, Option<gst::ClockTime>) {
        unsafe {
            let mut min = mem::MaybeUninit::uninit();
            let mut max = mem::MaybeUninit::uninit();
            ffi::gst_app_src_get_latency(self.to_glib_none().0, min.as_mut_ptr(), max.as_mut_ptr());
            (from_glib(min.assume_init()), from_glib(max.assume_init()))
        }
    }

    #[doc(alias = "do-timestamp")]
    #[doc(alias = "gst_base_src_set_do_timestamp")]
    pub fn set_do_timestamp(&self, timestamp: bool) {
        unsafe {
            gst_base::ffi::gst_base_src_set_do_timestamp(
                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc,
                timestamp.into_glib(),
            );
        }
    }

    #[doc(alias = "do-timestamp")]
    #[doc(alias = "gst_base_src_get_do_timestamp")]
    pub fn do_timestamp(&self) -> bool {
        unsafe {
            from_glib(gst_base::ffi::gst_base_src_get_do_timestamp(
                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc
            ))
        }
    }

    #[doc(alias = "do-timestamp")]
    pub fn connect_do_timestamp_notify<F: Fn(&Self) + Send + Sync + 'static>(
        &self,
        f: F,
    ) -> glib::SignalHandlerId {
        unsafe extern "C" fn notify_do_timestamp_trampoline<
            F: Fn(&AppSrc) + Send + Sync + 'static,
        >(
            this: *mut ffi::GstAppSrc,
            _param_spec: glib::ffi::gpointer,
            f: glib::ffi::gpointer,
        ) {
            unsafe {
                let f: &F = &*(f as *const F);
                f(&AppSrc::from_glib_borrow(this))
            }
        }
        unsafe {
            let f: Box<F> = Box::new(f);
            glib::signal::connect_raw(
                self.as_ptr() as *mut _,
                b"notify::do-timestamp\0".as_ptr() as *const _,
                Some(mem::transmute::<*const (), unsafe extern "C" fn()>(
                    notify_do_timestamp_trampoline::<F> as *const (),
                )),
                Box::into_raw(f),
            )
        }
    }

    #[doc(alias = "set-automatic-eos")]
    #[doc(alias = "gst_base_src_set_automatic_eos")]
    pub fn set_automatic_eos(&self, automatic_eos: bool) {
        unsafe {
            gst_base::ffi::gst_base_src_set_automatic_eos(
                self.as_ptr() as *mut gst_base::ffi::GstBaseSrc,
                automatic_eos.into_glib(),
            );
        }
    }

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

// rustdoc-stripper-ignore-next
/// A [builder-pattern] type to construct [`AppSrc`] objects.
///
/// [builder-pattern]: https://doc.rust-lang.org/1.0.0/style/ownership/builders.html
#[must_use = "The builder must be built to be used"]
pub struct AppSrcBuilder<'a> {
    builder: gst::gobject::GObjectBuilder<'a, AppSrc>,
    callbacks: Option<AppSrcCallbacks>,
    automatic_eos: Option<bool>,
}

impl<'a> AppSrcBuilder<'a> {
    // rustdoc-stripper-ignore-next
    /// Build the [`AppSrc`].
    ///
    /// # Panics
    ///
    /// This panics if the [`AppSrc`] doesn't have all the given properties or
    /// property values of the wrong type are provided.
    #[must_use = "Building the object from the builder is usually expensive and is not expected to have side effects"]
    pub fn build(self) -> AppSrc {
        let appsrc = self.builder.build().unwrap();

        if let Some(callbacks) = self.callbacks {
            appsrc.set_callbacks(callbacks);
        }

        if let Some(automatic_eos) = self.automatic_eos {
            appsrc.set_automatic_eos(automatic_eos);
        }

        appsrc
    }

    pub fn automatic_eos(self, automatic_eos: bool) -> Self {
        Self {
            automatic_eos: Some(automatic_eos),
            ..self
        }
    }

    pub fn block(self, block: bool) -> Self {
        Self {
            builder: self.builder.property("block", block),
            ..self
        }
    }

    pub fn callbacks(self, callbacks: AppSrcCallbacks) -> Self {
        Self {
            callbacks: Some(callbacks),
            ..self
        }
    }

    pub fn caps(self, caps: &'a gst::Caps) -> Self {
        Self {
            builder: self.builder.property("caps", caps),
            ..self
        }
    }

    pub fn do_timestamp(self, do_timestamp: bool) -> Self {
        Self {
            builder: self.builder.property("do-timestamp", do_timestamp),
            ..self
        }
    }

    pub fn duration(self, duration: u64) -> Self {
        Self {
            builder: self.builder.property("duration", duration),
            ..self
        }
    }

    pub fn format(self, format: gst::Format) -> Self {
        Self {
            builder: self.builder.property("format", format),
            ..self
        }
    }

    #[cfg(feature = "v1_18")]
    #[cfg_attr(docsrs, doc(cfg(feature = "v1_18")))]
    pub fn handle_segment_change(self, handle_segment_change: bool) -> Self {
        Self {
            builder: self
                .builder
                .property("handle-segment-change", handle_segment_change),
            ..self
        }
    }

    pub fn is_live(self, is_live: bool) -> Self {
        Self {
            builder: self.builder.property("is-live", is_live),
            ..self
        }
    }

    #[cfg(feature = "v1_20")]
    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
    pub fn leaky_type(self, leaky_type: crate::AppLeakyType) -> Self {
        Self {
            builder: self.builder.property("leaky-type", leaky_type),
            ..self
        }
    }

    #[cfg(feature = "v1_20")]
    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
    pub fn max_buffers(self, max_buffers: u64) -> Self {
        Self {
            builder: self.builder.property("max-buffers", max_buffers),
            ..self
        }
    }

    pub fn max_bytes(self, max_bytes: u64) -> Self {
        Self {
            builder: self.builder.property("max-bytes", max_bytes),
            ..self
        }
    }

    pub fn max_latency(self, max_latency: i64) -> Self {
        Self {
            builder: self.builder.property("max-latency", max_latency),
            ..self
        }
    }

    #[cfg(feature = "v1_20")]
    #[cfg_attr(docsrs, doc(cfg(feature = "v1_20")))]
    pub fn max_time(self, max_time: gst::ClockTime) -> Self {
        Self {
            builder: self.builder.property("max-time", max_time),
            ..self
        }
    }

    pub fn min_latency(self, min_latency: i64) -> Self {
        Self {
            builder: self.builder.property("min-latency", min_latency),
            ..self
        }
    }

    pub fn min_percent(self, min_percent: u32) -> Self {
        Self {
            builder: self.builder.property("min-percent", min_percent),
            ..self
        }
    }

    pub fn size(self, size: i64) -> Self {
        Self {
            builder: self.builder.property("size", size),
            ..self
        }
    }

    pub fn stream_type(self, stream_type: crate::AppStreamType) -> Self {
        Self {
            builder: self.builder.property("stream-type", stream_type),
            ..self
        }
    }

    #[cfg(feature = "v1_28")]
    #[cfg_attr(docsrs, doc(cfg(feature = "v1_28")))]
    pub fn silent(self, silent: bool) -> Self {
        Self {
            builder: self.builder.property("silent", silent),
            ..self
        }
    }

    // rustdoc-stripper-ignore-next
    /// Sets property `name` to the given value `value`.
    ///
    /// Overrides any default or previously defined value for `name`.
    #[inline]
    pub fn property(self, name: &'a str, value: impl Into<glib::Value> + 'a) -> Self {
        Self {
            builder: self.builder.property(name, value),
            ..self
        }
    }

    // rustdoc-stripper-ignore-next
    /// Sets property `name` to the given string value `value`.
    #[inline]
    pub fn property_from_str(self, name: &'a str, value: &'a str) -> Self {
        Self {
            builder: self.builder.property_from_str(name, value),
            ..self
        }
    }

    gst::impl_builder_gvalue_extra_setters!(property_and_name);
}

#[cfg(test)]
mod tests {
    use gst::prelude::*;

    use super::*;

    #[test]
    fn builder_caps_lt() {
        gst::init().unwrap();

        let caps = &gst::Caps::new_any();
        {
            let stream_type = "random-access".to_owned();
            let appsrc = AppSrc::builder()
                .property_from_str("stream-type", &stream_type)
                .caps(caps)
                .build();
            assert_eq!(
                appsrc.property::<crate::AppStreamType>("stream-type"),
                crate::AppStreamType::RandomAccess
            );
            assert!(appsrc.property::<gst::Caps>("caps").is_any());
        }

        let stream_type = &"random-access".to_owned();
        {
            let caps = &gst::Caps::new_any();
            let appsrc = AppSrc::builder()
                .property_from_str("stream-type", stream_type)
                .caps(caps)
                .build();
            assert_eq!(
                appsrc.property::<crate::AppStreamType>("stream-type"),
                crate::AppStreamType::RandomAccess
            );
            assert!(appsrc.property::<gst::Caps>("caps").is_any());
        }
    }
}