gstreamer/
promise.rs

1// Take a look at the license at the top of the repository in the LICENSE file.
2
3use std::{
4    ops::Deref,
5    pin::Pin,
6    ptr,
7    task::{Context, Poll},
8};
9
10use glib::translate::*;
11
12use crate::{PromiseResult, Structure, StructureRef, ffi};
13
14glib::wrapper! {
15    ///
16    /// const GstStructure *reply;
17    /// GstPromise *p;
18    /// if (gst_promise_wait (promise) != GST_PROMISE_RESULT_REPLIED)
19    ///  return; // interrupted or expired value
20    /// reply = gst_promise_get_reply (promise);
21    /// if (error in reply)
22    ///  return; // propagate error
23    /// p = gst_promise_new_with_change_func (another_promise_change_func, user_data, notify);
24    /// pass p to promise-using API
25    /// ]|
26    ///
27    /// Each [`Promise`][crate::Promise] starts out with a [`PromiseResult`][crate::PromiseResult] of
28    /// [`PromiseResult::Pending`][crate::PromiseResult::Pending] and only ever transitions once
29    /// into one of the other [`PromiseResult`][crate::PromiseResult]'s.
30    ///
31    /// In order to support multi-threaded code, [`reply()`][Self::reply()],
32    /// [`interrupt()`][Self::interrupt()] and [`expire()`][Self::expire()] may all be from
33    /// different threads with some restrictions and the final result of the promise
34    /// is whichever call is made first. There are two restrictions on ordering:
35    ///
36    /// 1. That [`reply()`][Self::reply()] and [`interrupt()`][Self::interrupt()] cannot be called
37    /// after [`expire()`][Self::expire()]
38    /// 2. That [`reply()`][Self::reply()] and [`interrupt()`][Self::interrupt()]
39    /// cannot be called twice.
40    ///
41    /// The change function set with [`with_change_func()`][Self::with_change_func()] is
42    /// called directly from either the [`reply()`][Self::reply()],
43    /// [`interrupt()`][Self::interrupt()] or [`expire()`][Self::expire()] and can be called
44    /// from an arbitrary thread. [`Promise`][crate::Promise] using APIs can restrict this to
45    /// a single thread or a subset of threads but that is entirely up to the API
46    /// that uses [`Promise`][crate::Promise].
47    #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
48    #[doc(alias = "GstPromise")]
49    pub struct Promise(Shared<ffi::GstPromise>);
50
51    match fn {
52        ref => |ptr| ffi::gst_mini_object_ref(ptr as *mut _),
53        unref => |ptr| ffi::gst_mini_object_unref(ptr as *mut _),
54        type_ => || ffi::gst_promise_get_type(),
55    }
56}
57
58#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
59pub enum PromiseError {
60    Interrupted,
61    Expired,
62    Other(PromiseResult),
63}
64
65impl Promise {
66    ///
67    /// # Returns
68    ///
69    /// a new [`Promise`][crate::Promise]
70    #[doc(alias = "gst_promise_new")]
71    pub fn new() -> Promise {
72        assert_initialized_main_thread!();
73        unsafe { from_glib_full(ffi::gst_promise_new()) }
74    }
75
76    /// `func` will be called exactly once when transitioning out of
77    /// [`PromiseResult::Pending`][crate::PromiseResult::Pending] into any of the other [`PromiseResult`][crate::PromiseResult]
78    /// states.
79    /// ## `func`
80    /// a `GstPromiseChangeFunc` to call
81    /// ## `notify`
82    /// notification function that `user_data` is no longer needed
83    ///
84    /// # Returns
85    ///
86    /// a new [`Promise`][crate::Promise]
87    #[doc(alias = "gst_promise_new_with_change_func")]
88    pub fn with_change_func<F>(func: F) -> Promise
89    where
90        F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
91    {
92        assert_initialized_main_thread!();
93        let user_data: Box<Option<F>> = Box::new(Some(func));
94
95        unsafe extern "C" fn trampoline<
96            F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
97        >(
98            promise: *mut ffi::GstPromise,
99            user_data: glib::ffi::gpointer,
100        ) {
101            unsafe {
102                let user_data: &mut Option<F> = &mut *(user_data as *mut _);
103                let callback = user_data.take().unwrap();
104
105                let promise: Borrowed<Promise> = from_glib_borrow(promise);
106
107                let res = match promise.wait() {
108                    PromiseResult::Replied => Ok(promise.get_reply()),
109                    PromiseResult::Interrupted => Err(PromiseError::Interrupted),
110                    PromiseResult::Expired => Err(PromiseError::Expired),
111                    PromiseResult::Pending => {
112                        panic!("Promise resolved but returned Pending");
113                    }
114                    err => Err(PromiseError::Other(err)),
115                };
116
117                callback(res);
118            }
119        }
120
121        unsafe extern "C" fn free_user_data<
122            F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
123        >(
124            user_data: glib::ffi::gpointer,
125        ) {
126            unsafe {
127                let _: Box<Option<F>> = Box::from_raw(user_data as *mut _);
128            }
129        }
130
131        unsafe {
132            from_glib_full(ffi::gst_promise_new_with_change_func(
133                Some(trampoline::<F>),
134                Box::into_raw(user_data) as *mut _,
135                Some(free_user_data::<F>),
136            ))
137        }
138    }
139
140    pub fn new_future() -> (Self, PromiseFuture) {
141        use futures_channel::oneshot;
142
143        // We only use the channel as a convenient waker
144        let (sender, receiver) = oneshot::channel();
145        let promise = Self::with_change_func(move |_res| {
146            let _ = sender.send(());
147        });
148
149        (promise.clone(), PromiseFuture(promise, receiver))
150    }
151
152    /// Expire a `self`. This will wake up any waiters with
153    /// [`PromiseResult::Expired`][crate::PromiseResult::Expired]. Called by a message loop when the parent
154    /// message is handled and/or destroyed (possibly unanswered).
155    #[doc(alias = "gst_promise_expire")]
156    pub fn expire(&self) {
157        unsafe {
158            ffi::gst_promise_expire(self.to_glib_none().0);
159        }
160    }
161
162    #[doc(alias = "gst_promise_get_reply")]
163    pub fn get_reply(&self) -> Option<&StructureRef> {
164        unsafe {
165            let s = ffi::gst_promise_get_reply(self.to_glib_none().0);
166            if s.is_null() {
167                None
168            } else {
169                Some(StructureRef::from_glib_borrow(s))
170            }
171        }
172    }
173
174    /// Interrupt waiting for a `self`. This will wake up any waiters with
175    /// [`PromiseResult::Interrupted`][crate::PromiseResult::Interrupted]. Called when the consumer does not want
176    /// the value produced anymore.
177    #[doc(alias = "gst_promise_interrupt")]
178    pub fn interrupt(&self) {
179        unsafe {
180            ffi::gst_promise_interrupt(self.to_glib_none().0);
181        }
182    }
183
184    /// Retrieve the reply set on `self`. `self` must be in
185    /// [`PromiseResult::Replied`][crate::PromiseResult::Replied] and the returned structure is owned by `self`
186    ///
187    /// # Returns
188    ///
189    /// The reply set on `self`
190    #[doc(alias = "gst_promise_reply")]
191    pub fn reply(&self, s: Option<Structure>) {
192        unsafe {
193            ffi::gst_promise_reply(
194                self.to_glib_none().0,
195                s.map(|s| s.into_glib_ptr()).unwrap_or(ptr::null_mut()),
196            );
197        }
198    }
199
200    /// Wait for `self` to move out of the [`PromiseResult::Pending`][crate::PromiseResult::Pending] state.
201    /// If `self` is not in [`PromiseResult::Pending`][crate::PromiseResult::Pending] then it will return
202    /// immediately with the current result.
203    ///
204    /// # Returns
205    ///
206    /// the result of the promise
207    #[doc(alias = "gst_promise_wait")]
208    pub fn wait(&self) -> PromiseResult {
209        unsafe { from_glib(ffi::gst_promise_wait(self.to_glib_none().0)) }
210    }
211}
212
213impl Default for Promise {
214    fn default() -> Self {
215        Self::new()
216    }
217}
218
219unsafe impl Send for Promise {}
220unsafe impl Sync for Promise {}
221
222#[derive(Debug)]
223pub struct PromiseFuture(Promise, futures_channel::oneshot::Receiver<()>);
224
225pub struct PromiseReply(Promise);
226
227impl std::future::Future for PromiseFuture {
228    type Output = Result<Option<PromiseReply>, PromiseError>;
229
230    fn poll(mut self: Pin<&mut Self>, context: &mut Context) -> Poll<Self::Output> {
231        match Pin::new(&mut self.1).poll(context) {
232            Poll::Ready(Err(_)) => panic!("Sender dropped before callback was called"),
233            Poll::Ready(Ok(())) => {
234                let res = match self.0.wait() {
235                    PromiseResult::Replied => {
236                        if self.0.get_reply().is_none() {
237                            Ok(None)
238                        } else {
239                            Ok(Some(PromiseReply(self.0.clone())))
240                        }
241                    }
242                    PromiseResult::Interrupted => Err(PromiseError::Interrupted),
243                    PromiseResult::Expired => Err(PromiseError::Expired),
244                    PromiseResult::Pending => {
245                        panic!("Promise resolved but returned Pending");
246                    }
247                    err => Err(PromiseError::Other(err)),
248                };
249                Poll::Ready(res)
250            }
251            Poll::Pending => Poll::Pending,
252        }
253    }
254}
255
256impl futures_core::future::FusedFuture for PromiseFuture {
257    fn is_terminated(&self) -> bool {
258        self.1.is_terminated()
259    }
260}
261
262impl Deref for PromiseReply {
263    type Target = StructureRef;
264
265    #[inline]
266    fn deref(&self) -> &StructureRef {
267        self.0.get_reply().expect("Promise without reply")
268    }
269}
270
271impl std::fmt::Debug for PromiseReply {
272    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
273        let mut debug = f.debug_tuple("PromiseReply");
274
275        match self.0.get_reply() {
276            Some(reply) => debug.field(reply),
277            None => debug.field(&"<no reply>"),
278        }
279        .finish()
280    }
281}
282
283#[cfg(test)]
284mod tests {
285    use std::{sync::mpsc::channel, thread};
286
287    use super::*;
288
289    #[test]
290    fn test_change_func() {
291        crate::init().unwrap();
292
293        let (sender, receiver) = channel();
294        let promise = Promise::with_change_func(move |res| {
295            sender.send(res.map(|s| s.map(ToOwned::to_owned))).unwrap();
296        });
297
298        thread::spawn(move || {
299            promise.reply(Some(crate::Structure::new_empty("foo/bar")));
300        });
301
302        let res = receiver.recv().unwrap();
303        let res = res.expect("promise failed").expect("promise returned None");
304        assert_eq!(res.name(), "foo/bar");
305    }
306}
gstreamer/promise.rs

gstreamer/
promise.rs
