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