futures 0.1.0

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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

use std::mem;

use {Future, IntoFuture, Task, Poll};
use util::Collapsed;

/// 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.
pub struct Collect<I>
    where I: IntoIterator + Send + 'static,
          I::Item: IntoFuture,
          I::IntoIter: Send + 'static,
{
    cur: Option<Collapsed<<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 + Send + 'static,
          I::Item: IntoFuture,
          I::IntoIter: Send + 'static,
{
    let mut i = i.into_iter();
    Collect {
        cur: i.next().map(IntoFuture::into_future).map(Collapsed::Start),
        remaining: i,
        result: Vec::new(),
    }
}

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


    fn poll(&mut self, task: &mut Task) -> Poll<Self::Item, Self::Error> {
        let mut task = task.scoped();
        loop {
            match self.cur {
                Some(ref mut cur) => {
                    match try_poll!(cur.poll(&mut task)) {
                        Ok(e) => self.result.push(e),

                        // 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 Poll::Err(e)
                        }
                    }
                }
                None => {
                    return Poll::Ok(mem::replace(&mut self.result, Vec::new()))
                }
            }

            task.ready();

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

    fn schedule(&mut self, task: &mut Task) {
        if let Some(ref mut cur) = self.cur {
            cur.schedule(task);
        }
    }

    fn tailcall(&mut self)
                -> Option<Box<Future<Item=Self::Item, Error=Self::Error>>> {
        if let Some(ref mut cur) = self.cur {
            cur.collapse();
        }
        None
    }

}