gstreamer-rs/gstreamer/src/promise.rs

215 lines
6.4 KiB
Rust
Raw Normal View History

2018-03-15 09:43:35 +00:00
// Copyright (C) 2018 Sebastian Dröge <sebastian@centricular.com>
//
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use glib::translate::*;
2019-03-19 07:58:20 +00:00
use glib_sys;
use gst_sys;
2018-04-25 08:10:06 +00:00
use PromiseResult;
use Structure;
use StructureRef;
2018-03-15 09:43:35 +00:00
use std::marker::PhantomData;
use std::pin::Pin;
use std::ptr;
use std::task::{Context, Poll};
2018-03-15 09:43:35 +00:00
glib_wrapper! {
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
2019-03-19 07:58:20 +00:00
pub struct Promise(Shared<gst_sys::GstPromise>);
2018-03-15 09:43:35 +00:00
match fn {
2019-03-19 07:58:20 +00:00
ref => |ptr| gst_sys::gst_mini_object_ref(ptr as *mut _),
unref => |ptr| gst_sys::gst_mini_object_unref(ptr as *mut _),
get_type => || gst_sys::gst_promise_get_type(),
2018-03-15 09:43:35 +00:00
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum PromiseError {
Interrupted,
Expired,
Other(PromiseResult),
}
2018-03-15 09:43:35 +00:00
impl Promise {
pub fn new() -> Promise {
assert_initialized_main_thread!();
2019-03-19 07:58:20 +00:00
unsafe { from_glib_full(gst_sys::gst_promise_new()) }
2018-03-15 09:43:35 +00:00
}
pub fn new_with_change_func<F>(func: F) -> Promise
2018-04-01 08:30:03 +00:00
where
F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
2018-04-01 08:30:03 +00:00
{
2020-03-22 14:18:47 +00:00
assert_initialized_main_thread!();
let user_data: Box<Option<F>> = Box::new(Some(func));
2018-03-15 09:43:35 +00:00
unsafe extern "C" fn trampoline<
F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
>(
2019-03-19 07:58:20 +00:00
promise: *mut gst_sys::GstPromise,
user_data: glib_sys::gpointer,
2018-03-15 09:43:35 +00:00
) {
let user_data: &mut Option<F> = &mut *(user_data as *mut _);
2018-03-15 09:43:35 +00:00
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);
2018-03-15 09:43:35 +00:00
}
unsafe extern "C" fn free_user_data<
F: FnOnce(Result<Option<&StructureRef>, PromiseError>) + Send + 'static,
>(
2019-03-19 07:58:20 +00:00
user_data: glib_sys::gpointer,
2018-03-15 09:43:35 +00:00
) {
let _: Box<Option<F>> = Box::from_raw(user_data as *mut _);
2018-03-15 09:43:35 +00:00
}
unsafe {
2019-03-19 07:58:20 +00:00
from_glib_full(gst_sys::gst_promise_new_with_change_func(
Some(trampoline::<F>),
2018-03-15 09:43:35 +00:00
Box::into_raw(user_data) as *mut _,
Some(free_user_data::<F>),
2018-03-15 09:43:35 +00:00
))
}
}
pub fn new_future<'a>() -> (Self, PromiseFuture<'a>) {
use futures_channel::oneshot;
// We only use the channel as a convenient waker
let (sender, receiver) = oneshot::channel();
let promise = Self::new_with_change_func(move |_res| {
let _ = sender.send(());
});
(
promise.clone(),
PromiseFuture(promise, receiver, PhantomData),
)
}
2018-03-15 09:43:35 +00:00
pub fn expire(&self) {
unsafe {
2019-03-19 07:58:20 +00:00
gst_sys::gst_promise_expire(self.to_glib_none().0);
2018-03-15 09:43:35 +00:00
}
}
pub fn get_reply(&self) -> Option<&StructureRef> {
unsafe {
2019-03-19 07:58:20 +00:00
let s = gst_sys::gst_promise_get_reply(self.to_glib_none().0);
2018-03-15 09:43:35 +00:00
if s.is_null() {
None
} else {
Some(StructureRef::from_glib_borrow(s))
}
}
}
pub fn interrupt(&self) {
unsafe {
2019-03-19 07:58:20 +00:00
gst_sys::gst_promise_interrupt(self.to_glib_none().0);
2018-03-15 09:43:35 +00:00
}
}
pub fn reply(&self, s: Option<Structure>) {
2018-03-15 09:43:35 +00:00
unsafe {
gst_sys::gst_promise_reply(
self.to_glib_none().0,
s.map(|s| s.into_ptr()).unwrap_or(ptr::null_mut()),
);
2018-03-15 09:43:35 +00:00
}
}
pub fn wait(&self) -> PromiseResult {
2019-03-19 07:58:20 +00:00
unsafe { from_glib(gst_sys::gst_promise_wait(self.to_glib_none().0)) }
2018-03-15 09:43:35 +00:00
}
}
impl Default for Promise {
fn default() -> Self {
Self::new()
}
}
unsafe impl Send for Promise {}
unsafe impl Sync for Promise {}
#[derive(Debug)]
pub struct PromiseFuture<'a>(
Promise,
futures_channel::oneshot::Receiver<()>,
PhantomData<&'a StructureRef>,
);
impl<'a> std::future::Future for PromiseFuture<'a> {
type Output = Result<Option<&'a StructureRef>, 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 => unsafe {
let s = gst_sys::gst_promise_get_reply(self.0.to_glib_none().0);
if s.is_null() {
Ok(None)
} else {
Ok(Some(StructureRef::from_glib_borrow(s)))
}
},
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,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::mpsc::channel;
use std::thread;
#[test]
fn test_change_func() {
::init().unwrap();
let (sender, receiver) = channel();
let promise = Promise::new_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("foo/bar", &[])));
});
let res = receiver.recv().unwrap();
let res = res.expect("promise failed").expect("promise returned None");
assert_eq!(res.get_name(), "foo/bar");
}
}