/// Processes a block and returns the updated context and a boolean flag indicating
/// whether new dynamic data sources have been added to the subgraph.
Processes a block and returns the updated context and a boolean flag indicating
whether new dynamic data sources have been added to the subgraph.
Processes a block and returns the updated context and a boolean flag indicating whether new dynamic data sources have been added to the subgraph .
fn process_block<B, S, T>(
    logger: Logger,
    ctx: IndexingContext<B, S, T>,
    block_stream_cancel_handle: CancelHandle,
    block: EthereumBlockWithTriggers,
) -> impl Future<Item = (IndexingContext<B, S, T>, bool), Error = CancelableError<Error>>
where
    B: BlockStreamBuilder,
    S: ChainStore + Store,
    T: RuntimeHostBuilder,
{
    let calls = block.calls;
    let triggers = block.triggers;
    let block = block.ethereum_block;
    let logger = logger.new(o!(
        "block_number" => format!("{:?}", block.block.number.unwrap()),
        "block_hash" => format!("{:?}", block.block.hash.unwrap())
    ));
    if triggers.len() == 1 {
        info!(logger, "1 trigger found in this block for this subgraph");
    } else if triggers.len() > 1 {
        info!(
            logger,
            "{} triggers found in this block for this subgraph",
            triggers.len()
        );
    }
    // Obtain current and new block pointer (after this block is processed)
    let block = Arc::new(block);
    let block_ptr_now = EthereumBlockPointer::to_parent(&block);
    let block_ptr_after = EthereumBlockPointer::from(&*block);
    let block_ptr_for_new_data_sources = block_ptr_after.clone();
    // Clone a few things to pass into futures
    let logger1 = logger.clone();
    let logger2 = logger.clone();
    let logger3 = logger.clone();
    let logger4 = logger.clone();
    // Process events one after the other, passing in entity operations
    // collected previously to every new event being processed
    process_triggers(
        logger.clone(),
        ctx,
        BlockState::default(),
        block.clone(),
        triggers,
    )
    .and_then(|(ctx, block_state)| {
        // Instantiate dynamic data sources
        create_dynamic_data_sources(logger1, ctx, block_state).from_err()
    })
    .and_then(move |(ctx, block_state, data_sources, runtime_hosts)| {
        // Reprocess the triggers from this block that match the new data sources
        let block_with_calls = EthereumBlockWithCalls {
            ethereum_block: block.deref().clone(),
            calls,
        };
        future::result(<B>::Stream::parse_triggers(
            EthereumLogFilter::from_data_sources_opt(&data_sources),
            EthereumCallFilter::from_data_sources_opt(&data_sources),
            EthereumBlockFilter::from_data_sources_opt(&data_sources),
            false,
            block_with_calls,
        ))
        .from_err()
        .and_then(move |block_with_triggers| {
            let triggers = block_with_triggers.triggers;
            if triggers.len() == 1 {
                info!(
                    logger,
                    "1 trigger found in this block for the new data sources"
                );
            } else if triggers.len() > 1 {
                info!(
                    logger,
                    "{} triggers found in this block for the new data sources",
                    triggers.len()
                );
            }
            process_triggers_in_runtime_hosts::<T>(
                logger2.clone(),
                runtime_hosts.clone(),
                block_state,
                block.clone(),
                triggers,
            )
            .map(move |block_state| (ctx, block_state, data_sources, runtime_hosts))
        })
    })
    .and_then(move |(ctx, block_state, data_sources, runtime_hosts)| {
        // Add entity operations for the new data sources to the block state
        // and add runtimes for the data sources to the subgraph instance
        future::result(
            persist_dynamic_data_sources(
                logger3,
                ctx,
                block_state,
                data_sources,
                runtime_hosts,
                block_ptr_for_new_data_sources,
            )
            .map(move |(ctx, block_state, data_sources_created)| {
                // If new data sources have been added, indicate that the subgraph
                // needs to be restarted after this block
                (ctx, block_state, data_sources_created)
            }),
        )
        .from_err()
    })
    // Apply entity operations and advance the stream
    .and_then(move |(ctx, block_state, needs_restart)| {
        // Avoid writing to store if block stream has been canceled
        if block_stream_cancel_handle.is_canceled() {
            return Err(CancelableError::Cancel);
        }
        if !block_state.entity_operations.is_empty() {
            info!(
                logger4,
                "Applying {} entity operation(s)",
                block_state.entity_operations.len()
            );
        }
        // Transact entity operations into the store and update the
        // subgraph's block stream pointer
        ctx.inputs
            .store
            .transact_block_operations(
                ctx.inputs.deployment_id.clone(),
                block_ptr_now,
                block_ptr_after,
                block_state.entity_operations,
            )
            .map(|_| (ctx, needs_restart))
            .map_err(|e| {
                format_err!("Error while processing block stream for a subgraph: {}", e).into()
            })
    })
}
fn process_block < B , S , T > ( logger : Logger , ctx : IndexingContext < B , S , T > , block_stream_cancel_handle : CancelHandle , block : EthereumBlockWithTriggers , ) -> impl Future < Item = ( IndexingContext < B , S , T > , bool ) , Error = CancelableError < Error > > where B : BlockStreamBuilder , S : ChainStore + Store , T : RuntimeHostBuilder , { let calls = block . calls ; let triggers = block . triggers ; let block = block . ethereum_block ; let logger = logger . new ( o ! ( " " format ( " " block block number unwrap ( ) ) " " format ( " " block block hash unwrap ( ) ) ) ) ; if triggers . len ( ) == 1 { info ! ( logger " " ) ; } else if triggers . len ( ) > 1 { info ! ( logger " " triggers len ( ) ) ; } // Obtain current and new block pointer (after this block is processed) let block = Arc :: new ( block ) ; let block_ptr_now = EthereumBlockPointer :: to_parent ( & block ) ; let block_ptr_after = EthereumBlockPointer :: from ( & * block ) ; let block_ptr_for_new_data_sources = block_ptr_after . clone ( ) ; // Clone a few things to pass into futures let logger1 = logger . clone ( ) ; let logger2 = logger . clone ( ) ; let logger3 = logger . clone ( ) ; let logger4 = logger . clone ( ) ; // Process events one after the other, passing in entity operations // collected previously to every new event being processed process_triggers ( logger . clone ( ) , ctx , BlockState :: default ( ) , block . clone ( ) , triggers , ) . and_then ( | ( ctx , block_state ) | { // Instantiate dynamic data sources create_dynamic_data_sources ( logger1 , ctx , block_state ) . from_err ( ) } ) . and_then ( move | ( ctx , block_state , data_sources , runtime_hosts ) | { // Reprocess the triggers from this block that match the new data sources let block_with_calls = EthereumBlockWithCalls { ethereum_block : block . deref ( ) . clone ( ) , calls , } ; future :: result ( < B > :: Stream :: parse_triggers ( EthereumLogFilter :: from_data_sources_opt ( & data_sources ) , EthereumCallFilter :: from_data_sources_opt ( & data_sources ) , EthereumBlockFilter :: from_data_sources_opt ( & data_sources ) , false , block_with_calls , ) ) . from_err ( ) . and_then ( move | block_with_triggers | { let triggers = block_with_triggers . triggers ; if triggers . len ( ) == 1 { info ! ( logger " " ) ; } else if triggers . len ( ) > 1 { info ! ( logger " " triggers len ( ) ) ; } process_triggers_in_runtime_hosts :: < T > ( logger2 . clone ( ) , runtime_hosts . clone ( ) , block_state , block . clone ( ) , triggers , ) . map ( move | block_state | ( ctx , block_state , data_sources , runtime_hosts ) ) } ) } ) . and_then ( move | ( ctx , block_state , data_sources , runtime_hosts ) | { // Add entity operations for the new data sources to the block state // and add runtimes for the data sources to the subgraph instance future :: result ( persist_dynamic_data_sources ( logger3 , ctx , block_state , data_sources , runtime_hosts , block_ptr_for_new_data_sources , ) . map ( move | ( ctx , block_state , data_sources_created ) | { // If new data sources have been added, indicate that the subgraph // needs to be restarted after this block ( ctx , block_state , data_sources_created ) } ) , ) . from_err ( ) } ) // Apply entity operations and advance the stream . and_then ( move | ( ctx , block_state , needs_restart ) | { // Avoid writing to store if block stream has been canceled if block_stream_cancel_handle . is_canceled ( ) { return Err ( CancelableError :: Cancel ) ; } if ! block_state . entity_operations . is_empty ( ) { info ! ( logger4 " " block_state entity_operations len ( ) ) ; } // Transact entity operations into the store and update the // subgraph's block stream pointer ctx . inputs . store . transact_block_operations ( ctx . inputs . deployment_id . clone ( ) , block_ptr_now , block_ptr_after , block_state . entity_operations , ) . map ( | _ | ( ctx , needs_restart ) ) . map_err ( | e | { format_err ! ( " " e ) . into ( ) } ) } ) }
Processes a block and returns the updated context and a boolean flag indicating
whether new dynamic data sources have been added to the subgraph.
Processes a block and returns the updated context and a boolean flag indicating whether new dynamic data sources have been added to the subgraph .
logger
Logger
ctx
IndexingContext<B, S, T>
block_stream_cancel_handle
CancelHandle
block
EthereumBlockWithTriggers