reth_rpc_eth_api/helpers/
subscriptions.rs

1//! Streams subscriptions providers for `eth_subscribe`.
2
3use crate::{EthApiTypes, RpcConvert, RpcNodeCore, RpcReceipt};
4use alloy_consensus::{transaction::TxHashRef, BlockHeader, TxReceipt};
5use alloy_rpc_types_eth::{pubsub::TransactionReceiptsParams, Filter, Log};
6use futures::StreamExt;
7use reth_chain_state::CanonStateSubscriptions;
8use reth_primitives_traits::TransactionMeta;
9use reth_rpc_convert::{transaction::ConvertReceiptInput, RpcHeader};
10use reth_rpc_eth_types::logs_utils;
11use tracing::error;
12
13/// Provides streams subscriptions for `eth_subscribe`.
14///
15/// Override the default methods to inject additional data sources (e.g. flashblocks).
16pub trait EthSubscriptions:
17    RpcNodeCore + EthApiTypes<RpcConvert: RpcConvert<Primitives = Self::Primitives>>
18{
19    /// Returns a stream that yields matching logs from canonical chain updates.
20    fn log_stream(&self, filter: Filter) -> impl futures::Stream<Item = Log> + Send + Unpin {
21        self.provider()
22            .canonical_state_stream()
23            .map(move |canon_state| canon_state.block_receipts())
24            .flat_map(futures::stream::iter)
25            .flat_map(move |(block_receipts, removed)| {
26                let all_logs = logs_utils::matching_block_logs_with_tx_hashes(
27                    &filter,
28                    block_receipts.block,
29                    block_receipts.timestamp,
30                    block_receipts.tx_receipts.iter().map(|(tx, receipt)| (*tx, receipt)),
31                    removed,
32                );
33                futures::stream::iter(all_logs)
34            })
35    }
36
37    /// Returns a stream that yields new block headers from canonical chain updates.
38    fn header_stream(
39        &self,
40    ) -> impl futures::Stream<Item = RpcHeader<Self::NetworkTypes>> + Send + Unpin {
41        let converter = self.converter();
42        self.provider().canonical_state_stream().flat_map(move |new_chain| {
43            let headers = new_chain
44                .committed()
45                .blocks_iter()
46                .filter_map(|block| {
47                    match converter.convert_header(block.clone_sealed_header(), block.rlp_length())
48                    {
49                        Ok(header) => Some(header),
50                        Err(err) => {
51                            error!(target = "rpc", %err, "Failed to convert header");
52                            None
53                        }
54                    }
55                })
56                .collect::<Vec<_>>();
57            futures::stream::iter(headers)
58        })
59    }
60
61    /// Returns a stream that yields matching transaction receipts from canonical chain updates.
62    fn transaction_receipts_stream(
63        &self,
64        filter: TransactionReceiptsParams,
65    ) -> impl futures::Stream<Item = Vec<RpcReceipt<Self::NetworkTypes>>> + Send + Unpin {
66        let converter = self.converter();
67        self.provider().canonical_state_stream().flat_map(move |new_chain| {
68            let results: Vec<_> = new_chain
69                .committed()
70                .blocks_and_receipts()
71                .filter_map(|(block, receipts)| {
72                    let block_hash = block.hash();
73                    let block_number = block.number();
74                    let base_fee = block.base_fee_per_gas();
75                    let excess_blob_gas = block.excess_blob_gas();
76                    let timestamp = block.timestamp();
77
78                    let mut gas_used: u64 = 0;
79                    let mut next_log_index: usize = 0;
80
81                    let inputs: Vec<_> = block
82                        .transactions_recovered()
83                        .zip(receipts.iter())
84                        .enumerate()
85                        .filter_map(|(idx, (tx, receipt))| {
86                            let gas_used_before = gas_used;
87                            let next_log_index_before = next_log_index;
88                            let cumulative_gas_used = receipt.cumulative_gas_used();
89
90                            gas_used = cumulative_gas_used;
91                            next_log_index += receipt.logs().len();
92
93                            let matches = match &filter.transaction_hashes {
94                                Some(hashes) if !hashes.is_empty() => hashes.contains(tx.tx_hash()),
95                                _ => true,
96                            };
97
98                            matches.then(|| ConvertReceiptInput {
99                                tx,
100                                gas_used: cumulative_gas_used - gas_used_before,
101                                next_log_index: next_log_index_before,
102                                meta: TransactionMeta {
103                                    tx_hash: *tx.tx_hash(),
104                                    index: idx as u64,
105                                    block_hash,
106                                    block_number,
107                                    base_fee,
108                                    excess_blob_gas,
109                                    timestamp,
110                                },
111                                receipt: receipt.clone(),
112                            })
113                        })
114                        .collect();
115
116                    if inputs.is_empty() {
117                        return None;
118                    }
119
120                    match converter.convert_receipts_with_block(inputs, block.sealed_block()) {
121                        Ok(rpc_receipts) => Some(rpc_receipts),
122                        Err(err) => {
123                            error!(target = "rpc", %err, "Failed to convert receipts");
124                            None
125                        }
126                    }
127                })
128                .collect();
129
130            futures::stream::iter(results)
131        })
132    }
133}
reth_rpc_eth_api/helpers/subscriptions.rs

reth_rpc_eth_api/helpers/
subscriptions.rs
