reth_consensus_debug_client/providers/
rpc.rs1use crate::BlockProvider;
2use alloy_consensus::BlockHeader;
3use alloy_provider::{Network, Provider, ProviderBuilder};
4use futures::StreamExt;
5use reth_node_api::Block;
6use reth_tracing::tracing::warn;
7use std::sync::Arc;
8use tokio::sync::mpsc::Sender;
9
10#[derive(derive_more::Debug, Clone)]
13pub struct RpcBlockProvider<N: Network, PrimitiveBlock> {
14 #[debug(skip)]
15 provider: Arc<dyn Provider<N>>,
16 url: String,
17 #[debug(skip)]
18 convert: Arc<dyn Fn(N::BlockResponse) -> PrimitiveBlock + Send + Sync>,
19}
20
21impl<N: Network, PrimitiveBlock> RpcBlockProvider<N, PrimitiveBlock> {
22 pub async fn new(
24 rpc_url: &str,
25 convert: impl Fn(N::BlockResponse) -> PrimitiveBlock + Send + Sync + 'static,
26 ) -> eyre::Result<Self> {
27 Ok(Self {
28 provider: Arc::new(ProviderBuilder::default().connect(rpc_url).await?),
29 url: rpc_url.to_string(),
30 convert: Arc::new(convert),
31 })
32 }
33}
34
35impl<N: Network, PrimitiveBlock> BlockProvider for RpcBlockProvider<N, PrimitiveBlock>
36where
37 PrimitiveBlock: Block + 'static,
38{
39 type Block = PrimitiveBlock;
40
41 async fn subscribe_blocks(&self, tx: Sender<Self::Block>) {
42 let mut stream = match self.provider.subscribe_blocks().await {
43 Ok(sub) => sub.into_stream(),
44 Err(err) => {
45 warn!(
46 target: "consensus::debug-client",
47 %err,
48 url=%self.url,
49 "Failed to subscribe to blocks",
50 );
51 return;
52 }
53 };
54 while let Some(header) = stream.next().await {
55 match self.get_block(header.number()).await {
56 Ok(block) => {
57 if tx.send(block).await.is_err() {
58 break;
60 }
61 }
62 Err(err) => {
63 warn!(
64 target: "consensus::debug-client",
65 %err,
66 url=%self.url,
67 "Failed to fetch a block",
68 );
69 }
70 }
71 }
72 }
73
74 async fn get_block(&self, block_number: u64) -> eyre::Result<Self::Block> {
75 let block = self
76 .provider
77 .get_block_by_number(block_number.into())
78 .full()
79 .await?
80 .ok_or_else(|| eyre::eyre!("block not found by number {}", block_number))?;
81 Ok((self.convert)(block))
82 }
83}