Merge #1486: Overhaul stats: Improve events package (part 3)

d98d93f refactor: [#1485] simplify event mods in pakages using the events pkg (Jose Celano)
97eb20a test: [#1485] add unit tests to events pkg (Jose Celano)
5a087d0 refactor: [#1485] replace Arc<Box<dyn Sender>> with Arc<dyn Sender> to avoid double allocations (Jose Celano)
8f5b57e refactor: [#1485] replace Arc<Option> with Option<Arc> in events pkg (Jose Celano)
c2df95f refactor: [#1485] rename trait fn send_event to send (Jose Celano)
f546bc1 refactor: [#1485] normalize event type parameter name in events pkg (Jose Celano)
3057f48 refactor: [#1485] decouple events traits from tokio broadcast channel implementation (Jose Celano)
e3703c1 feat: [#1485] add Receiver trait to events package (Jose Celano)
d4343c0 fix: [#1485] remove duplicate const (Jose Celano)

Pull request description:

  Continuation of #1483

  ### Subtasks

  - [x] `Receiver` trait.
  - [x] Decouple `Sender` and `Receiver` traits from the tokio broadcast implementation. Wrapper for `SendError` and `RecvError`.
  - [x] Return an `Option<Arc<...>>` instead of an `Arc<Option<...>>` in the `EventBus` implementation.
  - [x] Replace `Arc<Box<dyn Sender>>` with `Arc<dyn Sender>` to avoid double allocations.
  - [x] Add unit tests.

ACKs for top commit:
  josecelano:
    ACK d98d93f

Tree-SHA512: 6798238ff6dc2f6e06ecd077c2f0ddd56cd1e52bbb803273535bde6eec5c1334603d26c78daedb82eac886c5146fc575430520c0d29e7565e0fcb32ecdfe112c

Author: josecelano

packages/events/Cargo.toml

@@ -16,7 +16,7 @@ version.workspace = true
 
 [dependencies]
 futures = "0"
-tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] }
+tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync", "time"] }
 
 [dev-dependencies]
 mockall = "0"

packages/events/src/broadcaster.rs

@@ -1,36 +1,117 @@
 use futures::future::BoxFuture;
 use futures::FutureExt;
-use tokio::sync::broadcast::error::SendError;
 use tokio::sync::broadcast::{self};
 
-use crate::sender::Sender;
+use crate::receiver::{Receiver, RecvError};
+use crate::sender::{SendError, Sender};
 
 const CHANNEL_CAPACITY: usize = 32768;
 
-/// An event sender implementation using a broadcast channel.
-#[derive(Clone)]
-pub struct Broadcaster<E: Sync + Send + Clone> {
-    pub(crate) sender: broadcast::Sender<E>,
+/// An event sender and receiver implementation using a broadcast channel.
+#[derive(Clone, Debug)]
+pub struct Broadcaster<Event: Sync + Send + Clone> {
+    pub(crate) sender: broadcast::Sender<Event>,
 }
 
-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> {
+impl<Event: Sync + Send + Clone> Default for Broadcaster<Event> {
     fn default() -> Self {
-        let (sender, _) = broadcast::channel(CHANNEL_CAPACITY);
+        let (sender, _receiver) = broadcast::channel(CHANNEL_CAPACITY);
         Self { sender }
     }
 }
 
-impl<E: Sync + Send + Clone> Broadcaster<E> {
+impl<Event: Sync + Send + Clone> Broadcaster<Event> {
     #[must_use]
-    pub fn subscribe(&self) -> broadcast::Receiver<E> {
+    pub fn subscribe(&self) -> broadcast::Receiver<Event> {
         self.sender.subscribe()
     }
 }
+
+impl<Event: Sync + Send + Clone> Sender for Broadcaster<Event> {
+    type Event = Event;
+
+    fn send(&self, event: Event) -> BoxFuture<'_, Option<Result<usize, SendError<Event>>>> {
+        async move { Some(self.sender.send(event).map_err(std::convert::Into::into)) }.boxed()
+    }
+}
+
+impl<Event: Sync + Send + Clone> Receiver for broadcast::Receiver<Event> {
+    type Event = Event;
+
+    fn recv(&mut self) -> BoxFuture<'_, Result<Self::Event, RecvError>> {
+        async move { self.recv().await.map_err(std::convert::Into::into) }.boxed()
+    }
+}
+
+impl<Event> From<broadcast::error::SendError<Event>> for SendError<Event> {
+    fn from(err: broadcast::error::SendError<Event>) -> Self {
+        SendError(err.0)
+    }
+}
+
+impl From<broadcast::error::RecvError> for RecvError {
+    fn from(err: broadcast::error::RecvError) -> Self {
+        match err {
+            broadcast::error::RecvError::Lagged(amt) => RecvError::Lagged(amt),
+            broadcast::error::RecvError::Closed => RecvError::Closed,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use tokio::time::{timeout, Duration};
+
+    use super::*;
+
+    #[tokio::test]
+    async fn it_should_allow_sending_an_event_and_received_it() {
+        let broadcaster = Broadcaster::<String>::default();
+
+        let mut receiver = broadcaster.subscribe();
+
+        let event = "test";
+
+        let _unused = broadcaster.send(event.to_owned()).await.unwrap().unwrap();
+
+        let received_event = receiver.recv().await.unwrap();
+
+        assert_eq!(received_event, event);
+    }
+
+    #[tokio::test]
+    async fn it_should_return_the_number_of_receivers_when_and_event_is_sent() {
+        let broadcaster = Broadcaster::<String>::default();
+        let mut _receiver = broadcaster.subscribe();
+
+        let number_of_receivers = broadcaster.send("test".into()).await;
+
+        assert!(matches!(number_of_receivers, Some(Ok(1))));
+    }
+
+    #[tokio::test]
+    async fn it_should_fail_when_trying_tos_send_with_no_subscribers() {
+        let event = String::from("test");
+
+        let broadcaster = Broadcaster::<String>::default();
+
+        let result: Result<usize, SendError<String>> = broadcaster.send(event).await.unwrap();
+
+        assert!(matches!(result, Err(SendError::<String>(_event))));
+    }
+
+    #[tokio::test]
+    async fn it_should_allow_subscribing_multiple_receivers() {
+        let broadcaster = Broadcaster::<u8>::default();
+        let mut r1 = broadcaster.subscribe();
+        let mut r2 = broadcaster.subscribe();
+
+        let _ = broadcaster.send(1).await;
+
+        let val1 = timeout(Duration::from_secs(1), r1.recv()).await.unwrap().unwrap();
+        let val2 = timeout(Duration::from_secs(1), r2.recv()).await.unwrap().unwrap();
+
+        assert_eq!(val1, 1);
+        assert_eq!(val2, 1);
+    }
+}

packages/events/src/bus.rs

@@ -1,44 +1,101 @@
 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> {
+#[derive(Clone, Debug)]
+pub struct EventBus<Event: Sync + Send + Clone + 'static> {
     pub enable_sender: bool,
-    pub broadcaster: Broadcaster<E>,
+    pub broadcaster: Broadcaster<Event>,
 }
 
-impl<E: Sync + Send + Clone + 'static> Default for EventBus<E> {
+impl<Event: Sync + Send + Clone + 'static> Default for EventBus<Event> {
     fn default() -> Self {
         let enable_sender = true;
-        let broadcaster = Broadcaster::<E>::default();
+        let broadcaster = Broadcaster::<Event>::default();
 
         Self::new(enable_sender, broadcaster)
     }
 }
 
-impl<E: Sync + Send + Clone + 'static> EventBus<E> {
+impl<Event: Sync + Send + Clone + 'static> EventBus<Event> {
     #[must_use]
-    pub fn new(enable_sender: bool, broadcaster: Broadcaster<E>) -> Self {
+    pub fn new(enable_sender: bool, broadcaster: Broadcaster<Event>) -> Self {
         Self {
             enable_sender,
             broadcaster,
         }
     }
 
     #[must_use]
-    pub fn sender(&self) -> Arc<Option<Box<dyn sender::Sender<Event = E>>>> {
+    pub fn sender(&self) -> Option<Arc<dyn sender::Sender<Event = Event>>> {
         if self.enable_sender {
-            Arc::new(Some(Box::new(self.broadcaster.clone())))
+            Some(Arc::new(self.broadcaster.clone()))
         } else {
-            Arc::new(None)
+            None
         }
     }
 
     #[must_use]
-    pub fn receiver(&self) -> Receiver<E> {
-        self.broadcaster.subscribe()
+    pub fn receiver(&self) -> Box<dyn receiver::Receiver<Event = Event>> {
+        Box::new(self.broadcaster.subscribe())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use tokio::time::{timeout, Duration};
+
+    use super::*;
+
+    #[tokio::test]
+    async fn it_should_provide_an_event_sender_when_enabled() {
+        let bus = EventBus::<String>::new(true, Broadcaster::default());
+
+        assert!(bus.sender().is_some());
+    }
+
+    #[tokio::test]
+    async fn it_should_not_provide_event_sender_when_disabled() {
+        let bus = EventBus::<String>::new(false, Broadcaster::default());
+
+        assert!(bus.sender().is_none());
+    }
+
+    #[tokio::test]
+    async fn it_should_enabled_by_default() {
+        let bus = EventBus::<String>::default();
+
+        assert!(bus.sender().is_some());
+    }
+
+    #[tokio::test]
+    async fn it_should_allow_sending_events_that_are_received_by_receivers() {
+        let bus = EventBus::<String>::default();
+        let sender = bus.sender().unwrap();
+        let mut receiver = bus.receiver();
+
+        let event = "hello".to_string();
+
+        let _unused = sender.send(event.clone()).await.unwrap().unwrap();
+
+        let result = timeout(Duration::from_secs(1), receiver.recv()).await;
+
+        assert_eq!(result.unwrap().unwrap(), event);
+    }
+
+    #[tokio::test]
+    async fn it_should_send_a_closed_events_to_receivers_when_sender_is_dropped() {
+        let bus = EventBus::<String>::default();
+
+        let mut receiver = bus.receiver();
+
+        let future = receiver.recv();
+
+        drop(bus); // explicitly drop sender
+
+        let result = timeout(Duration::from_secs(1), future).await;
+
+        assert!(matches!(result.unwrap(), Err(crate::receiver::RecvError::Closed)));
     }
 }

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,38 @@
+use std::fmt;
+
+use futures::future::BoxFuture;
+#[cfg(test)]
+use mockall::{automock, predicate::str};
+
+/// 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>>;
+}
+
+/// An error returned from the [`recv`] function on a [`Receiver`].
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub enum RecvError {
+    /// There are no more active senders implying no further messages will ever
+    /// be sent.
+    Closed,
+
+    /// The receiver lagged too far behind. Attempting to receive again will
+    /// return the oldest message still retained by the channel.
+    ///
+    /// Includes the number of skipped messages.
+    Lagged(u64),
+}
+
+impl fmt::Display for RecvError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            RecvError::Closed => write!(f, "channel closed"),
+            RecvError::Lagged(amt) => write!(f, "channel lagged by {amt}"),
+        }
+    }
+}
+
+impl std::error::Error for RecvError {}

packages/events/src/sender.rs

@@ -1,15 +1,38 @@
+use std::fmt;
+
 use futures::future::BoxFuture;
 #[cfg(test)]
 use mockall::{automock, predicate::str};
-use tokio::sync::broadcast::error::SendError;
-
-/// Target for tracing crate logs.
-pub const EVENTS_TARGET: &str = "EVENTS";
 
 /// A trait for sending events.
 #[cfg_attr(test, automock(type Event=();))]
 pub trait Sender: Sync + Send {
     type Event: Send + Clone;
 
-    fn send_event(&self, event: Self::Event) -> BoxFuture<'_, Option<Result<usize, SendError<Self::Event>>>>;
+    /// Sends an event to all active receivers.
+    ///
+    /// Returns a future that resolves to an `Option<Result<usize, SendError<Self::Event>>>`:
+    ///
+    /// - `Some(Ok(n))` — the event was successfully sent to `n` receivers.
+    /// - `Some(Err(e))` — an error occurred while sending the event.
+    /// - `None` — the sender is inactive or disconnected, and the event was not sent.
+    ///
+    /// The `Option` allows implementations to express cases where sending is not possible
+    /// (e.g., when the sender is disabled or there are no active receivers).
+    ///
+    /// The `usize` typically represents the number of receivers the message was delivered to,
+    /// but its semantics may vary depending on the concrete implementation.
+    fn send(&self, event: Self::Event) -> BoxFuture<'_, Option<Result<usize, SendError<Self::Event>>>>;
+}
+
+/// Error returned by the [`send`] function on a [`Sender`].
+#[derive(Debug)]
+pub struct SendError<Event>(pub Event);
+
+impl<Event> fmt::Display for SendError<Event> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "channel closed")
+    }
 }
+
+impl<Event: fmt::Debug> std::error::Error for SendError<Event> {}

packages/http-tracker-core/benches/helpers/util.rs

@@ -20,8 +20,8 @@ use bittorrent_tracker_core::whitelist::authorization::WhitelistAuthorization;
 use bittorrent_tracker_core::whitelist::repository::in_memory::InMemoryWhitelist;
 use futures::future::BoxFuture;
 use mockall::mock;
-use tokio::sync::broadcast::error::SendError;
 use torrust_tracker_configuration::{Configuration, Core};
+use torrust_tracker_events::sender::SendError;
 use torrust_tracker_primitives::peer::Peer;
 use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch};
 use torrust_tracker_test_helpers::configuration;
@@ -127,6 +127,6 @@ mock! {
     impl torrust_tracker_events::sender::Sender for HttpStatsEventSender {
         type Event = Event;
 
-        fn send_event(&self, event: Event) -> BoxFuture<'static,Option<Result<usize,SendError<Event> > > > ;
+        fn send(&self, event: Event) -> BoxFuture<'static,Option<Result<usize,SendError<Event> > > > ;
     }
 }