Method apply

crates/execution/flashblocks-node/src/extension.rs:31–91

Applies the extension to the supplied hooks.

(self: Box<Self>, hooks: NodeHooks)

impl BaseNodeExtension for FlashblocksExtension {
    /// Applies the extension to the supplied hooks.
    fn apply(self: Box<Self>, hooks: NodeHooks) -> NodeHooks {
        let Some(cfg) = self.config else {
            info!(message = "flashblocks integration is disabled");
            return hooks;
        };

        let state = cfg.state;
        let mut subscriber = FlashblocksSubscriber::new(
            Arc::clone(&state),
            cfg.websocket_url,
            cfg.subscriber_ping_interval,
        );

        let state_for_canonical = Arc::clone(&state);
        let state_for_rpc = Arc::clone(&state);
        let state_for_start = state;

        // Start state processor, subscriber, and canonical subscription after node is started
        let hooks = hooks.add_node_started_hook(move |ctx| {
            info!(message = "Starting Flashblocks state processor");
            state_for_start.start(ctx.provider().clone());
            subscriber.start();

            let mut canonical_stream =
                BroadcastStream::new(ctx.provider().subscribe_to_canonical_state());
            tokio::spawn(async move {
                while let Some(Ok(notification)) = canonical_stream.next().await {
                    let committed = notification.committed();
                    for block in committed.blocks_iter() {
                        state_for_canonical.on_canonical_block_received(block.as_ref().clone());
                    }
                }
            });

            Ok(())
        });

        // Extend with RPC modules
        hooks.add_rpc_module(move |ctx| {
            info!(message = "Starting Flashblocks RPC");

            let api_ext = EthApiExt::new(
                ctx.registry.eth_api().clone(),
                ctx.registry.eth_handlers().filter.clone(),
                Arc::clone(&state_for_rpc),
            );
            ctx.modules.replace_configured(api_ext.into_rpc())?;

            // Register the eth_subscribe subscription endpoint for flashblocks
            // Uses replace_configured since eth_subscribe already exists from reth's standard module
            // Pass eth_api to enable proxying standard subscription types to reth's implementation
            let eth_pubsub = EthPubSub::new(
                ctx.registry.eth_api().clone(),
                ctx.node().task_executor.clone(),
                state_for_rpc,
            );
            ctx.modules.replace_configured(eth_pubsub.into_rpc())?;