Make Event Scanner Stream Events Back to the Caller Instead of Invoking Callbacks (#63)

Author: 0xNeshi

CONTRIBUTING.md

@@ -6,7 +6,7 @@ Thanks for your interest in contributing! This guide explains how to set up your
 
 ## Project Overview
 
-`event-scanner` is a Rust library for monitoring EVM-based smart contract events. It is built on the `alloy` ecosystem and provides in-memory scanning with retry-aware callback execution. See `README.md` for features, usage, examples, and testing notes.
+`event-scanner` is a Rust library for monitoring and streaming EVM-based smart contract events. It is built on the `alloy` ecosystem and provides in-memory scanning. See `README.md` for features, usage, examples, and testing notes.
 
 - Workspace manifest: `Cargo.toml`
 - Library code: `src/`

Cargo.lock

@@ -81,12 +81,12 @@ async fn run_scanner(ws_url: alloy::transports::http::reqwest::Url, contract: al
         .with_event(MyContract::SomeEvent::SIGNATURE)
         .with_callback(Arc::new(CounterCallback { processed: Arc::new(AtomicUsize::new(0)) }));
 
-    let mut scanner = EventScannerBuilder::new()
+    let mut client = EventScannerBuilder::new()
         .with_event_filter(filter)
         .connect_ws::<Ethereum>(ws_url)
         .await?;
 
-    scanner.start(BlockNumberOrTag::Latest, None).await?;
+    client.start_scanner(BlockNumberOrTag::Latest, None).await?;
     Ok(())
 }
 ```

README.md

@@ -14,9 +14,8 @@ path = "main.rs"
 alloy.workspace = true
 alloy-node-bindings.workspace = true
 tokio.workspace = true
+tokio-stream.workspace = true
 anyhow.workspace = true
-async-trait.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
-hex.workspace = true
 event-scanner = { path = "../.." }

examples/historical_scanning/Cargo.toml

@@ -1,27 +1,16 @@
-use std::{sync::Arc, time::Duration};
+use std::time::Duration;
 
 use alloy::{
-    eips::BlockNumberOrTag, network::Ethereum, providers::ProviderBuilder, rpc::types::Log, sol,
-    sol_types::SolEvent,
+    eips::BlockNumberOrTag, network::Ethereum, providers::ProviderBuilder, sol, sol_types::SolEvent,
 };
 use alloy_node_bindings::Anvil;
-use async_trait::async_trait;
-use event_scanner::{EventCallback, EventFilter, event_scanner::EventScannerBuilder};
+use event_scanner::{EventFilter, event_scanner::EventScanner};
 
 use tokio::time::sleep;
+use tokio_stream::StreamExt;
 use tracing::info;
 use tracing_subscriber::EnvFilter;
 
-struct CounterCallback;
-
-#[async_trait]
-impl EventCallback for CounterCallback {
-    async fn on_event(&self, log: &Log) -> anyhow::Result<()> {
-        info!("Callback successfully executed with event {:?}", log.inner.data);
-        Ok(())
-    }
-}
-
 sol! {
     #[allow(missing_docs)]
     #[sol(rpc, bytecode="608080604052346015576101b0908161001a8239f35b5f80fdfe6080806040526004361015610012575f80fd5b5f3560e01c90816306661abd1461016157508063a87d942c14610145578063d732d955146100ad5763e8927fbc14610048575f80fd5b346100a9575f3660031901126100a9575f5460018101809111610095576020817f7ca2ca9527391044455246730762df008a6b47bbdb5d37a890ef78394535c040925f55604051908152a1005b634e487b7160e01b5f52601160045260245ffd5b5f80fd5b346100a9575f3660031901126100a9575f548015610100575f198101908111610095576020817f53a71f16f53e57416424d0d18ccbd98504d42a6f98fe47b09772d8f357c620ce925f55604051908152a1005b60405162461bcd60e51b815260206004820152601860248201527f436f756e742063616e6e6f74206265206e6567617469766500000000000000006044820152606490fd5b346100a9575f3660031901126100a95760205f54604051908152f35b346100a9575f3660031901126100a9576020905f548152f3fea2646970667358221220b846b706f79f5ae1fc4a4238319e723a092f47ce4051404186424739164ab02264736f6c634300081e0033")]
@@ -61,18 +50,22 @@ async fn main() -> anyhow::Result<()> {
 
     let increase_filter = EventFilter::new()
         .with_contract_address(*contract_address)
-        .with_event(Counter::CountIncreased::SIGNATURE)
-        .with_callback(Arc::new(CounterCallback));
+        .with_event(Counter::CountIncreased::SIGNATURE);
 
     let _ = counter_contract.increase().send().await?.get_receipt().await?;
 
-    let mut scanner = EventScannerBuilder::new()
-        .with_event_filter(increase_filter)
-        .connect_ws::<Ethereum>(anvil.ws_endpoint_url())
-        .await?;
+    let mut client = EventScanner::new().connect_ws::<Ethereum>(anvil.ws_endpoint_url()).await?;
+
+    let mut stream = client.create_event_stream(increase_filter);
 
     sleep(Duration::from_secs(10)).await;
-    scanner.start(BlockNumberOrTag::Number(0), None).await.expect("failed to start scanner");
+    client.start_scanner(BlockNumberOrTag::Number(0), None).await.expect("failed to start scanner");
+
+    while let Some(Ok(logs)) = stream.next().await {
+        for log in logs {
+            info!("Callback successfully executed with event {:?}", log.inner.data);
+        }
+    }
 
     Ok(())
 }

examples/historical_scanning/main.rs

@@ -14,9 +14,8 @@ path = "main.rs"
 alloy.workspace = true
 alloy-node-bindings.workspace = true
 tokio.workspace = true
+tokio-stream.workspace = true
 anyhow.workspace = true
-async-trait.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
-hex.workspace = true
 event-scanner = { path = "../.." }

examples/simple_counter/Cargo.toml

@@ -1,52 +1,42 @@
-use std::{sync::Arc, time::Duration};
+use std::time::Duration;
 
 use alloy::{
-    eips::BlockNumberOrTag, network::Ethereum, providers::ProviderBuilder, rpc::types::Log, sol,
-    sol_types::SolEvent,
+    eips::BlockNumberOrTag, network::Ethereum, providers::ProviderBuilder, sol, sol_types::SolEvent,
 };
 use alloy_node_bindings::Anvil;
-use async_trait::async_trait;
-use event_scanner::{EventCallback, EventFilter, event_scanner::EventScannerBuilder};
+use event_scanner::{EventFilter, event_scanner::EventScanner};
 
 use tokio::time::sleep;
+use tokio_stream::StreamExt;
 use tracing::info;
 use tracing_subscriber::EnvFilter;
 
-struct CounterCallback;
-
-#[async_trait]
-impl EventCallback for CounterCallback {
-    async fn on_event(&self, log: &Log) -> anyhow::Result<()> {
-        info!("Callback successfully executed with event {:?}", log.inner.data);
-        Ok(())
-    }
-}
-
 sol! {
     #[allow(missing_docs)]
     #[sol(rpc, bytecode="608080604052346015576101b0908161001a8239f35b5f80fdfe6080806040526004361015610012575f80fd5b5f3560e01c90816306661abd1461016157508063a87d942c14610145578063d732d955146100ad5763e8927fbc14610048575f80fd5b346100a9575f3660031901126100a9575f5460018101809111610095576020817f7ca2ca9527391044455246730762df008a6b47bbdb5d37a890ef78394535c040925f55604051908152a1005b634e487b7160e01b5f52601160045260245ffd5b5f80fd5b346100a9575f3660031901126100a9575f548015610100575f198101908111610095576020817f53a71f16f53e57416424d0d18ccbd98504d42a6f98fe47b09772d8f357c620ce925f55604051908152a1005b60405162461bcd60e51b815260206004820152601860248201527f436f756e742063616e6e6f74206265206e6567617469766500000000000000006044820152606490fd5b346100a9575f3660031901126100a95760205f54604051908152f35b346100a9575f3660031901126100a9576020905f548152f3fea2646970667358221220b846b706f79f5ae1fc4a4238319e723a092f47ce4051404186424739164ab02264736f6c634300081e0033")]
-contract Counter {
-    uint256 public count;
+    contract Counter {
+        uint256 public count;
 
-    event CountIncreased(uint256 newCount);
-    event CountDecreased(uint256 newCount);
+        event CountIncreased(uint256 newCount);
+        event CountDecreased(uint256 newCount);
 
-    function increase() public {
-        count += 1;
-        emit CountIncreased(count);
-    }
+        function increase() public {
+            count += 1;
+            emit CountIncreased(count);
+        }
 
-    function decrease() public {
-        require(count > 0, "Count cannot be negative");
-        count -= 1;
-        emit CountDecreased(count);
-    }
+        function decrease() public {
+            require(count > 0, "Count cannot be negative");
+            count -= 1;
+            emit CountDecreased(count);
+        }
 
-    function getCount() public view returns (uint256) {
-        return count;
+        function getCount() public view returns (uint256) {
+            return count;
+        }
     }
 }
-}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     let _ = tracing_subscriber::fmt().with_env_filter(EnvFilter::from_default_env()).try_init();
@@ -61,16 +51,17 @@ async fn main() -> anyhow::Result<()> {
 
     let increase_filter = EventFilter::new()
         .with_contract_address(*contract_address)
-        .with_event(Counter::CountIncreased::SIGNATURE)
-        .with_callback(Arc::new(CounterCallback));
+        .with_event(Counter::CountIncreased::SIGNATURE);
+
+    let mut client = EventScanner::new().connect_ws::<Ethereum>(anvil.ws_endpoint_url()).await?;
 
-    let mut scanner = EventScannerBuilder::new()
-        .with_event_filter(increase_filter)
-        .connect_ws::<Ethereum>(anvil.ws_endpoint_url())
-        .await?;
+    let mut stream = client.create_event_stream(increase_filter);
 
     let task_1 = tokio::spawn(async move {
-        scanner.start(BlockNumberOrTag::Latest, None).await.expect("failed to start scanner");
+        client
+            .start_scanner(BlockNumberOrTag::Latest, None)
+            .await
+            .expect("failed to start scanner");
     });
 
     let task_2 = tokio::spawn(async move {
@@ -83,6 +74,12 @@ async fn main() -> anyhow::Result<()> {
         }
     });
 
+    while let Some(Ok(logs)) = stream.next().await {
+        for log in logs {
+            info!("Callback successfully executed with event {:?}", log.inner.data);
+        }
+    }
+
     task_1.await.ok();
     task_2.await.ok();
 

examples/simple_counter/main.rs

@@ -89,7 +89,7 @@ const DEFAULT_BLOCKS_READ_PER_EPOCH: usize = 1000;
 const DEFAULT_BLOCK_CONFIRMATIONS: u64 = 0;
 // const BACK_OFF_MAX_RETRIES: u64 = 5;
 
-const MAX_BUFFERED_MESSAGES: usize = 50000;
+pub const MAX_BUFFERED_MESSAGES: usize = 50000;
 
 const DEFAULT_REORG_REWIND_DEPTH: u64 = 0;
 
@@ -291,7 +291,7 @@ impl<N: Network> ConnectedBlockRangeScanner<N> {
     /// # Errors
     ///
     /// Returns an error if the subscription service fails to start.
-    pub fn run(&self) -> anyhow::Result<BlockRangeScannerClient> {
+    pub fn run(&self) -> Result<BlockRangeScannerClient, Error> {
         let (service, cmd_tx) = Service::new(self.config.clone(), self.provider.clone());
         tokio::spawn(async move {
             service.run().await;
@@ -1017,7 +1017,7 @@ mod tests {
         while let Some(result) = receiver.next().await {
             match result {
                 Ok(range) => {
-                    println!("Received block range: [{range:?}]");
+                    info!("Received block range: [{range:?}]");
                     if block_range_start == 0 {
                         block_range_start = *range.start();
                     }