Refine blockchain adapter

Author: cjdsellers

crates/adapters/blockchain/src/data/client.rs

@@ -30,10 +30,6 @@ use nautilus_model::{
     defi::{DefiData, SharedChain, validation::validate_address},
     identifiers::{ClientId, Venue},
 };
-use tokio::sync::{
-    mpsc::{UnboundedReceiver, UnboundedSender},
-    oneshot,
-};
 
 use crate::{
     config::BlockchainDataClientConfig,
@@ -63,13 +59,13 @@ pub struct BlockchainDataClient {
     /// Channel receiver for messages from the HyperSync client.
     hypersync_rx: Option<tokio::sync::mpsc::UnboundedReceiver<BlockchainMessage>>,
     /// Channel sender for commands to be processed asynchronously.
-    command_tx: UnboundedSender<DefiDataCommand>,
+    command_tx: tokio::sync::mpsc::UnboundedSender<DefiDataCommand>,
     /// Channel receiver for commands to be processed asynchronously.
-    command_rx: Option<UnboundedReceiver<DefiDataCommand>>,
+    command_rx: Option<tokio::sync::mpsc::UnboundedReceiver<DefiDataCommand>>,
     /// Background task for processing messages.
     process_task: Option<tokio::task::JoinHandle<()>>,
     /// Oneshot channel sender for graceful shutdown signal.
-    shutdown_tx: Option<oneshot::Sender<()>>,
+    shutdown_tx: Option<tokio::sync::oneshot::Sender<()>>,
 }
 
 impl BlockchainDataClient {
@@ -122,7 +118,7 @@ impl BlockchainDataClient {
             return;
         };
 
-        let (shutdown_tx, shutdown_rx) = oneshot::channel();
+        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
         self.shutdown_tx = Some(shutdown_tx);
 
         let mut core_client = self.core_client.take().unwrap();
@@ -205,7 +201,6 @@ impl BlockchainDataClient {
                                                 Ok(swap) => Some(DataEvent::DeFi(DefiData::PoolSwap(swap))),
                                                 Err(e) => {
                                                     tracing::error!("Error processing pool swap event: {e}");
-                                                    println!("FAIL");
                                                     None
                                                 }
                                             }
@@ -217,7 +212,6 @@ impl BlockchainDataClient {
                                     }
                                 }
                                 BlockchainMessage::BurnEvent(burn_event) => {
-                                    println!("BURN EVENT-->>> {:?}", burn_event);
                                     match core_client.get_pool(&burn_event.pool_address) {
                                         Ok(pool) => {
                                             let dex_extended = get_dex_extended(core_client.chain.name, &pool.dex.name).expect("Failed to get dex extended");
@@ -240,7 +234,6 @@ impl BlockchainDataClient {
                                     }
                                 }
                                 BlockchainMessage::MintEvent(mint_event) => {
-                                    println!("MINT EVENT-->>> {:?}", mint_event);
                                     match core_client.get_pool(&mint_event.pool_address) {
                                         Ok(pool) => {
                                             let dex_extended = get_dex_extended(core_client.chain.name,&pool.dex.name).expect("Failed to get dex extended");
@@ -266,8 +259,6 @@ impl BlockchainDataClient {
 
                             if let Some(event) = data_event {
                                 core_client.send_data(event);
-                            }else{
-                                println!("NO EVENT");
                             }
                         } else {
                             tracing::debug!("HyperSync data channel closed");
@@ -496,7 +487,7 @@ impl BlockchainDataClient {
     pub async fn await_process_task_close(&mut self) {
         if let Some(handle) = self.process_task.take() {
             if let Err(e) = handle.await {
-                tracing::error!("Process task join error: {}", e);
+                tracing::error!("Process task join error: {e}");
             }
         }
     }

crates/adapters/blockchain/src/data/core.rs

@@ -23,7 +23,6 @@ use nautilus_model::defi::{
     Block, Blockchain, DefiData, DexType, Pool, PoolLiquidityUpdate, PoolSwap, SharedChain,
     SharedDex, SharedPool, Token, validation::validate_address,
 };
-use tokio::sync::mpsc::UnboundedSender;
 
 use crate::{
     cache::BlockchainCache,
@@ -63,7 +62,7 @@ pub struct BlockchainDataClientCore {
     /// Interface for interacting with ERC20 token contracts.
     tokens: Erc20Contract,
     /// Channel sender for publishing data events to the `AsyncRunner`.
-    data_sender: UnboundedSender<DataEvent>,
+    data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
     /// Client for the HyperSync data indexing service.
     pub hypersync_client: HyperSyncClient,
     /// Optional WebSocket RPC client for direct blockchain node communication.
@@ -77,7 +76,7 @@ impl BlockchainDataClientCore {
     pub fn new(
         chain: SharedChain,
         config: BlockchainDataClientConfig,
-        hypersync_tx: UnboundedSender<BlockchainMessage>,
+        hypersync_tx: tokio::sync::mpsc::UnboundedSender<BlockchainMessage>,
     ) -> Self {
         let cache = BlockchainCache::new(chain.clone());
         let rpc_client = if !config.use_hypersync_for_live_data && config.wss_rpc_url.is_some() {

crates/adapters/blockchain/src/data/subscription.rs

@@ -13,9 +13,7 @@
 //  limitations under the License.
 // -------------------------------------------------------------------------------------------------
 
-use std::collections::HashMap;
-
-use ahash::{HashSet, HashSetExt};
+use ahash::{AHashMap, AHashSet};
 use alloy::primitives::{Address, keccak256};
 use nautilus_model::defi::DexType;
 
@@ -25,30 +23,30 @@ use nautilus_model::defi::DexType;
 /// and maintains the event signature encodings for efficient filtering.
 #[derive(Debug)]
 pub struct DefiDataSubscriptionManager {
-    subscribed_pool_swaps: HashMap<DexType, HashSet<Address>>,
-    pool_swap_event_encoded: HashMap<DexType, String>,
-    subscribed_pool_mints: HashMap<DexType, HashSet<Address>>,
-    pool_mint_event_encoded: HashMap<DexType, String>,
-    subscribed_pool_burns: HashMap<DexType, HashSet<Address>>,
-    pool_burn_event_encoded: HashMap<DexType, String>,
+    subscribed_pool_swaps: AHashMap<DexType, AHashSet<Address>>,
+    pool_swap_event_encoded: AHashMap<DexType, String>,
+    subscribed_pool_mints: AHashMap<DexType, AHashSet<Address>>,
+    pool_mint_event_encoded: AHashMap<DexType, String>,
+    subscribed_pool_burns: AHashMap<DexType, AHashSet<Address>>,
+    pool_burn_event_encoded: AHashMap<DexType, String>,
 }
 
 impl DefiDataSubscriptionManager {
     /// Creates a new [`DefiDataSubscriptionManager`] instance.
     pub fn new() -> Self {
         Self {
-            subscribed_pool_burns: HashMap::new(),
-            subscribed_pool_mints: HashMap::new(),
-            subscribed_pool_swaps: HashMap::new(),
-            pool_swap_event_encoded: HashMap::new(),
-            pool_burn_event_encoded: HashMap::new(),
-            pool_mint_event_encoded: HashMap::new(),
+            subscribed_pool_burns: AHashMap::new(),
+            subscribed_pool_mints: AHashMap::new(),
+            subscribed_pool_swaps: AHashMap::new(),
+            pool_swap_event_encoded: AHashMap::new(),
+            pool_burn_event_encoded: AHashMap::new(),
+            pool_mint_event_encoded: AHashMap::new(),
         }
     }
 
     /// Gets all unique contract addresses subscribed for any event type for a given DEX.
     pub fn get_subscribed_dex_contract_addresses(&self, dex: &DexType) -> Vec<Address> {
-        let mut unique_addresses = HashSet::new();
+        let mut unique_addresses = AHashSet::new();
 
         if let Some(addresses) = self.subscribed_pool_swaps.get(dex) {
             unique_addresses.extend(addresses.iter().cloned());
@@ -105,39 +103,39 @@ impl DefiDataSubscriptionManager {
         mint_event_signature: &str,
         burn_event_signature: &str,
     ) {
-        self.subscribed_pool_swaps.insert(dex, HashSet::new());
+        self.subscribed_pool_swaps.insert(dex, AHashSet::new());
         let swap_event_hash = keccak256(swap_event_signature.as_bytes());
         let encoded_swap_event = format!(
             "0x{encoded_hash}",
             encoded_hash = hex::encode(swap_event_hash)
         );
         self.pool_swap_event_encoded.insert(dex, encoded_swap_event);
 
-        self.subscribed_pool_mints.insert(dex, HashSet::new());
+        self.subscribed_pool_mints.insert(dex, AHashSet::new());
         let mint_event_hash = keccak256(mint_event_signature.as_bytes());
         let encoded_mint_event = format!(
             "0x{encoded_hash}",
             encoded_hash = hex::encode(mint_event_hash)
         );
         self.pool_mint_event_encoded.insert(dex, encoded_mint_event);
 
-        self.subscribed_pool_burns.insert(dex, HashSet::new());
+        self.subscribed_pool_burns.insert(dex, AHashSet::new());
         let burn_event_hash = keccak256(burn_event_signature.as_bytes());
         let encoded_burn_event = format!(
             "0x{encoded_hash}",
             encoded_hash = hex::encode(burn_event_hash)
         );
         self.pool_burn_event_encoded.insert(dex, encoded_burn_event);
 
-        tracing::info!("Registered dex for subscriptions: {:?}", dex);
+        tracing::info!("Registered DEX for subscriptions: {dex:?}");
     }
 
     /// Subscribes to swap events for a specific pool address on a DEX.
     pub fn subscribe_swaps(&mut self, dex: DexType, address: Address) {
         if let Some(pool_set) = self.subscribed_pool_swaps.get_mut(&dex) {
             pool_set.insert(address);
         } else {
-            tracing::error!("Dex not registered for swap subscriptions: {:?}", dex);
+            tracing::error!("DEX not registered for swap subscriptions: {dex:?}");
         }
     }
 
@@ -146,7 +144,7 @@ impl DefiDataSubscriptionManager {
         if let Some(pool_set) = self.subscribed_pool_mints.get_mut(&dex) {
             pool_set.insert(address);
         } else {
-            tracing::error!("Dex not registered for mint subscriptions: {:?}", dex);
+            tracing::error!("DEX not registered for mint subscriptions: {dex:?}");
         }
     }
 
@@ -155,7 +153,7 @@ impl DefiDataSubscriptionManager {
         if let Some(pool_set) = self.subscribed_pool_burns.get_mut(&dex) {
             pool_set.insert(address);
         } else {
-            tracing::warn!("Dex not registered for burn subscriptions: {:?}", dex);
+            tracing::warn!("DEX not registered for burn subscriptions: {dex:?}");
         }
     }
 
@@ -164,7 +162,7 @@ impl DefiDataSubscriptionManager {
         if let Some(pool_set) = self.subscribed_pool_swaps.get_mut(&dex) {
             pool_set.remove(&address);
         } else {
-            tracing::error!("Dex not registered for swap subscriptions: {:?}", dex);
+            tracing::error!("DEX not registered for swap subscriptions: {dex:?}");
         }
     }
 
@@ -173,7 +171,7 @@ impl DefiDataSubscriptionManager {
         if let Some(pool_set) = self.subscribed_pool_mints.get_mut(&dex) {
             pool_set.remove(&address);
         } else {
-            tracing::error!("Dex not registered for mint subscriptions: {:?}", dex);
+            tracing::error!("DEX not registered for mint subscriptions: {dex:?}");
         }
     }
 
@@ -182,11 +180,15 @@ impl DefiDataSubscriptionManager {
         if let Some(pool_set) = self.subscribed_pool_burns.get_mut(&dex) {
             pool_set.remove(&address);
         } else {
-            tracing::error!("Dex not registered for burn subscriptions: {:?}", dex);
+            tracing::error!("DEX not registered for burn subscriptions: {dex:?}");
         }
     }
 }
 
+////////////////////////////////////////////////////////////////////////////////
+// Tests
+////////////////////////////////////////////////////////////////////////////////
+
 #[cfg(test)]
 mod tests {
     use alloy::primitives::address;

docs/developer_guide/rust.md

@@ -387,6 +387,7 @@ Use consistent test module structure with section separators:
 ////////////////////////////////////////////////////////////////////////////////
 // Tests
 ////////////////////////////////////////////////////////////////////////////////
+
 #[cfg(test)]
 mod tests {
     use rstest::rstest;