codesynapse_grpc/
event_bus.rs

1use tokio::sync::broadcast;
2
3#[derive(Clone, Debug, PartialEq)]
4pub enum Event {
5    NodeAdded {
6        id: String,
7        label: String,
8        source_file: String,
9    },
10    EdgeAdded {
11        source: String,
12        target: String,
13        relation: String,
14    },
15    NodeRemoved {
16        id: String,
17    },
18    GraphReset,
19}
20
21const CHANNEL_CAPACITY: usize = 256;
22
23pub struct EventBus {
24    tx: broadcast::Sender<Event>,
25}
26
27impl EventBus {
28    pub fn new() -> Self {
29        let (tx, _) = broadcast::channel(CHANNEL_CAPACITY);
30        Self { tx }
31    }
32
33    pub fn subscribe(&self) -> broadcast::Receiver<Event> {
34        self.tx.subscribe()
35    }
36
37    pub fn emit(&self, event: Event) {
38        let _ = self.tx.send(event);
39    }
40}
41
42impl Default for EventBus {
43    fn default() -> Self {
44        Self::new()
45    }
46}
47
48#[cfg(test)]
49mod tests {
50    use super::*;
51
52    #[tokio::test]
53    async fn test_subscribe_and_receive_graph_reset() {
54        let bus = EventBus::new();
55        let mut rx = bus.subscribe();
56        bus.emit(Event::GraphReset);
57        let ev = rx.recv().await.unwrap();
58        assert_eq!(ev, Event::GraphReset);
59    }
60
61    #[tokio::test]
62    async fn test_node_added_event_roundtrip() {
63        let bus = EventBus::new();
64        let mut rx = bus.subscribe();
65        bus.emit(Event::NodeAdded {
66            id: "n1".to_string(),
67            label: "Foo".to_string(),
68            source_file: "a.rs".to_string(),
69        });
70        let ev = rx.recv().await.unwrap();
71        assert!(matches!(ev, Event::NodeAdded { ref id, .. } if id == "n1"));
72    }
73
74    #[tokio::test]
75    async fn test_edge_added_event_roundtrip() {
76        let bus = EventBus::new();
77        let mut rx = bus.subscribe();
78        bus.emit(Event::EdgeAdded {
79            source: "a".to_string(),
80            target: "b".to_string(),
81            relation: "calls".to_string(),
82        });
83        let ev = rx.recv().await.unwrap();
84        assert!(matches!(ev, Event::EdgeAdded { ref source, .. } if source == "a"));
85    }
86
87    #[tokio::test]
88    async fn test_multiple_subscribers_both_receive() {
89        let bus = EventBus::new();
90        let mut rx1 = bus.subscribe();
91        let mut rx2 = bus.subscribe();
92        bus.emit(Event::GraphReset);
93        assert_eq!(rx1.recv().await.unwrap(), Event::GraphReset);
94        assert_eq!(rx2.recv().await.unwrap(), Event::GraphReset);
95    }
96
97    #[tokio::test]
98    async fn test_multiple_events_received_in_order() {
99        let bus = EventBus::new();
100        let mut rx = bus.subscribe();
101        bus.emit(Event::NodeAdded {
102            id: "1".to_string(),
103            label: "A".to_string(),
104            source_file: "f.rs".to_string(),
105        });
106        bus.emit(Event::NodeAdded {
107            id: "2".to_string(),
108            label: "B".to_string(),
109            source_file: "f.rs".to_string(),
110        });
111        let first = rx.recv().await.unwrap();
112        let second = rx.recv().await.unwrap();
113        assert!(matches!(first, Event::NodeAdded { ref id, .. } if id == "1"));
114        assert!(matches!(second, Event::NodeAdded { ref id, .. } if id == "2"));
115    }
116
117    #[test]
118    fn test_emit_with_no_subscribers_does_not_panic() {
119        let bus = EventBus::new();
120        bus.emit(Event::GraphReset);
121    }
122}
codesynapse_grpc/event_bus.rs

codesynapse_grpc/
event_bus.rs
