Remove Fuse wrappers from events module
@@ -19,7 +19,6 @@ pub mod error;
pub mod network;
use futures::{Async, Poll, Stream};
-use futures::stream::Fuse;
use std::time::SystemTime;
use std::cmp::Ordering;
@@ -79,9 +78,10 @@ where
S2: Stream,
S3: Stream,
{
- timeout: Fuse<S1>,
- network: Fuse<S2>,
- api: Fuse<S3>,
+ done: bool,
+ timeout: S1,
+ network: S2,
+ api: S3,
}
impl<S1, S2, S3> EventsAggregator<S1, S2, S3>
@@ -92,9 +92,10 @@ where
{
pub fn new(timeout: S1, network: S2, api: S3) -> EventsAggregator<S1, S2, S3> {
EventsAggregator {
- network: network.fuse(),
- timeout: timeout.fuse(),
- api: api.fuse(),
+ done: false,
+ network: network,
+ timeout: timeout,
+ api: api,
}
}
}
@@ -115,33 +116,47 @@ where
type Error = S1::Error;
fn poll(&mut self) -> Poll<Option<Event>, Self::Error> {
- let mut stream_finished = false;
+ if self.done {
+ Ok(Async::Ready(None))
+ } else {
// Check timeout events
match self.timeout.poll()? {
- Async::Ready(Some(item)) => return Ok(Async::Ready(Some(Event::Timeout(item)))),
+ Async::Ready(Some(item)) => {
+ return Ok(Async::Ready(Some(Event::Timeout(item))));
+ }
// Just finish stream
- Async::Ready(None) => stream_finished = true,
+ Async::Ready(None) => {
+ self.done = true;
+ return Ok(Async::Ready(None));
+ }
Async::NotReady => {}
};
// Check network events
match self.network.poll()? {
- Async::Ready(Some(item)) => return Ok(Async::Ready(Some(Event::Network(item)))),
+ Async::Ready(Some(item)) => {
+ return Ok(Async::Ready(Some(Event::Network(item))));
+ }
// Just finish stream
- Async::Ready(None) => stream_finished = true,
+ Async::Ready(None) => {
+ self.done = true;
+ return Ok(Async::Ready(None));
+ }
Async::NotReady => {}
};
// Check api events
match self.api.poll()? {
- Async::Ready(Some(item)) => return Ok(Async::Ready(Some(Event::Api(item)))),
+ Async::Ready(Some(item)) => {
+ return Ok(Async::Ready(Some(Event::Api(item))));
+ }
// Just finish stream
- Async::Ready(None) => stream_finished = true,
+ Async::Ready(None) => {
+ self.done = true;
+ return Ok(Async::Ready(None));
+ }
Async::NotReady => {}
};
- Ok(if stream_finished {
- Async::Ready(None)
- } else {
- Async::NotReady
- })
+ Ok(Async::NotReady)
+ }
}
}
