futures 0.1.3

An implementation of futures and streams featuring zero allocations, composability, and iterator-like interfaces.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

use std::prelude::v1::*;

use std::mem;

use {Future, IntoFuture, Poll, Async};

/// A future which takes a list of futures and resolves with a vector of the
/// completed values.
///
/// This future is created with the `collect` method.
#[must_use = "futures do nothing unless polled"]
pub struct Collect<I>
    where I: IntoIterator,
          I::Item: IntoFuture,
{
    cur: Option<<I::Item as IntoFuture>::Future>,
    remaining: I::IntoIter,
    result: Vec<<I::Item as IntoFuture>::Item>,
}

/// Creates a future which represents a collection of the results of the futures
/// given.
///
/// The returned future will execute each underlying future one at a time,
/// collecting the results into a destination `Vec<T>`. If any future returns
/// an error then all other futures will be canceled and an error will be
/// returned immediately. If all futures complete successfully, however, then
/// the returned future will succeed with a `Vec` of all the successful results.
///
/// Note that this function does **not** attempt to execute each future in
/// parallel, they are all executed in sequence.
///
/// # Examples
///
/// ```
/// use futures::*;
///
/// let f = collect(vec![
///     finished::<u32, u32>(1),
///     finished::<u32, u32>(2),
///     finished::<u32, u32>(3),
/// ]);
/// let f = f.map(|x| {
///     assert_eq!(x, [1, 2, 3]);
/// });
///
/// let f = collect(vec![
///     finished::<u32, u32>(1).boxed(),
///     failed::<u32, u32>(2).boxed(),
///     finished::<u32, u32>(3).boxed(),
/// ]);
/// let f = f.then(|x| {
///     assert_eq!(x, Err(2));
///     x
/// });
/// ```
pub fn collect<I>(i: I) -> Collect<I>
    where I: IntoIterator,
          I::Item: IntoFuture,
{
    let mut i = i.into_iter();
    Collect {
        cur: i.next().map(IntoFuture::into_future),
        remaining: i,
        result: Vec::new(),
    }
}

impl<I> Future for Collect<I>
    where I: IntoIterator,
          I::Item: IntoFuture,
{
    type Item = Vec<<I::Item as IntoFuture>::Item>;
    type Error = <I::Item as IntoFuture>::Error;


    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        loop {
            match self.cur {
                Some(ref mut cur) => {
                    match cur.poll() {
                        Ok(Async::Ready(e)) => self.result.push(e),
                        Ok(Async::NotReady) => return Ok(Async::NotReady),

                        // If we hit an error, drop all our associated resources
                        // ASAP.
                        Err(e) => {
                            for f in self.remaining.by_ref() {
                                drop(f);
                            }
                            for f in self.result.drain(..) {
                                drop(f);
                            }
                            return Err(e)
                        }
                    }
                }
                None => {
                    return Ok(Async::Ready(mem::replace(&mut self.result, Vec::new())))
                }
            }

            self.cur = self.remaining.next()
                           .map(IntoFuture::into_future);
        }
    }
}