Port BusStream to futures 0.2

gstreamer/Cargo.toml

@@ -22,7 +22,7 @@ gstreamer-sys = { git = "https://github.com/sdroege/gstreamer-sys", features = [
 glib = { git = "https://github.com/gtk-rs/glib" }
 num-rational = { version = "0.1.38", default-features = false, features = [] }
 lazy_static = "1.0"
-futures = { version = "0.1", optional = true }
+futures-core = { version = "0.2", optional = true }
 muldiv = "0.1.1"
 
 [build-dependencies.rustdoc-stripper]
@@ -36,6 +36,7 @@ v1_14 = ["gstreamer-sys/v1_14", "v1_12"]
 embed-lgpl-docs = ["rustdoc-stripper"]
 purge-lgpl-docs = ["rustdoc-stripper"]
 dox = ["gstreamer-sys/dox", "glib/dox", "futures"]
+futures = ["futures-core"]
 default-features = []
 
 [badges]

gstreamer/src/bus.rs

@@ -143,32 +143,31 @@ impl Bus {
 #[cfg(any(feature = "futures", feature = "dox"))]
 mod futures {
     use super::*;
-    use futures;
-    use futures::stream::Stream;
-    use futures::task::Task;
-    use futures::{Async, Poll};
+    use futures_core::stream::Stream;
+    use futures_core::task::{Context, Waker};
+    use futures_core::{Async, Poll};
     use std::sync::{Arc, Mutex};
 
-    pub struct BusStream(Bus, Arc<Mutex<Option<Task>>>);
+    pub struct BusStream(Bus, Arc<Mutex<Option<Waker>>>);
 
     impl BusStream {
         pub fn new(bus: &Bus) -> Self {
             skip_assert_initialized!();
-            let task = Arc::new(Mutex::new(None));
-            let task_clone = Arc::clone(&task);
+            let waker = Arc::new(Mutex::new(None));
+            let waker_clone = Arc::clone(&waker);
 
             bus.set_sync_handler(move |_, _| {
-                let mut task = task_clone.lock().unwrap();
-                if let Some(task) = task.take() {
+                let mut waker = waker_clone.lock().unwrap();
+                if let Some(waker) = waker.take() {
                     // FIXME: Force type...
-                    let task: Task = task;
-                    task.notify();
+                    let waker: Waker = waker;
+                    waker.wake();
                 }
 
                 BusSyncReply::Pass
             });
 
-            BusStream(bus.clone(), task)
+            BusStream(bus.clone(), waker)
         }
     }
 
@@ -182,15 +181,16 @@ mod futures {
         type Item = Message;
         type Error = ();
 
-        fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-            let mut task = self.1.lock().unwrap();
+        fn poll_next(&mut self, ctx: &mut Context) -> Poll<Option<Self::Item>, Self::Error> {
+            let BusStream(ref bus, ref waker) = *self;
 
-            let msg = self.0.pop();
+            let msg = bus.pop();
             if let Some(msg) = msg {
                 Ok(Async::Ready(Some(msg)))
             } else {
-                *task = Some(futures::task::current());
-                Ok(Async::NotReady)
+                let mut waker = waker.lock().unwrap();
+                *waker = Some(ctx.waker().clone());
+                Ok(Async::Pending)
             }
         }
     }

gstreamer/src/lib.rs

@@ -26,7 +26,7 @@ extern crate glib;
 extern crate num_rational;
 
 #[cfg(any(feature = "futures", feature = "dox"))]
-extern crate futures;
+extern crate futures_core;
 
 extern crate muldiv;