Bring stream combinators over to 0.3

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! 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/future/flatten_sink.rs

@@ -66,7 +66,7 @@ impl<F> Sink for FlattenSink<F> where F: Future, <F as Future>::Item: Sink<SinkE
 
     fn poll_close(&mut self, cx: &mut task::Context) -> Result<Async<()>, Self::SinkError> {
         if let State::Ready(ref mut s) = self.st {
-            try_ready!(s.poll_close(cx));
+            ready!(s.poll_close(cx));
         }
         self.st = State::Closed;
         return Ok(Async::Ready(()));

futures-util/src/io/lines.rs

@@ -46,7 +46,7 @@ impl<A> Stream for Lines<A>
     type Error = io::Error;
 
     fn poll(&mut self, cx: &mut task::Context) -> Poll<Option<String>, io::Error> {
-        let n = try_ready!(self.io.read_line(&mut self.line));
+        let n = ready!(self.io.read_line(&mut self.line));
         if n == 0 && self.line.len() == 0 {
             return Ok(None.into())
         }

futures-util/src/io/read_until.rs

@@ -62,7 +62,7 @@ impl<A> Future for ReadUntil<A>
                 // and just return it, as we are finished.
                 // If we hit "would block" then all the read data so far
                 // is in our buffer, and otherwise we propagate errors.
-                try_ready!(a.read_until(byte, buf));
+                ready!(a.read_until(byte, buf));
             },
             State::Empty => panic!("poll ReadUntil after it's done"),
         }

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/sink/buffer.rs

@@ -44,11 +44,11 @@ impl<S: Sink> Buffer<S> {
     }
 
     fn try_empty_buffer(&mut self, cx: &mut task::Context) -> Poll<(), S::SinkError> {
-        try_ready!(self.sink.poll_ready(cx));
+        ready!(self.sink.poll_ready(cx));
         while let Some(item) = self.buf.pop_front() {
             self.sink.start_send(item)?;
             if self.buf.len() != 0 {
-                try_ready!(self.sink.poll_ready(cx));
+                ready!(self.sink.poll_ready(cx));
             }
         }
         Ok(Async::Ready(()))
@@ -93,13 +93,13 @@ impl<S: Sink> Sink for Buffer<S> {
     }
 
     fn poll_flush(&mut self, cx: &mut task::Context) -> Poll<(), Self::SinkError> {
-        try_ready!(self.try_empty_buffer(cx));
+        ready!(self.try_empty_buffer(cx));
         debug_assert!(self.buf.is_empty());
         self.sink.poll_flush(cx)
     }
 
     fn poll_close(&mut self, cx: &mut task::Context) -> Poll<(), Self::SinkError> {
-        try_ready!(self.try_empty_buffer(cx));
+        ready!(self.try_empty_buffer(cx));
         debug_assert!(self.buf.is_empty());
         self.sink.poll_close(cx)
     }

futures-util/src/sink/send.rs

@@ -61,7 +61,7 @@ impl<S: Sink> Future for Send<S> {
 
         // we're done sending the item, but want to block on flushing the
         // sink
-        try_ready!(self.sink_mut().poll_flush(cx));
+        ready!(self.sink_mut().poll_flush(cx));
 
         // now everything's emptied, so return the sink for further use
         Ok(Async::Ready(self.take_sink()))

futures-util/src/sink/send_all.rs

@@ -75,18 +75,18 @@ impl<T, U> Future for SendAll<T, U>
         // If we've got an item buffered already, we need to write it to the
         // sink before we can do anything else
         if let Some(item) = self.buffered.take() {
-            try_ready!(self.try_start_send(cx, item))
+            ready!(self.try_start_send(cx, item))
         }
 
         loop {
             match self.stream_mut().poll_next(cx)? {
-                Async::Ready(Some(item)) => try_ready!(self.try_start_send(cx, item)),
+                Async::Ready(Some(item)) => ready!(self.try_start_send(cx, item)),
                 Async::Ready(None) => {
-                    try_ready!(self.sink_mut().poll_flush(cx));
+                    ready!(self.sink_mut().poll_flush(cx));
                     return Ok(Async::Ready(self.take_result()))
                 }
                 Async::Pending => {
-                    try_ready!(self.sink_mut().poll_flush(cx));
+                    ready!(self.sink_mut().poll_flush(cx));
                     return Ok(Async::Pending)
                 }
             }

futures-util/src/sink/with.rs

@@ -143,12 +143,12 @@ impl<S, U, Fut, F> Sink for With<S, U, Fut, F>
     }
 
     fn poll_flush(&mut self, cx: &mut task::Context) -> Poll<(), Self::SinkError> {
-        try_ready!(self.poll(cx));
+        ready!(self.poll(cx));
         self.sink.poll_flush(cx).map_err(Into::into)
     }
 
     fn poll_close(&mut self, cx: &mut task::Context) -> Poll<(), Self::SinkError> {
-        try_ready!(self.poll(cx));
+        ready!(self.poll(cx));
         self.sink.poll_close(cx).map_err(Into::into)
     }
 }

futures-util/src/sink/with_flat_map.rs

@@ -71,7 +71,7 @@ where
             }
         }
         if let Some(mut stream) = self.stream.take() {
-            while let Some(x) = try_ready!(stream.poll_next(cx)) {
+            while let Some(x) = ready!(stream.poll_next(cx)) {
                 match self.sink.poll_ready(cx)? {
                     Async::Ready(()) => self.sink.start_send(x)?,
                     Async::Pending => {

futures-util/src/stream/buffer_unordered.rs

@@ -96,7 +96,7 @@ impl<S> Stream for BufferUnordered<S>
         }
 
         // Try polling a new future
-        if let Some(val) = try_ready!(self.queue.poll_next(cx)) {
+        if let Some(val) = ready!(self.queue.poll_next(cx)) {
             return Ok(Async::Ready(Some(val)));
         }
 

futures-util/src/stream/buffered.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);
 }
 
@@ -109,7 +106,7 @@ impl<S> Stream for Buffered<S>
         }
 
         // Try polling a new future
-        if let Some(val) = try_ready!(self.queue.poll_next(cx)) {
+        if let Some(val) = ready!(self.queue.poll_next(cx)) {
             return Ok(Async::Ready(Some(val)));
         }