fn handle_subgraph_events<B, S, T>(
        logger: Logger,
        receiver: Receiver<SubgraphAssignmentProviderEvent>,
        store: Arc<S>,
        host_builder: T,
        block_stream_builder: B,
        elastic_config: Option<ElasticLoggingConfig>,
    ) where
        S: Store + ChainStore,
        T: RuntimeHostBuilder,
        B: BlockStreamBuilder + 'static,
    {
        // Subgraph instance shutdown senders
        let instances: SharedInstanceKeepAliveMap = Default::default();

        tokio::spawn(receiver.for_each(move |event| {
            use self::SubgraphAssignmentProviderEvent::*;

            match event {
                SubgraphStart(manifest) => {
                    // Write subgraph logs to the terminal and, if enabled, Elasticsearch
                    let term_logger = logger.new(o!("subgraph_id" => manifest.id.to_string()));
                    let logger = elastic_config
                        .clone()
                        .map(|elastic_config| {
                            split_logger(
                                term_logger.clone(),
                                elastic_logger(
                                    ElasticDrainConfig {
                                        general: elastic_config,
                                        index: String::from("subgraph-logs"),
                                        document_type: String::from("log"),
                                        subgraph_id: manifest.id.clone(),
                                        flush_interval: Duration::from_secs(5),
                                    },
                                    term_logger.clone(),
                                ),
                            )
                        })
                        .unwrap_or(term_logger);

                    info!(
                        logger,
                        "Start subgraph";
                        "data_sources" => manifest.data_sources.len()
                    );

                    Self::start_subgraph(
                        logger.clone(),
                        instances.clone(),
                        host_builder.clone(),
                        block_stream_builder.clone(),
                        store.clone(),
                        manifest,
                    )
                    .map_err(|err| error!(logger, "Failed to start subgraph: {}", err))
                    .ok();
                }
                SubgraphStop(id) => {
                    info!(logger, "Stopping subgraph"; "subgraph_id" => id.to_string());
                    Self::stop_subgraph(instances.clone(), id);
                }
            };

            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, T>(
        logger: Logger,
        receiver: Receiver<SubgraphAssignmentProviderEvent>,
        store: Arc<S>,
        host_builder: T,
        block_stream_builder: B,
        elastic_config: Option<ElasticLoggingConfig>,
    ) where
        S: Store + ChainStore,
        T: RuntimeHostBuilder,
        B: BlockStreamBuilder + 'static,
    {
        // Subgraph instance shutdown senders
        let instances: SharedInstanceKeepAliveMap = Default::default();
        tokio::spawn(receiver.for_each(move |event| {
            use self::SubgraphAssignmentProviderEvent::*;
            match event {
                SubgraphStart(manifest) => {
                    // Write subgraph logs to the terminal and, if enabled, Elasticsearch
                    let term_logger = logger.new(o!("subgraph_id" => manifest.id.to_string()));
                    let logger = elastic_config
                        .clone()
                        .map(|elastic_config| {
                            split_logger(
                                term_logger.clone(),
                                elastic_logger(
                                    ElasticDrainConfig {
                                        general: elastic_config,
                                        index: String::from("subgraph-logs"),
                                        document_type: String::from("log"),
                                        subgraph_id: manifest.id.clone(),
                                        flush_interval: Duration::from_secs(5),
                                    },
                                    term_logger.clone(),
                                ),
                            )
                        })
                        .unwrap_or(term_logger);
                    info!(
                        logger,
                        "Start subgraph";
                        "data_sources" => manifest.data_sources.len()
                    );
                    Self::start_subgraph(
                        logger.clone(),
                        instances.clone(),
                        host_builder.clone(),
                        block_stream_builder.clone(),
                        store.clone(),
                        manifest,
                    )
                    .map_err(|err| error!(logger, "Failed to start subgraph: {}", err))
                    .ok();
                }
                SubgraphStop(id) => {
                    info!(logger, "Stopping subgraph"; "subgraph_id" => id.to_string());
                    Self::stop_subgraph(instances.clone(), id);
                }
            };
            Ok(())
        }));
    }
fn handle_subgraph_events < B , S , T > ( logger : Logger , receiver : Receiver < SubgraphAssignmentProviderEvent > , store : Arc < S > , host_builder : T , block_stream_builder : B , elastic_config : Option < ElasticLoggingConfig > , ) where S : Store + ChainStore , T : RuntimeHostBuilder , B : BlockStreamBuilder + ' static , { // Subgraph instance shutdown senders let instances : SharedInstanceKeepAliveMap = Default :: default ( ) ; tokio :: spawn ( receiver . for_each ( move | event | { use self :: SubgraphAssignmentProviderEvent :: * ; match event { SubgraphStart ( manifest ) => { // Write subgraph logs to the terminal and, if enabled, Elasticsearch let term_logger = logger . new ( o ! ( " " manifest id to_string ( ) ) ) ; let logger = elastic_config . clone ( ) . map ( | elastic_config | { split_logger ( term_logger . clone ( ) , elastic_logger ( ElasticDrainConfig { general : elastic_config , index : String :: from ( " " ) , document_type : String :: from ( " " ) , subgraph_id : manifest . id . clone ( ) , flush_interval : Duration :: from_secs ( 5 ) , } , term_logger . clone ( ) , ) , ) } ) . unwrap_or ( term_logger ) ; info ! ( logger " " " " manifest data_sources len ( ) ) ; Self :: start_subgraph ( logger . clone ( ) , instances . clone ( ) , host_builder . clone ( ) , block_stream_builder . clone ( ) , store . clone ( ) , manifest , ) . map_err ( | err | error ! ( logger " " err ) ) . ok ( ) ; } SubgraphStop ( id ) => { info ! ( logger " " " " id to_string ( ) ) ; Self :: stop_subgraph ( instances . clone ( ) , id ) ; } } ; Ok ( ( ) ) } ) ) ; }
Handle incoming events from subgraph providers.
Handle incoming events from subgraph providers .
logger
Logger
receiver
Receiver<SubgraphAssignmentProviderEvent>
store
Arc<S>
host_builder
T
block_stream_builder
B
elastic_config
Option<ElasticLoggingConfig>
logger
Logger
receiver
Receiver<SubgraphAssignmentProviderEvent>
store
Arc<S>
host_builder
T
block_stream_builder
B
elastic_config
Option<ElasticLoggingConfig>