Repository: rust-lang/futures-rs (rust)

Issue:
Add recv method to mpsc::Receiver
Add recv method to `Receiver` how was done in std and tokio.

Pull request description:
channel: Add recv
Implements `recv()` for `Receiver` and `UnboundedReceiver`, as discussed in #2944.

Since we're reimplementing `Next` anyways we might aswell get the benefits so this returns a new `RecvError`
instead of an `Option` for better symmetry with `try_recv` and other crates like [async_channel](https://docs.rs/async-channel/latest/async_channel/struct.Receiver.html#method.recv).

To be fair tokio [uses an option](https://docs.rs/tokio/latest/tokio/sync/mpsc/struct.Receiver.html#method.recv), but if we want to do comparisons then the style of [std::mpsc](https://doc.rust-lang.org/std/sync/mpsc/struct.Receiver.html#method.recv) is probably the one to immitate, and they are using a dedicated error. 
  
---
Closes #2184

Patch:
diff --git a/futures-channel/src/mpsc/mod.rs b/futures-channel/src/mpsc/mod.rs
index c5e6bada..69bb88ff 100644
--- a/futures-channel/src/mpsc/mod.rs
+++ b/futures-channel/src/mpsc/mod.rs
@@ -78,9 +78,11 @@
 // happens-before semantics required for the acquire / release semantics used
 // by the queue structure.
 
+use core::future::Future;
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::__internal::AtomicWaker;
 use futures_core::task::{Context, Poll, Waker};
+use futures_core::FusedFuture;
 use std::fmt;
 use std::pin::Pin;
 use std::sync::atomic::AtomicUsize;
@@ -167,7 +169,7 @@ enum SendErrorKind {
     Disconnected,
 }
 
-/// The error type returned from [`try_recv`](Receiver::try_recv).
+/// Error returned by [`Receiver::try_recv`] or [`UnboundedReceiver::try_recv`].
 #[derive(PartialEq, Eq, Clone, Copy, Debug)]
 pub enum TryRecvError {
     /// The channel is empty but not closed.
@@ -177,6 +179,18 @@ pub enum TryRecvError {
     Closed,
 }
 
+/// Error returned by the future returned by [`Receiver::recv()`] or [`UnboundedReceiver::recv()`].
+/// Received when the channel is empty and closed.
+#[derive(PartialEq, Eq, Clone, Copy, Debug)]
+pub struct RecvError;
+
+/// Future returned by [`Receiver::recv()`] or [`UnboundedReceiver::recv()`].
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct Recv<'a, St: ?Sized> {
+    stream: &'a mut St,
+}
+
 impl fmt::Display for SendError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         if self.is_full() {
@@ -189,6 +203,14 @@ impl fmt::Display for SendError {
 
 impl std::error::Error for SendError {}
 
+impl fmt::Display for RecvError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "receive failed because channel is empty and closed")
+    }
+}
+
+impl std::error::Error for RecvError {}
+
 impl SendError {
     /// Returns `true` if this error is a result of the channel being full.
     pub fn is_full(&self) -> bool {
@@ -979,6 +1001,12 @@ impl<T> fmt::Debug for UnboundedSender<T> {
  */
 
 impl<T> Receiver<T> {
+    /// Waits for a message from the channel.
+    /// If the channel is empty and closed, returns [`RecvError`].
+    pub fn recv(&mut self) -> Recv<'_, Self> {
+        Recv::new(self)
+    }
+
     /// Closes the receiving half of a channel, without dropping it.
     ///
     /// This prevents any further messages from being sent on the channel while
@@ -1121,6 +1149,31 @@ impl<T> Stream for Receiver<T> {
     }
 }
 
+impl<St: ?Sized + Unpin> Unpin for Recv<'_, St> {}
+impl<'a, St: ?Sized + Stream + Unpin> Recv<'a, St> {
+    fn new(stream: &'a mut St) -> Self {
+        Self { stream }
+    }
+}
+
+impl<St: ?Sized + FusedStream + Unpin> FusedFuture for Recv<'_, St> {
+    fn is_terminated(&self) -> bool {
+        self.stream.is_terminated()
+    }
+}
+
+impl<St: ?Sized + Stream + Unpin> Future for Recv<'_, St> {
+    type Output = Result<St::Item, RecvError>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        match Pin::new(&mut self.stream).poll_next(cx) {
+            Poll::Ready(Some(msg)) => Poll::Ready(Ok(msg)),
+            Poll::Ready(None) => Poll::Ready(Err(RecvError)),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
 impl<T> Drop for Receiver<T> {
     fn drop(&mut self) {
         // Drain the channel of all pending messages
@@ -1164,6 +1217,12 @@ impl<T> fmt::Debug for Receiver<T> {
 }
 
 impl<T> UnboundedReceiver<T> {
+    /// Waits for a message from the channel.
+    /// If the channel is empty and closed, returns [`RecvError`].
+    pub fn recv(&mut self) -> Recv<'_, Self> {
+        Recv::new(self)
+    }
+
     /// Closes the receiving half of a channel, without dropping it.
     ///
     /// This prevents any further messages from being sent on the channel while
diff --git a/futures-util/src/io/mod.rs b/futures-util/src/io/mod.rs
index 96d63958..bf46dcaa 100644
--- a/futures-util/src/io/mod.rs
+++ b/futures-util/src/io/mod.rs
@@ -21,7 +21,7 @@
 use crate::compat::Compat;
 use crate::future::assert_future;
 use crate::stream::assert_stream;
-use std::{pin::Pin, ptr, string::String, vec::Vec};
+use std::{pin::Pin, string::String, vec::Vec};
 
 // Re-export some types from `std::io` so that users don't have to deal
 // with conflicts when `use`ing `futures::io` and `std::io`.
@@ -34,14 +34,6 @@ pub use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite};
 // https://github.com/rust-lang/rust/blob/master/src/libstd/sys_common/io.rs#L1
 const DEFAULT_BUF_SIZE: usize = 8 * 1024;
 
-/// Initializes a buffer if necessary.
-///
-/// A buffer is currently always initialized.
-#[inline]
-unsafe fn initialize<R: AsyncRead>(_reader: &R, buf: &mut [u8]) {
-    unsafe { ptr::write_bytes(buf.as_mut_ptr(), 0, buf.len()) }
-}
-
 mod allow_std;
 pub use self::allow_std::AllowStdIo;
 
diff --git a/futures-util/src/io/read_to_end.rs b/futures-util/src/io/read_to_end.rs
index 919d7d13..82a36d21 100644
--- a/futures-util/src/io/read_to_end.rs
+++ b/futures-util/src/io/read_to_end.rs
@@ -3,6 +3,7 @@ use futures_core::ready;
 use futures_core::task::{Context, Poll};
 use futures_io::AsyncRead;
 use std::io;
+use std::iter;
 use std::pin::Pin;
 use std::vec::Vec;
 
@@ -55,12 +56,10 @@ pub(super) fn read_to_end_internal<R: AsyncRead + ?Sized>(
     let mut g = Guard { len: buf.len(), buf };
     loop {
         if g.len == g.buf.len() {
-            unsafe {
-                g.buf.reserve(32);
-                let capacity = g.buf.capacity();
-                g.buf.set_len(capacity);
-                super::initialize(&rd, &mut g.buf[g.len..]);
-            }
+            g.buf.reserve(32);
+            let spare_capacity = g.buf.capacity() - g.buf.len();
+            // FIXME: switch to `Vec::resize` once rust-lang/rust#120050 is fixed
+            g.buf.extend(iter::repeat(0).take(spare_capacity));
         }
 
         let buf = &mut g.buf[g.len..];
diff --git a/futures-util/src/lib.rs b/futures-util/src/lib.rs
index 96e0c297..93d7c5e2 100644
--- a/futures-util/src/lib.rs
+++ b/futures-util/src/lib.rs
@@ -327,5 +327,4 @@ pub mod lock;
 mod abortable;
 
 mod fns;
-mod macros;
 mod unfold_state;
diff --git a/futures-util/src/macros.rs b/futures-util/src/macros.rs
deleted file mode 100644
index 0f5cba99..00000000
--- a/futures-util/src/macros.rs
+++ /dev/null
@@ -1,31 +0,0 @@
-/// Pins a value on the stack.
-///
-/// Can safely pin values that are not `Unpin` by taking ownership.
-///
-/// **Note:** Since Rust 1.68, this macro is soft-deprecated in favor of
-/// [`pin!`](https://doc.rust-lang.org/std/pin/macro.pin.html) macro
-/// in the standard library.
-///
-/// # Example
-///
-/// ```rust
-/// # use futures_util::pin_mut;
-/// # use core::pin::Pin;
-/// # struct Foo {}
-/// let foo = Foo { /* ... */ };
-/// pin_mut!(foo);
-/// let _: Pin<&mut Foo> = foo;
-/// ```
-#[macro_export]
-macro_rules! pin_mut {
-    ($($x:ident),* $(,)?) => { $(
-        // Move the value to ensure that it is owned
-        let mut $x = $x;
-        // Shadow the original binding so that it can't be directly accessed
-        // ever again.
-        #[allow(unused_mut)]
-        let mut $x = unsafe {
-            $crate::__private::Pin::new_unchecked(&mut $x)
-        };
-    )* }
-}
diff --git a/futures-util/src/sink/with.rs b/futures-util/src/sink/with.rs
index 86d3dcc7..5a2c8a08 100644
--- a/futures-util/src/sink/with.rs
+++ b/futures-util/src/sink/with.rs
@@ -3,7 +3,7 @@ use core::marker::PhantomData;
 use core::pin::Pin;
 use futures_core::future::Future;
 use futures_core::ready;
-use futures_core::stream::Stream;
+use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Context, Poll};
 use futures_sink::Sink;
 use pin_project_lite::pin_project;
@@ -74,6 +74,17 @@ where
     delegate_stream!(sink);
 }
 
+impl<S, Item, U, Fut, F> FusedStream for With<S, Item, U, Fut, F>
+where
+    S: FusedStream + Sink<Item>,
+    F: FnMut(U) -> Fut,
+    Fut: Future,
+{
+    fn is_terminated(&self) -> bool {
+        self.sink.is_terminated()
+    }
+}
+
 impl<Si, Item, U, Fut, F, E> With<Si, Item, U, Fut, F>
 where
     Si: Sink<Item>,
diff --git a/futures/src/lib.rs b/futures/src/lib.rs
index 71e71c23..2c31e76d 100644
--- a/futures/src/lib.rs
+++ b/futures/src/lib.rs
@@ -119,7 +119,6 @@ pub use futures_util::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteEx
 
 // Macro reexports
 pub use futures_core::ready; // Readiness propagation
-pub use futures_util::pin_mut;
 #[cfg(feature = "std")]
 #[cfg(feature = "async-await")]
 pub use futures_util::select;


Test patch:
diff --git a/futures-channel/tests/mpsc.rs b/futures-channel/tests/mpsc.rs
index 1ef778b7..847f3a2b 100644
--- a/futures-channel/tests/mpsc.rs
+++ b/futures-channel/tests/mpsc.rs
@@ -4,7 +4,7 @@ use futures::future::{poll_fn, FutureExt};
 use futures::sink::{Sink, SinkExt};
 use futures::stream::{Stream, StreamExt};
 use futures::task::{Context, Poll};
-use futures_channel::mpsc::TryRecvError;
+use futures_channel::mpsc::{RecvError, TryRecvError};
 use futures_test::task::{new_count_waker, noop_context};
 use std::pin::pin;
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -429,6 +429,49 @@ fn stress_poll_ready() {
     stress(16);
 }
 
+#[test]
+fn test_bounded_recv() {
+    let (dropped_tx, dropped_rx) = oneshot::channel();
+    let (tx, mut rx) = mpsc::channel(1);
+    thread::spawn(move || {
+        block_on(async move {
+            send_one_two_three(tx).await;
+            dropped_tx.send(()).unwrap();
+        });
+    });
+
+    let res = block_on(async move {
+        let mut res = Vec::new();
+        for _ in 0..3 {
+            res.push(rx.recv().await.unwrap());
+        }
+        dropped_rx.await.unwrap();
+        assert_eq!(rx.recv().await, Err(RecvError));
+        res
+    });
+    assert_eq!(res, [1, 2, 3]);
+}
+
+#[test]
+fn test_unbounded_recv() {
+    let (mut tx, mut rx) = mpsc::unbounded();
+
+    let res = block_on(async move {
+        let mut res = Vec::new();
+        for i in 1..=3 {
+            tx.send(i).await.unwrap();
+        }
+        drop(tx);
+
+        for _ in 0..3 {
+            res.push(rx.recv().await.unwrap());
+        }
+        assert_eq!(rx.recv().await, Err(RecvError));
+        res
+    });
+    assert_eq!(res, [1, 2, 3]);
+}
+
 #[test]
 fn try_send_1() {
     const N: usize = if cfg!(miri) { 100 } else { 3000 };