Bring stream combinators over to 0.3

Author: aturon

futures-core/src/lib.rs

@@ -40,6 +40,28 @@ macro_rules! pinned_field {
     )
 }
 
+#[macro_export]
+macro_rules! unsafe_pinned {
+    ($f:tt -> $t:ty) => (
+        fn $f<'a>(self: &'a mut PinMut<Self>) -> PinMut<'a, $t> {
+            unsafe {
+                pinned_field!(self, $f)
+            }
+        }
+    )
+}
+
+#[macro_export]
+macro_rules! unsafe_unpinned {
+    ($f:tt -> $t:ty) => (
+        fn $f<'a>(self: &'a mut PinMut<Self>) -> &'a mut $t {
+            unsafe {
+                &mut ::core::mem::PinMut::get_mut(self.reborrow()).$f
+            }
+        }
+    )
+}
+
 mod poll;
 pub use poll::Poll;
 

futures-core/src/poll.rs

@@ -1,3 +1,14 @@
+/// A macro for extracting the successful type of a `Poll<T>`.
+///
+/// This macro bakes in propagation of `Pending` signals by returning early.
+#[macro_export]
+macro_rules! try_ready {
+    ($e:expr) => (match $e {
+        $crate::Poll::Ready(t) => t,
+        $crate::Poll::Pending => return $crate::Poll::Pending,
+    })
+}
+
 /// Indicates whether a value is available, or if the current task has been
 /// scheduled for later wake-up instead.
 #[derive(Copy, Clone, Debug, PartialEq)]

futures-util/src/lib.rs

@@ -72,12 +72,15 @@ pub mod io;
 #[cfg(feature = "std")]
 pub use io::{AsyncReadExt, AsyncWriteExt};
 
-// pub mod stream;
-// pub use stream::StreamExt;
+pub mod stream;
+pub use stream::StreamExt;
 
 // pub mod sink;
 // pub use sink::SinkExt;
 
+mod pin_mut;
+pub use pin_mut::{PinMutExt, OptionExt};
+
 pub mod prelude {
     //! Prelude containing the extension traits, which add functionality to
     //! existing asynchronous types.

futures-util/src/pin_mut.rs

@@ -0,0 +1,29 @@
+use core::mem::PinMut;
+
+/// Useful additions to the `PinMut` type
+pub trait PinMutExt<T> {
+    /// Overwrite the referenced data, dropping the existing content
+    fn assign(&mut self, data: T);
+}
+
+impl<'a, T> PinMutExt<T> for PinMut<'a, T> {
+    fn assign(&mut self, data: T) {
+        unsafe { *PinMut::get_mut(self.reborrow()) = data }
+    }
+}
+
+/// Useful additions to the `Option` type, for working with `PinMut`
+pub trait OptionExt<'a, T> {
+    /// Push `PinMut` through an `Option`
+    fn as_pin_mut(self) -> Option<PinMut<'a, T>>;
+}
+
+impl<'a, T> OptionExt<'a, T> for PinMut<'a, Option<T>> {
+    fn as_pin_mut(self) -> Option<PinMut<'a, T>> {
+        unsafe {
+            PinMut::get_mut(self).as_mut().map(|x| {
+                PinMut::new_unchecked(x)
+            })
+        }
+    }
+}

futures-util/src/stream/and_then.rs

@@ -1,8 +1,8 @@
+use core::mem::PinMut;
 use std::fmt;
 
-use futures_core::{Async, IntoFuture, Poll, Stream};
+use futures_core::{Future, Poll, Stream};
 use futures_core::task;
-use futures_sink::{Sink};
 
 use stream::{Fuse, FuturesOrdered};
 
@@ -13,12 +13,9 @@ use stream::{Fuse, FuturesOrdered};
 /// results in the order that they were pulled out of the original stream. This
 /// is created by the `Stream::buffered` method.
 #[must_use = "streams do nothing unless polled"]
