oxicode_sdk/
event_bus.rs

1//! Generic event bus for pub/sub communication.
2//!
3//! Provides a type-safe, broadcast-based event bus that works with any
4//! event type implementing `Clone + Send + 'static`. This is the
5//! generalized version of the kernel's `EventBus<KernelEvent>`.
6//!
7//! # Example
8//!
9//! ```ignore
10//! use oxicode_sdk::EventBus;
11//!
12//! #[derive(Debug, Clone)]
13//! struct MyEvent { name: String }
14//!
15//! let bus: EventBus<MyEvent> = EventBus::new(256);
16//! let mut rx = bus.subscribe();
17//! bus.publish(MyEvent { name: "hello".into() }).unwrap();
18//! let event = rx.try_recv().unwrap();
19//! assert_eq!(event.name, "hello");
20//! ```
21
22use std::fmt;
23
24/// Generic broadcast-based event bus.
25///
26/// Wraps `tokio::sync::broadcast` for type-safe pub/sub event distribution.
27/// Any number of subscribers can consume events concurrently.
28pub struct EventBus<E: Clone + Send + 'static> {
29    tx: tokio::sync::broadcast::Sender<E>,
30}
31
32impl<E: Clone + Send + 'static> EventBus<E> {
33    /// Create a new event bus with the given broadcast channel capacity.
34    pub fn new(capacity: usize) -> Self {
35        let (tx, _rx) = tokio::sync::broadcast::channel(capacity);
36        // Drop _rx — the channel stays open as long as tx exists.
37        drop(_rx);
38        Self { tx }
39    }
40
41    /// Subscribe to events. Returns a receiver that will receive all
42    /// events published after this call.
43    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<E> {
44        self.tx.subscribe()
45    }
46
47    /// Publish an event to all subscribers.
48    ///
49    /// Returns `Ok(())` even if there are zero subscribers.
50    /// Returns an error only if all receivers have been dropped.
51    pub fn publish(&self, event: E) -> anyhow::Result<()> {
52        let _ = self.tx.send(event);
53        Ok(())
54    }
55
56    /// Returns the number of active subscribers.
57    pub fn subscriber_count(&self) -> usize {
58        self.tx.receiver_count()
59    }
60}
61
62impl<E: Clone + Send + 'static> Clone for EventBus<E> {
63    fn clone(&self) -> Self {
64        Self {
65            tx: self.tx.clone(),
66        }
67    }
68}
69
70impl<E: Clone + Send + 'static> fmt::Debug for EventBus<E> {
71    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
72        f.debug_struct("EventBus")
73            .field("subscribers", &self.tx.receiver_count())
74            .finish()
75    }
76}
77
78#[cfg(test)]
79mod tests {
80    use super::*;
81
82    #[derive(Debug, Clone, PartialEq, Eq)]
83    struct TestEvent {
84        name: String,
85        value: i32,
86    }
87
88    #[test]
89    fn test_new_bus() {
90        let bus: EventBus<TestEvent> = EventBus::new(16);
91        assert_eq!(bus.subscriber_count(), 0);
92    }
93
94    #[test]
95    fn test_publish_with_no_subscribers() {
96        let bus: EventBus<TestEvent> = EventBus::new(16);
97        assert!(
98            bus.publish(TestEvent {
99                name: "test".into(),
100                value: 1
101            })
102            .is_ok()
103        );
104    }
105
106    #[tokio::test]
107    async fn test_single_subscriber() {
108        let bus: EventBus<TestEvent> = EventBus::new(16);
109        let mut rx = bus.subscribe();
110        assert_eq!(bus.subscriber_count(), 1);
111
112        bus.publish(TestEvent {
113            name: "hello".into(),
114            value: 42,
115        })
116        .unwrap();
117
118        let event = rx.recv().await.unwrap();
119        assert_eq!(event.name, "hello");
120        assert_eq!(event.value, 42);
121    }
122
123    #[tokio::test]
124    async fn test_multiple_subscribers() {
125        let bus: EventBus<TestEvent> = EventBus::new(16);
126        let mut rx1 = bus.subscribe();
127        let mut rx2 = bus.subscribe();
128
129        bus.publish(TestEvent {
130            name: "broadcast".into(),
131            value: 99,
132        })
133        .unwrap();
134
135        let e1 = rx1.recv().await.unwrap();
136        let e2 = rx2.recv().await.unwrap();
137        assert_eq!(e1, e2);
138    }
139
140    #[tokio::test]
141    async fn test_late_subscriber_misses_events() {
142        let bus: EventBus<TestEvent> = EventBus::new(16);
143
144        bus.publish(TestEvent {
145            name: "early".into(),
146            value: 1,
147        })
148        .unwrap();
149
150        // Subscribe after publish — should NOT receive the event
151        let mut rx = bus.subscribe();
152        assert!(rx.try_recv().is_err());
153    }
154
155    #[test]
156    fn test_clone() {
157        let bus: EventBus<TestEvent> = EventBus::new(16);
158        let bus2 = bus.clone();
159        // Both share the same underlying channel
160        assert_eq!(bus.subscriber_count(), 0);
161        assert_eq!(bus2.subscriber_count(), 0);
162
163        let _rx = bus.subscribe();
164        // Visible from clone
165        assert_eq!(bus2.subscriber_count(), 1);
166    }
167
168    #[tokio::test]
169    async fn test_generic_with_string() {
170        let bus: EventBus<String> = EventBus::new(64);
171        let mut rx = bus.subscribe();
172        bus.publish("hello world".into()).unwrap();
173        let msg = rx.recv().await.unwrap();
174        assert_eq!(msg, "hello world");
175    }
176}
oxicode_sdk/event_bus.rs

oxicode_sdk/
event_bus.rs
