feat: Add Multiple providers (#140)

Co-authored-by: Nenad <[email protected]>

Author: LeoPatOZ

README.md

@@ -59,20 +59,24 @@ event-scanner = "0.4.0-alpha"
 Create an event stream for the given event filters registered with the `EventScanner`:
 
 ```rust
-use alloy::{network::Ethereum, sol_types::SolEvent};
-use event_scanner::{EventFilter, EventScannerBuilder, Message};
+use alloy::{network::Ethereum, providers::{Provider, ProviderBuilder}, sol_types::SolEvent};
+use event_scanner::{EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder};
 use tokio_stream::StreamExt;
 
 use crate::MyContract;
 
 async fn run_scanner(
-    ws_url: alloy::transports::http::reqwest::Url,
+    ws_url: &str,
     contract: alloy::primitives::Address,
 ) -> Result<(), Box<dyn std::error::Error>> {
+    // Connect to provider
+    let provider = ProviderBuilder::new().connect(ws_url).await?;
+    let robust_provider = RobustProviderBuilder::new(provider).build().await?;
+    
     // Configure scanner with custom batch size (optional)
     let mut scanner = EventScannerBuilder::live()
         .max_block_range(500)  // Process up to 500 blocks per batch
-        .connect_ws::<Ethereum>(ws_url).await?;
+        .connect(robust_provider);
 
     // Register an event listener
     let filter = EventFilter::new()
@@ -109,48 +113,56 @@ async fn run_scanner(
 
 ### Building a Scanner
 
-`EventScannerBuilder` provides mode-specific constructors and a functions to configure settings before connecting.
-Once configured, connect using one of:
+`EventScannerBuilder` provides mode-specific constructors and functions to configure settings before connecting.
+Once configured, connect using:
 
-- `connect_ws::<Ethereum>(ws_url)`
-- `connect_ipc::<Ethereum>(path)`
-- `connect::<Ethereum>(provider)`
+- `connect(provider)` - Connect using a `RobustProvider` wrapping your alloy provider or using an alloy provider directly
 
 This will connect the `EventScanner` and allow you to create event streams and start scanning in various [modes](#scanning-modes).
 
 ```rust
+use alloy::providers::{Provider, ProviderBuilder};
+use event_scanner::robust_provider::RobustProviderBuilder;
+
+// Connect to provider (example with WebSocket)
+let provider = ProviderBuilder::new().connect("ws://localhost:8545").await?;
+
 // Live streaming mode
 let scanner = EventScannerBuilder::live()
     .max_block_range(500)  // Optional: set max blocks per read (default: 1000)
     .block_confirmations(12)  // Optional: set block confirmations (default: 12)
-    .connect_ws::<Ethereum>(ws_url).await?;
+    .connect(provider.clone());
 
 // Historical block range mode
 let scanner = EventScannerBuilder::historic()
     .from_block(1_000_000)
     .to_block(2_000_000)
     .max_block_range(500)
-    .connect_ws::<Ethereum>(ws_url).await?;
+    .connect(provider.clone());
+
+// we can also wrap the provider in a RobustProvider
+// for more advanced configurations like retries and fallbacks
+let robust_provider = RobustProviderBuilder::new(provider).build().await?;
 
 // Latest events mode
 let scanner = EventScannerBuilder::latest(100)
     // .from_block(1_000_000)  // Optional: set start of search range
     // .to_block(2_000_000)    // Optional: set end of search range
     .max_block_range(500)
-    .connect_ws::<Ethereum>(ws_url).await?;
+    .connect(robust_provider.clone());
 
 // Sync from block then switch to live mode
 let scanner = EventScannerBuilder::sync()
     .from_block(100)
     .max_block_range(500)
     .block_confirmations(12)
-    .connect_ws::<Ethereum>(ws_url).await?;
+    .connect(robust_provider.clone());
 
 // Sync the latest 60 events then switch to live mode
 let scanner = EventScannerBuilder::sync()
     .from_latest(60)
     .block_confirmations(12)
-    .connect_ws::<Ethereum>(ws_url).await?;
+    .connect(robust_provider);
 ```
 
 Invoking `scanner.start()` starts the scanner in the specified mode.

examples/historical_scanning/main.rs

@@ -1,7 +1,9 @@
-use alloy::{network::Ethereum, providers::ProviderBuilder, sol, sol_types::SolEvent};
+use alloy::{providers::ProviderBuilder, sol, sol_types::SolEvent};
 use alloy_node_bindings::Anvil;
 
-use event_scanner::{EventFilter, EventScannerBuilder, Message};
+use event_scanner::{
+    EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder,
+};
 use tokio_stream::StreamExt;
 use tracing::{error, info};
 use tracing_subscriber::EnvFilter;
@@ -38,9 +40,11 @@ async fn main() -> anyhow::Result<()> {
 
     let anvil = Anvil::new().block_time_f64(0.1).try_spawn()?;
     let wallet = anvil.wallet();
-    let provider =
-        ProviderBuilder::new().wallet(wallet.unwrap()).connect(anvil.endpoint().as_str()).await?;
-    let counter_contract = Counter::deploy(provider).await?;
+    let provider = ProviderBuilder::new()
+        .wallet(wallet.unwrap())
+        .connect(anvil.ws_endpoint_url().as_str())
+        .await?;
+    let counter_contract = Counter::deploy(provider.clone()).await?;
 
     let contract_address = counter_contract.address();
 
@@ -50,8 +54,14 @@ async fn main() -> anyhow::Result<()> {
 
     let _ = counter_contract.increase().send().await?.get_receipt().await?;
 
-    let mut scanner =
-        EventScannerBuilder::historic().connect_ws::<Ethereum>(anvil.ws_endpoint_url()).await?;
+    let robust_provider = RobustProviderBuilder::new(provider)
+        .max_timeout(std::time::Duration::from_secs(30))
+        .max_retries(5)
+        .min_delay(std::time::Duration::from_millis(500))
+        .build()
+        .await?;
+
+    let mut scanner = EventScannerBuilder::historic().connect(robust_provider).await?;
 
     let mut stream = scanner.subscribe(increase_filter);
 

examples/latest_events_scanning/main.rs

@@ -1,6 +1,8 @@
-use alloy::{network::Ethereum, providers::ProviderBuilder, sol, sol_types::SolEvent};
+use alloy::{providers::ProviderBuilder, sol, sol_types::SolEvent};
 use alloy_node_bindings::Anvil;
-use event_scanner::{EventFilter, EventScannerBuilder, Message};
+use event_scanner::{
+    EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder,
+};
 use tokio_stream::StreamExt;
 use tracing::{error, info};
 use tracing_subscriber::EnvFilter;
@@ -37,18 +39,26 @@ async fn main() -> anyhow::Result<()> {
 
     let anvil = Anvil::new().block_time_f64(0.5).try_spawn()?;
     let wallet = anvil.wallet();
-    let provider =
-        ProviderBuilder::new().wallet(wallet.unwrap()).connect(anvil.endpoint().as_str()).await?;
-    let counter_contract = Counter::deploy(provider).await?;
+    let provider = ProviderBuilder::new()
+        .wallet(wallet.unwrap())
+        .connect(anvil.ws_endpoint_url().as_str())
+        .await?;
+    let counter_contract = Counter::deploy(provider.clone()).await?;
 
     let contract_address = counter_contract.address();
 
     let increase_filter = EventFilter::new()
         .contract_address(*contract_address)
         .event(Counter::CountIncreased::SIGNATURE);
 
-    let mut scanner =
-        EventScannerBuilder::latest(5).connect_ws::<Ethereum>(anvil.ws_endpoint_url()).await?;
+    let robust_provider = RobustProviderBuilder::new(provider)
+        .max_timeout(std::time::Duration::from_secs(30))
+        .max_retries(5)
+        .min_delay(std::time::Duration::from_millis(500))
+        .build()
+        .await?;
+
+    let mut scanner = EventScannerBuilder::latest(5).connect(robust_provider).await?;
 
     let mut stream = scanner.subscribe(increase_filter);
 

examples/live_scanning/main.rs

@@ -1,6 +1,8 @@
-use alloy::{network::Ethereum, providers::ProviderBuilder, sol, sol_types::SolEvent};
+use alloy::{providers::ProviderBuilder, sol, sol_types::SolEvent};
 use alloy_node_bindings::Anvil;
-use event_scanner::{EventFilter, EventScannerBuilder, Message};
+use event_scanner::{
+    EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder,
+};
 
 use tokio_stream::StreamExt;
 use tracing::{error, info};
@@ -38,18 +40,26 @@ async fn main() -> anyhow::Result<()> {
 
     let anvil = Anvil::new().block_time(1).try_spawn()?;
     let wallet = anvil.wallet();
-    let provider =
-        ProviderBuilder::new().wallet(wallet.unwrap()).connect(anvil.endpoint().as_str()).await?;
-    let counter_contract = Counter::deploy(provider).await?;
+    let provider = ProviderBuilder::new()
+        .wallet(wallet.unwrap())
+        .connect(anvil.ws_endpoint_url().as_str())
+        .await?;
+    let counter_contract = Counter::deploy(provider.clone()).await?;
 
     let contract_address = counter_contract.address();
 
     let increase_filter = EventFilter::new()
         .contract_address(*contract_address)
         .event(Counter::CountIncreased::SIGNATURE);
 
-    let mut scanner =
-        EventScannerBuilder::live().connect_ws::<Ethereum>(anvil.ws_endpoint_url()).await?;
+    let robust_provider = RobustProviderBuilder::new(provider)
+        .max_timeout(std::time::Duration::from_secs(30))
+        .max_retries(5)
+        .min_delay(std::time::Duration::from_millis(500))
+        .build()
+        .await?;
+
+    let mut scanner = EventScannerBuilder::live().connect(robust_provider).await?;
 
     let mut stream = scanner.subscribe(increase_filter);
 

examples/sync_from_block_scanning/main.rs

@@ -1,8 +1,10 @@
 use std::time::Duration;
 
-use alloy::{network::Ethereum, providers::ProviderBuilder, sol, sol_types::SolEvent};
+use alloy::{providers::ProviderBuilder, sol, sol_types::SolEvent};
 use alloy_node_bindings::Anvil;
-use event_scanner::{EventFilter, EventScannerBuilder, Message};
+use event_scanner::{
+    EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder,
+};
 use tokio::time::sleep;
 use tokio_stream::StreamExt;
 use tracing::{error, info};
@@ -40,9 +42,11 @@ async fn main() -> anyhow::Result<()> {
 
     let anvil = Anvil::new().block_time(1).try_spawn()?;
     let wallet = anvil.wallet();
-    let provider =
-        ProviderBuilder::new().wallet(wallet.unwrap()).connect(anvil.endpoint().as_str()).await?;
-    let counter_contract = Counter::deploy(provider).await?;
+    let provider = ProviderBuilder::new()
+        .wallet(wallet.unwrap())
+        .connect(anvil.ws_endpoint_url().as_str())
+        .await?;
+    let counter_contract = Counter::deploy(provider.clone()).await?;
 
     let contract_address = counter_contract.address();
 
@@ -56,11 +60,15 @@ async fn main() -> anyhow::Result<()> {
         info!("Historical event {} created", i + 1);
     }
 
-    let mut scanner = EventScannerBuilder::sync()
-        .from_block(0)
-        .connect_ws::<Ethereum>(anvil.ws_endpoint_url())
+    let robust_provider = RobustProviderBuilder::new(provider)
+        .max_timeout(Duration::from_secs(30))
+        .max_retries(5)
+        .min_delay(Duration::from_millis(500))
+        .build()
         .await?;
 
+    let mut scanner = EventScannerBuilder::sync().from_block(0).connect(robust_provider).await?;
+
     let mut stream = scanner.subscribe(increase_filter);
 
     info!("Starting sync scanner...");

examples/sync_from_latest_scanning/main.rs

@@ -1,6 +1,8 @@
-use alloy::{network::Ethereum, providers::ProviderBuilder, sol, sol_types::SolEvent};
+use alloy::{providers::ProviderBuilder, sol, sol_types::SolEvent};
 use alloy_node_bindings::Anvil;
-use event_scanner::{EventFilter, EventScannerBuilder, Message};
+use event_scanner::{
+    EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder,
+};
 
 use tokio_stream::StreamExt;
 use tracing::{error, info};
@@ -38,21 +40,27 @@ async fn main() -> anyhow::Result<()> {
 
     let anvil = Anvil::new().block_time_f64(0.5).try_spawn()?;
     let wallet = anvil.wallet();
-    let provider =
-        ProviderBuilder::new().wallet(wallet.unwrap()).connect(anvil.endpoint().as_str()).await?;
-    let counter_contract = Counter::deploy(provider).await?;
+    let provider = ProviderBuilder::new()
+        .wallet(wallet.unwrap())
+        .connect(anvil.ws_endpoint_url().as_str())
+        .await?;
+    let counter_contract = Counter::deploy(provider.clone()).await?;
 
     let contract_address = counter_contract.address();
 
     let increase_filter = EventFilter::new()
         .contract_address(*contract_address)
         .event(Counter::CountIncreased::SIGNATURE);
 
-    let mut client = EventScannerBuilder::sync()
-        .from_latest(5)
-        .connect_ws::<Ethereum>(anvil.ws_endpoint_url())
+    let robust_provider = RobustProviderBuilder::new(provider)
+        .max_timeout(std::time::Duration::from_secs(30))
+        .max_retries(5)
+        .min_delay(std::time::Duration::from_millis(500))
+        .build()
         .await?;
 
+    let mut client = EventScannerBuilder::sync().from_latest(5).connect(robust_provider).await?;
+
     let mut stream = client.subscribe(increase_filter);
 
     for _ in 0..10 {