fix: integrate latest into scanner api refactor

Author: LeoPatOZ

examples/latest_events_scanning/main.rs

@@ -1,18 +1,15 @@
 use alloy::{network::Ethereum, providers::ProviderBuilder, sol, sol_types::SolEvent};
 use alloy_node_bindings::Anvil;
-use event_scanner::{
-    EventFilter,
-    event_scanner::{EventScanner, EventScannerMessage},
-};
-
+use event_scanner::{EventFilter, EventScanner, EventScannerMessage};
 use tokio_stream::StreamExt;
 use tracing::{error, info};
 use tracing_subscriber::EnvFilter;
 
 sol! {
     #[allow(missing_docs)]
-    #[sol(rpc, bytecode="608080604052346015576101b0908161001a8239f35b5f80fdfe6080806040526004361015610012575f80fd5b5f3560e01c90816306661abd1461016157508063a87d942c14610145578063d732d955146100ad5763e8927fbc14610048575f80fd5b346100a9575f3660031901126100a9575f5460018101809111610095576020817f7ca2ca9527391044455246730762df008a6b47bbdb5d37a890ef78394535c040925f55604051908152a1005b634e487b7160e01b5f52601160045260245ffd5b5f80fd5b346100a9575f3660031901126100a9575f548015610100575f198101908111610095576020817f53a71f16f53e57416424d0d18ccbd98504d42a6f98fe47b09772d8f357c620ce925f55604051908152a1005b60405162461bcd60e51b815260206004820152601860248201527f436f756e742063616e6e6f74206265206e6567617469766500000000000000006044820152606490fd5b346100a9575f3660031901126100a95760205f54604051908152f35b346100a9575f3660031901126100a9576020905f548152f3fea2646970667358221220b846b706f79f5ae1fc4a4238319e723a092f47ce4051404186424739164ab02264736f6c634300081e0033")]
-    contract Counter {
+    #[sol(rpc,
+bytecode="608080604052346015576101b0908161001a8239f35b5f80fdfe6080806040526004361015610012575f80fd5b5f3560e01c90816306661abd1461016157508063a87d942c14610145578063d732d955146100ad5763e8927fbc14610048575f80fd5b346100a9575f3660031901126100a9575f5460018101809111610095576020817f7ca2ca9527391044455246730762df008a6b47bbdb5d37a890ef78394535c040925f55604051908152a1005b634e487b7160e01b5f52601160045260245ffd5b5f80fd5b346100a9575f3660031901126100a9575f548015610100575f198101908111610095576020817f53a71f16f53e57416424d0d18ccbd98504d42a6f98fe47b09772d8f357c620ce925f55604051908152a1005b60405162461bcd60e51b815260206004820152601860248201527f436f756e742063616e6e6f74206265206e6567617469766500000000000000006044820152606490fd5b346100a9575f3660031901126100a95760205f54604051908152f35b346100a9575f3660031901126100a9576020905f548152f3fea2646970667358221220b846b706f79f5ae1fc4a4238319e723a092f47ce4051404186424739164ab02264736f6c634300081e0033"
+)]     contract Counter {
         uint256 public count;
 
         event CountIncreased(uint256 newCount);
@@ -51,31 +48,29 @@ async fn main() -> anyhow::Result<()> {
         .with_contract_address(*contract_address)
         .with_event(Counter::CountIncreased::SIGNATURE);
 
-    let mut client = EventScanner::new().connect_ws::<Ethereum>(anvil.ws_endpoint_url()).await?;
+    let mut scanner =
+        EventScanner::latest().count(5).connect_ws::<Ethereum>(anvil.ws_endpoint_url()).await?;
 
-    let mut stream = client.create_event_stream(increase_filter);
+    let mut stream = scanner.create_event_stream(increase_filter);
 
-    for _ in 0..10 {
+    for _ in 0..8 {
         _ = counter_contract.increase().send().await?;
     }
 
-    tokio::spawn(async move {
-        client.scan_latest(5).await.expect("failed to start scanner");
-    });
+    scanner.start().await?;
 
-    // only the last 5 events will be streamed
     while let Some(message) = stream.next().await {
         match message {
             EventScannerMessage::Data(logs) => {
                 for log in logs {
-                    info!("Callback successfully executed with event {:?}", log.inner.data);
+                    info!("Received event: {:?}", log.inner.data);
                 }
             }
             EventScannerMessage::Error(e) => {
                 error!("Received error: {}", e);
             }
             EventScannerMessage::Status(info) => {
-                info!("Received info: {:?}", info);
+                info!("Received status: {:?}", info);
             }
         }
     }

src/event_lib/modes/latest.rs

@@ -21,10 +21,10 @@ use super::{BaseConfig, BaseConfigBuilder};
 pub struct LatestScannerConfig {
     base: BaseConfig,
     // Defatuls to 1
-    count: u64,
-    // Defaults to Earliest
-    from_block: BlockNumberOrTag,
+    count: usize,
     // Defaults to Latest
+    from_block: BlockNumberOrTag,
+    // Defaults to Earliest
     to_block: BlockNumberOrTag,
     // Defaults to 0
     block_confirmations: u64,
@@ -49,8 +49,8 @@ impl LatestScannerConfig {
         Self {
             base: BaseConfig::new(),
             count: 1,
-            from_block: BlockNumberOrTag::Earliest,
-            to_block: BlockNumberOrTag::Latest,
+            from_block: BlockNumberOrTag::Latest,
+            to_block: BlockNumberOrTag::Earliest,
             block_confirmations: DEFAULT_BLOCK_CONFIRMATIONS,
             switch_to_live: false,
         }
@@ -63,7 +63,7 @@ impl LatestScannerConfig {
     }
 
     #[must_use]
-    pub fn count(mut self, count: u64) -> Self {
+    pub fn count(mut self, count: usize) -> Self {
         self.count = count;
         self
     }
@@ -189,7 +189,9 @@ impl<N: Network> LatestEventScanner<N> {
     /// * `EventScannerMessage::ServiceShutdown` - if the service is already shutting down.
     #[allow(clippy::unused_async)]
     pub async fn start(self) -> Result<(), EventScannerError> {
-        unimplemented!()
+        self.inner
+            .stream_latest(self.config.count, self.config.from_block, self.config.to_block)
+            .await
     }
 }
 
@@ -202,8 +204,8 @@ mod tests {
         let config = LatestScannerConfig::new();
 
         assert_eq!(config.count, 1);
-        assert!(matches!(config.from_block, BlockNumberOrTag::Earliest));
-        assert!(matches!(config.to_block, BlockNumberOrTag::Latest));
+        assert!(matches!(config.from_block, BlockNumberOrTag::Latest));
+        assert!(matches!(config.to_block, BlockNumberOrTag::Earliest));
         assert_eq!(config.block_confirmations, DEFAULT_BLOCK_CONFIRMATIONS);
         assert!(!config.switch_to_live);
     }

src/event_lib/scanner.rs

@@ -183,7 +183,7 @@ impl<N: Network> EventScannerService<N> {
     /// * Returns `EventScannerError` if the scanner fails to start or fetching logs fails.
     ///
     /// [`ScannerStatus::ReorgDetected`]: crate::types::ScannerStatus::ReorgDetected
-    pub async fn scan_latest<T: Into<BlockNumberOrTag>>(
+    pub async fn stream_latest<T: Into<BlockNumberOrTag>>(
         self,
         count: usize,
         start_height: T,

tests/common.rs

@@ -10,7 +10,10 @@ use alloy::{
     sol_types::SolEvent,
 };
 use alloy_node_bindings::{Anvil, AnvilInstance};
-use event_scanner::{EventFilter, EventScanner, EventScannerMessage, LiveEventScanner};
+use event_scanner::{
+    EventFilter, EventScanner, EventScannerMessage, HistoricEventScanner, LiveEventScanner,
+    SyncEventScanner, event_lib::modes::LatestEventScanner,
+};
 use tokio_stream::wrappers::ReceiverStream;
 
 // Shared test contract used across integration tests
@@ -42,23 +45,32 @@ sol! {
     }
 }
 
-pub struct LiveScannerSetup<P>
+pub struct ScannerSetup<S, P>
 where
     P: Provider<Ethereum> + Clone,
 {
     pub provider: RootProvider,
     pub contract: TestCounter::TestCounterInstance<Arc<P>>,
-    pub scanner: LiveEventScanner<Ethereum>,
+    pub scanner: S,
     pub stream: ReceiverStream<EventScannerMessage>,
     pub anvil: AnvilInstance,
 }
 
+pub type LiveScannerSetup<P> = ScannerSetup<LiveEventScanner<Ethereum>, P>;
+pub type HistoricScannerSetup<P> = ScannerSetup<HistoricEventScanner<Ethereum>, P>;
+pub type SyncScannerSetup<P> = ScannerSetup<SyncEventScanner<Ethereum>, P>;
+pub type LatestScannerSetup<P> = ScannerSetup<LatestEventScanner<Ethereum>, P>;
+
 #[allow(clippy::missing_errors_doc)]
-pub async fn setup_live_scanner(
+pub async fn setup_common(
     block_interval: Option<f64>,
     filter: Option<EventFilter>,
-    confirmations: u64,
-) -> anyhow::Result<LiveScannerSetup<impl Provider<Ethereum> + Clone>> {
+) -> anyhow::Result<(
+    AnvilInstance,
+    RootProvider,
+    TestCounter::TestCounterInstance<Arc<RootProvider>>,
+    EventFilter,
+)> {
     let anvil = spawn_anvil(block_interval)?;
     let provider = build_provider(&anvil).await?;
     let contract = deploy_counter(Arc::new(provider.clone())).await?;
@@ -69,15 +81,87 @@ pub async fn setup_live_scanner(
 
     let filter = filter.unwrap_or(default_filter);
 
+    Ok((anvil, provider, contract, filter))
+}
+
+#[allow(clippy::missing_errors_doc)]
+pub async fn setup_live_scanner(
+    block_interval: Option<f64>,
+    filter: Option<EventFilter>,
+    confirmations: u64,
+) -> anyhow::Result<LiveScannerSetup<impl Provider<Ethereum> + Clone>> {
+    let (anvil, provider, contract, filter) = setup_common(block_interval, filter).await?;
+
     let mut scanner = EventScanner::live()
         .block_confirmations(confirmations)
         .connect_ws(anvil.ws_endpoint_url())
         .await?;
 
     let stream = scanner.create_event_stream(filter);
 
-    // return anvil otherwise it doesnt live long enough...
-    Ok(LiveScannerSetup { provider, contract, scanner, stream, anvil })
+    Ok(ScannerSetup { provider, contract, scanner, stream, anvil })
+}
+
+#[allow(clippy::missing_errors_doc)]
+pub async fn setup_sync_scanner(
+    block_interval: Option<f64>,
+    filter: Option<EventFilter>,
+    confirmations: u64,
+) -> anyhow::Result<SyncScannerSetup<impl Provider<Ethereum> + Clone>> {
+    let (anvil, provider, contract, filter) = setup_common(block_interval, filter).await?;
+
+    let mut scanner = EventScanner::sync()
+        .block_confirmations(confirmations)
+        .connect_ws(anvil.ws_endpoint_url())
+        .await?;
+
+    let stream = scanner.create_event_stream(filter);
+
+    Ok(ScannerSetup { provider, contract, scanner, stream, anvil })
+}
+
+#[allow(clippy::missing_errors_doc)]
+pub async fn setup_historic_scanner(
+    block_interval: Option<f64>,
+    filter: Option<EventFilter>,
+    from: BlockNumberOrTag,
+    to: BlockNumberOrTag,
+) -> anyhow::Result<HistoricScannerSetup<impl Provider<Ethereum> + Clone>> {
+    let (anvil, provider, contract, filter) = setup_common(block_interval, filter).await?;
+
+    let mut scanner = EventScanner::historic()
+        .from_block(from)
+        .to_block(to)
+        .connect_ws(anvil.ws_endpoint_url())
+        .await?;
+
+    let stream = scanner.create_event_stream(filter);
+
+    Ok(ScannerSetup { provider, contract, scanner, stream, anvil })
+}
+
+#[allow(clippy::missing_errors_doc)]
+pub async fn setup_latest_scanner(
+    block_interval: Option<f64>,
+    filter: Option<EventFilter>,
+    count: usize,
+    from: Option<BlockNumberOrTag>,
+    to: Option<BlockNumberOrTag>,
+) -> anyhow::Result<LatestScannerSetup<impl Provider<Ethereum> + Clone>> {
+    let (anvil, provider, contract, filter) = setup_common(block_interval, filter).await?;
+    let mut builder = EventScanner::latest().count(count);
+    if let Some(f) = from {
+        builder = builder.from_block(f);
+    }
+    if let Some(t) = to {
+        builder = builder.to_block(t);
+    }
+
+    let mut scanner = builder.connect_ws(anvil.ws_endpoint_url()).await?;
+
+    let stream = scanner.create_event_stream(filter);
+
+    Ok(ScannerSetup { provider, contract, scanner, stream, anvil })
 }
 
 #[allow(clippy::missing_errors_doc)]

tests/historic_mode/basic.rs

@@ -6,41 +6,31 @@ use std::{
     time::Duration,
 };
 
-use alloy::{network::Ethereum, sol_types::SolEvent};
-use event_scanner::{EventFilter, EventScanner, EventScannerMessage};
+use event_scanner::EventScannerMessage;
 use tokio::time::timeout;
 use tokio_stream::StreamExt;
 
-use crate::common::{TestCounter, build_provider, deploy_counter, spawn_anvil};
+use crate::common::{TestCounter, setup_historic_scanner};
 
 #[tokio::test]
 async fn processes_events_within_specified_historical_range() -> anyhow::Result<()> {
-    let anvil = spawn_anvil(Some(0.1))?;
-    let provider = build_provider(&anvil).await?;
-    let contract = deploy_counter(provider.clone()).await?;
-    let contract_address = *contract.address();
-
-    let filter = EventFilter::new()
-        .with_contract_address(contract_address)
-        .with_event(TestCounter::CountIncreased::SIGNATURE);
-
-    let receipt = contract.increase().send().await?.get_receipt().await?;
-    let start_block = receipt.block_number.expect("receipt should contain block number");
-    let mut end_block = 0;
-
+    let setup = setup_historic_scanner(
+        Some(0.1),
+        None,
+        alloy::eips::BlockNumberOrTag::Earliest,
+        alloy::eips::BlockNumberOrTag::Latest,
+    )
+    .await?;
+
+    let contract = setup.contract.clone();
     let expected_event_count = 4;
 
-    for _ in 1..expected_event_count {
-        let receipt = contract.increase().send().await?.get_receipt().await?;
-        end_block = receipt.block_number.expect("receipt should contain block number");
+    for _ in 0..expected_event_count {
+        contract.increase().send().await?.watch().await?;
     }
 
-    let mut scanner = EventScanner::historic()
-        .from_block(start_block)
-        .to_block(end_block)
-        .connect_ws::<Ethereum>(anvil.ws_endpoint_url())
-        .await?;
-    let mut stream = scanner.create_event_stream(filter).take(expected_event_count);
+    let scanner = setup.scanner;
+    let mut stream = setup.stream.take(expected_event_count);
 
     tokio::spawn(async move { scanner.run().await });
 

tests/historic_to_live/basic.rs

@@ -1,36 +1,22 @@
-use alloy::{network::Ethereum, primitives::U256, sol_types::SolEvent};
-use event_scanner::{EventFilter, EventScanner, assert_next, types::ScannerStatus};
+use alloy::primitives::U256;
+use event_scanner::{assert_next, types::ScannerStatus};
 
-use crate::common::{TestCounter, build_provider, deploy_counter, spawn_anvil};
+use crate::common::{TestCounter, setup_sync_scanner};
 
 #[tokio::test]
 async fn replays_historical_then_switches_to_live() -> anyhow::Result<()> {
-    let anvil = spawn_anvil(Some(0.1))?;
-    let provider = build_provider(&anvil).await?;
-    let contract = deploy_counter(provider).await?;
-    let contract_address = *contract.address();
+    let setup = setup_sync_scanner(Some(0.1), None, 0).await?;
+    let contract = setup.contract.clone();
 
     let historical_events = 3;
     let live_events = 2;
 
-    let receipt = contract.increase().send().await?.get_receipt().await?;
-    let first_historical_block =
-        receipt.block_number.expect("historical receipt should contain block number");
-
-    for _ in 1..historical_events {
+    for _ in 0..historical_events {
         contract.increase().send().await?.watch().await?;
     }
 
-    let filter = EventFilter::new()
-        .with_contract_address(contract_address)
-        .with_event(TestCounter::CountIncreased::SIGNATURE);
-
-    let mut scanner = EventScanner::sync()
-        .from_block(first_historical_block)
-        .connect_ws::<Ethereum>(anvil.ws_endpoint_url())
-        .await?;
-
-    let mut stream = scanner.create_event_stream(filter);
+    let scanner = setup.scanner;
+    let mut stream = setup.stream;
 
     tokio::spawn(async move { scanner.start().await });