updated commands

paradigmxyz · May 11, 2024 · a6f2112 · a6f2112
1 parent 1a039d6
commit a6f2112
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 5 deletions.
diff --git a/bin/reth/src/commands/debug_cmd/execution.rs b/bin/reth/src/commands/debug_cmd/execution.rs
@@ -11,7 +11,7 @@ use crate::{
     utils::get_single_header,
 };
 use clap::Parser;
-use futures::{stream::select as stream_select, StreamExt};
+use futures::stream::select as stream_select;
 use reth_beacon_consensus::EthBeaconConsensus;
 use reth_cli_runner::CliContext;
 use reth_config::{config::EtlConfig, Config};
@@ -264,8 +264,8 @@ impl Command {
 
         let pipeline_events = pipeline.events();
         let events = stream_select(
-            network.event_listener().map(Into::into),
-            pipeline_events.map(Into::into),
+            reth_node_events::node::handle_broadcast_stream(network.event_listener()),
+            reth_node_events::node::handle_broadcast_stream(pipeline_events),
         );
         ctx.task_executor.spawn_critical(
             "events task",

diff --git a/bin/reth/src/commands/import.rs b/bin/reth/src/commands/import.rs
@@ -11,7 +11,7 @@ use crate::{
 };
 use clap::Parser;
 use eyre::Context;
-use futures::{Stream, StreamExt};
+use futures::Stream;
 use reth_beacon_consensus::EthBeaconConsensus;
 use reth_config::{config::EtlConfig, Config};
 use reth_consensus::Consensus;
@@ -300,7 +300,7 @@ where
         )
         .build(provider_factory, static_file_producer);
 
-    let events = pipeline.events().map(Into::into);
+    let events = reth_node_events::node::handle_broadcast_stream(pipeline.events());
 
     Ok((pipeline, events))
 }