stream to start

Author: LeoPatOZ

README.md

@@ -130,11 +130,8 @@ let scanner = EventScanner::latest()
 - `block_read_limit(usize)` – Sets the maximum number of blocks to process per read operation. This prevents RPC provider errors from overly large block range queries.
 - Connect with `connect_ws::<Ethereum>(url)`, `connect_ipc::<Ethereum>(path)`, or `connect(provider)`.
 
-**Mode-specific APIs:**
-- Live: `client.stream()` – Start streaming new blocks
-- Historical: `client.stream()` – Process historical range (configured during setup)
-- Sync: `client.stream()` – Process historical then transition to live
-- Latest: `client.stream()` – Process a set number of events
+**Starting the Scanner:**
+Invoking `scanner.start()` starts the scanner in the specified mode.
 
 ### Defining Event Filters
 

examples/historical_scanning/main.rs

@@ -60,7 +60,7 @@ async fn main() -> anyhow::Result<()> {
     let mut stream = client.create_event_stream(increase_filter);
 
     sleep(Duration::from_secs(10)).await;
-    client.stream().await.expect("failed to start scanner");
+    client.run().await.expect("failed to start scanner");
 
     while let Some(message) = stream.next().await {
         match message {

examples/simple_counter/main.rs

@@ -57,7 +57,7 @@ async fn main() -> anyhow::Result<()> {
     let mut stream = client.create_event_stream(increase_filter);
 
     let task_1 = tokio::spawn(async move {
-        client.stream().await.expect("failed to start scanner");
+        client.start().await.expect("failed to start scanner");
     });
 
     let task_2 = tokio::spawn(async move {

src/event_lib/modes/historic.rs

@@ -114,7 +114,7 @@ impl<N: Network> HistoricEventScanner<N> {
     /// # Errors
     ///
     /// * `EventScannerMessage::ServiceShutdown` - if the service is already shutting down.
-    pub async fn stream(self) -> Result<(), EventScannerError> {
+    pub async fn run(self) -> Result<(), EventScannerError> {
         self.inner.stream_historical(self.config.from_block, self.config.to_block).await
     }
 }

src/event_lib/modes/latest.rs

@@ -188,7 +188,7 @@ impl<N: Network> LatestEventScanner<N> {
     ///
     /// * `EventScannerMessage::ServiceShutdown` - if the service is already shutting down.
     #[allow(clippy::unused_async)]
-    pub async fn stream(self) -> Result<(), EventScannerError> {
+    pub async fn start(self) -> Result<(), EventScannerError> {
         unimplemented!()
     }
 }

src/event_lib/modes/live.rs

@@ -101,7 +101,7 @@ impl<N: Network> LiveEventScanner<N> {
     /// # Errors
     ///
     /// * `EventScannerMessage::ServiceShutdown` - if the service is already shutting down.
-    pub async fn stream(self) -> Result<(), EventScannerError> {
+    pub async fn start(self) -> Result<(), EventScannerError> {
         self.inner.stream_live(self.config.block_confirmations).await
     }
 }

src/event_lib/modes/sync.rs

@@ -114,7 +114,7 @@ impl<N: Network> SyncEventScanner<N> {
     /// # Errors
     ///
     /// * `EventScannerMessage::ServiceShutdown` - if the service is already shutting down.
-    pub async fn stream(self) -> Result<(), EventScannerError> {
+    pub async fn start(self) -> Result<(), EventScannerError> {
         self.inner.stream_from(self.config.from_block, self.config.block_confirmations).await
     }
 }

tests/historic_mode/basic.rs

@@ -42,7 +42,7 @@ async fn processes_events_within_specified_historical_range() -> anyhow::Result<
         .await?;
     let mut stream = client.create_event_stream(filter).take(expected_event_count);
 
-    tokio::spawn(async move { client.stream().await });
+    tokio::spawn(async move { client.run().await });
 
     let event_count = Arc::new(AtomicUsize::new(0));
     let event_count_clone = Arc::clone(&event_count);

tests/live_mode/basic.rs

@@ -28,7 +28,7 @@ async fn basic_single_event_scanning() -> anyhow::Result<()> {
     let mut client = EventScanner::live().connect_ws::<Ethereum>(anvil.ws_endpoint_url()).await?;
     let mut stream = client.create_event_stream(filter).take(expected_event_count);
 
-    tokio::spawn(async move { client.stream().await });
+    tokio::spawn(async move { client.start().await });
 
     for _ in 0..expected_event_count {
         contract.increase().send().await?.watch().await?;
@@ -88,7 +88,7 @@ async fn multiple_contracts_same_event_isolate_callbacks() -> anyhow::Result<()>
     let a_stream = client.create_event_stream(a_filter);
     let b_stream = client.create_event_stream(b_filter);
 
-    tokio::spawn(async move { client.stream().await });
+    tokio::spawn(async move { client.start().await });
 
     for _ in 0..expected_events_a {
         a.increase().send().await?.watch().await?;
@@ -160,7 +160,7 @@ async fn multiple_events_same_contract() -> anyhow::Result<()> {
     let mut incr_stream = client.create_event_stream(increase_filter).take(expected_incr_events);
     let mut decr_stream = client.create_event_stream(decrease_filter).take(expected_decr_events);
 
-    tokio::spawn(async move { client.stream().await });
+    tokio::spawn(async move { client.start().await });
 
     for _ in 0..expected_incr_events {
         contract.increase().send().await?.watch().await?;
@@ -227,7 +227,7 @@ async fn signature_matching_ignores_irrelevant_events() -> anyhow::Result<()> {
 
     let mut stream = client.create_event_stream(filter).take(num_of_events);
 
-    tokio::spawn(async move { client.stream().await });
+    tokio::spawn(async move { client.start().await });
 
     for _ in 0..num_of_events {
         contract.increase().send().await?.watch().await?;
@@ -260,7 +260,7 @@ async fn live_filters_malformed_signature_graceful() -> anyhow::Result<()> {
 
     let mut stream = client.create_event_stream(filter).take(num_of_events);
 
-    tokio::spawn(async move { client.stream().await });
+    tokio::spawn(async move { client.start().await });
 
     for _ in 0..num_of_events {
         contract.increase().send().await?.watch().await?;

tests/live_mode/optional_fields.rs

@@ -27,7 +27,7 @@ async fn track_all_events_from_contract() -> anyhow::Result<()> {
 
     let mut stream = client.create_event_stream(filter).take(expected_event_count);
 
-    tokio::spawn(async move { client.stream().await });
+    tokio::spawn(async move { client.start().await });
 
     // Generate both increase and decrease events
     for _ in 0..expected_event_count {
@@ -67,7 +67,7 @@ async fn track_all_events_in_block_range() -> anyhow::Result<()> {
 
     let mut stream = client.create_event_stream(filter).take(expected_event_count);
 
-    tokio::spawn(async move { client.stream().await });
+    tokio::spawn(async move { client.start().await });
 
     // Generate events from our contract
     for _ in 0..expected_event_count {
@@ -112,7 +112,7 @@ async fn mixed_optional_and_required_filters() -> anyhow::Result<()> {
         client.create_event_stream(specific_filter).take(expected_specific_count);
     let mut all_stream = client.create_event_stream(all_events_filter).take(expected_all_count);
 
-    tokio::spawn(async move { client.stream().await });
+    tokio::spawn(async move { client.start().await });
 
     // First increase the counter to have some balance
     for _ in 0..expected_all_count {