Merge pull request #1162 from Nemo157/extended-compat

Extend compatibility to Stream and Sink

Author: MajorBreakfast

.rustfmt.toml

@@ -0,0 +1,2 @@
+max_width = 80
+comment_width = 80

.travis.yml

@@ -35,6 +35,17 @@ matrix:
         - cargo build --manifest-path futures-sink/Cargo.toml --no-default-features
         - cargo build --manifest-path futures-util/Cargo.toml --no-default-features
 
+    - name: cargo build --all-features
+      rust: nightly
+      script:
+        - cargo build --manifest-path futures/Cargo.toml --all-features
+        - cargo build --manifest-path futures-core/Cargo.toml --all-features
+        - cargo build --manifest-path futures-channel/Cargo.toml --all-features
+        - cargo build --manifest-path futures-executor/Cargo.toml --all-features
+        - cargo build --manifest-path futures-io/Cargo.toml --all-features
+        - cargo build --manifest-path futures-sink/Cargo.toml --all-features
+        - cargo build --manifest-path futures-util/Cargo.toml --all-features
+
     - name: cargo build --target=thumbv6m-none-eabi
       rust: nightly
       install:

futures-util/src/compat/compat.rs

@@ -1,20 +1,20 @@
-/// Converts a futures 0.3 `TryFuture` into a futures 0.1 `Future`
-/// and vice versa.
+/// Converts a futures 0.3 `TryFuture`, `TryStream` or `Sink` into a futures 0.1
+/// `Future` and vice versa.
 #[derive(Debug)]
 #[must_use = "futures do nothing unless polled"]
