Merge pull request #619 from srijs/feat/futures-unordered-from-iterator

Implement FromIterator for FuturesUnordered/FuturesOrdered

Author: alexcrichton

src/stream/futures_ordered.rs

@@ -1,6 +1,7 @@
 use std::cmp::{Eq, PartialEq, PartialOrd, Ord, Ordering};
 use std::collections::BinaryHeap;
 use std::fmt::{self, Debug};
+use std::iter::FromIterator;
 
 use {Async, Future, IntoFuture, Poll, Stream};
 use stream::FuturesUnordered;
@@ -198,3 +199,15 @@ impl<T: Debug> Debug for FuturesOrdered<T>
         write!(fmt, "FuturesOrdered {{ ... }}")
     }
 }
+
+impl<F: Future> FromIterator<F> for FuturesOrdered<F> {
+    fn from_iter<T>(iter: T) -> Self 
+        where T: IntoIterator<Item = F>
+    {
+        let mut new = FuturesOrdered::new();
+        for future in iter.into_iter() {
+            new.push(future);
+        }
+        new
+    }
+}

src/stream/futures_unordered.rs

@@ -1,5 +1,6 @@
 use std::cell::UnsafeCell;
 use std::fmt::{self, Debug};
+use std::iter::FromIterator;
 use std::marker::PhantomData;
 use std::mem;
 use std::ptr;
@@ -427,6 +428,18 @@ impl<T> Drop for FuturesUnordered<T> {
     }
 }
 
+impl<F: Future> FromIterator<F> for FuturesUnordered<F> {
+    fn from_iter<T>(iter: T) -> Self 
+        where T: IntoIterator<Item = F>
+    {
+        let mut new = FuturesUnordered::new();
+        for future in iter.into_iter() {
+            new.push(future);
+        }
+        new
+    }
+}
+
 impl<T> Inner<T> {
     /// The enqueue function from the 1024cores intrusive MPSC queue algorithm.
     fn enqueue(&self, node: *const Node<T>) {

tests/futures_ordered.rs

@@ -48,6 +48,20 @@ fn works_2() {
     assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_ready());
 }
 
+#[test]
+fn from_iterator() {
+    use futures::future::ok;
+    use futures::stream::FuturesOrdered;
+
+    let stream = vec![
+        ok::<u32, ()>(1),
+        ok::<u32, ()>(2),
+        ok::<u32, ()>(3)
+    ].into_iter().collect::<FuturesOrdered<_>>();
+    assert_eq!(stream.len(), 3);
+    assert_eq!(stream.collect().wait(), Ok(vec![1,2,3]));
+}
+
 #[test]
 fn queue_never_unblocked() {
     let (_a_tx, a_rx) = oneshot::channel::<Box<Any+Send>>();

tests/futures_unordered.rs

@@ -46,6 +46,20 @@ fn works_2() {
     assert!(spawn.poll_stream_notify(&support::notify_noop(), 0).unwrap().is_ready());
 }
 
+#[test]
+fn from_iterator() {
+    use futures::future::ok;
+    use futures::stream::FuturesUnordered;
+
+    let stream = vec![
+        ok::<u32, ()>(1),
+        ok::<u32, ()>(2),
+        ok::<u32, ()>(3)
+    ].into_iter().collect::<FuturesUnordered<_>>();
+    assert_eq!(stream.len(), 3);
+    assert_eq!(stream.collect().wait(), Ok(vec![1,2,3]));
+}
+
 #[test]
 fn finished_future_ok() {
     let (_a_tx, a_rx) = oneshot::channel::<Box<Any+Send>>();