gstreamer/

promise.rs

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

use std::{
    ops::Deref,
    pin::Pin,
    ptr,
    task::{Context, Poll},
};

use glib::translate::*;

use crate::{ffi, PromiseResult, Structure, StructureRef};

glib::wrapper! {
    /// The [`Promise`][crate::Promise] object implements the container for values that may
    /// be available later. i.e. a Future or a Promise in
    /// <https://en.wikipedia.org/wiki/Futures_and_promises>.
    /// As with all Future/Promise-like functionality, there is the concept of the
    /// producer of the value and the consumer of the value.
    ///
    /// A [`Promise`][crate::Promise] is created with [`new()`][Self::new()] by the consumer and passed
    /// to the producer to avoid thread safety issues with the change callback.
    /// A [`Promise`][crate::Promise] can be replied to with a value (or an error) by the producer
    /// with [`reply()`][Self::reply()]. The exact value returned is defined by the API
    /// contract of the producer and [`None`] may be a valid reply.
    /// [`interrupt()`][Self::interrupt()] is for the consumer to
    /// indicate to the producer that the value is not needed anymore and producing
    /// that value can stop. The [`PromiseResult::Expired`][crate::PromiseResult::Expired] state set by a call
    /// to [`expire()`][Self::expire()] indicates to the consumer that a value will never
    /// be produced and is intended to be called by a third party that implements
    /// some notion of message handling such as [`Bus`][crate::Bus].
    /// A callback can also be installed at [`Promise`][crate::Promise] creation for
    /// result changes with [`with_change_func()`][Self::with_change_func()].
    /// The change callback can be used to chain `GstPromises`'s together as in the
    /// following example.
    ///
    ///
    /// **⚠️ The following code is in C ⚠️**
    ///
    /// ```C
    /// const GstStructure *reply;
    /// GstPromise *p;
    /// if (gst_promise_wait (promise) != GST_PROMISE_RESULT_REPLIED)
    ///   return; // interrupted or expired value
    /// reply = gst_promise_get_reply (promise);
    /// if (error in reply)
    ///   return; // propagate error
    /// p = gst_promise_new_with_change_func (another_promise_change_func, user_data, notify);
    /// pass p to promise-using API
    /// ```
    ///
    /// Each [`Promise`][crate::Promise] starts out with a [`PromiseResult`][crate::PromiseResult] of
    /// [`PromiseResult::Pending`][crate::PromiseResult::Pending] and only ever transitions once
    /// into one of the other [`PromiseResult`][crate::PromiseResult]'s.
    ///
    /// In order to support multi-threaded code, [`reply()`][Self::reply()],
    /// [`interrupt()`][Self::interrupt()] and [`expire()`][Self::expire()] may all be from
    /// different threads with some restrictions and the final result of the promise
    /// is whichever call is made first. There are two restrictions on ordering:
    ///
    /// 1. That [`reply()`][Self::reply()] and [`interrupt()`][Self::interrupt()] cannot be called
    /// after [`expire()`][Self::expire()]
    /// 2. That [`reply()`][Self::reply()] and [`interrupt()`][Self::interrupt()]
    /// cannot be called twice.
    ///
    /// The change function set with [`with_change_func()`][Self::with_change_func()] is
    /// called directly from either the [`reply()`][Self::reply()],
    /// [`interrupt()`][Self::interrupt()] or [`expire()`][Self::expire()] and can be called
    /// from an arbitrary thread. [`Promise`][crate::Promise] using APIs can restrict this to
    /// a single thread or a subset of threads but that is entirely up to the API
    /// that uses [`Promise`][crate::Promise].
    #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
    #[doc(alias = "GstPromise")]
    pub struct Promise(Shared<ffi::GstPromise>);

    match fn {
        ref => |ptr| ffi::gst_mini_object_ref(ptr as *mut _),
        unref => |ptr| ffi::gst_mini_object_unref(ptr as *mut _),
        type_ => || ffi::gst_promise_get_type(),
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum PromiseError {
    Interrupted,
    Expired,
    Other(PromiseResult),
}

impl Promise {
    ///
    /// # Returns
    ///
    /// a new [`Promise`][crate::Promise]
    #[doc(alias = "gst_promise_new")]
    pub fn new() -> Promise {
        assert_initialized_main_thread!();
        unsafe { from_glib_full(ffi::gst_promise_new()) }
    }

    /// `func` will be called exactly once when transitioning out of
    /// [`PromiseResult::Pending`][crate::PromiseResult::Pending] into any of the other [`PromiseResult`][crate::PromiseResult]
    /// states.
    /// ## `func`
    /// a `GstPromiseChangeFunc` to call
    /// ## `notify`
    /// notification function that `user_data` is no longer needed
    ///
    /// # Returns
    ///
    /// a new [`Promise`][crate::Promise]
    #[doc(alias = "gst_promise_new_with_change_func")]
    pub fn with_change_func<F>(func: F) -> Promise
    where
        F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
    {
        assert_initialized_main_thread!();
        let user_data: Box<Option<F>> = Box::new(Some(func));

        unsafe extern "C" fn trampoline<
            F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
        >(
            promise: *mut ffi::GstPromise,
            user_data: glib::ffi::gpointer,
        ) {
            let user_data: &mut Option<F> = &mut *(user_data as *mut _);
            let callback = user_data.take().unwrap();

            let promise: Borrowed<Promise> = from_glib_borrow(promise);

            let res = match promise.wait() {
                PromiseResult::Replied => Ok(promise.get_reply()),
                PromiseResult::Interrupted => Err(PromiseError::Interrupted),
                PromiseResult::Expired => Err(PromiseError::Expired),
                PromiseResult::Pending => {
                    panic!("Promise resolved but returned Pending");
                }
                err => Err(PromiseError::Other(err)),
            };

            callback(res);
        }

        unsafe extern "C" fn free_user_data<
            F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
        >(
            user_data: glib::ffi::gpointer,
        ) {
            let _: Box<Option<F>> = Box::from_raw(user_data as *mut _);
        }

        unsafe {
            from_glib_full(ffi::gst_promise_new_with_change_func(
                Some(trampoline::<F>),
                Box::into_raw(user_data) as *mut _,
                Some(free_user_data::<F>),
            ))
        }
    }

    pub fn new_future() -> (Self, PromiseFuture) {
        use futures_channel::oneshot;

        // We only use the channel as a convenient waker
        let (sender, receiver) = oneshot::channel();
        let promise = Self::with_change_func(move |_res| {
            let _ = sender.send(());
        });

        (promise.clone(), PromiseFuture(promise, receiver))
    }

    /// Expire a `self`. This will wake up any waiters with
    /// [`PromiseResult::Expired`][crate::PromiseResult::Expired]. Called by a message loop when the parent
    /// message is handled and/or destroyed (possibly unanswered).
    #[doc(alias = "gst_promise_expire")]
    pub fn expire(&self) {
        unsafe {
            ffi::gst_promise_expire(self.to_glib_none().0);
        }
    }

    #[doc(alias = "gst_promise_get_reply")]
    pub fn get_reply(&self) -> Option<&StructureRef> {
        unsafe {
            let s = ffi::gst_promise_get_reply(self.to_glib_none().0);
            if s.is_null() {
                None
            } else {
                Some(StructureRef::from_glib_borrow(s))
            }
        }
    }

    /// Interrupt waiting for a `self`. This will wake up any waiters with
    /// [`PromiseResult::Interrupted`][crate::PromiseResult::Interrupted]. Called when the consumer does not want
    /// the value produced anymore.
    #[doc(alias = "gst_promise_interrupt")]
    pub fn interrupt(&self) {
        unsafe {
            ffi::gst_promise_interrupt(self.to_glib_none().0);
        }
    }

    /// Retrieve the reply set on `self`. `self` must be in
    /// [`PromiseResult::Replied`][crate::PromiseResult::Replied] and the returned structure is owned by `self`
    ///
    /// # Returns
    ///
    /// The reply set on `self`
    #[doc(alias = "gst_promise_reply")]
    pub fn reply(&self, s: Option<Structure>) {
        unsafe {
            ffi::gst_promise_reply(
                self.to_glib_none().0,
                s.map(|s| s.into_glib_ptr()).unwrap_or(ptr::null_mut()),
            );
        }
    }

    /// Wait for `self` to move out of the [`PromiseResult::Pending`][crate::PromiseResult::Pending] state.
    /// If `self` is not in [`PromiseResult::Pending`][crate::PromiseResult::Pending] then it will return
    /// immediately with the current result.
    ///
    /// # Returns
    ///
    /// the result of the promise
    #[doc(alias = "gst_promise_wait")]
    pub fn wait(&self) -> PromiseResult {
        unsafe { from_glib(ffi::gst_promise_wait(self.to_glib_none().0)) }
    }
}

impl Default for Promise {
    fn default() -> Self {
        Self::new()
    }
}

unsafe impl Send for Promise {}
unsafe impl Sync for Promise {}

#[derive(Debug)]
pub struct PromiseFuture(Promise, futures_channel::oneshot::Receiver<()>);

pub struct PromiseReply(Promise);

impl std::future::Future for PromiseFuture {
    type Output = Result<Option<PromiseReply>, PromiseError>;

    fn poll(mut self: Pin<&mut Self>, context: &mut Context) -> Poll<Self::Output> {
        match Pin::new(&mut self.1).poll(context) {
            Poll::Ready(Err(_)) => panic!("Sender dropped before callback was called"),
            Poll::Ready(Ok(())) => {
                let res = match self.0.wait() {
                    PromiseResult::Replied => {
                        if self.0.get_reply().is_none() {
                            Ok(None)
                        } else {
                            Ok(Some(PromiseReply(self.0.clone())))
                        }
                    }
                    PromiseResult::Interrupted => Err(PromiseError::Interrupted),
                    PromiseResult::Expired => Err(PromiseError::Expired),
                    PromiseResult::Pending => {
                        panic!("Promise resolved but returned Pending");
                    }
                    err => Err(PromiseError::Other(err)),
                };
                Poll::Ready(res)
            }
            Poll::Pending => Poll::Pending,
        }
    }
}

impl futures_core::future::FusedFuture for PromiseFuture {
    fn is_terminated(&self) -> bool {
        self.1.is_terminated()
    }
}

impl Deref for PromiseReply {
    type Target = StructureRef;

    #[inline]
    fn deref(&self) -> &StructureRef {
        self.0.get_reply().expect("Promise without reply")
    }
}

impl std::fmt::Debug for PromiseReply {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut debug = f.debug_tuple("PromiseReply");

        match self.0.get_reply() {
            Some(reply) => debug.field(reply),
            None => debug.field(&"<no reply>"),
        }
        .finish()
    }
}

#[cfg(test)]
mod tests {
    use std::{sync::mpsc::channel, thread};

    use super::*;

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

        let (sender, receiver) = channel();
        let promise = Promise::with_change_func(move |res| {
            sender.send(res.map(|s| s.map(ToOwned::to_owned))).unwrap();
        });

        thread::spawn(move || {
            promise.reply(Some(crate::Structure::new_empty("foo/bar")));
        });

        let res = receiver.recv().unwrap();
        let res = res.expect("promise failed").expect("promise returned None");
        assert_eq!(res.name(), "foo/bar");
    }
}