kabu_node_exex/
reth_exex_worker.rs

1use kabu_core_blockchain::Blockchain;
2use kabu_types_blockchain::MempoolTx;
3use kabu_types_events::{MessageMempoolDataUpdate, NodeMempoolDataUpdate};
4use reth_ethereum_primitives::EthPrimitives;
5use reth_transaction_pool::{EthPooledTransaction, TransactionPool};
6use tokio::select;
7use tracing::{debug, error, info};
8
9pub async fn mempool_worker<Pool>(mempool: Pool, bc: Blockchain<EthPrimitives>) -> eyre::Result<()>
10where
11    Pool: TransactionPool<Transaction = EthPooledTransaction> + Clone + 'static,
12{
13    info!("Mempool worker started");
14    let mut tx_listener = mempool.new_transactions_listener();
15
16    let mempool_tx = bc.new_mempool_tx_channel();
17
18    // EthTxBuilder removed in new version
19
20    loop {
21        select! {
22            tx_notification = tx_listener.recv() => {
23                if let Some(tx_notification) = tx_notification {
24                    let tx_hash = *tx_notification.transaction.hash();
25                    // TODO: Implement proper transaction conversion from EthPooledTransaction to alloy RPC Transaction
26                    // For now, we're skipping the transaction details in the mempool update
27                    // This allows the system to track transaction hashes without full transaction data
28                    let update_msg: MessageMempoolDataUpdate<EthPrimitives> = MessageMempoolDataUpdate::new_with_source(NodeMempoolDataUpdate { tx_hash, mempool_tx: MempoolTx { tx: None, ..MempoolTx::default() } }, "exex".to_string());
29                    if let Err(e) =  mempool_tx.send(update_msg) {
30                        error!(error=?e.to_string(), "mempool_tx.send");
31                    }else{
32                        debug!(hash = ?tx_notification.transaction.hash(), "Received pool tx");
33                    }
34                }
35            }
36        }
37    }
38}
kabu_node_exex/reth_exex_worker.rs

kabu_node_exex/
reth_exex_worker.rs
