Make Eventfilter fields optional (#60)

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

Author: 3 people

README.md

@@ -74,11 +74,10 @@ impl EventCallback for CounterCallback {
 }
 
 async fn run_scanner(ws_url: alloy::transports::http::reqwest::Url, contract: alloy::primitives::Address) -> anyhow::Result<()> {
-    let filter = EventFilter {
-        contract_address: contract,
-        event: MyContract::SomeEvent::SIGNATURE.to_owned(),
-        callback: Arc::new(CounterCallback { processed: Arc::new(AtomicUsize::new(0)) }),
-    };
+    let filter = EventFilter::new()
+        .with_contract_address(contract)
+        .with_event(MyContract::SomeEvent::SIGNATURE)
+        .with_callback(Arc::new(CounterCallback { processed: Arc::new(AtomicUsize::new(0)) }));
 
     let mut scanner = EventScannerBuilder::new()
         .with_event_filter(filter)
@@ -111,16 +110,67 @@ Once configured, connect using either `connect_ws::<Ethereum>(ws_url)` or `conne
 
 Create an `EventFilter` for each contract/event pair you want to track. The filter bundles the contract address, the event signature (from `SolEvent::SIGNATURE`), and an `Arc<dyn EventCallback + Send + Sync>`.
 
+Both `contract_address` and `event` fields are optional, allowing for flexible event tracking.
+
+You can construct EventFilters using either the builder pattern (recommended) or direct struct construction:
+
+### Builder Pattern (Recommended)
+
+```rust
+// Track a specific event from a specific contract
+let specific_filter = EventFilter::new()
+    .with_contract_address(*counter_contract.address())
+    .with_event(Counter::CountIncreased::SIGNATURE)
+    .with_callback(Arc::new(CounterCallback));
+
+// Track ALL events from a specific contract
+let all_contract_events_filter = EventFilter::new()
+    .with_contract_address(*counter_contract.address())
+    .with_callback(Arc::new(AllEventsCallback));
+
+// Track ALL events from ALL contracts in the block range
+let all_events_filter = EventFilter::new()
+    .with_callback(Arc::new(GlobalEventsCallback));
+```
+
+### Direct Struct Construction
+
 ```rust
-let filter = EventFilter {
-    contract_address: *counter_contract.address(),
-    event: Counter::CountIncreased::SIGNATURE.to_owned(),
+// Track a specific event from a specific contract (traditional usage)
+let specific_filter = EventFilter {
+    contract_address: Some(*counter_contract.address()),
+    event: Some(Counter::CountIncreased::SIGNATURE.to_owned()),
     callback: Arc::new(CounterCallback),
 };
+
+// Track ALL events from a specific contract
+let all_contract_events_filter = EventFilter {
+    contract_address: Some(*counter_contract.address()),
+    event: None, // Will track all events from this contract
+    callback: Arc::new(AllEventsCallback),
+};
+
+// Track ALL events from ALL contracts in the block range
+let all_events_filter = EventFilter {
+    contract_address: None, // Will track events from all contracts
+    event: None,            // Will track all event types
+    callback: Arc::new(GlobalEventsCallback),
+};
 ```
 
 Register multiple filters by calling either `with_event_filter` repeatedly or `with_event_filters` once.
 
+#### Use Cases for Optional Fields
+
+The optional `contract_address` and `event` fields enable several powerful use cases:
+
+- **Track all events from a specific contract**: Set `contract_address` but leave `event` as `None`
+- **Track all events across all contracts**: Set both `contract_address` and `event` as `None`
+- **Track specific events from specific contracts**: Set both fields (traditional usage)
+- **Mixed filtering**: Use multiple filters with different optional field combinations
+
+This flexibility allows you to build sophisticated event monitoring systems that can track events at different granularities depending on your application's needs.
+
 
 ### Scanning Modes
 

examples/historical_scanning/main.rs

@@ -59,11 +59,10 @@ async fn main() -> anyhow::Result<()> {
 
     let contract_address = counter_contract.address();
 
-    let increase_filter = EventFilter {
-        contract_address: *contract_address,
-        event: Counter::CountIncreased::SIGNATURE.to_owned(),
-        callback: Arc::new(CounterCallback),
-    };
+    let increase_filter = EventFilter::new()
+        .with_contract_address(*contract_address)
+        .with_event(Counter::CountIncreased::SIGNATURE)
+        .with_callback(Arc::new(CounterCallback));
 
     let _ = counter_contract.increase().send().await?.get_receipt().await?;
 

examples/simple_counter/main.rs

@@ -59,11 +59,10 @@ async fn main() -> anyhow::Result<()> {
 
     let contract_address = counter_contract.address();
 
-    let increase_filter = EventFilter {
-        contract_address: *contract_address,
-        event: Counter::CountIncreased::SIGNATURE.to_owned(),
-        callback: Arc::new(CounterCallback),
-    };
+    let increase_filter = EventFilter::new()
+        .with_contract_address(*contract_address)
+        .with_event(Counter::CountIncreased::SIGNATURE)
+        .with_callback(Arc::new(CounterCallback));
 
     let mut scanner = EventScannerBuilder::new()
         .with_event_filter(increase_filter)

src/event_filter.rs

@@ -0,0 +1,82 @@
+use std::sync::Arc;
+
+use alloy::primitives::Address;
+
+use crate::callback::EventCallback;
+
+#[derive(Clone)]
+pub struct EventFilter {
+    /// Contract address to filter events from. If None, events from all contracts will be tracked.
+    pub contract_address: Option<Address>,
+    /// Human-readable event signature, e.g. "Transfer(address,address,uint256)".
+    /// If None, all events from the specified contract(s) will be tracked.
+    pub event: Option<String>,
+    pub callback: Arc<dyn EventCallback + Send + Sync>,
+}
+
+impl EventFilter {
+    /// Creates a new `EventFilter` builder.
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// use event_scanner::{EventFilter, EventCallback};
+    /// use std::sync::Arc;
+    /// use alloy::primitives::Address;
+    ///
+    /// # struct MyCallback;
+    /// # #[async_trait::async_trait]
+    /// # impl EventCallback for MyCallback {
+    /// #     async fn on_event(&self, _log: &alloy::rpc::types::Log) -> anyhow::Result<()> { Ok(()) }
+    /// # }
+    /// # async fn example() -> anyhow::Result<()> {
+    /// let contract_address = Address::ZERO;
+    /// let callback = Arc::new(MyCallback);
+    /// let filter = EventFilter::new()
+    ///     .with_contract_address(contract_address)
+    ///     .with_event("Transfer(address,address,uint256)")
+    ///     .with_callback(callback);
+    /// # Ok(())
+    /// # }
+    /// ```
+    #[must_use]
+    #[allow(clippy::new_ret_no_self)]
+    pub fn new() -> EventFilterBuilder {
+        EventFilterBuilder::default()
+    }
+}
+
+/// Builder for constructing `EventFilter` instances with optional fields.
+#[derive(Default)]
+pub struct EventFilterBuilder {
+    contract_address: Option<Address>,
+    event: Option<String>,
+}
+
+impl EventFilterBuilder {
+    /// Sets the contract address to filter events from.
+    /// If not set, events from all contracts will be tracked.
+    #[must_use]
+    pub fn with_contract_address(mut self, contract_address: Address) -> Self {
+        self.contract_address = Some(contract_address);
+        self
+    }
+
+    /// Sets the event signature to filter specific events.
+    /// If not set, all events from the specified contract(s) will be tracked.
+    #[must_use]
+    pub fn with_event(mut self, event: impl Into<String>) -> Self {
+        self.event = Some(event.into());
+        self
+    }
+
+    /// Sets the callback for processing events and builds the `EventFilter`.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the callback is not set, as it's required for event processing.
+    #[must_use]
+    pub fn with_callback(self, callback: Arc<dyn EventCallback + Send + Sync>) -> EventFilter {
+        EventFilter { contract_address: self.contract_address, event: self.event, callback }
+    }
+}

src/event_scanner.rs

@@ -3,7 +3,7 @@ use std::{collections::HashMap, sync::Arc};
 use crate::{
     block_range_scanner::{self, BlockRangeScanner, ConnectedBlockRangeScanner},
     callback::strategy::{CallbackStrategy, StateSyncAwareStrategy},
-    types::EventFilter,
+    event_filter::EventFilter,
 };
 use alloy::{
     eips::BlockNumberOrTag,
@@ -134,8 +134,8 @@ pub struct EventScanner<N: Network> {
 
 #[derive(Hash, Eq, PartialEq)]
 struct EventIdentifier {
-    contract_address: Address,
-    event: String,
+    contract_address: Option<Address>,
+    event: Option<String>,
 }
 
 impl<N: Network> EventScanner<N> {
@@ -165,12 +165,8 @@ impl<N: Network> EventScanner<N> {
 
             let callback = filter.callback.clone();
             let strategy = self.callback_strategy.clone();
-            Self::spawn_event_callback_task_executors(
-                receiver,
-                callback,
-                strategy,
-                filter.event.clone(),
-            );
+            let event_name = filter.event.clone().unwrap_or_else(|| "all events".to_string());
+            Self::spawn_event_callback_task_executors(receiver, callback, strategy, event_name);
 
             event_channels.insert(unique_event, sender);
         }
@@ -225,20 +221,31 @@ impl<N: Network> EventScanner<N> {
         event_channels: &HashMap<EventIdentifier, mpsc::Sender<Log>>,
     ) -> anyhow::Result<()> {
         for event_filter in &self.tracked_events {
-            let filter = Filter::new()
-                .address(event_filter.contract_address)
-                .event(event_filter.event.as_str())
-                .from_block(from_block)
-                .to_block(to_block);
+            let mut filter = Filter::new().from_block(from_block).to_block(to_block);
+
+            // Add contract address filter if specified
+            if let Some(contract_address) = event_filter.contract_address {
+                filter = filter.address(contract_address);
+            }
+
+            // Add event signature filter if specified
+            if let Some(event_signature) = &event_filter.event {
+                filter = filter.event(event_signature.as_str());
+            }
 
             match self.block_range_scanner.provider().get_logs(&filter).await {
                 Ok(logs) => {
                     if logs.is_empty() {
                         continue;
                     }
+                    let contract_display = event_filter
+                        .contract_address
+                        .map_or_else(|| "all contracts".to_string(), |addr| format!("{addr:?}"));
+                    let event_display = event_filter.event.as_deref().map_or("all events", |s| s);
+
                     info!(
-                        contract = ?event_filter.contract_address,
-                        event = %event_filter.event,
+                        contract = %contract_display,
+                        event = %event_display,
                         log_count = logs.len(),
                         from_block,
                         to_block,
@@ -253,17 +260,34 @@ impl<N: Network> EventScanner<N> {
                     if let Some(sender) = event_channels.get(&event_identifier) {
                         for log in logs {
                             if let Err(e) = sender.send(log).await {
-                                warn!(event = %event_filter.event, error = %e, "failed to enqueue log for processing");
+                                let contract_display = event_filter.contract_address.map_or_else(
+                                    || "all contracts".to_string(),
+                                    |addr| format!("{addr:?}"),
+                                );
+                                let event_display =
+                                    event_filter.event.as_deref().map_or("all events", |s| s);
+                                warn!(contract = %contract_display, event = %event_display, error = %e, "failed to enqueue log for processing");
                             }
                         }
                     } else {
-                        warn!(event = %event_filter.event, "no channel found for event type");
+                        let contract_display = event_filter.contract_address.map_or_else(
+                            || "all contracts".to_string(),
+                            |addr| format!("{addr:?}"),
+                        );
+                        let event_display =
+                            event_filter.event.as_deref().map_or("all events", |s| s);
+                        warn!(contract = %contract_display, event = %event_display, "no channel found for event type");
                     }
                 }
                 Err(e) => {
+                    let contract_display = event_filter
+                        .contract_address
+                        .map_or_else(|| "all contracts".to_string(), |addr| format!("{addr:?}"));
+                    let event_display = event_filter.event.as_deref().map_or("all events", |s| s);
+
                     error!(
-                        contract = ?event_filter.contract_address,
-                        event = %event_filter.event,
+                        contract = %contract_display,
+                        event = %event_display,
                         error = %e,
                         from_block,
                         to_block,

src/lib.rs

@@ -1,7 +1,7 @@
 pub mod block_range_scanner;
 pub mod callback;
+pub mod event_filter;
 pub mod event_scanner;
-pub mod types;
 
 pub use crate::{
     callback::{
@@ -11,5 +11,5 @@ pub use crate::{
             StateSyncConfig,
         },
     },
-    types::EventFilter,
+    event_filter::{EventFilter, EventFilterBuilder},
 };

src/types.rs

@@ -7,7 +7,7 @@ use std::{
 };
 
 use alloy::{eips::BlockNumberOrTag, network::Ethereum, sol_types::SolEvent};
-use event_scanner::{event_scanner::EventScannerBuilder, types::EventFilter};
+use event_scanner::{event_filter::EventFilter, event_scanner::EventScannerBuilder};
 use tokio::time::{sleep, timeout};
 
 use crate::{
@@ -26,8 +26,8 @@ async fn processes_events_within_specified_historical_range() -> anyhow::Result<
     let callback = Arc::new(BasicCounterCallback { count: Arc::clone(&event_count) });
 
     let filter = EventFilter {
-        contract_address,
-        event: TestCounter::CountIncreased::SIGNATURE.to_owned(),
+        contract_address: Some(contract_address),
+        event: Some(TestCounter::CountIncreased::SIGNATURE.to_owned()),
         callback,
     };
 

tests/historic_mode/basic.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use alloy::{eips::BlockNumberOrTag, network::Ethereum, sol_types::SolEvent};
-use event_scanner::{event_scanner::EventScannerBuilder, types::EventFilter};
+use event_scanner::{event_filter::EventFilter, event_scanner::EventScannerBuilder};
 use tokio::time::{Duration, sleep, timeout};
 
 use crate::{
@@ -31,8 +31,8 @@ async fn replays_historical_then_switches_to_live() -> anyhow::Result<()> {
     let callback = Arc::new(EventOrderingCallback { counts: Arc::clone(&event_new_counts) });
 
     let filter = EventFilter {
-        contract_address,
-        event: TestCounter::CountIncreased::SIGNATURE.to_owned(),
+        contract_address: Some(contract_address),
+        event: Some(TestCounter::CountIncreased::SIGNATURE.to_owned()),
         callback,
     };