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 {
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
}))
}
}
#[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,
}
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> Default for FuturesOrdered<T> where T: Future {
fn default() -> Self {
FuturesOrdered::new()
}
}
impl<T> FuturesOrdered<T>
where T: Future
{
pub fn new() -> FuturesOrdered<T> {
FuturesOrdered {
in_progress: FuturesUnordered::new(),
queued_results: BinaryHeap::new(),
next_incoming_index: 0,
next_outgoing_index: 0,
}
}
pub fn len(&self) -> usize {
self.in_progress.len() + self.queued_results.len()
}
pub fn is_empty(&self) -> bool {
self.in_progress.is_empty() && self.queued_results.is_empty()
}
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> {
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() {
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
}
}