      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
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    190
    191
    192
    193
    194
    195
    196
    197
    198
    199
    200
    201
    202
    203
    204
    205
    206
    207
    208
    209
    210
    211
    212
    213

    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;

    #[derive(Debug)]
    struct OrderWrapper<T> {
        item: T,
        index: usize,
    }

    impl<T> PartialEq for OrderWrapper<T> {
        fn eq(&self, other: &Self) -> bool {
            self.index == other.index
        }
    }

    impl<T> Eq for OrderWrapper<T> {}

    impl<T> PartialOrd for OrderWrapper<T> {
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
            Some(self.cmp(other))
        }
    }

    impl<T> Ord for OrderWrapper<T> {
        fn cmp(&self, other: &Self) -> Ordering {
            // BinaryHeap is a max heap, so compare backwards here.
            other.index.cmp(&self.index)
        }
    }

    impl<T> Future for OrderWrapper<T>
        where T: Future
    {
        type Item = OrderWrapper<T::Item>;
        type Error = T::Error;

        fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
            let result = try_ready!(self.item.poll());
            Ok(Async::Ready(OrderWrapper {
                item: result,
                index: self.index
            }))
        }
    }

    /// An unbounded queue of futures.
    ///
    /// This "combinator" is similar to `FuturesUnordered`, but it imposes an order
    /// on top of the set of futures. While futures in the set will race to
    /// completion in parallel, results will only be returned in the order their
    /// originating futures were added to the queue.
    ///
    /// Futures are pushed into this queue and their realized values are yielded in
    /// order. This structure is optimized to manage a large number of futures.
    /// Futures managed by `FuturesOrdered` will only be polled when they generate
    /// notifications. This reduces the required amount of work needed to coordinate
    /// large numbers of futures.
    ///
    /// When a `FuturesOrdered` is first created, it does not contain any futures.
    /// Calling `poll` in this state will result in `Ok(Async::Ready(None))` to be
    /// returned. Futures are submitted to the queue using `push`; however, the
    /// future will **not** be polled at this point. `FuturesOrdered` will only
    /// poll managed futures when `FuturesOrdered::poll` is called. As such, it
    /// is important to call `poll` after pushing new futures.
    ///
    /// If `FuturesOrdered::poll` returns `Ok(Async::Ready(None))` this means that
    /// the queue is currently not managing any futures. A future may be submitted
    /// to the queue at a later time. At that point, a call to
    /// `FuturesOrdered::poll` will either return the future's resolved value
    /// **or** `Ok(Async::NotReady)` if the future has not yet completed. When
    /// multiple futures are submitted to the queue, `FuturesOrdered::poll` will
    /// return `Ok(Async::NotReady)` until the first future completes, even if
    /// some of the later futures have already completed.
    ///
    /// Note that you can create a ready-made `FuturesOrdered` via the
    /// `futures_ordered` function in the `stream` module, or you can start with an
    /// empty queue with the `FuturesOrdered::new` constructor.
    #[must_use = "streams do nothing unless polled"]
    pub struct FuturesOrdered<T>
        where T: Future
    {
        in_progress: FuturesUnordered<OrderWrapper<T>>,
        queued_results: BinaryHeap<OrderWrapper<T::Item>>,
        next_incoming_index: usize,
        next_outgoing_index: usize,
    }

    /// Converts a list of futures into a `Stream` of results from the futures.
    ///
    /// This function will take an list of futures (e.g. a vector, an iterator,
    /// etc), and return a stream. The stream will yield items as they become
    /// available on the futures internally, in the order that their originating
    /// futures were submitted to the queue. If the futures complete out of order,
    /// items will be stored internally within `FuturesOrdered` until all preceding
    /// items have been yielded.
    ///
    /// Note that the returned queue can also be used to dynamically push more
    /// futures into the queue as they become available.
    pub fn futures_ordered<I>(futures: I) -> FuturesOrdered<<I::Item as IntoFuture>::Future>
        where I: IntoIterator,
              I::Item: IntoFuture
    {
        let mut queue = FuturesOrdered::new();

        for future in futures {
            queue.push(future.into_future());
        }

        return queue
    }

    impl<T> FuturesOrdered<T>
        where T: Future
    {
        /// Constructs a new, empty `FuturesOrdered`
        ///
        /// The returned `FuturesOrdered` does not contain any futures and, in this
        /// state, `FuturesOrdered::poll` will return `Ok(Async::Ready(None))`.
        pub fn new() -> FuturesOrdered<T> {
            FuturesOrdered {
                in_progress: FuturesUnordered::new(),
                queued_results: BinaryHeap::new(),
                next_incoming_index: 0,
                next_outgoing_index: 0,
            }
        }

        /// Returns the number of futures contained in the queue.
        ///
        /// This represents the total number of in-flight futures, both
        /// those currently processing and those that have completed but
        /// which are waiting for earlier futures to complete.
        pub fn len(&self) -> usize {
            self.in_progress.len() + self.queued_results.len()
        }

        /// Returns `true` if the queue contains no futures
        pub fn is_empty(&self) -> bool {
            self.in_progress.is_empty() && self.queued_results.is_empty()
        }

        /// Push a future into the queue.
        ///
        /// This function submits the given future to the internal set for managing.
        /// This function will not call `poll` on the submitted future. The caller
        /// must ensure that `FuturesOrdered::poll` is called in order to receive
        /// task notifications.
        pub fn push(&mut self, future: T) {
            let wrapped = OrderWrapper {
                item: future,
                index: self.next_incoming_index,
            };
            self.next_incoming_index += 1;
            self.in_progress.push(wrapped);
        }
    }

    impl<T> Stream for FuturesOrdered<T>
        where T: Future
    {
        type Item = T::Item;
        type Error = T::Error;

        fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
            // Get any completed futures from the unordered set.
            loop {
                match self.in_progress.poll()? {
                    Async::Ready(Some(result)) => self.queued_results.push(result),
                    Async::Ready(None) | Async::NotReady => break,
                }
            }

            if let Some(next_result) = self.queued_results.peek() {
                // PeekMut::pop is not stable yet QQ
                if next_result.index != self.next_outgoing_index {
                    return Ok(Async::NotReady);
                }
            } else if !self.in_progress.is_empty() {
                return Ok(Async::NotReady);
            } else {
                return Ok(Async::Ready(None));
            }

            let next_result = self.queued_results.pop().unwrap();
            self.next_outgoing_index += 1;
            Ok(Async::Ready(Some(next_result.item)))
        }
    }

    impl<T: Debug> Debug for FuturesOrdered<T>
        where T: Future
    {
        fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
            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
        }
    }

# Help

## Keyboard Shortcuts

`?`
  Show this help dialog
`S`
  Focus the search field
`↑`
  Move up in search results
`↓`
  Move down in search results
`↹`
  Switch tab
`⏎`
  Go to active search result
`+`
  Expand all sections
`-`
  Collapse all sections

## Search Tricks

Prefix searches with a type followed by a colon (e.g. `fn:`) to restrict the search to a given type.

Accepted types are: `fn`, `mod`, `struct`, `enum`, `trait`, `type`, `macro`, and `const`.

Search functions by type signature (e.g. `vec -> usize` or `* -> vec`)