sergio/takana

1//! Bus de eventos in-process. Cualquier subsistema (watcher, init, supervisor futuro) puede
2//! `publish(event)` para que **todas** las conexiones del bus de agente abiertas reciban una
3//! copia. No hay tópicos: el filtrado lo hace el cliente.
4//!
5//! Implementación: lista de canales `mpsc::Sender<Event>`, uno por conexión. Cuando un
6//! `send` falla (el peer cerró), el slot se marca caído y se purga en la siguiente publicación.
7//!
8//! Concurrencia: un `Mutex` sobre el vector basta. La cadencia esperada es baja (mutaciones
9//! humanas y builds, no logs de tracing); cuando se vuelva crítico, sustituiremos por
10//! `tokio::broadcast` o un sharded broadcaster.
11
12use std::sync::mpsc::{self, Receiver, Sender};
13use std::sync::{Arc, Mutex};
14
15use takana_core::proto::Event;
16
17#[derive(Clone)]
18pub struct EventBus {
19    subs: Arc<Mutex<Vec<Sender<Event>>>>,
20}
21
22impl Default for EventBus {
23    fn default() -> Self {
24        Self::new()
25    }
26}
27
28impl EventBus {
29    pub fn new() -> Self {
30        Self { subs: Arc::new(Mutex::new(Vec::new())) }
31    }
32
33    /// Registra una nueva suscripción. El `Receiver` lo consume el thread de la conexión;
34    /// cuando ese thread termina (peer cerró), el `Sender` se desreferencia y queda muerto
35    /// implícitamente. La purga real ocurre en `publish` (perezosa).
36    pub fn subscribe(&self) -> Receiver<Event> {
37        let (tx, rx) = mpsc::channel();
38        if let Ok(mut g) = self.subs.lock() {
39            g.push(tx);
40        }
41        rx
42    }
43
44    /// Difunde un evento a todos los suscriptores vivos. Los muertos (send falla) se
45    /// purgan en sitio.
46    pub fn publish(&self, ev: &Event) {
47        let Ok(mut g) = self.subs.lock() else { return };
48        g.retain(|tx| tx.send(ev.clone()).is_ok());
49    }
50
51    /// Sólo para tests / introspección.
52    #[cfg(test)]
53    pub fn live_subs(&self) -> usize {
54        self.subs.lock().map(|g| g.len()).unwrap_or(0)
55    }
56}
57
58#[cfg(test)]
59mod tests {
60    use super::*;
61    use takana_core::proto::Event;
62
63    #[test]
64    fn publish_reaches_all_subs() {
65        let bus = EventBus::new();
66        let a = bus.subscribe();
67        let b = bus.subscribe();
68        bus.publish(&Event::Crashed { service: "web".into(), code: 1 });
69        assert!(matches!(a.try_recv().unwrap(), Event::Crashed { code: 1, .. }));
70        assert!(matches!(b.try_recv().unwrap(), Event::Crashed { code: 1, .. }));
71    }
72
73    #[test]
74    fn dead_subs_get_purged() {
75        let bus = EventBus::new();
76        let a = bus.subscribe();
77        let b = bus.subscribe();
78        assert_eq!(bus.live_subs(), 2);
79        drop(b); // simula desconexión del peer
80        bus.publish(&Event::Crashed { service: "x".into(), code: 0 });
81        // a sigue vivo; b se purgó al fallar el send.
82        assert!(a.try_recv().is_ok());
83        assert_eq!(bus.live_subs(), 1);
84    }
85
86    #[test]
87    fn no_subs_is_noop() {
88        let bus = EventBus::new();
89        bus.publish(&Event::Crashed { service: "x".into(), code: 0 });
90        // No panic, no nada.
91    }
92}