use futures::{Async, Future, IntoFuture, Poll, Stream};
use std::mem;
#[derive(Debug)]
#[must_use = "streams do nothing unless polled"]
pub struct FoldWhile<S, F, G, Fut, Gut, T>
where
Fut: IntoFuture,
Gut: IntoFuture,
{
stream: S,
f: F,
state: State<T, Fut::Future, Gut::Future>,
g: G,
}
#[derive(Debug)]
enum State<T, F, G>
where
F: Future,
G: Future,
{
Empty,
Ready(T),
Processing(F),
Checking(G),
}
pub fn new<S, F, G, Fut, Gut, T>(s: S, t: T, f: F, g: G) -> FoldWhile<S, F, G, Fut, Gut, T>
where
S: Stream,
F: FnMut(T, S::Item) -> Fut,
G: FnMut(T) -> Gut,
Fut: IntoFuture<Item = T>,
Gut: IntoFuture<Item = (bool, T)>,
S::Error: From<Fut::Error> + From<Gut::Error>,
{
FoldWhile {
stream: s,
f: f,
state: State::Ready(t),
g: g,
}
}
impl<S, F, G, Fut, Gut, T> Future for FoldWhile<S, F, G, Fut, Gut, T>
where
S: Stream,
F: FnMut(T, S::Item) -> Fut,
G: FnMut(T) -> Gut,
Fut: IntoFuture<Item = T>,
Gut: IntoFuture<Item = (bool, T)>,
S::Error: From<Fut::Error> + From<Gut::Error>,
{
type Item = T;
type Error = S::Error;
fn poll(&mut self) -> Poll<T, S::Error> {
loop {
match mem::replace(&mut self.state, State::Empty) {
State::Empty => panic!("cannot poll Fold twice"),
State::Ready(state) => match self.stream.poll()? {
Async::Ready(Some(e)) => {
let future = (self.f)(state, e);
let future = future.into_future();
self.state = State::Processing(future);
}
Async::Ready(None) => return Ok(Async::Ready(state)),
Async::NotReady => {
self.state = State::Ready(state);
return Ok(Async::NotReady);
}
},
State::Processing(mut fut) => match fut.poll()? {
Async::Ready(state) => {
self.state = State::Checking((self.g)(state).into_future())
}
Async::NotReady => {
self.state = State::Processing(fut);
return Ok(Async::NotReady);
}
},
State::Checking(mut gut) => match gut.poll()? {
Async::Ready((true, state)) => self.state = State::Ready(state),
Async::Ready((false, state)) => return Ok(Async::Ready(state)),
Async::NotReady => {
self.state = State::Checking(gut);
return Ok(Async::NotReady);
}
},
}
}
}
}