feat: add fallback provider logic to event scanners

Author: LeoPatOZ

src/block_range_scanner.rs

@@ -217,7 +217,9 @@ impl BlockRangeScanner {
         let provider =
             RootProvider::<N>::new(ClientBuilder::default().ws(WsConnect::new(ws_url)).await?);
 
-        let fallback_providers = self.connect_all_fallbacks::<N>().await?;
+        let fallback_providers =
+            Self::connect_all_fallbacks::<N>(&self.fallback_ws_urls, &self.fallback_ipc_paths)
+                .await?;
 
         Ok(self.connect_with_fallbacks(provider, fallback_providers))
     }
@@ -237,7 +239,9 @@ impl BlockRangeScanner {
         let provider =
             RootProvider::<N>::new(ClientBuilder::default().ipc(IpcConnect::new(ipc_path)).await?);
 
-        let fallback_providers = self.connect_all_fallbacks::<N>().await?;
+        let fallback_providers =
+            Self::connect_all_fallbacks::<N>(&self.fallback_ws_urls, &self.fallback_ipc_paths)
+                .await?;
 
         Ok(self.connect_with_fallbacks(provider, fallback_providers))
     }
@@ -253,37 +257,16 @@ impl BlockRangeScanner {
         self,
         provider: RootProvider<N>,
     ) -> Result<ConnectedBlockRangeScanner<N>, RpcError<TransportErrorKind>> {
-        let fallback_providers = self.connect_all_fallbacks::<N>().await?;
+        let fallback_providers =
+            Self::connect_all_fallbacks::<N>(&self.fallback_ws_urls, &self.fallback_ipc_paths)
+                .await?;
 
         Ok(self.connect_with_fallbacks(provider, fallback_providers))
     }
 
-    /// Establishes connections to all configured fallback providers (both WebSocket and IPC).
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if any fallback connection fails
-    async fn connect_all_fallbacks<N: Network>(
-        &self,
-    ) -> Result<Vec<RootProvider<N>>, RpcError<TransportErrorKind>> {
-        let mut fallback_providers = Vec::new();
-
-        for url in &self.fallback_ws_urls {
-            let client = ClientBuilder::default().ws(WsConnect::new(url.clone())).await?;
-            fallback_providers.push(RootProvider::<N>::new(client));
-        }
-
-        for path in &self.fallback_ipc_paths {
-            let client = ClientBuilder::default().ipc(IpcConnect::new(path.clone())).await?;
-            fallback_providers.push(RootProvider::<N>::new(client));
-        }
-
-        Ok(fallback_providers)
-    }
-
     /// Connects to an existing provider with fallback providers
     #[must_use]
