Compat implementation

Author: tinaun

futures-util/Cargo.toml

@@ -19,6 +19,7 @@ name = "futures_util"
 [features]
 std = ["futures-core-preview/std", "futures-io-preview/std", "futures-sink-preview/std", "either/use_std", "slab"]
 default = ["std", "futures-core-preview/either", "futures-sink-preview/either"]
+compat = ["std", "futures"]
 bench = []
 nightly = []
 
@@ -29,6 +30,7 @@ futures-io-preview = { path = "../futures-io", version = "0.3.0-alpha.2", defaul
 futures-sink-preview = { path = "../futures-sink", version = "0.3.0-alpha.2", default-features = false}
 either = { version = "1.4", default-features = false }
 slab = { version = "0.4", optional = true }
+futures = { version = "0.1", optional = true }
 
 [dev-dependencies]
 futures-preview = { path = "../futures", version = "0.3.0-alpha.2" }

futures-util/src/compat/executor.rs

@@ -0,0 +1,47 @@
+
+pub trait ExecCompat: Executor01<
+        Compat<FutureObj<'static, ()>, BoxedExecutor>
+    > + Clone + Send + 'static
+{
+    fn compat(self) -> ExecutorCompat<Self> 
+        where Self: Sized;
+}
+
+impl<E> ExecCompat for E
+where E: Executor01<
+        Compat<FutureObj<'static, ()>, BoxedExecutor>
+      >,
+      E: Clone + Send + 'static
+{
+    fn compat(self) -> ExecutorCompat<Self> {
+        ExecutorCompat {
+            exec: self,
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct ExecutorCompat<E> {
+    exec: E
+}
+
+impl<E> Executor03 for ExecutorCompat<E> 
+    where E: Executor01<
+        Compat<FutureObj<'static, ()>, Box<Executor03>>
+    >,
+    E: Clone + Send + 'static,
+{
+    fn spawn_obj(&mut self, obj: FutureObj<'static, ()>) -> Result<(), task::SpawnObjError> {
+        
+        self.exec.execute(obj.compat(Box::new(self.clone())))
+                 .map_err(|exec_err| {
+                     use futures_core::task::{SpawnObjError, SpawnErrorKind};
+                     
+                     let fut = exec_err.into_future().compat().map(|_| ());
+                     SpawnObjError {
+                         kind: SpawnErrorKind::shutdown(),
+                         task: Box::new(fut).into(),   
+                     }
+                 })
+    }
+}

futures-util/src/compat/mod.rs

@@ -0,0 +1,132 @@
+//! Futures 0.1 / 0.3 shims
+//! 
+
+#![allow(missing_debug_implementations)]
+
+use futures::Future as Future01;
+use futures::Poll as Poll01;
+use futures::task as task01;
+use futures::task::Task as Task01;
+use futures::executor::{with_notify, NotifyHandle, Notify, UnsafeNotify};
+
+use futures_core::Future as Future03;
+use futures_core::TryFuture as TryFuture03;
+use futures_core::Poll as Poll03;
+use futures_core::task;
+use futures_core::task::Executor as Executor03;
+use futures_core::task::{Wake, Waker, LocalWaker, local_waker_from_nonlocal};
+
+use core::mem::PinMut;
+use core::marker::Unpin;
+use std::sync::Arc;
+
+/// Converts a futures 0.3 `TryFuture` into a futures 0.1 `Future`
+/// and vice versa.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless polled"]
+pub struct Compat<Fut, E> {
+    crate inner: Fut, 
+    crate executor: Option<E>,
+}
+
+impl<Fut, E> Compat<Fut, E> {
+    /// Returns the inner future.
+    pub fn into_inner(self) -> Fut {
+        self.inner
+    }
+
+    /// Creates a new `Compat`.
+    crate fn new(inner: Fut, executor: Option<E>) -> Compat<Fut, E> {
+        Compat {
+            inner,
+            executor
+        }
+    }
+}
+
+impl<T> Future03 for Compat<T, ()> where T: Future01 {
+    type Output = Result<T::Item, T::Error>;
+
+    fn poll(self: PinMut<Self>, cx: &mut task::Context) -> Poll03<Self::Output> {
+        use futures::Async;
+
+        let notify = &WakerToHandle(cx.waker());
+
+        with_notify(notify, 0, move || { 
+            unsafe {
+                match PinMut::get_mut_unchecked(self).inner.poll() {
+                    Ok(Async::Ready(t)) => Poll03::Ready(Ok(t)),
+                    Ok(Async::NotReady) => Poll03::Pending,
+                    Err(e) => Poll03::Ready(Err(e)),
+                }
+            }
+        })
+    }
+}
+
+struct NotifyWaker(Waker);
+
+#[derive(Clone)]
+struct WakerToHandle<'a>(&'a Waker);
+
+impl<'a> From<WakerToHandle<'a>> for NotifyHandle {
+    fn from(handle: WakerToHandle<'a>) -> NotifyHandle {
+        let ptr = Box::new(NotifyWaker(handle.0.clone()));
+
+        unsafe {
+            NotifyHandle::new(Box::into_raw(ptr))
+        }
+    }
+}
+
+impl Notify for NotifyWaker {
+    fn notify(&self, _: usize) {
+        self.0.wake();
+    }
+}
+
+unsafe impl UnsafeNotify for NotifyWaker {
+    unsafe fn clone_raw(&self) -> NotifyHandle {
+        WakerToHandle(&self.0).into()
+    }
+
+    unsafe fn drop_raw(&self) {
+        let ptr: *const dyn UnsafeNotify = self;
+        drop(Box::from_raw(ptr as *mut dyn UnsafeNotify));
+    }
+}
+
+
+
+
+impl<T, E> Future01 for Compat<T, E> where T: TryFuture03 + Unpin,
+    E: Executor03
+{
+    type Item = T::Ok;
+    type Error = T::Error;
+
+    fn poll(&mut self) -> Poll01<Self::Item, Self::Error> {
+        use futures::Async;
+
+        let waker = current_as_waker();
+        let mut cx = task::Context::new(&waker, self.executor.as_mut().unwrap());
+        match PinMut::new(&mut self.inner).try_poll(&mut cx) {
+            Poll03::Ready(Ok(t)) => Ok(Async::Ready(t)),
+            Poll03::Pending => Ok(Async::NotReady),
+            Poll03::Ready(Err(e)) => Err(e),
+        }
+    }
+}
+
+fn current_as_waker() -> LocalWaker {
+    let arc_waker = Arc::new(Current(task01::current()));
+    local_waker_from_nonlocal(arc_waker)
+}
+
+struct Current(Task01);
+
+impl Wake for Current {
+    fn wake(arc_self: &Arc<Self>) {
+        arc_self.0.notify();
+    }
+}

futures-util/src/future/mod.rs

@@ -57,6 +57,9 @@ pub use self::then::Then;
 mod inspect;
 pub use self::inspect::Inspect;
 
+mod never_error;
+pub use self::never_error::NeverError;
+
 mod with_executor;
 pub use self::with_executor::WithExecutor;
 
@@ -83,6 +86,8 @@ if_std! {
 
     mod shared;
     pub use self::shared::Shared;
+
+    use std::boxed::PinBox;
 }
 
 impl<T: ?Sized> FutureExt for T where T: Future {}
@@ -632,6 +637,21 @@ pub trait FutureExt: Future {
         Shared::new(self)
     }
 
+    /// Wrap the future in a Box, pinning it.
+    #[cfg(feature = "std")]
+    fn boxed(self) -> PinBox<Self>
+        where Self: Sized
+    {
+        PinBox::new(self)
+    }
+
+    /// Turns a `Future` into a `TryFuture` that never errors
+    fn never_error(self) -> NeverError<Self>
+        where Self: Sized
+    {
+        NeverError::new(self)
+    }
+
     /// Assigns the provided `Executor` to be used when spawning tasks
     /// from within the future.
     ///

futures-util/src/future/never_error.rs

@@ -0,0 +1,39 @@
+use core::marker::Unpin;
+use core::mem::PinMut;
+use futures_core::future::Future;
+use futures_core::task::{self, Poll};
+
+/// Future for the `never_error` combinator, turning a `Future` into a `TryFuture`.
+///
+/// This is created by the `FutureExt::never_error` method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless polled"]
+pub struct NeverError<Fut> {
+    future: Fut,
+}
+
+impl<Fut> NeverError<Fut> {
+    unsafe_pinned!(future: Fut);
+
+    /// Creates a new NeverError.
+    pub(super) fn new(future: Fut) -> NeverError<Fut> {
+        NeverError { future }
+    }
+}
+
+impl<Fut: Unpin> Unpin for NeverError<Fut> {}
+
+impl<Fut, T> Future for NeverError<Fut>
+    where Fut: Future<Output = T>,
+{
+    type Output = Result<T, ()>;
+
+    fn poll(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<Result<T, ()>> {
+        match self.future().poll(cx) {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(output) => {
+                Poll::Ready(Ok(output))
+            }
+        }
+    }
+}

futures-util/src/lib.rs

@@ -75,6 +75,9 @@ pub mod sink;
 
 pub mod task;
 
+#[cfg(feature = "compat")]
+pub mod compat;
+
 if_std! {
     // FIXME: currently async/await is only available with std
     pub mod async_await;

futures-util/src/try_future/mod.rs

@@ -6,6 +6,15 @@
 use futures_core::future::TryFuture;
 use futures_sink::Sink;
 
+#[cfg(feature = "compat")]
+use crate::compat::Compat;
+
+#[cfg(feature = "compat")]
+use futures_core::task::Executor;
+
+#[cfg(feature = "compat")]
+use core::marker::Unpin;
+
 /* TODO
 mod join;
 mod select;
@@ -477,6 +486,17 @@ pub trait TryFutureExt: TryFuture {
         UnwrapOrElse::new(self, f)
     }
 
+    /// Wraps a [`TryFuture`] into a future compatable with libraries using
+    /// futures 0.1 future definitons. Requires the `compat` feature to enable.
+    /// 
+    #[cfg(feature = "compat")]
+    fn compat<E>(self, executor: E) -> Compat<Self, E> 
+        where Self: Sized + Unpin,
+              E: Executor,
+    {
+        Compat::new(self, Some(executor))
+    }
+
     /// Wraps a [`TryFuture`] into a type that implements
     /// [`Future`](std::future::Future).
     ///
@@ -498,7 +518,7 @@ pub trait TryFutureExt: TryFuture {
     /// fn take_future(future: impl Future<Output = Result<T, E>>) { /* ... */ }
     ///
     /// take_future(make_try_future().into_future());
-    /// ```
+    /// ```    
     fn into_future(self) -> IntoFuture<Self>
         where Self: Sized,
     {