chore: Adds rpc_url to block_handlers and handlers

stakelens · Jul 19, 2024 · c96c0c6 · c96c0c6
1 parent b890f04
commit c96c0c6
Show file tree

Hide file tree

Showing 6 changed files with 38 additions and 25 deletions.
diff --git a/ghost-crab-macros/src/lib.rs b/ghost-crab-macros/src/lib.rs
@@ -28,6 +28,9 @@ pub fn block_handler(metadata: TokenStream, input: TokenStream) -> TokenStream {
     let config = get_config();
     let source = config.block_handlers.get(name).expect("Source not found.");
 
+    let rpc_url = config.networks.get(&source.network).expect("RPC url not found for network");
+    let rpc_url = Literal::string(&rpc_url);
+
     let step = Literal::u64_suffixed(source.step);
     let start_block = Literal::u64_suffixed(source.start_block);
     let network = Literal::string(&source.network);
@@ -65,6 +68,10 @@ pub fn block_handler(metadata: TokenStream, input: TokenStream) -> TokenStream {
                 #network
             }
 
+            fn rpc_url(&self) -> String {
+                #rpc_url
+            }
+
             fn start_block(&self) -> u64 {
                 #start_block
             }
@@ -150,6 +157,9 @@ fn create_handler(metadata: TokenStream, input: TokenStream, is_template: bool)
         execution_mode = source.execution_mode.clone().unwrap_or(ExecutionMode::Parallel);
     };
 
+    let rpc_url = config.networks.get(&network).expect("RPC url not found for network");
+    let rpc_url = Literal::string(&rpc_url);
+
     let abi = Literal::string(&abi);
     let network = Literal::string(&network);
 
@@ -211,6 +221,10 @@ fn create_handler(metadata: TokenStream, input: TokenStream, is_template: bool)
                 String::from(#network)
             }
 
+            fn rpc_url(&self) -> String {
+                #rpc_url
+            }
+
             fn execution_mode(&self) -> ExecutionMode {
                 #execution_mode
             }

diff --git a/ghost-crab/src/block_handler.rs b/ghost-crab/src/block_handler.rs
@@ -1,3 +1,4 @@
+use crate::cache::manager::RPC_MANAGER;
 use crate::indexer::TemplateManager;
 use crate::latest_block_manager::LatestBlockManager;
 use alloy::providers::Provider;
