feat: [torrust#1485] add Receiver trait to events package

Author: josecelano

packages/events/src/broadcaster.rs

@@ -1,8 +1,9 @@
 use futures::future::BoxFuture;
 use futures::FutureExt;
-use tokio::sync::broadcast::error::SendError;
+use tokio::sync::broadcast::error::{RecvError, SendError};
 use tokio::sync::broadcast::{self};
 
+use crate::receiver::Receiver;
 use crate::sender::Sender;
 
 const CHANNEL_CAPACITY: usize = 32768;
@@ -13,14 +14,6 @@ pub struct Broadcaster<E: Sync + Send + Clone> {
     pub(crate) sender: broadcast::Sender<E>,
 }
 
-impl<E: Sync + Send + Clone> Sender for Broadcaster<E> {
-    type Event = E;
-
-    fn send_event(&self, event: E) -> BoxFuture<'_, Option<Result<usize, SendError<E>>>> {
-        async move { Some(self.sender.send(event)) }.boxed()
-    }
-}
-
 impl<E: Sync + Send + Clone> Default for Broadcaster<E> {
     fn default() -> Self {
         let (sender, _) = broadcast::channel(CHANNEL_CAPACITY);
@@ -34,3 +27,19 @@ impl<E: Sync + Send + Clone> Broadcaster<E> {
         self.sender.subscribe()
     }
 }
+
+impl<E: Sync + Send + Clone> Sender for Broadcaster<E> {
+    type Event = E;
+
+    fn send_event(&self, event: E) -> BoxFuture<'_, Option<Result<usize, SendError<E>>>> {
+        async move { Some(self.sender.send(event)) }.boxed()
+    }
+}
+
+impl<E: Sync + Send + Clone> Receiver for broadcast::Receiver<E> {
+    type Event = E;
+
+    fn recv(&mut self) -> BoxFuture<'_, Result<Self::Event, RecvError>> {
+        async move { self.recv().await }.boxed()
+    }
+}

packages/events/src/bus.rs

@@ -1,9 +1,7 @@
 use std::sync::Arc;
 
-use tokio::sync::broadcast::Receiver;
-
 use crate::broadcaster::Broadcaster;
-use crate::sender;
+use crate::{receiver, sender};
 
 pub struct EventBus<E: Sync + Send + Clone + 'static> {
     pub enable_sender: bool,
@@ -38,7 +36,7 @@ impl<E: Sync + Send + Clone + 'static> EventBus<E> {
     }
 
     #[must_use]
-    pub fn receiver(&self) -> Receiver<E> {
-        self.broadcaster.subscribe()
+    pub fn receiver(&self) -> Box<dyn receiver::Receiver<Event = E>> {
+        Box::new(self.broadcaster.subscribe())
     }
 }

packages/events/src/lib.rs

@@ -1,5 +1,6 @@
 pub mod broadcaster;
 pub mod bus;
+pub mod receiver;
 pub mod sender;
 
 /// Target for tracing crate logs.

packages/events/src/receiver.rs

@@ -0,0 +1,12 @@
+use futures::future::BoxFuture;
+#[cfg(test)]
+use mockall::{automock, predicate::str};
+use tokio::sync::broadcast::error::RecvError;
+
+/// A trait for receiving events.
+#[cfg_attr(test, automock(type Event=();))]
+pub trait Receiver: Sync + Send {
+    type Event: Send + Clone;
+
+    fn recv(&mut self) -> BoxFuture<'_, Result<Self::Event, RecvError>>;
+}

packages/http-tracker-core/src/event/mod.rs

@@ -1,4 +1,5 @@
 pub mod bus;
+pub mod receiver;
 pub mod sender;
 
 use std::net::{IpAddr, SocketAddr};

packages/http-tracker-core/src/event/receiver.rs

@@ -0,0 +1,3 @@
+use super::Event;
+
+pub type Receiver = Box<dyn torrust_tracker_events::receiver::Receiver<Event = Event>>;

packages/http-tracker-core/src/statistics/event/listener.rs

@@ -1,16 +1,16 @@
 use std::sync::Arc;
 
-use tokio::sync::broadcast::{self, Receiver};
+use tokio::sync::broadcast::{self};
 use tokio::task::JoinHandle;
 use torrust_tracker_clock::clock::Time;
 
 use super::handler::handle_event;
-use crate::event::Event;
+use crate::event::receiver::Receiver;
 use crate::statistics::repository::Repository;
 use crate::{CurrentClock, HTTP_TRACKER_LOG_TARGET};
 
 #[must_use]
-pub fn run_event_listener(receiver: Receiver<Event>, repository: &Arc<Repository>) -> JoinHandle<()> {
+pub fn run_event_listener(receiver: Receiver, repository: &Arc<Repository>) -> JoinHandle<()> {
     let stats_repository = repository.clone();
 
     tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "Starting HTTP tracker core event listener");
@@ -22,7 +22,7 @@ pub fn run_event_listener(receiver: Receiver<Event>, repository: &Arc<Repository
     })
 }
 
-async fn dispatch_events(mut receiver: broadcast::Receiver<Event>, stats_repository: Arc<Repository>) {
+async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc<Repository>) {
     loop {
         match receiver.recv().await {
             Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,

packages/udp-tracker-core/src/event/mod.rs

@@ -1,4 +1,5 @@
 pub mod bus;
+pub mod receiver;
 pub mod sender;
 
 use std::net::SocketAddr;

packages/udp-tracker-core/src/event/receiver.rs

@@ -0,0 +1,3 @@
+use super::Event;
+
+pub type Receiver = Box<dyn torrust_tracker_events::receiver::Receiver<Event = Event>>;

packages/udp-tracker-core/src/statistics/event/listener.rs

@@ -1,16 +1,16 @@
 use std::sync::Arc;
 
-use tokio::sync::broadcast::{self, Receiver};
+use tokio::sync::broadcast::{self};
 use tokio::task::JoinHandle;
 use torrust_tracker_clock::clock::Time;
 
 use super::handler::handle_event;
-use crate::event::Event;
+use crate::event::receiver::Receiver;
 use crate::statistics::repository::Repository;
 use crate::{CurrentClock, UDP_TRACKER_LOG_TARGET};
 
 #[must_use]
-pub fn run_event_listener(receiver: Receiver<Event>, repository: &Arc<Repository>) -> JoinHandle<()> {
+pub fn run_event_listener(receiver: Receiver, repository: &Arc<Repository>) -> JoinHandle<()> {
     let stats_repository = repository.clone();
 
     tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Starting UDP tracker core event listener");
@@ -22,7 +22,7 @@ pub fn run_event_listener(receiver: Receiver<Event>, repository: &Arc<Repository
     })
 }
 
-async fn dispatch_events(mut receiver: broadcast::Receiver<Event>, stats_repository: Arc<Repository>) {
+async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc<Repository>) {
     loop {
         match receiver.recv().await {
             Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,