-pub struct Buffered<S>
-    where S: Stream,
-          S::Item: IntoFuture,
-{
+pub struct Buffered<S: Stream> {
     stream: Fuse<S>,
-    queue: FuturesOrdered<<S::Item as IntoFuture>::Future>,
+    queue: FuturesOrdered<S::Item>,
     max: usize,
 }
 
@@ -84,7 +81,7 @@ impl<S> Sink for Buffered<S>
 {
     type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
-    
+
     delegate_sink!(stream);
 }
 

futures-util/src/stream/buffered.rs

@@ -1,9 +1,9 @@
 use std::prelude::v1::*;
 use std::any::Any;
 use std::panic::{catch_unwind, UnwindSafe, AssertUnwindSafe};
-use std::mem;
+use std::mem::PinMut;
 
-use futures_core::{Poll, Async, Stream};
+use futures_core::{Poll, Stream};
 use futures_core::task;
 
 /// Stream for the `catch_unwind` combinator.
@@ -12,51 +12,29 @@ use futures_core::task;
 #[derive(Debug)]
 #[must_use = "streams do nothing unless polled"]
 pub struct CatchUnwind<S> where S: Stream {
-    state: CatchUnwindState<S>,
+    stream: S,
 }
 
 pub fn new<S>(stream: S) -> CatchUnwind<S>
     where S: Stream + UnwindSafe,
 {
-    CatchUnwind {
-        state: CatchUnwindState::Stream(stream),
-    }
+    CatchUnwind { stream }
 }
 
-#[derive(Debug)]
-enum CatchUnwindState<S> {
-    Stream(S),
-    Eof,
-    Done,
+impl<S: Stream> CatchUnwind<S> {
+    unsafe_pinned!(stream -> S);
 }
 
 impl<S> Stream for CatchUnwind<S>
     where S: Stream + UnwindSafe,
 {
-    type Item = Result<S::Item, S::Error>;
-    type Error = Box<Any + Send>;
+    type Item = Result<S::Item, Box<Any + Send>>;
 
-    fn poll_next(&mut self, cx: &mut task::Context) -> Poll<Option<Self::Item>, Self::Error> {
-        let mut stream = match mem::replace(&mut self.state, CatchUnwindState::Eof) {
-            CatchUnwindState::Done => panic!("cannot poll after eof"),
-            CatchUnwindState::Eof => {
-                self.state = CatchUnwindState::Done;
-                return Ok(Async::Ready(None));
-            }
-            CatchUnwindState::Stream(stream) => stream,
-        };
-        let res = catch_unwind(AssertUnwindSafe(|| (stream.poll_next(cx), stream)));
+    fn poll_next(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<Option<Self::Item>> {
+        let res = catch_unwind(AssertUnwindSafe(|| self.stream().poll_next(cx)));
         match res {
-            Err(e) => Err(e), // and state is already Eof
-            Ok((poll, stream)) => {
-                self.state = CatchUnwindState::Stream(stream);
-                match poll {
-                    Err(e) => Ok(Async::Ready(Some(Err(e)))),
-                    Ok(Async::Pending) => Ok(Async::Pending),
-                    Ok(Async::Ready(Some(r))) => Ok(Async::Ready(Some(Ok(r)))),
-                    Ok(Async::Ready(None)) => Ok(Async::Ready(None)),
-                }
-            }
+            Ok(poll) => poll.map(|opt| opt.map(Ok)),
+            Err(e) => Poll::Ready(Some(Err(e))),
         }
     }
 }

futures-util/src/stream/catch_unwind.rs

@@ -1,18 +1,9 @@
-use core::mem;
+use core::mem::PinMut;
 
-use futures_core::{Stream, Async, Poll};
-use futures_core::task;
+use {PinMutExt, OptionExt};
 
-/// State of chain stream.
-#[derive(Debug)]
-enum State<S1, S2> {
-    /// Emitting elements of first stream
-    First(S1, S2),
-    /// Emitting elements of second stream
-    Second(S2),
-    /// Temporary value to replace first with second
-    Temp,
-}
+use futures_core::{Stream, Poll};
+use futures_core::task;
 
 /// An adapter for chaining the output of two streams.
 ///
@@ -21,36 +12,37 @@ enum State<S1, S2> {
 #[derive(Debug)]
 #[must_use = "streams do nothing unless polled"]
 pub struct Chain<S1, S2> {
-    state: State<S1, S2>
+    first: Option<S1>,
+    second: S2,
 }
 
 pub fn new<S1, S2>(s1: S1, s2: S2) -> Chain<S1, S2>
-    where S1: Stream, S2: Stream<Item=S1::Item, Error=S1::Error>,
+    where S1: Stream, S2: Stream<Item=S1::Item>,
 {
-    Chain { state: State::First(s1, s2) }
+    Chain {
+        first: Some(s1),
+        second: s2,
+    }
+}
+
+// All interactions with `PinMut<Chain<..>>` happen through these methods
+impl<S1, S2> Chain<S1, S2> {
+    unsafe_pinned!(first -> Option<S1>);
+    unsafe_pinned!(second -> S2);
 }
 
 impl<S1, S2> Stream for Chain<S1, S2>
-    where S1: Stream, S2: Stream<Item=S1::Item, Error=S1::Error>,
+    where S1: Stream, S2: Stream<Item=S1::Item>,
 {
     type Item = S1::Item;
-    type Error = S1::Error;
 
-    fn poll_next(&mut self, cx: &mut task::Context) -> Poll<Option<Self::Item>, Self::Error> {
-        loop {
-            match self.state {
-                State::First(ref mut s1, ref _s2) => match s1.poll_next(cx) {
-                    Ok(Async::Ready(None)) => (), // roll
-                    x => return x,
-                },
-                State::Second(ref mut s2) => return s2.poll_next(cx),
-                State::Temp => unreachable!(),
+    fn poll_next(mut self: PinMut<Self>, cx: &mut task::Context) -> Poll<Option<Self::Item>> {
+        if let Some(first) = self.first().as_pin_mut() {
+            if let Some(item) = try_ready!(first.poll_next(cx)) {
+                return Poll::Ready(Some(item))
             }
-
-            self.state = match mem::replace(&mut self.state, State::Temp) {
-                State::First(_s1, s2) => State::Second(s2),
-                _ => unreachable!(),
-            };
         }
+        self.first().assign(None);
+        self.second().poll_next(cx)
     }
 }