-    fn connect_with_fallbacks<N: Network>(
+    pub fn connect_with_fallbacks<N: Network>(
         self,
         provider: RootProvider<N>,
         fallback_providers: Vec<RootProvider<N>>,
@@ -302,6 +285,30 @@ impl BlockRangeScanner {
             max_block_range: self.max_block_range,
         }
     }
+
+    /// Establishes connections to all configured fallback providers (both WebSocket and IPC).
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if any fallback connection fails
+    async fn connect_all_fallbacks<N: Network>(
+        fallback_ws_urls: &Vec<Url>,
+        fallback_ipc_paths: &Vec<String>,
+    ) -> Result<Vec<RootProvider<N>>, RpcError<TransportErrorKind>> {
+        let mut fallback_providers = Vec::new();
+
+        for url in fallback_ws_urls {
+            let client = ClientBuilder::default().ws(WsConnect::new(url.clone())).await?;
+            fallback_providers.push(RootProvider::<N>::new(client));
+        }
+
+        for path in fallback_ipc_paths {
+            let client = ClientBuilder::default().ipc(IpcConnect::new(path.clone())).await?;
+            fallback_providers.push(RootProvider::<N>::new(client));
+        }
+
+        Ok(fallback_providers)
+    }
 }
 
 pub struct ConnectedBlockRangeScanner<N: Network> {

src/event_scanner/scanner/mod.rs

@@ -365,9 +365,62 @@ impl EventScannerBuilder<SyncFromBlock> {
 }
 
 impl<M> EventScannerBuilder<M> {
+    /// Adds a fallback WebSocket URL to the scanner.
+    ///
+    /// The WebSocket connection will be established when calling the `connect` methods.
+    /// Multiple fallback providers can be added by calling this method multiple times.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// # use alloy::network::Ethereum;
+    /// # use event_scanner::EventScannerBuilder;
+    /// #
+    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
+    /// # let ws_url = "ws://localhost:8545".parse()?;
+    /// # let fallback_url = "ws://fallback:8545".parse()?;
+    /// let scanner = EventScannerBuilder::historic()
+    ///     .fallback_ws(fallback_url)
+    ///     .connect_ws::<Ethereum>(ws_url)
+    ///     .await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    #[must_use]
+    pub fn fallback_ws(mut self, url: Url) -> Self {
+        self.block_range_scanner = self.block_range_scanner.fallback_ws(url);
+        self
+    }
+
+    /// Adds a fallback IPC path to the scanner.
+    ///
+    /// The IPC connection will be established when calling the `connect` methods.
+    /// Multiple fallback providers can be added by calling this method multiple times.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// # use alloy::network::Ethereum;
+    /// # use event_scanner::EventScannerBuilder;
+    /// #
+    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
+    /// # let ws_url = "ws://localhost:8545".parse()?;
+    /// let scanner = EventScannerBuilder::historic()
+    ///     .fallback_ipc("/tmp/fallback.ipc".to_string())
+    ///     .connect_ws::<Ethereum>(ws_url)
+    ///     .await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    #[must_use]
+    pub fn fallback_ipc(mut self, path: String) -> Self {
+        self.block_range_scanner = self.block_range_scanner.fallback_ipc(path);
+        self
+    }
+
     /// Connects to the provider via WebSocket.
     ///
-    /// Final builder method: consumes the builder and returns the built [`HistoricEventScanner`].
+    /// Final builder method: consumes the builder and returns the built [`EventScanner`].
     ///
     /// # Errors
     ///
@@ -379,7 +432,7 @@ impl<M> EventScannerBuilder<M> {
 
     /// Connects to the provider via IPC.
     ///
-    /// Final builder method: consumes the builder and returns the built [`HistoricEventScanner`].
+    /// Final builder method: consumes the builder and returns the built [`EventScanner`].
     ///
     /// # Errors
     ///
@@ -394,18 +447,31 @@ impl<M> EventScannerBuilder<M> {
 
     /// Connects to an existing provider.
     ///
-    /// Final builder method: consumes the builder and returns the built [`HistoricEventScanner`].
+    /// Final builder method: consumes the builder and returns the built [`EventScanner`].
     ///
     /// # Errors
     ///
-    /// Returns an error if the connection fails
+    /// Returns an error if any fallback connection fails
     pub async fn connect<N: Network>(
         self,
         provider: RootProvider<N>,
     ) -> TransportResult<EventScanner<M, N>> {
         let block_range_scanner = self.block_range_scanner.connect::<N>(provider).await?;
         Ok(EventScanner { config: self.config, block_range_scanner, listeners: Vec::new() })
     }
+
+    /// Connects to an existing provider with fallback providers
+    ///
+    /// Final builder method: consumes the builder and returns the built [`EventScanner`].
+    pub fn connect_with_fallbacks<N: Network>(
+        self,
+        provider: RootProvider<N>,
+        fallback_providers: Vec<RootProvider<N>>,
+    ) -> EventScanner<M, N> {
+        let block_range_scanner =
+            self.block_range_scanner.connect_with_fallbacks::<N>(provider, fallback_providers);
+        EventScanner { config: self.config, block_range_scanner, listeners: Vec::new() }
+    }
 }
 
 impl<M, N: Network> EventScanner<M, N> {
@@ -485,4 +551,40 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_scanner_builder_fallback_methods() {
+        let ws_url: Url = "ws://fallback:8545".parse().unwrap();
+        let ipc_path = "/tmp/fallback.ipc".to_string();
+
+        let builder = EventScannerBuilder::historic()
+            .fallback_ws(ws_url.clone())
+            .fallback_ipc(ipc_path.clone());
+
+        assert_eq!(builder.block_range_scanner.fallback_ws_urls.len(), 1);
+        assert_eq!(builder.block_range_scanner.fallback_ws_urls[0], ws_url);
+        assert_eq!(builder.block_range_scanner.fallback_ipc_paths.len(), 1);
+        assert_eq!(builder.block_range_scanner.fallback_ipc_paths[0], ipc_path);
+    }
+
+    #[test]
+    fn test_scanner_builder_multiple_fallbacks() {
+        let ws_url1: Url = "ws://fallback1:8545".parse().unwrap();
+        let ws_url2: Url = "ws://fallback2:8545".parse().unwrap();
+        let ipc_path1 = "/tmp/fallback1.ipc".to_string();
+        let ipc_path2 = "/tmp/fallback2.ipc".to_string();
+
+        let builder = EventScannerBuilder::live()
+            .fallback_ws(ws_url1.clone())
+            .fallback_ws(ws_url2.clone())
+            .fallback_ipc(ipc_path1.clone())
+            .fallback_ipc(ipc_path2.clone());
+
+        assert_eq!(builder.block_range_scanner.fallback_ws_urls.len(), 2);
+        assert_eq!(builder.block_range_scanner.fallback_ws_urls[0], ws_url1);
+        assert_eq!(builder.block_range_scanner.fallback_ws_urls[1], ws_url2);
+        assert_eq!(builder.block_range_scanner.fallback_ipc_paths.len(), 2);
+        assert_eq!(builder.block_range_scanner.fallback_ipc_paths[0], ipc_path1);
+        assert_eq!(builder.block_range_scanner.fallback_ipc_paths[1], ipc_path2);
+    }
 }