Merge branch 'master' into fs-stream-fold

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 {