Merge pull request #478 from portgasd666/master

Add Future::join and Future::try_join

Author: yoshuawuyts

src/future/future/join.rs

@@ -0,0 +1,62 @@
+use std::pin::Pin;
+
+use async_macros::MaybeDone;
+use pin_project_lite::pin_project;
+
+use crate::task::{Context, Poll};
+use std::future::Future;
+
+pin_project! {
+    #[allow(missing_docs)]
+    #[allow(missing_debug_implementations)]
+    pub struct Join<L, R>
+    where
+        L: Future,
+        R: Future<Output = L::Output>
+    {
+        #[pin] left: MaybeDone<L>,
+        #[pin] right: MaybeDone<R>,
+    }
+}
+
+impl<L, R> Join<L, R>
+where
+    L: Future,
+    R: Future<Output = L::Output>,
+{
+    pub(crate) fn new(left: L, right: R) -> Self {
+        Self {
+            left: MaybeDone::new(left),
+            right: MaybeDone::new(right),
+        }
+    }
+}
+
+impl<L, R> Future for Join<L, R>
+where
+    L: Future,
+    R: Future<Output = L::Output>,
+{
+    type Output = (L::Output, R::Output);
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+
+        let mut left = this.left;
+        let mut right = this.right;
+
+        if Future::poll(Pin::new(&mut left), cx).is_ready() {
+            if right.as_ref().output().is_some() {
+                return Poll::Ready((left.take().unwrap(), right.take().unwrap()));
+            }
+        }
+
+        if Future::poll(Pin::new(&mut right), cx).is_ready() {
+            if left.as_ref().output().is_some() {
+                return Poll::Ready((left.take().unwrap(), right.take().unwrap()));
+            }
+        }
+
+        Poll::Pending
+    }
+}

src/future/future/mod.rs

@@ -3,6 +3,8 @@ cfg_unstable! {
     mod flatten;
     mod race;
     mod try_race;
+    mod join;
+    mod try_join;
 
     use std::time::Duration;
 
@@ -11,6 +13,8 @@ cfg_unstable! {
     use crate::future::IntoFuture;
     use race::Race;
     use try_race::TryRace;
+    use join::Join;
+    use try_join::TryJoin;
 }
 
 extension_trait! {
@@ -264,6 +268,90 @@ extension_trait! {
         {
             TryRace::new(self, other)
         }
+
+        #[doc = r#"
+            Waits for two similarly-typed futures to complete.
+
+            Awaits multiple futures simultaneously, returning the output of the
+            futures once both complete.
+
+            This function returns a new future which polls both futures
+            concurrently.
+
+            # Examples
+
+            ```
+            # async_std::task::block_on(async {
+            use async_std::prelude::*;
+            use async_std::future;
+
+            let a = future::ready(1u8);
+            let b = future::ready(2u8);
+
+            let f = a.join(b);
+            assert_eq!(f.await, (1u8, 2u8));
+            # });
+            ```
+        "#]
+        #[cfg(any(feature = "unstable", feature = "docs"))]
+        #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+        fn join<F>(
+            self,
+            other: F
+        ) -> impl Future<Output = (<Self as std::future::Future>::Output, <F as std::future::Future>::Output)> [Join<Self, F>]
+        where
+            Self: std::future::Future + Sized,
+            F: std::future::Future<Output = <Self as std::future::Future>::Output>,
+        {
+            Join::new(self, other)
+        }
+
+        #[doc = r#"
+            Waits for two similarly-typed fallible futures to complete.
+
+            Awaits multiple futures simultaneously, returning all results once
+            complete.
+
+            `try_join` is similar to [`join`], but returns an error immediately
+            if a future resolves to an error.
+
+            [`join`]: #method.join
+
+            # Examples
+
+            ```
+            # fn main() -> std::io::Result<()> { async_std::task::block_on(async {
+            #
+            use async_std::prelude::*;
+            use async_std::future;
+
+            let a = future::ready(Err("Error"));
+            let b = future::ready(Ok(1u8));
+
+            let f = a.try_join(b);
+            assert_eq!(f.await, Err("Error"));
+
+            let a = future::ready(Ok::<u8, String>(1u8));
+            let b = future::ready(Ok::<u8, String>(2u8));
+
+            let f = a.try_join(b);
+            assert_eq!(f.await, Ok((1u8, 2u8)));
+            #
+            # Ok(()) }) }
+            ```
+        "#]
+        #[cfg(any(feature = "unstable", feature = "docs"))]
+        #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+        fn try_join<F, T, E>(
+            self,
+            other: F
+        ) -> impl Future<Output = Result<(T, T), E>> [TryJoin<Self, F>]
+        where
+            Self: std::future::Future<Output = Result<T, E>> + Sized,
+            F: std::future::Future<Output = <Self as std::future::Future>::Output>,
+        {
+            TryJoin::new(self, other)
+        }
     }
 
     impl<F: Future + Unpin + ?Sized> Future for Box<F> {

src/future/future/try_join.rs

@@ -0,0 +1,72 @@
+use std::pin::Pin;
+
+use async_macros::MaybeDone;
+use pin_project_lite::pin_project;
+
+use crate::task::{Context, Poll};
+use std::future::Future;
+
+pin_project! {
+    #[allow(missing_docs)]
+    #[allow(missing_debug_implementations)]
+    pub struct TryJoin<L, R>
+    where
+        L: Future,
+        R: Future<Output = L::Output>
+    {
+        #[pin] left: MaybeDone<L>,
+        #[pin] right: MaybeDone<R>,
+    }
+}
+
+impl<L, R> TryJoin<L, R>
+where
+    L: Future,
+    R: Future<Output = L::Output>,
+{
+    pub(crate) fn new(left: L, right: R) -> Self {
+        Self {
+            left: MaybeDone::new(left),
+            right: MaybeDone::new(right),
+        }
+    }
+}
+
+impl<L, R, T, E> Future for TryJoin<L, R>
+where
+    L: Future<Output = Result<T, E>>,
+    R: Future<Output = L::Output>,
+{
+    type Output = Result<(T, T), E>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+
+        let mut left = this.left;
+        let mut right = this.right;
+
+        if Future::poll(Pin::new(&mut left), cx).is_ready() {
+            if left.as_ref().output().unwrap().is_err() {
+                return Poll::Ready(Err(left.take().unwrap().err().unwrap()));
+            } else if right.as_ref().output().is_some() {
+                return Poll::Ready(Ok((
+                    left.take().unwrap().ok().unwrap(),
+                    right.take().unwrap().ok().unwrap(),
+                )));
+            }
+        }
+
+        if Future::poll(Pin::new(&mut right), cx).is_ready() {
+            if right.as_ref().output().unwrap().is_err() {
+                return Poll::Ready(Err(right.take().unwrap().err().unwrap()));
+            } else if left.as_ref().output().is_some() {
+                return Poll::Ready(Ok((
+                    left.take().unwrap().ok().unwrap(),
+                    right.take().unwrap().ok().unwrap(),
+                )));
+            }
+        }
+
+        Poll::Pending
+    }
+}

