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    #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
16    #[doc(alias = "GstPromise")]
17    pub struct Promise(Shared<ffi::GstPromise>);
18
19    match fn {
20        ref => |ptr| ffi::gst_mini_object_ref(ptr as *mut _),
21        unref => |ptr| ffi::gst_mini_object_unref(ptr as *mut _),
22        type_ => || ffi::gst_promise_get_type(),
23    }
24}
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
27pub enum PromiseError {
28    Interrupted,
29    Expired,
30    Other(PromiseResult),
31}
32
33impl Promise {
34    #[doc(alias = "gst_promise_new")]
35    pub fn new() -> Promise {
36        assert_initialized_main_thread!();
37        unsafe { from_glib_full(ffi::gst_promise_new()) }
38    }
39
40    #[doc(alias = "gst_promise_new_with_change_func")]
41    pub fn with_change_func<F>(func: F) -> Promise
42    where
43        F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
44    {
45        assert_initialized_main_thread!();
46        let user_data: Box<Option<F>> = Box::new(Some(func));
47
48        unsafe extern "C" fn trampoline<
49            F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
50        >(
51            promise: *mut ffi::GstPromise,
52            user_data: glib::ffi::gpointer,
53        ) {
54            unsafe {
55                let user_data: &mut Option<F> = &mut *(user_data as *mut _);
56                let callback = user_data.take().unwrap();
57
58                let promise: Borrowed<Promise> = from_glib_borrow(promise);
59
60                let res = match promise.wait() {
61                    PromiseResult::Replied => Ok(promise.get_reply()),
62                    PromiseResult::Interrupted => Err(PromiseError::Interrupted),
63                    PromiseResult::Expired => Err(PromiseError::Expired),
64                    PromiseResult::Pending => {
65                        panic!("Promise resolved but returned Pending");
66                    }
67                    err => Err(PromiseError::Other(err)),
68                };
69
70                callback(res);
71            }
72        }
73
74        unsafe extern "C" fn free_user_data<
75            F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
76        >(
77            user_data: glib::ffi::gpointer,
78        ) {
79            unsafe {
80                let _: Box<Option<F>> = Box::from_raw(user_data as *mut _);
81            }
82        }
83
84        unsafe {
85            from_glib_full(ffi::gst_promise_new_with_change_func(
86                Some(trampoline::<F>),
87                Box::into_raw(user_data) as *mut _,
88                Some(free_user_data::<F>),
89            ))
90        }
91    }
92
93    pub fn new_future() -> (Self, PromiseFuture) {
94        use futures_channel::oneshot;
95
96        // We only use the channel as a convenient waker
97        let (sender, receiver) = oneshot::channel();
98        let promise = Self::with_change_func(move |_res| {
99            let _ = sender.send(());
100        });
101
102        (promise.clone(), PromiseFuture(promise, receiver))
103    }
104
105    #[doc(alias = "gst_promise_expire")]
106    pub fn expire(&self) {
107        unsafe {
108            ffi::gst_promise_expire(self.to_glib_none().0);
109        }
110    }
111
112    #[doc(alias = "gst_promise_get_reply")]
113    pub fn get_reply(&self) -> Option<&StructureRef> {
114        unsafe {
115            let s = ffi::gst_promise_get_reply(self.to_glib_none().0);
116            if s.is_null() {
117                None
118            } else {
119                Some(StructureRef::from_glib_borrow(s))
120            }
121        }
122    }
123
124    #[doc(alias = "gst_promise_interrupt")]
125    pub fn interrupt(&self) {
126        unsafe {
127            ffi::gst_promise_interrupt(self.to_glib_none().0);
128        }
129    }
130
131    #[doc(alias = "gst_promise_reply")]
132    pub fn reply(&self, s: Option<Structure>) {
133        unsafe {
134            ffi::gst_promise_reply(
135                self.to_glib_none().0,
136                s.map(|s| s.into_glib_ptr()).unwrap_or(ptr::null_mut()),
137            );
138        }
139    }
140
141    #[doc(alias = "gst_promise_wait")]
142    pub fn wait(&self) -> PromiseResult {
143        unsafe { from_glib(ffi::gst_promise_wait(self.to_glib_none().0)) }
144    }
145}
146
147impl Default for Promise {
148    fn default() -> Self {
149        Self::new()
150    }
151}
152
153unsafe impl Send for Promise {}
154unsafe impl Sync for Promise {}
155
156#[derive(Debug)]
157pub struct PromiseFuture(Promise, futures_channel::oneshot::Receiver<()>);
158
159pub struct PromiseReply(Promise);
160
161impl std::future::Future for PromiseFuture {
162    type Output = Result<Option<PromiseReply>, PromiseError>;
163
164    fn poll(mut self: Pin<&mut Self>, context: &mut Context) -> Poll<Self::Output> {
165        match Pin::new(&mut self.1).poll(context) {
166            Poll::Ready(Err(_)) => panic!("Sender dropped before callback was called"),
167            Poll::Ready(Ok(())) => {
168                let res = match self.0.wait() {
169                    PromiseResult::Replied => {
170                        if self.0.get_reply().is_none() {
171                            Ok(None)
172                        } else {
173                            Ok(Some(PromiseReply(self.0.clone())))
174                        }
175                    }
176                    PromiseResult::Interrupted => Err(PromiseError::Interrupted),
177                    PromiseResult::Expired => Err(PromiseError::Expired),
178                    PromiseResult::Pending => {
179                        panic!("Promise resolved but returned Pending");
180                    }
181                    err => Err(PromiseError::Other(err)),
182                };
183                Poll::Ready(res)
184            }
185            Poll::Pending => Poll::Pending,
186        }
187    }
188}
189
190impl futures_core::future::FusedFuture for PromiseFuture {
191    fn is_terminated(&self) -> bool {
192        self.1.is_terminated()
193    }
194}
195
196impl Deref for PromiseReply {
197    type Target = StructureRef;
198
199    #[inline]
200    fn deref(&self) -> &StructureRef {
201        self.0.get_reply().expect("Promise without reply")
202    }
203}
204
205impl std::fmt::Debug for PromiseReply {
206    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
207        let mut debug = f.debug_tuple("PromiseReply");
208
209        match self.0.get_reply() {
210            Some(reply) => debug.field(reply),
211            None => debug.field(&"<no reply>"),
212        }
213        .finish()
214    }
215}
216
217#[cfg(test)]
218mod tests {
219    use std::{sync::mpsc::channel, thread};
220
221    use super::*;
222
223    #[test]
224    fn test_change_func() {
225        crate::init().unwrap();
226
227        let (sender, receiver) = channel();
228        let promise = Promise::with_change_func(move |res| {
229            sender.send(res.map(|s| s.map(ToOwned::to_owned))).unwrap();
230        });
231
232        thread::spawn(move || {
233            promise.reply(Some(crate::Structure::new_empty("foo/bar")));
234        });
235
236        let res = receiver.recv().unwrap();
237        let res = res.expect("promise failed").expect("promise returned None");
238        assert_eq!(res.name(), "foo/bar");
239    }
240}
gstreamer/promise.rs

gstreamer/
promise.rs
