perf(providers): replace wake_by_ref with loop
#![allow(clippy::return_self_not_must_use)]
use crate::{JsonRpcClient, Middleware, PinBoxFut, Provider, ProviderError};
-
use ethers_core::types::{Transaction, TxHash, U256};
-
use futures_core::{stream::Stream, Future};
use futures_util::{stream, stream::FuturesUnordered, FutureExt, StreamExt};
use pin_project::pin_project;
@@ -37,8 +35,8 @@ enum FilterWatcherState<'a, R> {
}
#[must_use = "filters do nothing unless you stream them"]
-#[pin_project]
/// Streams data from an installed filter via `eth_getFilterChanges`
+#[pin_project]
pub struct FilterWatcher<'a, P, R> {
/// The filter's installed id on the ethereum node
pub id: U256,
@@ -47,7 +45,7 @@ pub struct FilterWatcher<'a, P, R> {
// The polling interval
interval: Box<dyn Stream<Item = ()> + Send + Unpin>,
-
+ /// statemachine driven by the Stream impl
state: FilterWatcherState<'a, R>,
}
@@ -79,8 +77,7 @@ where
}
}
-// Pattern for flattening the returned Vec of filter changes taken from
-// https://github.com/tomusdrw/rust-web3/blob/f043b222744580bf4be043da757ab0b300c3b2da/src/api/eth_filter.rs#L50-L67
+// Advances the filter's state machine
impl<'a, P, R> Stream for FilterWatcher<'a, P, R>
where
P: JsonRpcClient,
@@ -89,16 +86,14 @@ where
type Item = R;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
- let this = self.project();
+ let mut this = self.project();
let id = *this.id;
- *this.state = match this.state {
+ loop {
+ *this.state = match &mut this.state {
FilterWatcherState::WaitForInterval => {
// Wait the polling period
let _ready = futures_util::ready!(this.interval.poll_next_unpin(cx));
-
- // create a new instance of the future
- cx.waker().wake_by_ref();
let fut = Box::pin(this.provider.get_filter_changes(id));
FilterWatcherState::GetFilterChanges(fut)
}
@@ -107,8 +102,8 @@ where
// vector. Should we make this return a Result instead? Ideally if we're
// in a streamed loop we wouldn't want the loop to terminate if an error
// is encountered (since it might be a temporary error).
- let items: Vec<R> = futures_util::ready!(fut.as_mut().poll(cx)).unwrap_or_default();
- cx.waker().wake_by_ref();
+ let items: Vec<R> =
+ futures_util::ready!(fut.as_mut().poll(cx)).unwrap_or_default();
FilterWatcherState::NextItem(items.into_iter())
}
// Consume 1 element from the vector. If more elements are in the vector,
@@ -116,15 +111,13 @@ where
// filter changes again. Once the whole vector is consumed, it will poll again
// for new logs
FilterWatcherState::NextItem(iter) => {
- cx.waker().wake_by_ref();
- match iter.next() {
- Some(item) => return Poll::Ready(Some(item)),
- None => FilterWatcherState::WaitForInterval,
+ if let item @ Some(_) = iter.next() {
+ return Poll::Ready(item)
}
+ FilterWatcherState::WaitForInterval
}
};
-
- Poll::Pending
+ }
}
}
