gstreamer/element: Add call_async_future() that returns a future

The future would resolve into the return value of the closure that is
called asynchronously on the thread pool, and allows asynchronous
awaiting for it to finish.

  let res = element.call_async(|element| {
      element.set_state(gst::State::Playing)
  }).await;

  assert_eq!(res, Ok(gst::StateChangeSuccess::Success))

gstreamer/Cargo.toml

@@ -23,6 +23,8 @@ glib = { git = "https://github.com/gtk-rs/glib" }
 num-rational = { version = "0.2", default-features = false, features = [] }
 lazy_static = "1.0"
 futures-core = "0.3"
+futures-util = "0.3"
+futures-channel = "0.3"
 muldiv = "0.2"
 serde = { version = "1.0", optional = true }
 serde_bytes = { version = "0.11", optional = true }

gstreamer/src/element.rs

@@ -38,8 +38,14 @@ use StateChangeReturn;
 use StateChangeSuccess;
 
 use std::ffi::CStr;
+#[cfg(any(feature = "v1_10", feature = "dox"))]
+use std::future::Future;
+#[cfg(any(feature = "v1_10", feature = "dox"))]
+use std::marker::Unpin;
 use std::mem;
 use std::num::NonZeroU64;
+#[cfg(any(feature = "v1_10", feature = "dox"))]
+use std::pin::Pin;
 
 use libc;
 
@@ -248,6 +254,15 @@ pub trait ElementExtManual: 'static {
     fn call_async<F>(&self, func: F)
     where
         F: FnOnce(&Self) + Send + 'static;
+
+    #[cfg(any(feature = "v1_10", feature = "dox"))]
+    fn call_async_future<F, T>(
+        &self,
+        func: F,
+    ) -> Pin<Box<dyn Future<Output = T> + Unpin + Send + 'static>>
+    where
+        F: FnOnce(&Self) -> T + Send + 'static,
+        T: Send + 'static;
 }
 
 impl<O: IsA<Element>> ElementExtManual for O {
@@ -769,6 +784,27 @@ impl<O: IsA<Element>> ElementExtManual for O {
             );
         }
     }
+
+    #[cfg(any(feature = "v1_10", feature = "dox"))]
+    fn call_async_future<F, T>(
+        &self,
+        func: F,
+    ) -> Pin<Box<dyn Future<Output = T> + Unpin + Send + 'static>>
+    where
+        F: FnOnce(&Self) -> T + Send + 'static,
+        T: Send + 'static,
+    {
+        use futures_channel::oneshot;
+        use futures_util::future::FutureExt;
+
+        let (sender, receiver) = oneshot::channel();
+
+        self.call_async(move |element| {
+            let _ = sender.send(func(element));
+        });
+
+        Box::pin(receiver.map(|res| res.expect("sender dropped")))
+    }
 }
 
 impl ElementClass {

gstreamer/src/lib.rs

@@ -33,7 +33,9 @@ pub extern crate glib;
 
 extern crate num_rational;
 
+extern crate futures_channel;
 extern crate futures_core;
+extern crate futures_util;
 
 extern crate muldiv;