feat: Support conversions (#164)

Author: 0xNeshi

README.md

@@ -60,7 +60,7 @@ Create an event stream for the given event filters registered with the `EventSca
 
 ```rust
 use alloy::{network::Ethereum, providers::{Provider, ProviderBuilder}, sol_types::SolEvent};
-use event_scanner::{EventFilter, EventScannerBuilder, Message, robust_provider::RobustProvider};
+use event_scanner::{EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder};
 use tokio_stream::StreamExt;
 
 use crate::MyContract;
@@ -71,7 +71,7 @@ async fn run_scanner(
 ) -> Result<(), Box<dyn std::error::Error>> {
     // Connect to provider
     let provider = ProviderBuilder::new().connect(ws_url).await?;
-    let robust_provider = RobustProvider::new(provider);
+    let robust_provider = RobustProviderBuilder::new(provider).build().await?;
 
     // Configure scanner with custom batch size (optional)
     let mut scanner = EventScannerBuilder::live()
@@ -116,30 +116,33 @@ async fn run_scanner(
 `EventScannerBuilder` provides mode-specific constructors and functions to configure settings before connecting.
 Once configured, connect using:
 
-- `connect(robust_provider)` - Connect using a `RobustProvider` wrapping your alloy 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::RobustProvider;
+use event_scanner::robust_provider::RobustProviderBuilder;
 
 // Connect to provider (example with WebSocket)
 let provider = ProviderBuilder::new().connect("ws://localhost:8545").await?;
-let robust_provider = RobustProvider::new(provider);
 
 // 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(robust_provider.clone());
+    .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(robust_provider.clone());
+    .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)

examples/historical_scanning/main.rs

@@ -1,7 +1,9 @@
 use alloy::{providers::ProviderBuilder, sol, sol_types::SolEvent};
 use alloy_node_bindings::Anvil;
 
-use event_scanner::{EventFilter, EventScannerBuilder, Message, robust_provider::RobustProvider};
+use event_scanner::{
+    EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder,
+};
 use tokio_stream::StreamExt;
 use tracing::{error, info};
 use tracing_subscriber::EnvFilter;
@@ -52,12 +54,14 @@ async fn main() -> anyhow::Result<()> {
 
     let _ = counter_contract.increase().send().await?.get_receipt().await?;
 
-    let robust_provider = RobustProvider::new(provider)
+    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));
+        .min_delay(std::time::Duration::from_millis(500))
+        .build()
+        .await?;
 
-    let mut scanner = EventScannerBuilder::historic().connect(robust_provider);
+    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::{providers::ProviderBuilder, sol, sol_types::SolEvent};
 use alloy_node_bindings::Anvil;
-use event_scanner::{EventFilter, EventScannerBuilder, Message, robust_provider::RobustProvider};
+use event_scanner::{
+    EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder,
+};
 use tokio_stream::StreamExt;
 use tracing::{error, info};
 use tracing_subscriber::EnvFilter;
@@ -49,12 +51,14 @@ async fn main() -> anyhow::Result<()> {
         .contract_address(*contract_address)
         .event(Counter::CountIncreased::SIGNATURE);
 
-    let robust_provider = RobustProvider::new(provider)
+    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));
+        .min_delay(std::time::Duration::from_millis(500))
+        .build()
+        .await?;
 
-    let mut scanner = EventScannerBuilder::latest(5).connect(robust_provider);
+    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::{providers::ProviderBuilder, sol, sol_types::SolEvent};
 use alloy_node_bindings::Anvil;
-use event_scanner::{EventFilter, EventScannerBuilder, Message, robust_provider::RobustProvider};
+use event_scanner::{
+    EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder,
+};
 
 use tokio_stream::StreamExt;
 use tracing::{error, info};
@@ -50,12 +52,14 @@ async fn main() -> anyhow::Result<()> {
         .contract_address(*contract_address)
         .event(Counter::CountIncreased::SIGNATURE);
 
-    let robust_provider = RobustProvider::new(provider)
+    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));
+        .min_delay(std::time::Duration::from_millis(500))
+        .build()
+        .await?;
 
-    let mut scanner = EventScannerBuilder::live().connect(robust_provider);
+    let mut scanner = EventScannerBuilder::live().connect(robust_provider).await?;
 
     let mut stream = scanner.subscribe(increase_filter);
 

examples/sync_from_block_scanning/main.rs

@@ -2,7 +2,9 @@ use std::time::Duration;
 
 use alloy::{providers::ProviderBuilder, sol, sol_types::SolEvent};
 use alloy_node_bindings::Anvil;
