ethereum_adapter: Turn `logs_with_sigs` into a closure
@@ -114,43 +114,6 @@ where
})
}
- fn logs_with_sigs(
- &self,
- logger: &Logger,
- from: u64,
- to: u64,
- addresses: Vec<H160>,
- event_signatures: Vec<H256>,
- ) -> impl Future<Item = Vec<Log>, Error = graph::tokio_timer::timeout::Error<web3::error::Error>>
- {
- let eth_adapter = self.clone();
- let logger = logger.to_owned();
-
- retry("eth_getLogs RPC call", &logger)
- .when(|res: &Result<_, web3::error::Error>| match res {
- Ok(_) => false,
- Err(e) => !e.to_string().contains("ServerError(-32005)"),
- })
- .no_limit()
- .timeout_secs(60)
- .run(move || {
- // Create a log filter
- let log_filter: Filter = FilterBuilder::default()
- .from_block(from.into())
- .to_block(to.into())
- .address(addresses.clone())
- .topics(Some(event_signatures.clone()), None, None, None)
- .build();
-
- // Request logs from client
- let logger = logger.clone();
- eth_adapter.web3.eth().logs(log_filter).map(move |logs| {
- debug!(logger, "Received logs for blocks [{}, {}]", from, to);
- logs
- })
- })
- }
-
fn trace_stream(
self,
logger: &Logger,
@@ -192,6 +155,37 @@ where
to: u64,
log_filter: EthereumLogFilter,
) -> impl Future<Item = Vec<Log>, Error = Error> {
+ let eth_get_logs = move |eth_adapter: Self,
+ logger,
+ from: u64,
+ to: u64,
+ addresses: Vec<H160>,
+ event_signatures: Vec<H256>| {
+ retry("eth_getLogs RPC call", &logger)
+ .when(|res: &Result<_, web3::error::Error>| match res {
+ Ok(_) => false,
+ Err(e) => !e.to_string().contains("ServerError(-32005)"),
+ })
+ .no_limit()
+ .timeout_secs(60)
+ .run(move || {
+ // Create a log filter
+ let log_filter: Filter = FilterBuilder::default()
+ .from_block(from.into())
+ .to_block(to.into())
+ .address(addresses.clone())
+ .topics(Some(event_signatures.clone()), None, None, None)
+ .build();
+
+ // Request logs from client
+ let logger = logger.clone();
+ eth_adapter.web3.eth().logs(log_filter).map(move |logs| {
+ debug!(logger, "Received logs for blocks [{}, {}]", from, to);
+ logs
+ })
+ })
+ };
+
if from > to {
panic!(
"cannot produce a log stream on a backwards block range (from={}, to={})",
@@ -255,7 +249,14 @@ where
let logger = logger.clone();
let log_filter = log_filter.clone();
Some(
- eth.logs_with_sigs(&logger, start, end, addresses.clone(), event_sigs.clone())
+ eth_get_logs(
+ eth.clone(),
+ logger.clone(),
+ start,
+ end,
+ addresses.clone(),
+ event_sigs.clone(),
+ )
.map(move |logs| {
logs.into_iter()
.filter(move |log| log_filter.matches(log))
