ref: rename EventListener's sender -> subscriber

Author: 0xNeshi

src/event_scanner/listener.rs

@@ -4,5 +4,5 @@ use tokio::sync::mpsc::Sender;
 #[derive(Clone)]
 pub(crate) struct EventListener {
     pub filter: EventFilter,
-    pub sender: Sender<Message>,
+    pub subscriber: Sender<Message>,
 }

src/event_scanner/modes/common.rs

@@ -48,7 +48,7 @@ pub fn spawn_log_consumers<N: Network>(
     mode: ConsumerMode,
 ) {
     for listener in listeners.iter().cloned() {
-        let EventListener { filter, sender } = listener;
+        let EventListener { filter, subscriber } = listener;
 
         let provider = provider.clone();
         let base_filter = Filter::from(&filter);
@@ -72,7 +72,7 @@ pub fn spawn_log_consumers<N: Network>(
 
                                 match mode {
                                     ConsumerMode::Stream => {
-                                        if !sender.try_stream(logs).await {
+                                        if !subscriber.try_stream(logs).await {
                                             break;
                                         }
                                     }
@@ -90,19 +90,19 @@ pub fn spawn_log_consumers<N: Network>(
                                 }
                             }
                             Err(e) => {
-                                if !sender.try_stream(e).await {
+                                if !subscriber.try_stream(e).await {
                                     break;
                                 }
                             }
                         }
                     }
                     Ok(BlockRangeMessage::Error(e)) => {
-                        if !sender.try_stream(e).await {
+                        if !subscriber.try_stream(e).await {
                             break;
                         }
                     }
                     Ok(BlockRangeMessage::Status(status)) => {
-                        if !sender.try_stream(status).await {
+                        if !subscriber.try_stream(status).await {
                             break;
                         }
                     }
@@ -119,7 +119,7 @@ pub fn spawn_log_consumers<N: Network>(
                     collected.reverse(); // restore chronological order
                 }
 
-                _ = sender.try_stream(collected).await;
+                _ = subscriber.try_stream(collected).await;
             }
         });
     }

src/event_scanner/modes/historic.rs

@@ -107,7 +107,7 @@ impl<N: Network> HistoricEventScanner<N> {
     #[must_use]
     pub fn subscribe(&mut self, filter: EventFilter) -> ReceiverStream<Message> {
         let (sender, receiver) = mpsc::channel::<Message>(MAX_BUFFERED_MESSAGES);
-        self.listeners.push(EventListener { filter, sender });
+        self.listeners.push(EventListener { filter, subscriber: sender });
         ReceiverStream::new(receiver)
     }
 
@@ -199,7 +199,7 @@ mod tests {
         let provider = RootProvider::<Ethereum>::new(RpcClient::mocked(Asserter::new()));
         let mut scanner = HistoricScannerBuilder::new().connect::<Ethereum>(provider);
         let _stream = scanner.subscribe(EventFilter::new());
-        let sender = &scanner.listeners[0].sender;
+        let sender = &scanner.listeners[0].subscriber;
         assert_eq!(sender.capacity(), MAX_BUFFERED_MESSAGES);
     }
 }

src/event_scanner/modes/latest.rs

@@ -126,7 +126,7 @@ impl<N: Network> LatestEventScanner<N> {
     #[must_use]
     pub fn subscribe(&mut self, filter: EventFilter) -> ReceiverStream<Message> {
         let (sender, receiver) = mpsc::channel::<Message>(MAX_BUFFERED_MESSAGES);
-        self.listeners.push(EventListener { filter, sender });
+        self.listeners.push(EventListener { filter, subscriber: sender });
         ReceiverStream::new(receiver)
     }
 
@@ -243,7 +243,7 @@ mod tests {
         let provider = RootProvider::<Ethereum>::new(RpcClient::mocked(Asserter::new()));
         let mut scanner = LatestScannerBuilder::new().connect::<Ethereum>(provider);
         let _stream = scanner.subscribe(EventFilter::new());
-        let sender = &scanner.listeners[0].sender;
+        let sender = &scanner.listeners[0].subscriber;
         assert_eq!(sender.capacity(), MAX_BUFFERED_MESSAGES);
     }
 }

src/event_scanner/modes/live.rs

@@ -98,7 +98,7 @@ impl<N: Network> LiveEventScanner<N> {
     #[must_use]
     pub fn subscribe(&mut self, filter: EventFilter) -> ReceiverStream<Message> {
         let (sender, receiver) = mpsc::channel::<Message>(MAX_BUFFERED_MESSAGES);
-        self.listeners.push(EventListener { filter, sender });
+        self.listeners.push(EventListener { filter, subscriber: sender });
         ReceiverStream::new(receiver)
     }
 
@@ -188,7 +188,7 @@ mod tests {
         let provider = RootProvider::<Ethereum>::new(RpcClient::mocked(Asserter::new()));
         let mut scanner = LiveScannerBuilder::new().connect::<Ethereum>(provider);
         let _stream = scanner.subscribe(EventFilter::new());
-        let sender = &scanner.listeners[0].sender;
+        let sender = &scanner.listeners[0].subscriber;
         assert_eq!(sender.capacity(), MAX_BUFFERED_MESSAGES);
     }
 }

src/event_scanner/modes/sync.rs

@@ -107,7 +107,7 @@ impl<N: Network> SyncEventScanner<N> {
     #[must_use]
     pub fn subscribe(&mut self, filter: EventFilter) -> ReceiverStream<Message> {
         let (sender, receiver) = mpsc::channel::<Message>(MAX_BUFFERED_MESSAGES);
-        self.listeners.push(EventListener { filter, sender });
+        self.listeners.push(EventListener { filter, subscriber: sender });
         ReceiverStream::new(receiver)
     }
 
@@ -219,7 +219,7 @@ mod tests {
         let provider = RootProvider::<Ethereum>::new(RpcClient::mocked(Asserter::new()));
         let mut scanner = SyncScannerBuilder::new().connect::<Ethereum>(provider);
         let _stream = scanner.subscribe(EventFilter::new());
-        let sender = &scanner.listeners[0].sender;
+        let sender = &scanner.listeners[0].subscriber;
         assert_eq!(sender.capacity(), MAX_BUFFERED_MESSAGES);
     }
 }