Merge branch 'master' into 0.4-stream-try-fold-for-each

Author: taiki-e

.editorconfig

@@ -0,0 +1,6 @@
+[*.rs]
+end_of_line = lf
+insert_final_newline = true
+charset = utf-8
+indent_style = space
+indent_size = 4

futures-channel/src/mpsc/mod.rs

@@ -1021,8 +1021,10 @@ impl<T> Receiver<T> {
     /// only when you've otherwise arranged to be notified when the channel is
     /// no longer empty.
     ///
-    /// This function will panic if called after `try_next` or `poll_next` has
-    /// returned `None`.
+    /// This function returns:
+    /// * `Ok(Some(t))` when message is fetched
+    /// * `Ok(None)` when channel is closed and no messages left in the queue
+    /// * `Err(e)` when there are no messages available, but channel is not yet closed
     pub fn try_next(&mut self) -> Result<Option<T>, TryRecvError> {
         match self.next_message() {
             Poll::Ready(msg) => {
@@ -1033,7 +1035,10 @@ impl<T> Receiver<T> {
     }
 
     fn next_message(&mut self) -> Poll<Option<T>> {
-        let inner = self.inner.as_mut().expect("Receiver::next_message called after `None`");
+        let inner = match self.inner.as_mut() {
+            None => return Poll::Ready(None),
+            Some(inner) => inner,
+        };
         // Pop off a message
         match unsafe { inner.message_queue.pop_spin() } {
             Some(msg) => {
@@ -1169,8 +1174,10 @@ impl<T> UnboundedReceiver<T> {
     /// only when you've otherwise arranged to be notified when the channel is
     /// no longer empty.
     ///
-    /// This function will panic if called after `try_next` or `poll_next` has
-    /// returned `None`.
+    /// This function returns:
+    /// * `Ok(Some(t))` when message is fetched
+    /// * `Ok(None)` when channel is closed and no messages left in the queue
+    /// * `Err(e)` when there are no messages available, but channel is not yet closed
     pub fn try_next(&mut self) -> Result<Option<T>, TryRecvError> {
         match self.next_message() {
             Poll::Ready(msg) => {
@@ -1181,7 +1188,10 @@ impl<T> UnboundedReceiver<T> {
     }
 
     fn next_message(&mut self) -> Poll<Option<T>> {
-        let inner = self.inner.as_mut().expect("Receiver::next_message called after `None`");
+        let inner = match self.inner.as_mut() {
+            None => return Poll::Ready(None),
+            Some(inner) => inner,
+        };
         // Pop off a message
         match unsafe { inner.message_queue.pop_spin() } {
             Some(msg) => {

futures-channel/tests/mpsc-close.rs

@@ -276,3 +276,25 @@ fn stress_try_send_as_receiver_closes() {
     bg.join()
         .expect("background thread join");
 }
+
+#[test]
+fn unbounded_try_next_after_none() {
+    let (tx, mut rx) = mpsc::unbounded::<String>();
+    // Drop the sender, close the channel.
+    drop(tx);
+    // Receive the end of channel.
+    assert_eq!(Ok(None), rx.try_next().map_err(|_| ()));
+    // None received, check we can call `try_next` again.
+    assert_eq!(Ok(None), rx.try_next().map_err(|_| ()));
+}
+
+#[test]
+fn bounded_try_next_after_none() {
+    let (tx, mut rx) = mpsc::channel::<String>(17);
+    // Drop the sender, close the channel.
+    drop(tx);
+    // Receive the end of channel.
+    assert_eq!(Ok(None), rx.try_next().map_err(|_| ()));
+    // None received, check we can call `try_next` again.
+    assert_eq!(Ok(None), rx.try_next().map_err(|_| ()));
+}

futures-util/src/future/join.rs

@@ -76,15 +76,6 @@ macro_rules! generate {
 generate! {
     /// Future for the [`join`](join()) function.
     (Join, <Fut1, Fut2>),
-
-    /// Future for the [`join3`] function.
-    (Join3, <Fut1, Fut2, Fut3>),
-
-    /// Future for the [`join4`] function.
-    (Join4, <Fut1, Fut2, Fut3, Fut4>),
-
-    /// Future for the [`join5`] function.
-    (Join5, <Fut1, Fut2, Fut3, Fut4, Fut5>),
 }
 
 /// Joins the result of two futures, waiting for them both to complete.
@@ -116,111 +107,3 @@ where
     let f = Join::new(future1, future2);
     assert_future::<(Fut1::Output, Fut2::Output), _>(f)
 }
-
-/// Same as [`join`](join()), but with more futures.
-///
-/// # Examples
-///
-/// ```
-/// # futures::executor::block_on(async {
-/// use futures::future;
-///
-/// let a = async { 1 };
-/// let b = async { 2 };
-/// let c = async { 3 };
-/// let tuple = future::join3(a, b, c);
-///
-/// assert_eq!(tuple.await, (1, 2, 3));
-/// # });
-/// ```
-pub fn join3<Fut1, Fut2, Fut3>(
-    future1: Fut1,
-    future2: Fut2,
-    future3: Fut3,
-) -> Join3<Fut1, Fut2, Fut3>
-where
-    Fut1: Future,
-    Fut2: Future,
-    Fut3: Future,
-{
-    let f = Join3::new(future1, future2, future3);
-    assert_future::<(Fut1::Output, Fut2::Output, Fut3::Output), _>(f)
-}
-
-/// Same as [`join`](join()), but with more futures.
-///
-/// # Examples
-///
-/// ```
-/// # futures::executor::block_on(async {
-/// use futures::future;
-///
-/// let a = async { 1 };
-/// let b = async { 2 };
-/// let c = async { 3 };
-/// let d = async { 4 };
-/// let tuple = future::join4(a, b, c, d);
-///
-/// assert_eq!(tuple.await, (1, 2, 3, 4));
-/// # });
-/// ```
-pub fn join4<Fut1, Fut2, Fut3, Fut4>(
-    future1: Fut1,
-    future2: Fut2,
-    future3: Fut3,
-    future4: Fut4,
-) -> Join4<Fut1, Fut2, Fut3, Fut4>
-where
-    Fut1: Future,
-    Fut2: Future,
-    Fut3: Future,
-    Fut4: Future,
-{
-    let f = Join4::new(future1, future2, future3, future4);
-    assert_future::<(Fut1::Output, Fut2::Output, Fut3::Output, Fut4::Output), _>(f)
-}
-
-/// Same as [`join`](join()), but with more futures.
-///
-/// # Examples
-///
-/// ```
-/// # futures::executor::block_on(async {
-/// use futures::future;
-///
-/// let a = async { 1 };
-/// let b = async { 2 };
-/// let c = async { 3 };
-/// let d = async { 4 };
-/// let e = async { 5 };
-/// let tuple = future::join5(a, b, c, d, e);
-///
-/// assert_eq!(tuple.await, (1, 2, 3, 4, 5));
-/// # });
-/// ```
-pub fn join5<Fut1, Fut2, Fut3, Fut4, Fut5>(
-    future1: Fut1,
-    future2: Fut2,
-    future3: Fut3,
-    future4: Fut4,
-    future5: Fut5,
-) -> Join5<Fut1, Fut2, Fut3, Fut4, Fut5>
-where
-    Fut1: Future,
-    Fut2: Future,
-    Fut3: Future,
-    Fut4: Future,
-    Fut5: Future,
-{
-    let f = Join5::new(future1, future2, future3, future4, future5);
-    assert_future::<
-        (
-            Fut1::Output,
-            Fut2::Output,
-            Fut3::Output,
-            Fut4::Output,
-            Fut5::Output,
-        ),
-        _,
-    >(f)
-}

futures-util/src/future/mod.rs

@@ -72,7 +72,7 @@ mod ready;
 pub use self::ready::{err, ok, ready, Ready};
 
 mod join;
-pub use self::join::{join, join3, join4, join5, Join, Join3, Join4, Join5};
+pub use self::join::{join, Join};
 
 #[cfg(feature = "alloc")]
 mod join_all;
@@ -88,9 +88,7 @@ mod select_all;
 pub use self::select_all::{select_all, SelectAll};
 
 mod try_join;
-pub use self::try_join::{
-    try_join, try_join3, try_join4, try_join5, TryJoin, TryJoin3, TryJoin4, TryJoin5,
-};
+pub use self::try_join::{try_join, TryJoin};
 
 #[cfg(feature = "alloc")]
 mod try_join_all;

futures-util/src/future/select_all.rs

@@ -42,6 +42,13 @@ pub fn select_all<I>(iter: I) -> SelectAll<I::Item>
     assert_future::<(<I::Item as Future>::Output, usize, Vec<I::Item>), _>(ret)
 }
 
+impl<Fut> SelectAll<Fut> {
+    /// Consumes this combinator, returning the underlying futures.
+    pub fn into_inner(self) -> Vec<Fut> {
+        self.inner
+    }
+}
+
 impl<Fut: Future + Unpin> Future for SelectAll<Fut> {
     type Output = (Fut::Output, usize, Vec<Fut>);
 

futures-util/src/future/try_join.rs

@@ -92,15 +92,6 @@ macro_rules! generate {
 generate! {
     /// Future for the [`try_join`](try_join()) function.
     (TryJoin, <Fut1, Fut2>),
-
-    /// Future for the [`try_join3`] function.
-    (TryJoin3, <Fut1, Fut2, Fut3>),
-
-    /// Future for the [`try_join4`] function.
-    (TryJoin4, <Fut1, Fut2, Fut3, Fut4>),
-
-    /// Future for the [`try_join5`] function.
-    (TryJoin5, <Fut1, Fut2, Fut3, Fut4, Fut5>),
 }
 
 /// Joins the result of two futures, waiting for them both to complete or
@@ -152,105 +143,3 @@ where
 {
     assert_future::<Result<(Fut1::Ok, Fut2::Ok), Fut1::Error>, _>(TryJoin::new(future1, future2))
 }
-
-/// Same as [`try_join`](try_join()), but with more futures.
-///
-/// # Examples
-///
-/// ```
-/// # futures::executor::block_on(async {
-/// use futures::future;
-///
-/// let a = future::ready(Ok::<i32, i32>(1));
-/// let b = future::ready(Ok::<i32, i32>(2));
-/// let c = future::ready(Ok::<i32, i32>(3));
-/// let tuple = future::try_join3(a, b, c);
-///
-/// assert_eq!(tuple.await, Ok((1, 2, 3)));
-/// # });
-/// ```
-pub fn try_join3<Fut1, Fut2, Fut3>(
-    future1: Fut1,
-    future2: Fut2,
-    future3: Fut3,
-) -> TryJoin3<Fut1, Fut2, Fut3>
-where
-    Fut1: TryFuture,
-    Fut2: TryFuture<Error = Fut1::Error>,
-    Fut3: TryFuture<Error = Fut1::Error>,
-{
-    assert_future::<Result<(Fut1::Ok, Fut2::Ok, Fut3::Ok), Fut1::Error>, _>(TryJoin3::new(
-        future1, future2, future3,
-    ))
-}
-
-/// Same as [`try_join`](try_join()), but with more futures.
-///
-/// # Examples
-///
-/// ```
-/// # futures::executor::block_on(async {
-/// use futures::future;
-///
-/// let a = future::ready(Ok::<i32, i32>(1));
-/// let b = future::ready(Ok::<i32, i32>(2));
-/// let c = future::ready(Ok::<i32, i32>(3));
-/// let d = future::ready(Ok::<i32, i32>(4));
-/// let tuple = future::try_join4(a, b, c, d);
-///
-/// assert_eq!(tuple.await, Ok((1, 2, 3, 4)));
-/// # });
-/// ```
-pub fn try_join4<Fut1, Fut2, Fut3, Fut4>(
-    future1: Fut1,
-    future2: Fut2,
-    future3: Fut3,
-    future4: Fut4,
-) -> TryJoin4<Fut1, Fut2, Fut3, Fut4>
-where
-    Fut1: TryFuture,
-    Fut2: TryFuture<Error = Fut1::Error>,
-    Fut3: TryFuture<Error = Fut1::Error>,
-    Fut4: TryFuture<Error = Fut1::Error>,
-{
-    assert_future::<Result<(Fut1::Ok, Fut2::Ok, Fut3::Ok, Fut4::Ok), Fut1::Error>, _>(
-        TryJoin4::new(future1, future2, future3, future4),
-    )
-}
-
-/// Same as [`try_join`](try_join()), but with more futures.
-///
-/// # Examples
-///
-/// ```
-/// # futures::executor::block_on(async {
-/// use futures::future;
-///
-/// let a = future::ready(Ok::<i32, i32>(1));
-/// let b = future::ready(Ok::<i32, i32>(2));
-/// let c = future::ready(Ok::<i32, i32>(3));
-/// let d = future::ready(Ok::<i32, i32>(4));
-/// let e = future::ready(Ok::<i32, i32>(5));
-/// let tuple = future::try_join5(a, b, c, d, e);
-///
-/// assert_eq!(tuple.await, Ok((1, 2, 3, 4, 5)));
-/// # });
-/// ```
-pub fn try_join5<Fut1, Fut2, Fut3, Fut4, Fut5>(
-    future1: Fut1,
-    future2: Fut2,
-    future3: Fut3,
-    future4: Fut4,
-    future5: Fut5,
-) -> TryJoin5<Fut1, Fut2, Fut3, Fut4, Fut5>
-where
-    Fut1: TryFuture,
-    Fut2: TryFuture<Error = Fut1::Error>,
-    Fut3: TryFuture<Error = Fut1::Error>,
-    Fut4: TryFuture<Error = Fut1::Error>,
-    Fut5: TryFuture<Error = Fut1::Error>,
-{
-    assert_future::<Result<(Fut1::Ok, Fut2::Ok, Fut3::Ok, Fut4::Ok, Fut5::Ok), Fut1::Error>, _>(
-        TryJoin5::new(future1, future2, future3, future4, future5),
-    )
-}

futures-util/src/io/mod.rs

@@ -601,6 +601,17 @@ pub trait AsyncSeekExt: AsyncSeek {
     {
         assert_future::<Result<u64>, _>(Seek::new(self, pos))
     }
+
+    /// Creates a future which will return the current seek position from the
+    /// start of the stream.
+    ///
+    /// This is equivalent to `self.seek(SeekFrom::Current(0))`.
+    fn stream_position(&mut self) -> Seek<'_, Self>
+    where
+        Self: Unpin,
+    {
+        self.seek(SeekFrom::Current(0))
+    }
 }
 
 impl<S: AsyncSeek + ?Sized> AsyncSeekExt for S {}