instance_manager: Inline `process_triggers_in_runtime_hosts`
It's too small to exist by itself.
Add note to the loop.
@@ -459,6 +459,10 @@ where
// 1. Instantiate created data sources.
// 2. Process those data sources for the current block.
// Until no data sources are created or MAX_DATA_SOURCES is hit.
+
+ // Note that this algorithm processes data sources spawned on the same block _breadth
+ // first_ on the tree implied by the parent-child relationship between data sources. Only a
+ // very contrived subgraph would be able to observe this.
loop_fn(
(ctx, block_state),
move |(mut ctx, mut block_state)| -> Box<dyn Future<Item = _, Error = _> + Send> {
@@ -522,14 +526,21 @@ where
return Box::new(future::err(err.into()));
}
+ let logger = logger.clone();
+ let block = block.clone();
Box::new(
- process_triggers_in_runtime_hosts::<T>(
- logger.clone(),
- runtime_hosts,
- block_state,
+ stream::iter_ok(triggers)
+ .fold(block_state, move |block_state, trigger| {
+ // Process the triggers in each host in the same order the
+ // corresponding data sources have been created.
+ SubgraphInstance::<T>::process_trigger_in_runtime_hosts(
+ &logger,
+ runtime_hosts.iter().cloned(),
block.clone(),
- triggers,
+ trigger,
+ block_state,
)
+ })
.and_then(|block_state| future::ok(Loop::Continue((ctx, block_state)))),
)
},
@@ -608,31 +619,6 @@ where
})
}
-fn process_triggers_in_runtime_hosts<T>(
- logger: Logger,
- runtime_hosts: Vec<Arc<T::Host>>,
- block_state: BlockState,
- block: Arc<EthereumBlock>,
- triggers: Vec<EthereumTrigger>,
-) -> impl Future<Item = BlockState, Error = CancelableError<Error>>
-where
- T: RuntimeHostBuilder,
-{
- stream::iter_ok::<_, CancelableError<Error>>(triggers)
- // Process events from the block stream
- .fold(block_state, move |block_state, trigger| {
- // Process the log in each host in the same order the corresponding
- // data sources have been created
- SubgraphInstance::<T>::process_trigger_in_runtime_hosts(
- &logger,
- runtime_hosts.iter().cloned(),
- block.clone(),
- trigger,
- block_state,
- )
- })
-}
-
fn create_dynamic_data_sources<B, S, T>(
logger: Logger,
inputs: &IndexingInputs<B, S, T>,
