fn handle_subgraph_events<B, S, M>(
        logger_factory: LoggerFactory,
        receiver: Receiver<SubgraphAssignmentProviderEvent>,
        stores: HashMap<String, Arc<S>>,
        eth_adapters: HashMap<String, Arc<dyn EthereumAdapter>>,
        host_builder: impl RuntimeHostBuilder,
        block_stream_builder: B,
        metrics_registry: Arc<M>,
    ) where
        S: Store + ChainStore + SubgraphDeploymentStore + EthereumCallCache,
        B: BlockStreamBuilder,
        M: MetricsRegistry,
    {
        let metrics_registry_for_manager = metrics_registry.clone();
        let metrics_registry_for_subgraph = metrics_registry.clone();
        let manager_metrics = SubgraphInstanceManagerMetrics::new(metrics_registry_for_manager);

        // Subgraph instance shutdown senders
        let instances: SharedInstanceKeepAliveMap = Default::default();

        // Blocking due to store interactions. Won't be blocking after #905.
        graph::spawn_blocking(receiver.compat().try_for_each(move |event| {
            use self::SubgraphAssignmentProviderEvent::*;

            match event {
                SubgraphStart(manifest) => {
                    let logger = logger_factory.subgraph_logger(&manifest.id);
                    info!(
                        logger,
                        "Start subgraph";
                        "data_sources" => manifest.data_sources.len()
                    );
                    let network = manifest.network_name();
                    Self::start_subgraph(
                        logger.clone(),
                        instances.clone(),
                        host_builder.clone(),
                        block_stream_builder.clone(),
                        stores
                            .get(&network)
                            .expect(&format!(
                                "expected store that matches subgraph network: {}",
                                &network
                            ))
                            .clone(),
                        eth_adapters
                            .get(&network)
                            .expect(&format!(
                                "expected eth adapter that matches subgraph network: {}",
                                &network
                            ))
                            .clone(),
                        manifest,
                        metrics_registry_for_subgraph.clone(),
                    )
                    .map_err(|err| {
                        error!(
                            logger,
                            "Failed to start subgraph";
                            "error" => format!("{}", err),
                            "code" => LogCode::SubgraphStartFailure
                        )
                    })
                    .and_then(|_| {
                        manager_metrics.subgraph_count.inc();
                        Ok(())
                    })
                    .ok();
                }
                SubgraphStop(id) => {
                    let logger = logger_factory.subgraph_logger(&id);
                    info!(logger, "Stop subgraph");

                    Self::stop_subgraph(instances.clone(), id);
                    manager_metrics.subgraph_count.dec();
                }
            };

            futures03::future::ok(())
        }));
    }
/// Handle incoming events from subgraph providers.
Handle incoming events from subgraph providers.
Handle incoming events from subgraph providers .
fn handle_subgraph_events<B, S, M>(
        logger_factory: LoggerFactory,
        receiver: Receiver<SubgraphAssignmentProviderEvent>,
        stores: HashMap<String, Arc<S>>,
        eth_adapters: HashMap<String, Arc<dyn EthereumAdapter>>,
        host_builder: impl RuntimeHostBuilder,
        block_stream_builder: B,
        metrics_registry: Arc<M>,
    ) where
        S: Store + ChainStore + SubgraphDeploymentStore + EthereumCallCache,
        B: BlockStreamBuilder,
        M: MetricsRegistry,
    {
        let metrics_registry_for_manager = metrics_registry.clone();
        let metrics_registry_for_subgraph = metrics_registry.clone();
        let manager_metrics = SubgraphInstanceManagerMetrics::new(metrics_registry_for_manager);
        // Subgraph instance shutdown senders
        let instances: SharedInstanceKeepAliveMap = Default::default();
        // Blocking due to store interactions. Won't be blocking after #905.
        graph::spawn_blocking(receiver.compat().try_for_each(move |event| {
            use self::SubgraphAssignmentProviderEvent::*;
            match event {
                SubgraphStart(manifest) => {
                    let logger = logger_factory.subgraph_logger(&manifest.id);
                    info!(
                        logger,
                        "Start subgraph";
                        "data_sources" => manifest.data_sources.len()
                    );
                    let network = manifest.network_name();
                    Self::start_subgraph(
                        logger.clone(),
                        instances.clone(),
                        host_builder.clone(),
                        block_stream_builder.clone(),
                        stores
                            .get(&network)
                            .expect(&format!(
                                "expected store that matches subgraph network: {}",
                                &network
                            ))
                            .clone(),
                        eth_adapters
                            .get(&network)
                            .expect(&format!(
                                "expected eth adapter that matches subgraph network: {}",
                                &network
                            ))
                            .clone(),
                        manifest,
                        metrics_registry_for_subgraph.clone(),
                    )
                    .map_err(|err| {
                        error!(
                            logger,
                            "Failed to start subgraph";
                            "error" => format!("{}", err),
                            "code" => LogCode::SubgraphStartFailure
                        )
                    })
                    .and_then(|_| {
                        manager_metrics.subgraph_count.inc();
                        Ok(())
                    })
                    .ok();
                }
                SubgraphStop(id) => {
                    let logger = logger_factory.subgraph_logger(&id);
                    info!(logger, "Stop subgraph");
                    Self::stop_subgraph(instances.clone(), id);
                    manager_metrics.subgraph_count.dec();
                }
            };
            futures03::future::ok(())
        }));
    }
