test(tokio-util): add unit tests for EventSender (#11980)

crates/tokio-util/src/event_sender.rs

@@ -40,3 +40,96 @@ impl<T: Clone + Send + Sync + 'static> EventSender<T> {
         EventStream::new(self.sender.subscribe())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::{
+        task,
+        time::{timeout, Duration},
+    };
+    use tokio_stream::StreamExt;
+
+    #[tokio::test]
+    async fn test_event_broadcast_to_listener() {
+        let sender = EventSender::default();
+
+        // Create a listener for the events
+        let mut listener = sender.new_listener();
+
+        // Broadcast an event
+        sender.notify("event1");
+
+        // Check if the listener receives the event
+        let received_event = listener.next().await;
+        assert_eq!(received_event, Some("event1"));
+    }
+
+    #[tokio::test]
+    async fn test_event_no_listener() {
+        let sender = EventSender::default();
+
+        // Broadcast an event with no listeners
+        sender.notify("event2");
+
+        // Ensure it doesn't panic or fail when no listeners are present
+        // (this test passes if it runs without errors).
+    }
+
+    #[tokio::test]
+    async fn test_multiple_listeners_receive_event() {
+        let sender = EventSender::default();
+
+        // Create two listeners
+        let mut listener1 = sender.new_listener();
+        let mut listener2 = sender.new_listener();
+
+        // Broadcast an event
+        sender.notify("event3");
+
+        // Both listeners should receive the same event
+        let event1 = listener1.next().await;
+        let event2 = listener2.next().await;
+
+        assert_eq!(event1, Some("event3"));
+        assert_eq!(event2, Some("event3"));
+    }
+
+    #[tokio::test]
+    async fn test_bounded_channel_size() {
+        // Create a channel with size 2
+        let sender = EventSender::new(2);
+
+        // Create a listener
+        let mut listener = sender.new_listener();
+
+        // Broadcast 3 events, which exceeds the channel size
+        sender.notify("event4");
+        sender.notify("event5");
+        sender.notify("event6");
+
+        // Only the last two should be received due to the size limit
+        let received_event1 = listener.next().await;
+        let received_event2 = listener.next().await;
+
+        assert_eq!(received_event1, Some("event5"));
+        assert_eq!(received_event2, Some("event6"));
+    }
+
+    #[tokio::test]
+    async fn test_event_listener_timeout() {
+        let sender = EventSender::default();
+        let mut listener = sender.new_listener();
+
+        // Broadcast an event asynchronously
+        task::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(50)).await;
+            sender.notify("delayed_event");
+        });
+
+        // Use a timeout to ensure that the event is received within a certain time
+        let result = timeout(Duration::from_millis(100), listener.next()).await;
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap(), Some("delayed_event"));
+    }
+}