-use event_scanner::{EventFilter, EventScannerBuilder, Message, robust_provider::RobustProvider};
+use event_scanner::{
+    EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder,
+};
 use tokio::time::sleep;
 use tokio_stream::StreamExt;
 use tracing::{error, info};
@@ -58,12 +60,14 @@ async fn main() -> anyhow::Result<()> {
         info!("Historical event {} created", i + 1);
     }
 
-    let robust_provider = RobustProvider::new(provider)
+    let robust_provider = RobustProviderBuilder::new(provider)
         .max_timeout(Duration::from_secs(30))
         .max_retries(5)
-        .min_delay(Duration::from_millis(500));
+        .min_delay(Duration::from_millis(500))
+        .build()
+        .await?;
 
-    let mut scanner = EventScannerBuilder::sync().from_block(0).connect(robust_provider);
+    let mut scanner = EventScannerBuilder::sync().from_block(0).connect(robust_provider).await?;
 
     let mut stream = scanner.subscribe(increase_filter);
 

examples/sync_from_latest_scanning/main.rs

@@ -1,6 +1,8 @@
 use alloy::{providers::ProviderBuilder, sol, sol_types::SolEvent};
 use alloy_node_bindings::Anvil;
-use event_scanner::{EventFilter, EventScannerBuilder, Message, robust_provider::RobustProvider};
+use event_scanner::{
+    EventFilter, EventScannerBuilder, Message, robust_provider::RobustProviderBuilder,
+};
 
 use tokio_stream::StreamExt;
 use tracing::{error, info};
@@ -50,12 +52,14 @@ async fn main() -> anyhow::Result<()> {
         .contract_address(*contract_address)
         .event(Counter::CountIncreased::SIGNATURE);
 
-    let robust_provider = RobustProvider::new(provider)
+    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));
+        .min_delay(std::time::Duration::from_millis(500))
+        .build()
+        .await?;
 
-    let mut client = EventScannerBuilder::sync().from_latest(5).connect(robust_provider);
+    let mut client = EventScannerBuilder::sync().from_latest(5).connect(robust_provider).await?;
 
     let mut stream = client.subscribe(increase_filter);
 

src/block_range_scanner.rs

@@ -12,7 +12,7 @@
 //!         BlockRangeScanner, BlockRangeScannerClient, DEFAULT_BLOCK_CONFIRMATIONS,
 //!         DEFAULT_MAX_BLOCK_RANGE, Message,
 //!     },
-//!     robust_provider::RobustProvider,
+//!     robust_provider::RobustProviderBuilder,
 //! };
 //! use tokio::time::Duration;
 //! use tracing::{error, info};
@@ -24,8 +24,8 @@
 //!
 //!     // Configuration
 //!     let provider = ProviderBuilder::new().connect("ws://localhost:8546").await?;
-//!     let robust_provider = RobustProvider::new(provider);
-//!     let block_range_scanner = BlockRangeScanner::new().connect(robust_provider);
+//!     let robust_provider = RobustProviderBuilder::new(provider).build().await?;
+//!     let block_range_scanner = BlockRangeScanner::new().connect(robust_provider).await?;
 //!
 //!     // Create client to send subscribe command to block scanner
 //!     let client: BlockRangeScannerClient = block_range_scanner.run()?;
@@ -69,7 +69,7 @@ use tokio_stream::{StreamExt, wrappers::ReceiverStream};
 use crate::{
     ScannerMessage,
     error::ScannerError,
-    robust_provider::{Error as RobustProviderError, RobustProvider},
+    robust_provider::{Error as RobustProviderError, IntoRobustProvider, RobustProvider},
     types::{ScannerStatus, TryStream},
 };
 use alloy::{
@@ -148,9 +148,16 @@ impl BlockRangeScanner {
     }
 
     /// Connects to an existing provider
-    #[must_use]
-    pub fn connect<N: Network>(self, provider: RobustProvider<N>) -> ConnectedBlockRangeScanner<N> {
-        ConnectedBlockRangeScanner { provider, max_block_range: self.max_block_range }
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the provider connection fails.
+    pub async fn connect<N: Network>(
+        self,
+        provider: impl IntoRobustProvider<N>,
+    ) -> Result<ConnectedBlockRangeScanner<N>, ScannerError> {
+        let provider = provider.into_robust_provider().await?;
+        Ok(ConnectedBlockRangeScanner { provider, max_block_range: self.max_block_range })
     }
 }