Merge branch 'master' into fs-stream-enumerate

Author: montekki

docs/src/tutorial/all_together.md

@@ -115,7 +115,8 @@ async fn broker(mut events: Receiver<Event>) -> Result<()> {
             Event::Message { from, to, msg } => {
                 for addr in to {
                     if let Some(peer) = peers.get_mut(&addr) {
-                        peer.send(format!("from {}: {}\n", from, msg)).await?
+                        let msg = format!("from {}: {}\n", from, msg);
+                        peer.send(msg).await?
                     }
                 }
             }

docs/src/tutorial/clean_shutdown.md

@@ -115,7 +115,8 @@ Let's add waiting to the server:
 #             Event::Message { from, to, msg } => {
 #                 for addr in to {
 #                     if let Some(peer) = peers.get_mut(&addr) {
-#                         peer.send(format!("from {}: {}\n", from, msg)).await?
+#                         let msg = format!("from {}: {}\n", from, msg);
+#                         peer.send(msg).await?
 #                     }
 #                 }
 #             }
@@ -217,7 +218,8 @@ async fn broker(mut events: Receiver<Event>) -> Result<()> {
             Event::Message { from, to, msg } => {
                 for addr in to {
                     if let Some(peer) = peers.get_mut(&addr) {
-                        peer.send(format!("from {}: {}\n", from, msg)).await?
+                        let msg = format!("from {}: {}\n", from, msg);
+                        peer.send(msg).await?
                     }
                 }
             }

docs/src/tutorial/connecting_readers_and_writers.md

@@ -73,7 +73,8 @@ async fn broker(mut events: Receiver<Event>) -> Result<()> {
             Event::Message { from, to, msg } => {  // 3
                 for addr in to {
                     if let Some(peer) = peers.get_mut(&addr) {
-                        peer.send(format!("from {}: {}\n", from, msg)).await?
+                        let msg = format!("from {}: {}\n", from, msg);
+                        peer.send(msg).await?
                     }
                 }
             }

docs/src/tutorial/handling_disconnection.md

@@ -257,7 +257,8 @@ async fn broker(events: Receiver<Event>) {
             Event::Message { from, to, msg } => {
                 for addr in to {
                     if let Some(peer) = peers.get_mut(&addr) {
-                        peer.send(format!("from {}: {}\n", from, msg)).await
+                        let msg = format!("from {}: {}\n", from, msg);
+                        peer.send(msg).await
                             .unwrap() // 6
                     }
                 }

examples/a-chat/server.rs

@@ -139,9 +139,8 @@ async fn broker_loop(mut events: Receiver<Event>) {
             Event::Message { from, to, msg } => {
                 for addr in to {
                     if let Some(peer) = peers.get_mut(&addr) {
-                        peer.send(format!("from {}: {}\n", from, msg))
-                            .await
-                            .unwrap()
+                        let msg = format!("from {}: {}\n", from, msg);
+                        peer.send(msg).await.unwrap();
                     }
                 }
             }

examples/socket-timeouts.rs

@@ -10,7 +10,7 @@ async fn get() -> io::Result<Vec<u8>> {
 
     let mut buf = vec![];
 
-    io::timeout(Duration::from_secs(5), async {
+    io::timeout(Duration::from_secs(5), async move {
         stream.read_to_end(&mut buf).await?;
         Ok(buf)
     })

src/io/buf_read/fill_buf.rs

@@ -0,0 +1,32 @@
+use std::pin::Pin;
+
+use futures_io::AsyncBufRead;
+
+use crate::future::Future;
+use crate::io;
+use crate::task::{Context, Poll};
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct FillBufFuture<'a, R: ?Sized> {
+    reader: &'a mut R,
+}
+
+impl<'a, R: ?Sized> FillBufFuture<'a, R> {
+    pub(crate) fn new(reader: &'a mut R) -> Self {
+        Self { reader }
+    }
+}
+
+impl<'a, R: AsyncBufRead + Unpin + ?Sized> Future for FillBufFuture<'a, R> {
+    type Output = io::Result<&'a [u8]>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&'a [u8]>> {
+        let Self { reader } = &mut *self;
+        let result = Pin::new(reader).poll_fill_buf(cx);
+        // This is safe because:
+        // 1. The buffer is valid for the lifetime of the reader.
+        // 2. Output is unrelated to the wrapper (Self).
+        result.map_ok(|buf| unsafe { std::mem::transmute::<&'_ [u8], &'a [u8]>(buf) })
+    }
+}

src/io/buf_read/mod.rs

@@ -1,7 +1,9 @@
+mod fill_buf;
 mod lines;
 mod read_line;
 mod read_until;
 
+use fill_buf::FillBufFuture;
 pub use lines::Lines;
 use read_line::ReadLineFuture;
 use read_until::ReadUntilFuture;
@@ -41,6 +43,26 @@ cfg_if! {
 /// [`futures::io::AsyncBufRead`]:
 /// https://docs.rs/futures-preview/0.3.0-alpha.17/futures/io/trait.AsyncBufRead.html
 pub trait BufRead {
+    /// Returns the contents of the internal buffer, filling it with more data from the inner
+    /// reader if it is empty.
+    ///
+    /// This function is a lower-level call. It needs to be paired with the [`consume`] method to
+    /// function properly. When calling this method, none of the contents will be "read" in the
+    /// sense that later calling `read` may return the same contents. As such, [`consume`] must be
+    /// called with the number of bytes that are consumed from this buffer to ensure that the bytes
+    /// are never returned twice.
+    ///
+    /// [`consume`]: #tymethod.consume
+    ///
+    /// An empty buffer returned indicates that the stream has reached EOF.
+    // TODO: write a proper doctest with `consume`
+    fn fill_buf<'a>(&'a mut self) -> ret!('a, FillBufFuture, io::Result<&'a [u8]>)
+    where
+        Self: Unpin,
+    {
+        FillBufFuture::new(self)
+    }
+
     /// Reads all bytes into `buf` until the delimiter `byte` or EOF is reached.
     ///
     /// This function will read bytes from the underlying stream until the delimiter or EOF is

src/stream/stream/all.rs

@@ -1,43 +1,36 @@
-use crate::future::Future;
-use crate::task::{Context, Poll};
-
 use std::marker::PhantomData;
 use std::pin::Pin;
 
-#[derive(Debug)]
-pub struct AllFuture<'a, S, F, T>
-where
-    F: FnMut(T) -> bool,
-{
+use crate::future::Future;
+use crate::stream::Stream;
+use crate::task::{Context, Poll};
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct AllFuture<'a, S, F, T> {
     pub(crate) stream: &'a mut S,
     pub(crate) f: F,
     pub(crate) result: bool,
-    pub(crate) __item: PhantomData<T>,
+    pub(crate) _marker: PhantomData<T>,
 }
 
-impl<'a, S, F, T> AllFuture<'a, S, F, T>
-where
-    F: FnMut(T) -> bool,
-{
-    pin_utils::unsafe_pinned!(stream: &'a mut S);
-    pin_utils::unsafe_unpinned!(result: bool);
-    pin_utils::unsafe_unpinned!(f: F);
-}
+impl<S: Unpin, F, T> Unpin for AllFuture<'_, S, F, T> {}
 
 impl<S, F> Future for AllFuture<'_, S, F, S::Item>
 where
-    S: futures_core::stream::Stream + Unpin + Sized,
+    S: Stream + Unpin + Sized,
     F: FnMut(S::Item) -> bool,
 {
     type Output = bool;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        use futures_core::stream::Stream;
-        let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
+        let next = futures_core::ready!(Pin::new(&mut *self.stream).poll_next(cx));
+
         match next {
             Some(v) => {
-                let result = (self.as_mut().f())(v);
-                *self.as_mut().result() = result;
+                let result = (&mut self.f)(v);
+                self.result = result;
+
                 if result {
                     // don't forget to wake this task again to pull the next item from stream
                     cx.waker().wake_by_ref();

src/stream/stream/any.rs

@@ -1,43 +1,36 @@
-use crate::future::Future;
-use crate::task::{Context, Poll};
-
 use std::marker::PhantomData;
 use std::pin::Pin;
 
-#[derive(Debug)]
-pub struct AnyFuture<'a, S, F, T>
-where
-    F: FnMut(T) -> bool,
-{
+use crate::future::Future;
+use crate::stream::Stream;
+use crate::task::{Context, Poll};
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct AnyFuture<'a, S, F, T> {
     pub(crate) stream: &'a mut S,
     pub(crate) f: F,
     pub(crate) result: bool,
-    pub(crate) __item: PhantomData<T>,
+    pub(crate) _marker: PhantomData<T>,
 }
 
-impl<'a, S, F, T> AnyFuture<'a, S, F, T>
-where
-    F: FnMut(T) -> bool,
-{
-    pin_utils::unsafe_pinned!(stream: &'a mut S);
-    pin_utils::unsafe_unpinned!(result: bool);
-    pin_utils::unsafe_unpinned!(f: F);
-}
+impl<S: Unpin, F, T> Unpin for AnyFuture<'_, S, F, T> {}
 
 impl<S, F> Future for AnyFuture<'_, S, F, S::Item>
 where
-    S: futures_core::stream::Stream + Unpin + Sized,
+    S: Stream + Unpin + Sized,
     F: FnMut(S::Item) -> bool,
 {
     type Output = bool;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        use futures_core::stream::Stream;
-        let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
+        let next = futures_core::ready!(Pin::new(&mut *self.stream).poll_next(cx));
+
         match next {
             Some(v) => {
-                let result = (self.as_mut().f())(v);
-                *self.as_mut().result() = result;
+                let result = (&mut self.f)(v);
+                self.result = result;
+
                 if result {
                     Poll::Ready(true)
                 } else {

src/stream/stream/find.rs

@@ -0,0 +1,49 @@
+use crate::task::{Context, Poll};
+use std::marker::PhantomData;
+use std::pin::Pin;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct FindFuture<'a, S, P, T> {
+    stream: &'a mut S,
+    p: P,
+    __t: PhantomData<T>,
+}
+
+impl<'a, S, P, T> FindFuture<'a, S, P, T> {
+    pin_utils::unsafe_pinned!(stream: &'a mut S);
+    pin_utils::unsafe_unpinned!(p: P);
+
+    pub(super) fn new(stream: &'a mut S, p: P) -> Self {
+        FindFuture {
+            stream,
+            p,
+            __t: PhantomData,
+        }
+    }
+}
+
+impl<'a, S, P> futures_core::future::Future for FindFuture<'a, S, P, S::Item>
+where
+    S: futures_core::stream::Stream + Unpin + Sized,
+    P: FnMut(&S::Item) -> bool,
+{
+    type Output = Option<S::Item>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        use futures_core::stream::Stream;
+
+        let item = futures_core::ready!(self.as_mut().stream().poll_next(cx));
+
+        match item {
+            Some(v) => match (self.as_mut().p())(&v) {
+                true => Poll::Ready(Some(v)),
+                false => {
+                    cx.waker().wake_by_ref();
+                    Poll::Pending
+                }
+            },
+            None => Poll::Ready(None),
+        }
+    }
+}

src/stream/stream/min_by.rs

@@ -2,8 +2,10 @@ use std::cmp::Ordering;
 use std::pin::Pin;
 
 use crate::future::Future;
+use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
+#[doc(hidden)]
 #[allow(missing_debug_implementations)]
 pub struct MinByFuture<S, F, T> {
     stream: S,
@@ -27,7 +29,7 @@ impl<S, F, T> MinByFuture<S, F, T> {
 
 impl<S, F> Future for MinByFuture<S, F, S::Item>
 where
-    S: futures_core::stream::Stream + Unpin + Sized,
+    S: Stream + Unpin + Sized,
     S::Item: Copy,
     F: FnMut(&S::Item, &S::Item) -> Ordering,
 {