src/future/mod.rs

@@ -8,17 +8,17 @@
 //! operations converts multiple future into a single future that returns the
 //! first output.
 //!
-//! For operating on futures the following macros can be used:
+//! For operating on futures the following functions can be used:
 //!
 //! | Name               | Return signature | When does it return?     |
 //! | ---                | ---              | ---                      |
-//! | [`future::join!`]  | `(T1, T2)`       | Wait for all to complete
+//! | [`Future::join`]   | `(T1, T2)`       | Wait for all to complete
 //! | [`Future::race`]   | `T`              | Return on first value
 //!
 //! ## Fallible Futures Concurrency
 //!
 //! For operating on futures that return `Result` additional `try_` variants of
-//! the macros mentioned before can be used. These macros are aware of `Result`,
+//! the functions mentioned before can be used. These functions are aware of `Result`,
 //! and will behave slightly differently from their base variants.
 //!
 //! In the case of `try_join`, if any of the futures returns `Err` all
@@ -30,19 +30,19 @@
 //! means `try_race` will keep going until any one of the futures returns
 //! `Ok`, or _all_ futures have returned `Err`.
 //!
-//! However sometimes it can be useful to use the base variants of the macros
+//! However sometimes it can be useful to use the base variants of the functions
 //! even on futures that return `Result`. Here is an overview of operations that
 //! work on `Result`, and their respective semantics:
 //!
 //! | Name                   | Return signature               | When does it return? |
 //! | ---                    | ---                            | ---                  |
-//! | [`future::join!`]      | `(Result<T, E>, Result<T, E>)` | Wait for all to complete
-//! | [`future::try_join!`]  | `Result<(T1, T2), E>`          | Return on first `Err`, wait for all to complete
+//! | [`Future::join`]       | `(Result<T, E>, Result<T, E>)` | Wait for all to complete
+//! | [`Future::try_join`]   | `Result<(T1, T2), E>`          | Return on first `Err`, wait for all to complete
 //! | [`Future::race`]       | `Result<T, E>`                 | Return on first value
 //! | [`Future::try_race`]   | `Result<T, E>`                 | Return on first `Ok`, reject on last Err
 //!
-//! [`future::join!`]: macro.join.html
-//! [`future::try_join!`]: macro.try_join.html
+//! [`Future::join`]: trait.Future.html#method.join
+//! [`Future::try_join`]: trait.Future.html#method.try_join
 //! [`Future::race`]: trait.Future.html#method.race
 //! [`Future::try_race`]: trait.Future.html#method.try_race