fn handle_subgraph_events < B , S , M > ( logger_factory : LoggerFactory , receiver : Receiver < SubgraphAssignmentProviderEvent > , stores : HashMap < String , Arc < S > > , eth_adapters : HashMap < String , Arc < dyn EthereumAdapter > > , host_builder : impl RuntimeHostBuilder , block_stream_builder : B , metrics_registry : Arc < M > , ) where S : Store + ChainStore + SubgraphDeploymentStore + EthereumCallCache , B : BlockStreamBuilder , M : MetricsRegistry , { let metrics_registry_for_manager = metrics_registry . clone ( ) ; let metrics_registry_for_subgraph = metrics_registry . clone ( ) ; let manager_metrics = SubgraphInstanceManagerMetrics :: new ( metrics_registry_for_manager ) ; // Subgraph instance shutdown senders let instances : SharedInstanceKeepAliveMap = Default :: default ( ) ; // Blocking due to store interactions. Won't be blocking after #905. graph :: spawn_blocking ( receiver . compat ( ) . try_for_each ( move | event | { use self :: SubgraphAssignmentProviderEvent :: * ; match event { SubgraphStart ( manifest ) => { let logger = logger_factory . subgraph_logger ( & manifest . id ) ; info ! ( logger " " " " manifest data_sources len ( ) ) ; let network = manifest . network_name ( ) ; Self :: start_subgraph ( logger . clone ( ) , instances . clone ( ) , host_builder . clone ( ) , block_stream_builder . clone ( ) , stores . get ( & network ) . expect ( & format ! ( " " network ) ) . clone ( ) , eth_adapters . get ( & network ) . expect ( & format ! ( " " network ) ) . clone ( ) , manifest , metrics_registry_for_subgraph . clone ( ) , ) . map_err ( | err | { error ! ( logger " " " " format ( " " err ) " " LogCode SubgraphStartFailure ) } ) . and_then ( | _ | { manager_metrics . subgraph_count . inc ( ) ; Ok ( ( ) ) } ) . ok ( ) ; } SubgraphStop ( id ) => { let logger = logger_factory . subgraph_logger ( & id ) ; info ! ( logger " " ) ; Self :: stop_subgraph ( instances . clone ( ) , id ) ; manager_metrics . subgraph_count . dec ( ) ; } } ; futures03 :: future :: ok ( ( ) ) } ) ) ; }
Handle incoming events from subgraph providers.
Handle incoming events from subgraph providers .
logger_factory
LoggerFactory
receiver
Receiver<SubgraphAssignmentProviderEvent>
stores
HashMap<String, Arc<S>>
eth_adapters
HashMap<String, Arc<dyn EthereumAdapter>>
host_builder
impl RuntimeHostBuilder
block_stream_builder
B
metrics_registry
Arc<M>
logger_factory
LoggerFactory
receiver
Receiver<SubgraphAssignmentProviderEvent>
stores
HashMap<String, Arc<S>>
eth_adapters
HashMap<String, Arc<dyn EthereumAdapter>>
host_builder
impl RuntimeHostBuilder
block_stream_builder
B
metrics_registry
Arc<M>