Port futures_util::future::join_all to the new API

Author: tomhoule

futures-util/src/future/disabled/join_all.rs

@@ -0,0 +1,121 @@
+//! Definition of the `JoinAll` combinator, waiting for all of a list of futures
+//! to finish.
+
+use std::fmt;
+use std::future::Future;
+use std::iter::FromIterator;
+use std::mem;
+use std::pin::{Pin, Unpin};
+use std::prelude::v1::*;
+use std::task::Poll;
+
+#[derive(Debug)]
+enum ElemState<F>
+where
+    F: Future,
+{
+    Pending(F),
+    Done(F::Output),
+}
+
+/// A future which takes a list of futures and resolves with a vector of the
+/// completed values.
+///
+/// This future is created with the `join_all` method.
+#[must_use = "futures do nothing unless polled"]
+pub struct JoinAll<F>
+where
+    F: Future,
+{
+    elems: Vec<ElemState<F>>,
+}
+
+impl<F> fmt::Debug for JoinAll<F>
+where
+    F: Future + fmt::Debug,
+    F::Output: fmt::Debug,
+{
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        fmt.debug_struct("JoinAll")
+            .field("elems", &self.elems)
+            .finish()
+    }
+}
+
+/// Creates a future which represents a collection of the outputs of the futures
+/// given.
+///
+/// The returned future will drive execution for all of its underlying futures,
+/// collecting the results into a destination `Vec<T>` in the same order as they
+/// were provided.
+///
+/// # Examples
+///
+/// ```
+/// use futures_util::future::{FutureExt, join_all, ready};
+///
+/// let f = join_all(vec![
+///     ready::<u32>(1),
+///     ready::<u32>(2),
+///     ready::<u32>(3),
+/// ]);
+/// let f = f.map(|x| {
+///     assert_eq!(x, [1, 2, 3]);
+/// });
+/// ```
+pub fn join_all<I>(i: I) -> JoinAll<I::Item>
+where
+    I: IntoIterator,
+    I::Item: Future,
+{
+    let elems = i.into_iter().map(ElemState::Pending).collect();
+    JoinAll { elems }
+}
+
+impl<F> Future for JoinAll<F>
+where
+    F: Future + Unpin,
+    F::Output: Unpin,
+{
+    type Output = Vec<F::Output>;
+
+    fn poll(
+        mut self: Pin<&mut Self>,
+        lw: &::std::task::LocalWaker,
+    ) -> Poll<Self::Output> {
+        let mut all_done = true;
+
+        for elem in self.as_mut().elems.iter_mut() {
+            match elem {
+                ElemState::Pending(ref mut t) => match Pin::new(t).poll(lw) {
+                    Poll::Ready(v) => *elem = ElemState::Done(v),
+                    Poll::Pending => {
+                        all_done = false;
+                        continue;
+                    }
+                },
+                ElemState::Done(ref mut _v) => (),
+            };
+        }
+
+        if all_done {
+            let elems = mem::replace(&mut self.elems, Vec::new());
+            let result = elems
+                .into_iter()
+                .map(|e| match e {
+                    ElemState::Done(t) => t,
+                    _ => unreachable!(),
+                })
+                .collect();
+            Poll::Ready(result)
+        } else {
+            Poll::Pending
+        }
+    }
+}
+
+impl<F: Future> FromIterator<F> for JoinAll<F> {
+    fn from_iter<T: IntoIterator<Item = F>>(iter: T) -> Self {
+        join_all(iter)
+    }
+}

futures-util/src/future/join_all.rs

@@ -79,11 +79,11 @@ mod remote_handle;
 #[cfg(feature = "std")]
 pub use self::remote_handle::{Remote, RemoteHandle};
 
-// ToDo
-// #[cfg(feature = "std")]
-// mod join_all;
-// #[cfg(feature = "std")]
-// pub use self::join_all::{join_all, JoinAll};
+#[cfg(feature = "std")]
+mod join_all;
+
+#[cfg(feature = "std")]
+pub use self::join_all::{join_all, JoinAll};
 
 // #[cfg(feature = "std")]
 // mod select_all;

futures-util/src/future/mod.rs

@@ -0,0 +1,46 @@
+#![feature(async_await, pin, arbitrary_self_types, futures_api)]
+
+extern crate futures_util;
+extern crate futures;
+
+use futures_util::future::*;
+use std::future::Future;
+use futures::executor::block_on;
+use std::fmt::Debug;
+use std::pin::Unpin;
+
+fn assert_done<T: PartialEq + Debug, F: FnOnce() -> Box<Future<Output=T> + Unpin>>(actual_fut: F, expected: T) {
+    let output = block_on(actual_fut());
+
+    assert_eq!(output, expected);
+}
+
+#[test]
+fn collect_collects() {
+    assert_done(|| Box::new(join_all(vec![ready(1), ready(2)])), vec![1, 2]);
+    assert_done(|| Box::new(join_all(vec![ready(1)])), vec![1]);
+    // REVIEW: should this be implemented?
+    // assert_done(|| Box::new(join_all(Vec::<i32>::new())), vec![]);
+
+    // TODO: needs more tests
+}
+
+#[test]
+fn join_all_iter_lifetime() {
+    // In futures-rs version 0.1, this function would fail to typecheck due to an overly
+    // conservative type parameterization of `JoinAll`.
+    fn sizes<'a>(bufs: Vec<&'a [u8]>) -> Box<Future<Output=Vec<usize>> + Unpin> {
+        let iter = bufs.into_iter().map(|b| ready::<usize>(b.len()));
+        Box::new(join_all(iter))
+    }
+
+    assert_done(|| sizes(vec![&[1,2,3], &[], &[0]]), vec![3 as usize, 0, 1]);
+}
+
+#[test]
+fn join_all_from_iter() {
+    assert_done(
+        || Box::new(vec![ready(1), ready(2)].into_iter().collect::<JoinAll<_>>()),
+        vec![1, 2],
+    )
+}

futures/tests/join_all.rs

@@ -243,34 +243,6 @@ fn join_incomplete() {
     })
 }
 
-#[test]
-fn collect_collects() {
-    assert_done(|| join_all(vec![f_ok(1), f_ok(2)]), Ok(vec![1, 2]));
-    assert_done(|| join_all(vec![f_ok(1)]), Ok(vec![1]));
-    assert_done(|| join_all(Vec::<Result<i32, u32>>::new()), Ok(vec![]));
-
-    // TODO: needs more tests
-}
-
-#[test]
-fn join_all_iter_lifetime() {
-    // In futures-rs version 0.1, this function would fail to typecheck due to an overly
-    // conservative type parameterization of `JoinAll`.
-    fn sizes<'a>(bufs: Vec<&'a [u8]>) -> Box<Future<Item=Vec<usize>, Error=()> + 'static> {
-        let iter = bufs.into_iter().map(|b| future::ok::<usize, ()>(b.len()));
-        Box::new(join_all(iter))
-    }
-
-    assert_done(|| sizes(vec![&[1,2,3], &[], &[0]]), Ok(vec![3, 0, 1]));
-}
-
-#[test]
-fn join_all_from_iter() {
-    assert_done(
-        || vec![f_ok(1), f_ok(2)].into_iter().collect::<JoinAll<_>>(),
-        Ok(vec![1, 2]),
-    )
-}
 
 #[test]
 fn select2() {