Update futures code to futures 0.3

Also clean it up a bit.

examples/Cargo.toml

@@ -19,7 +19,7 @@ gstreamer-rtsp-server-sys = { git = "https://gitlab.freedesktop.org/gstreamer/gs
 gtk = { git = "https://github.com/gtk-rs/gtk", optional = true }
 gdk = { git = "https://github.com/gtk-rs/gdk", optional = true }
 gio = { git = "https://github.com/gtk-rs/gio", optional = true }
-futures-preview = { version = "0.2", optional = true }
+futures-preview = { version = "0.3.0-alpha", optional = true }
 byte-slice-cast = "0.2"
 failure = "0.1"
 failure_derive = "0.1"

examples/src/bin/futures.rs

@@ -7,8 +7,10 @@ extern crate gstreamer as gst;
 use gst::prelude::*;
 
 extern crate futures;
-use futures::executor::block_on;
+use futures::executor::LocalPool;
+use futures::future;
 use futures::prelude::*;
+use futures::task::SpawnExt;
 
 use std::env;
 
@@ -29,42 +31,45 @@ fn example_main() {
         .set_state(gst::State::Playing)
         .expect("Unable to set the pipeline to the `Playing` state");
 
-    // BusStream implements the Stream trait, but Stream::for_each is
+    // Use a LocalPool as executor. This runs single threaded on this very thread.
-    // calling a closure for each item and returns a Future that resolves
+    let mut pool = LocalPool::new();
-    // when the stream is done or an error has happened
-    let messages = gst::BusStream::new(&bus)
-        .for_each(|msg| {
-            use gst::MessageView;
 
-            // Determine whether we want to resolve the future, or we still have
+    // We use an AbortHandle for having a Future that runs forever
-            // to wait. The future is resolved when either an error occurs, or the
+    // until we call handle.abort() to quit our event loop
-            // pipeline succeeded execution (got an EOS event).
+    let (quit_handle, quit_registration) = future::AbortHandle::new_pair();
-            let quit = match msg.view() {
+    let quit_future = future::Abortable::new(future::empty::<()>(), quit_registration).map(|_| ());
-                MessageView::Eos(..) => true,
-                MessageView::Error(err) => {
-                    println!(
-                        "Error from {:?}: {} ({:?})",
-                        err.get_src().map(|s| s.get_path_string()),
-                        err.get_error(),
-                        err.get_debug()
-                    );
-                    true
-                }
-                _ => false,
-            };
 
-            if quit {
+    // BusStream implements the Stream trait. Stream::for_each is calling a closure for each item
-                Err(()) // This resolves the future that is returned by for_each
+    // and returns a Future that resolves when the stream is done
-                        // FIXME: At the moment, EOS messages also result in the future to be resolved
+    let messages = gst::BusStream::new(&bus).for_each(move |msg| {
-                        // by an error. This should probably be changed in the future.
+        use gst::MessageView;
-            } else {
+
-                Ok(()) // Continue - do not resolve the future yet.
+        // Determine whether we want to quit: on EOS or error message
+        // we quit, otherwise simply continue.
+        match msg.view() {
+            MessageView::Eos(..) => quit_handle.abort(),
+            MessageView::Error(err) => {
+                println!(
+                    "Error from {:?}: {} ({:?})",
+                    err.get_src().map(|s| s.get_path_string()),
+                    err.get_error(),
+                    err.get_debug()
+                );
+                quit_handle.abort();
             }
-        })
+            _ => (),
-        .and_then(|_| Ok(()));
+        };
 
-    // Synchronously wait on the future we created above.
+        // New future to resolve for each message: nothing here
-    let _ = block_on(messages);
+        future::ready(())
+    });
+
+    // Spawn our message handling stream
+    pool.spawner().spawn(messages).unwrap();
+
+    // And run until something is quitting the loop, i.e. an EOS
+    // or error message is received above
+    pool.run_until(quit_future);
 
     pipeline
         .set_state(gst::State::Null)

examples/src/bin/glib-futures.rs

@@ -4,6 +4,7 @@ use gst::prelude::*;
 extern crate glib;
 
 extern crate futures;
+use futures::future;
 use futures::prelude::*;
 
 use std::env;
@@ -12,12 +13,18 @@ use std::env;
 mod examples_common;
 
 fn example_main() {
+    // Get the default main context and make it also the thread default, then create
+    // a main loop for it
     let ctx = glib::MainContext::default();
+    ctx.push_thread_default();
+    let loop_ = glib::MainLoop::new(Some(&ctx), false);
 
+    // Read the pipeline to launch from the commandline, using the launch syntax.
     let pipeline_str = env::args().collect::<Vec<String>>()[1..].join(" ");
 
     gst::init().unwrap();
 
+    // Create a pipeline from the launch-syntax given on the cli.
     let pipeline = gst::parse_launch(&pipeline_str).unwrap();
     let bus = pipeline.get_bus().unwrap();
 
@@ -25,37 +32,44 @@ fn example_main() {
         .set_state(gst::State::Playing)
         .expect("Unable to set the pipeline to the `Playing` state");
 
-    let messages = gst::BusStream::new(&bus)
+    // BusStream implements the Stream trait. Stream::for_each is calling a closure for each item
-        .for_each(|msg| {
+    // and returns a Future that resolves when the stream is done
-            use gst::MessageView;
+    let loop_clone = loop_.clone();
+    let messages = gst::BusStream::new(&bus).for_each(move |msg| {
+        use gst::MessageView;
 
-            let quit = match msg.view() {
+        // Determine whether we want to quit: on EOS or error message
-                MessageView::Eos(..) => true,
+        // we quit, otherwise simply continue.
-                MessageView::Error(err) => {
+        match msg.view() {
-                    println!(
+            MessageView::Eos(..) => loop_clone.quit(),
-                        "Error from {:?}: {} ({:?})",
+            MessageView::Error(err) => {
-                        err.get_src().map(|s| s.get_path_string()),
+                println!(
-                        err.get_error(),
+                    "Error from {:?}: {} ({:?})",
-                        err.get_debug()
+                    err.get_src().map(|s| s.get_path_string()),
-                    );
+                    err.get_error(),
-                    true
+                    err.get_debug()
-                }
+                );
-                _ => false,
+                loop_clone.quit();
-            };
-
-            if quit {
-                Err(())
-            } else {
-                Ok(())
             }
-        })
+            _ => (),
-        .and_then(|_| Ok(()));
+        }
 
-    let _ = ctx.block_on(messages);
+        // New future to resolve for each message: nothing here
+        future::ready(())
+    });
+
+    // Spawn our message handling stream
+    ctx.spawn_local(messages);
+
+    // And run until something is quitting the loop, i.e. an EOS
+    // or error message is received above
+    loop_.run();
 
     pipeline
         .set_state(gst::State::Null)
         .expect("Unable to set the pipeline to the `Null` state");
+
+    ctx.pop_thread_default();
 }
 
 fn main() {

gstreamer/Cargo.toml

@@ -22,7 +22,7 @@ gstreamer-sys = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs-s
 glib = { git = "https://github.com/gtk-rs/glib" }
 num-rational = { version = "0.2", default-features = false, features = [] }
 lazy_static = "1.0"
-futures-core-preview = { version = "0.2", optional = true }
+futures-core-preview = { version = "0.3.0-alpha", optional = true }
 muldiv = "0.2"
 serde = { version = "1.0", optional = true }
 serde_bytes = { version = "0.11", optional = true }

gstreamer/src/bus.rs

@@ -201,7 +201,8 @@ mod futures {
     use super::*;
     use futures_core::stream::Stream;
     use futures_core::task::{Context, Waker};
-    use futures_core::{Async, Poll};
+    use futures_core::Poll;
+    use std::pin::Pin;
     use std::sync::{Arc, Mutex};
 
     #[derive(Debug)]
@@ -236,18 +237,17 @@ mod futures {
 
     impl Stream for BusStream {
         type Item = Message;
-        type Error = ();
 
-        fn poll_next(&mut self, ctx: &mut Context) -> Poll<Option<Self::Item>, Self::Error> {
+        fn poll_next(self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Option<Self::Item>> {
             let BusStream(ref bus, ref waker) = *self;
 
             let msg = bus.pop();
             if let Some(msg) = msg {
-                Ok(Async::Ready(Some(msg)))
+                Poll::Ready(Some(msg))
             } else {
                 let mut waker = waker.lock().unwrap();
                 *waker = Some(ctx.waker().clone());
-                Ok(Async::Pending)
+                Poll::Pending
             }
         }
     }