test: [#1485] add unit tests to events pkg

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

@@ -7,15 +7,15 @@ use crate::sender::{SendError, Sender};
 
 const CHANNEL_CAPACITY: usize = 32768;
 
-/// An event sender implementation using a broadcast channel.
-#[derive(Clone)]
+/// 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<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 }
     }
 }
@@ -57,3 +57,61 @@ impl From<broadcast::error::RecvError> for RecvError {
         }
     }
 }
+
+#[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

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use crate::broadcaster::Broadcaster;
 use crate::{receiver, sender};
 
+#[derive(Clone, Debug)]
 pub struct EventBus<Event: Sync + Send + Clone + 'static> {
     pub enable_sender: bool,
     pub broadcaster: Broadcaster<Event>,
@@ -40,3 +41,61 @@ impl<Event: Sync + Send + Clone + 'static> EventBus<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/sender.rs

@@ -9,6 +9,19 @@ use mockall::{automock, predicate::str};
 pub trait Sender: Sync + Send {
     type Event: Send + Clone;
 
+    /// 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>>>>;
 }