soothe_client/appkit/
broadcaster.rs

1//! Generic SSE-style pub/sub fan-out for application backends.
2
3use std::collections::HashMap;
4use std::sync::Mutex;
5
6use serde_json::Value;
7use tokio::sync::mpsc;
8use uuid::Uuid;
9
10/// One Server-Sent Event payload. The event type vocabulary
11/// (`delta` / `complete` / `query_error` / `status_change` / …) is app-defined.
12#[derive(Debug, Clone)]
13pub struct SseEvent {
14    /// Application-defined event type name.
15    pub event_type: String,
16    /// Structured event payload.
17    pub data: Value,
18}
19
20/// Generic, string-keyed pub/sub fan-out for SSE-style event delivery.
21///
22/// Applications map their own session ids at the boundary. Non-blocking sends drop on a
23/// full subscriber channel so a slow consumer cannot stall the broadcaster.
24pub struct SseBroadcaster {
25    subscribers: Mutex<HashMap<String, HashMap<String, mpsc::Sender<SseEvent>>>>,
26}
27
28impl Default for SseBroadcaster {
29    fn default() -> Self {
30        Self::new()
31    }
32}
33
34impl SseBroadcaster {
35    /// Creates an empty broadcaster.
36    pub fn new() -> Self {
37        Self {
38            subscribers: Mutex::new(HashMap::new()),
39        }
40    }
41
42    /// Registers a new subscriber for `app_key`.
43    ///
44    /// Returns `(subscriber_id, receiver)`. The channel is buffered (cap 100) so a
45    /// transient slow consumer does not drop events immediately.
46    pub fn subscribe(&self, app_key: &str) -> (String, mpsc::Receiver<SseEvent>) {
47        let (tx, rx) = mpsc::channel(100);
48        let subscriber_id = Uuid::new_v4().to_string();
49
50        let mut subs = self.subscribers.lock().expect("broadcaster mutex poisoned");
51        subs.entry(app_key.to_string())
52            .or_default()
53            .insert(subscriber_id.clone(), tx);
54
55        (subscriber_id, rx)
56    }
57
58    /// Removes a subscriber by id and closes its channel.
59    pub fn unsubscribe(&self, app_key: &str, subscriber_id: &str) {
60        let mut subs = self.subscribers.lock().expect("broadcaster mutex poisoned");
61        let Some(app_subs) = subs.get_mut(app_key) else {
62            return;
63        };
64        app_subs.remove(subscriber_id);
65        if app_subs.is_empty() {
66            subs.remove(app_key);
67        }
68    }
69
70    /// Sends an event to all subscribers for `app_key`.
71    ///
72    /// Non-blocking: a full subscriber channel is skipped (drop-on-full) so one slow
73    /// consumer cannot block the others.
74    pub fn broadcast(&self, app_key: &str, event: SseEvent) {
75        let subs = self.subscribers.lock().expect("broadcaster mutex poisoned");
76        let Some(app_subs) = subs.get(app_key) else {
77            return;
78        };
79        for tx in app_subs.values() {
80            let _ = tx.try_send(event.clone());
81        }
82    }
83
84    /// Closes all subscriber channels for `app_key` and removes the entry.
85    pub fn close(&self, app_key: &str) {
86        let mut subs = self.subscribers.lock().expect("broadcaster mutex poisoned");
87        subs.remove(app_key);
88    }
89
90    /// Closes every subscriber channel across all sessions.
91    pub fn close_all(&self) {
92        let mut subs = self.subscribers.lock().expect("broadcaster mutex poisoned");
93        subs.clear();
94    }
95}
soothe_client/appkit/broadcaster.rs

soothe_client/appkit/
broadcaster.rs