-pub struct Compat<Fut, Ex> {
-    crate future: Fut,
+pub struct Compat<T, Ex> {
+    crate inner: T,
     crate executor: Option<Ex>,
 }
 
-impl<Fut, Ex> Compat<Fut, Ex> {
-    /// Returns the inner future.
-    pub fn into_inner(self) -> Fut {
-        self.future
+impl<T, Ex> Compat<T, Ex> {
+    /// Returns the inner item.
+    pub fn into_inner(self) -> T {
+        self.inner
     }
 
     /// Creates a new `Compat`.
-    crate fn new(future: Fut, executor: Option<Ex>) -> Compat<Fut, Ex> {
-        Compat { future, executor }
+    crate fn new(inner: T, executor: Option<Ex>) -> Compat<T, Ex> {
+        Compat { inner, executor }
     }
 }

futures-util/src/compat/compat01to03.rs

@@ -1,28 +1,48 @@
 use super::Compat;
-use futures::Async as Async01;
-use futures::Future as Future01;
-use futures::executor::{self as executor01, NotifyHandle as NotifyHandle01,
-                        Notify as Notify01, UnsafeNotify as UnsafeNotify01};
-use futures_core::Future as Future03;
-use futures_core::task as task03;
+use futures::{
+    executor::{
+        self as executor01, Notify as Notify01, NotifyHandle as NotifyHandle01,
+        UnsafeNotify as UnsafeNotify01,
+    },
+    Async as Async01, Future as Future01, Stream as Stream01,
+};
+use futures_core::{task as task03, Future as Future03, Stream as Stream03};
 use std::mem::PinMut;
 
 impl<Fut: Future01> Future03 for Compat<Fut, ()> {
     type Output = Result<Fut::Item, Fut::Error>;
 
     fn poll(
         self: PinMut<Self>,
-        cx: &mut task03::Context
+        cx: &mut task03::Context,
     ) -> task03::Poll<Self::Output> {
         let notify = &WakerToHandle(cx.waker());
 
         executor01::with_notify(notify, 0, move || {
-            unsafe {
-                match PinMut::get_mut_unchecked(self).future.poll() {
-                    Ok(Async01::Ready(t)) => task03::Poll::Ready(Ok(t)),
-                    Ok(Async01::NotReady) => task03::Poll::Pending,
-                    Err(e) => task03::Poll::Ready(Err(e)),
-                }
+            match unsafe { PinMut::get_mut_unchecked(self) }.inner.poll() {
+                Ok(Async01::Ready(t)) => task03::Poll::Ready(Ok(t)),
+                Ok(Async01::NotReady) => task03::Poll::Pending,
+                Err(e) => task03::Poll::Ready(Err(e)),
+            }
+        })
+    }
+}
+
+impl<St: Stream01> Stream03 for Compat<St, ()> {
+    type Item = Result<St::Item, St::Error>;
+
+    fn poll_next(
+        self: PinMut<Self>,
+        cx: &mut task03::Context,
+    ) -> task03::Poll<Option<Self::Item>> {
+        let notify = &WakerToHandle(cx.waker());
+
+        executor01::with_notify(notify, 0, move || {
+            match unsafe { PinMut::get_mut_unchecked(self) }.inner.poll() {
+                Ok(Async01::Ready(Some(t))) => task03::Poll::Ready(Some(Ok(t))),
+                Ok(Async01::Ready(None)) => task03::Poll::Ready(None),
+                Ok(Async01::NotReady) => task03::Poll::Pending,
+                Err(e) => task03::Poll::Ready(Some(Err(e))),
             }
         })
     }
@@ -37,9 +57,7 @@ impl<'a> From<WakerToHandle<'a>> for NotifyHandle01 {
     fn from(handle: WakerToHandle<'a>) -> NotifyHandle01 {
         let ptr = Box::new(NotifyWaker(handle.0.clone()));
 
-        unsafe {
-            NotifyHandle01::new(Box::into_raw(ptr))
-        }
+        unsafe { NotifyHandle01::new(Box::into_raw(ptr)) }
     }
 }
 

futures-util/src/compat/compat03to01.rs

@@ -1,29 +1,87 @@
 use super::Compat;
-use futures::Future as Future01;
-use futures::Poll as Poll01;
-use futures::task as task01;
-use futures::Async as Async01;
-use futures_core::TryFuture as TryFuture03;
-use futures_core::task as task03;
-use std::marker::Unpin;
-use std::mem::PinMut;
-use std::sync::Arc;
+use futures::{
+    task as task01, Async as Async01, AsyncSink as AsyncSink01,
+    Future as Future01, Poll as Poll01, Sink as Sink01,
+    StartSend as StartSend01, Stream as Stream01,
+};
+use futures_core::{
+    task as task03, TryFuture as TryFuture03, TryStream as TryStream03,
+};
+use futures_sink::Sink as Sink03;
+use std::{marker::Unpin, mem::PinMut, sync::Arc};
 
 impl<Fut, Ex> Future01 for Compat<Fut, Ex>
-where Fut: TryFuture03 + Unpin,
-      Ex: task03::Executor
+where
+    Fut: TryFuture03 + Unpin,
+    Ex: task03::Executor,
 {
     type Item = Fut::Ok;
     type Error = Fut::Error;
 
     fn poll(&mut self) -> Poll01<Self::Item, Self::Error> {
-        let waker = current_as_waker();
-        let mut cx = task03::Context::new(&waker, self.executor.as_mut().unwrap());
-        match PinMut::new(&mut self.future).try_poll(&mut cx) {
+        with_context(self, |inner, cx| match inner.try_poll(cx) {
             task03::Poll::Ready(Ok(t)) => Ok(Async01::Ready(t)),
             task03::Poll::Pending => Ok(Async01::NotReady),
             task03::Poll::Ready(Err(e)) => Err(e),
-        }
+        })
+    }
+}
+
+impl<St, Ex> Stream01 for Compat<St, Ex>
+where
+    St: TryStream03 + Unpin,
+    Ex: task03::Executor,
+{
+    type Item = St::Ok;
+    type Error = St::Error;
+
+    fn poll(&mut self) -> Poll01<Option<Self::Item>, Self::Error> {
+        with_context(self, |inner, cx| match inner.try_poll_next(cx) {
+            task03::Poll::Ready(None) => Ok(Async01::Ready(None)),
+            task03::Poll::Ready(Some(Ok(t))) => Ok(Async01::Ready(Some(t))),
+            task03::Poll::Pending => Ok(Async01::NotReady),
+            task03::Poll::Ready(Some(Err(e))) => Err(e),
+        })
+    }
+}
+
+impl<T, E> Sink01 for Compat<T, E>
+where
+    T: Sink03 + Unpin,
+    E: task03::Executor,
+{
+    type SinkItem = T::SinkItem;
+    type SinkError = T::SinkError;
+
+    fn start_send(
+        &mut self,
+        item: Self::SinkItem,
+    ) -> StartSend01<Self::SinkItem, Self::SinkError> {
+        with_context(self, |mut inner, cx| {
+            match inner.reborrow().poll_ready(cx) {
+                task03::Poll::Ready(Ok(())) => {
+                    inner.start_send(item).map(|()| AsyncSink01::Ready)
+                }
+                task03::Poll::Pending => Ok(AsyncSink01::NotReady(item)),
+                task03::Poll::Ready(Err(e)) => Err(e),
+            }
+        })
+    }
+
+    fn poll_complete(&mut self) -> Poll01<(), Self::SinkError> {
+        with_context(self, |inner, cx| match inner.poll_flush(cx) {
+            task03::Poll::Ready(Ok(())) => Ok(Async01::Ready(())),
+            task03::Poll::Pending => Ok(Async01::NotReady),
+            task03::Poll::Ready(Err(e)) => Err(e),
+        })
+    }
+
+    fn close(&mut self) -> Poll01<(), Self::SinkError> {
+        with_context(self, |inner, cx| match inner.poll_close(cx) {
+            task03::Poll::Ready(Ok(())) => Ok(Async01::Ready(())),
+            task03::Poll::Pending => Ok(Async01::NotReady),
+            task03::Poll::Ready(Err(e)) => Err(e),
+        })
     }
 }
 
@@ -39,3 +97,15 @@ impl task03::Wake for Current {
         arc_self.0.notify();
     }
 }
+
+fn with_context<T, E, R, F>(compat: &mut Compat<T, E>, f: F) -> R
+where
+    T: Unpin,
+    E: task03::Executor,
+    F: FnOnce(PinMut<T>, &mut task03::Context) -> R,
+{
+    let waker = current_as_waker();
+    let executor = compat.executor.as_mut().unwrap();
+    let mut cx = task03::Context::new(&waker, executor);
+    f(PinMut::new(&mut compat.inner), &mut cx)
+}

futures-util/src/compat/future01ext.rs

@@ -9,7 +9,7 @@ pub trait Future01CompatExt: Future01 {
     /// futures 0.3 `Future<Output = Result<T, E>>`.
     fn compat(self) -> Compat<Self, ()> where Self: Sized {
         Compat {
-            future: self,
+            inner: self,
             executor: None,
         }
     }

futures-util/src/compat/mod.rs

@@ -13,3 +13,6 @@ mod compat03to01;
 
 mod future01ext;
 pub use self::future01ext::Future01CompatExt;
+
+mod stream01ext;
+pub use self::stream01ext::Stream01CompatExt;

futures-util/src/compat/stream01ext.rs

@@ -0,0 +1,19 @@
+use super::Compat;
+use futures::Stream as Stream01;
+
+impl<St: Stream01> Stream01CompatExt for St {}
+
+/// Extension trait for futures 0.1 [`Stream`][Stream01]
+pub trait Stream01CompatExt: Stream01 {
+    /// Converts a futures 0.1 [`Stream<Item = T, Error = E>`][Stream01] into a
+    /// futures 0.3 [`Stream<Item = Result<T, E>>`][Stream03].
+    fn compat(self) -> Compat<Self, ()>
+    where
+        Self: Sized,
+    {
+        Compat {
+            inner: self,
+            executor: None,
+        }
+    }
+}

futures-util/src/sink/mod.rs

@@ -9,6 +9,12 @@ use futures_core::future::Future;
 use futures_core::stream::Stream;
 use futures_sink::Sink;
 
+#[cfg(feature = "compat")]
+use crate::compat::Compat;
+
+#[cfg(feature = "compat")]
+use futures_core::task::Executor;
+
 mod close;
 pub use self::close::Close;
 
@@ -248,4 +254,14 @@ pub trait SinkExt: Sink {
     {
         Either::Right(self)
     }
+
+    /// Wraps a [`Sink`] into a sink compatible with libraries using
+    /// futures 0.1 `Sink`. Requires the `compat` feature to be enabled.
+    #[cfg(feature = "compat")]
+    fn compat<E>(self, executor: E) -> Compat<Self, E>
+        where Self: Sized + Unpin,
+              E: Executor,
+    {
+        Compat::new(self, Some(executor))
+    }
 }

futures-util/src/try_stream/mod.rs

@@ -7,6 +7,12 @@ use core::marker::Unpin;
 use futures_core::future::TryFuture;
 use futures_core::stream::TryStream;
 
+#[cfg(feature = "compat")]
+use crate::compat::Compat;
+
+#[cfg(feature = "compat")]
+use futures_core::task::Executor;
+
 mod err_into;
 pub use self::err_into::ErrInto;
 
@@ -363,4 +369,14 @@ pub trait TryStreamExt: TryStream {
     {
         TryBufferUnordered::new(self, n)
     }
+
+    /// Wraps a [`TryStream`] into a stream compatible with libraries using
+    /// futures 0.1 `Stream`. Requires the `compat` feature to be enabled.
+    #[cfg(feature = "compat")]
+    fn compat<E>(self, executor: E) -> Compat<Self, E>
+        where Self: Sized + Unpin,
+              E: Executor,
+    {
+        Compat::new(self, Some(executor))
+    }
 }

futures/src/lib.rs

@@ -90,6 +90,7 @@ pub mod compat {
         Executor01As03,
         Executor01CompatExt,
         Future01CompatExt,
+        Stream01CompatExt,
     };
 }