@@ -24,23 +25,27 @@ pub trait BlockHandler {
     async fn handle(&self, params: BlockContext);
     fn step(&self) -> u64;
     fn network(&self) -> String;
+    fn rpc_url(&self) -> String;
     fn start_block(&self) -> u64;
     fn execution_mode(&self) -> ExecutionMode;
 }
 
 pub struct BlockConfig {
     pub handler: BlockHandlerInstance,
-    pub provider: RootProvider<Http<Client>>,
     pub templates: TemplateManager,
 }
 
 pub async fn process_logs_block(
-    BlockConfig { handler, provider, templates }: BlockConfig,
+    BlockConfig { handler, templates }: BlockConfig,
 ) -> Result<(), TransportError> {
     let step = handler.step();
+    let network = handler.network();
+    let rpc_url = handler.rpc_url();
     let start_block = handler.start_block();
     let execution_mode = handler.execution_mode();
 
+    let provider = RPC_MANAGER.lock().await.get_or_create(network, rpc_url).await;
+
     let mut current_block = start_block;
     let mut latest_block_manager =
         LatestBlockManager::new(provider.clone(), Duration::from_secs(10));

diff --git a/ghost-crab/src/cache/manager.rs b/ghost-crab/src/cache/manager.rs
@@ -1,5 +1,4 @@
 use super::rpc_proxy::RpcWithCache;
-use crate::config;
 use alloy::providers::ProviderBuilder;
 use alloy::providers::RootProvider;
 use alloy::transports::http::{Client, Http};
@@ -14,7 +13,6 @@ pub static RPC_MANAGER: Lazy<Arc<Mutex<RPCManager>>> =
 pub struct RPCManager {
     current_port: u16,
     rpcs: HashMap<String, RootProvider<Http<Client>>>,
-    config: config::Config,
 }
 
 impl Default for RPCManager {
@@ -25,12 +23,15 @@ impl Default for RPCManager {
 
 impl RPCManager {
     pub fn new() -> Self {
-        RPCManager { rpcs: HashMap::new(), current_port: 3001, config: config::load() }
+        RPCManager { rpcs: HashMap::new(), current_port: 3001 }
     }
 
-    pub async fn get(&mut self, network: String) -> RootProvider<Http<Client>> {
-        let rpc_url = self.config.networks.get(&network).unwrap();
-        let provider = self.rpcs.get(rpc_url);
+    pub async fn get_or_create(
+        &mut self,
+        network: String,
+        rpc_url: String,
+    ) -> RootProvider<Http<Client>> {
+        let provider = self.rpcs.get(&rpc_url);
 
         match provider {
             Some(value) => value.clone(),

diff --git a/ghost-crab/src/handler.rs b/ghost-crab/src/handler.rs
@@ -22,6 +22,7 @@ pub trait Handler {
     fn get_source(&self) -> String;
     fn is_template(&self) -> bool;
     fn network(&self) -> String;
+    fn rpc_url(&self) -> String;
     fn execution_mode(&self) -> ExecutionMode;
     fn get_event_signature(&self) -> String;
 }
@@ -32,6 +33,5 @@ pub struct HandlerConfig {
     pub step: u64,
     pub address: String,
     pub handler: HandleInstance,
-    pub provider: RootProvider<Http<Client>>,
     pub templates: TemplateManager,
 }
diff --git a/ghost-crab/src/indexer.rs b/ghost-crab/src/indexer.rs
@@ -1,5 +1,4 @@
 use crate::block_handler::{process_logs_block, BlockConfig, BlockHandlerInstance};
-use crate::cache::manager::RPC_MANAGER;
 use crate::config;
 use crate::handler::{HandleInstance, HandlerConfig};
 use crate::process_logs::process_logs;
@@ -19,14 +18,11 @@ pub struct Template {
 
 impl TemplateManager {
     pub async fn start(&self, template: Template) {
-        let provider = RPC_MANAGER.lock().await.get(template.handler.network()).await;
-
         self.tx
             .send(HandlerConfig {
                 start_block: template.start_block,
                 address: template.address.clone(),
                 step: 10_000,
-                provider,
                 handler: template.handler,
                 templates: self.clone(),
             })
@@ -73,27 +69,19 @@ impl Indexer {
             return;
         }
 
-        let provider = RPC_MANAGER.lock().await.get(handler.network()).await;
         let source = self.config.data_sources.get(&handler.get_source()).unwrap();
 
         self.handlers.push(HandlerConfig {
             start_block: source.start_block,
             address: source.address.clone(),
             step: 10_000,
-            provider,
             handler,
             templates: self.templates.clone(),
         });
     }
 
     pub async fn load_block_handler(&mut self, handler: BlockHandlerInstance) {
-        let provider = RPC_MANAGER.lock().await.get(handler.network()).await;
-
-        self.block_handlers.push(BlockConfig {
-            handler,
-            provider,
-            templates: self.templates.clone(),
-        });
+        self.block_handlers.push(BlockConfig { handler, templates: self.templates.clone() });
     }
 
     pub async fn start(mut self) {

diff --git a/ghost-crab/src/process_logs.rs b/ghost-crab/src/process_logs.rs
@@ -1,3 +1,4 @@
+use crate::cache::manager::RPC_MANAGER;
 use crate::handler::{Context, HandlerConfig};
 use crate::latest_block_manager::LatestBlockManager;
 use alloy::primitives::Address;
@@ -8,13 +9,17 @@ use ghost_crab_common::config::ExecutionMode;
 use std::time::Duration;
 
 pub async fn process_logs(
-    HandlerConfig { start_block, step, address, handler, provider, templates }: HandlerConfig,
+    HandlerConfig { start_block, step, address, handler, templates }: HandlerConfig,
 ) -> Result<(), TransportError> {
-    let mut current_block = start_block;
+    let network = handler.network();
+    let rpc_url = handler.rpc_url();
+    let execution_mode = handler.execution_mode();
     let event_signature = handler.get_event_signature();
+
+    let provider = RPC_MANAGER.lock().await.get_or_create(network, rpc_url).await;
+    let mut current_block = start_block;
     let address = address.parse::<Address>().unwrap();
 
-    let execution_mode = handler.execution_mode();
     let mut latest_block_manager =
         LatestBlockManager::new(provider.clone(), Duration::from_secs(10));