kabu_core_blockchain/
blockchain.rs

1use crate::blockchain_tokens::add_default_tokens_to_market;
2use alloy::primitives::ChainId;
3use influxdb::WriteQuery;
4use kabu_types_blockchain::{ChainParameters, Mempool};
5use kabu_types_entities::AccountNonceAndBalanceState;
6use kabu_types_events::{LoomTask, MarketEvents, MempoolEvents, MessageHealthEvent, MessageMempoolDataUpdate, MessageTxCompose};
7use kabu_types_market::Market;
8use reth_ethereum_primitives::EthPrimitives;
9use reth_node_types::NodePrimitives;
10use std::sync::Arc;
11use tokio::sync::{RwLock, broadcast};
12use tracing::error;
13
14#[derive(Clone)]
15pub struct Blockchain<N: NodePrimitives = EthPrimitives> {
16    chain_id: ChainId,
17    chain_parameters: ChainParameters,
18    market: Arc<RwLock<Market>>,
19    mempool: Arc<RwLock<Mempool<N>>>,
20    account_nonce_and_balance: Arc<RwLock<AccountNonceAndBalanceState>>,
21
22    new_mempool_tx_channel: broadcast::Sender<MessageMempoolDataUpdate<N>>,
23    market_events_channel: broadcast::Sender<MarketEvents>,
24    mempool_events_channel: broadcast::Sender<MempoolEvents>,
25    tx_compose_channel: broadcast::Sender<MessageTxCompose<N>>,
26
27    pool_health_monitor_channel: broadcast::Sender<MessageHealthEvent>,
28    influxdb_write_channel: Option<broadcast::Sender<WriteQuery>>,
29    tasks_channel: broadcast::Sender<LoomTask>,
30}
31
32impl Blockchain<EthPrimitives> {
33    pub fn new(chain_id: ChainId) -> Blockchain<EthPrimitives> {
34        Self::new_with_config(chain_id, true)
35    }
36
37    pub fn new_with_config(chain_id: ChainId, enable_influxdb: bool) -> Blockchain<EthPrimitives> {
38        let new_mempool_tx_channel: broadcast::Sender<MessageMempoolDataUpdate<EthPrimitives>> = broadcast::channel(5000).0;
39
40        let market_events_channel: broadcast::Sender<MarketEvents> = broadcast::channel(100).0;
41        let mempool_events_channel: broadcast::Sender<MempoolEvents> = broadcast::channel(2000).0;
42        let tx_compose_channel: broadcast::Sender<MessageTxCompose<EthPrimitives>> = broadcast::channel(2000).0;
43
44        let pool_health_monitor_channel: broadcast::Sender<MessageHealthEvent> = broadcast::channel(1000).0;
45        let influx_write_channel = if enable_influxdb { Some(broadcast::channel(1000).0) } else { None };
46        let tasks_channel: broadcast::Sender<LoomTask> = broadcast::channel(1000).0;
47
48        let mut market_instance = Market::default();
49
50        if let Err(error) = add_default_tokens_to_market(&mut market_instance, chain_id) {
51            error!(%error, "Failed to add default tokens to market");
52        }
53
54        Blockchain {
55            chain_id,
56            chain_parameters: ChainParameters::ethereum(),
57            market: Arc::new(RwLock::new(market_instance)),
58            mempool: Arc::new(RwLock::new(Mempool::<EthPrimitives>::new())),
59            account_nonce_and_balance: Arc::new(RwLock::new(AccountNonceAndBalanceState::new())),
60            new_mempool_tx_channel,
61            market_events_channel,
62            mempool_events_channel,
63            pool_health_monitor_channel,
64            tx_compose_channel,
65            influxdb_write_channel: influx_write_channel,
66            tasks_channel,
67        }
68    }
69}
70
71impl<N: NodePrimitives> Blockchain<N> {
72    pub fn chain_id(&self) -> u64 {
73        self.chain_id
74    }
75
76    pub fn chain_parameters(&self) -> ChainParameters {
77        self.chain_parameters.clone()
78    }
79
80    pub fn market(&self) -> Arc<RwLock<Market>> {
81        self.market.clone()
82    }
83
84    pub fn mempool(&self) -> Arc<RwLock<Mempool<N>>> {
85        self.mempool.clone()
86    }
87
88    pub fn nonce_and_balance(&self) -> Arc<RwLock<AccountNonceAndBalanceState>> {
89        self.account_nonce_and_balance.clone()
90    }
91
92    pub fn new_mempool_tx_channel(&self) -> broadcast::Sender<MessageMempoolDataUpdate<N>> {
93        self.new_mempool_tx_channel.clone()
94    }
95
96    pub fn market_events_channel(&self) -> broadcast::Sender<MarketEvents> {
97        self.market_events_channel.clone()
98    }
99
100    pub fn mempool_events_channel(&self) -> broadcast::Sender<MempoolEvents> {
101        self.mempool_events_channel.clone()
102    }
103
104    pub fn tx_compose_channel(&self) -> broadcast::Sender<MessageTxCompose<N>> {
105        self.tx_compose_channel.clone()
106    }
107
108    pub fn health_monitor_channel(&self) -> broadcast::Sender<MessageHealthEvent> {
109        self.pool_health_monitor_channel.clone()
110    }
111
112    pub fn influxdb_write_channel(&self) -> Option<broadcast::Sender<WriteQuery>> {
113        self.influxdb_write_channel.clone()
114    }
115
116    pub fn tasks_channel(&self) -> broadcast::Sender<LoomTask> {
117        self.tasks_channel.clone()
118    }
119}
kabu_core_blockchain/blockchain.rs

kabu_core_blockchain/
blockchain.rs
