ironflow_api/
sse.rs

1//! Server-Sent Events broadcaster for real-time event streaming.
2//!
3//! [`SseBroadcaster`] implements [`EventSubscriber`] and forwards events into
4//! a [`tokio::sync::broadcast`] channel. The SSE route creates receivers from
5//! this channel and streams them to connected clients.
6
7use tokio::sync::broadcast;
8
9use ironflow_engine::notify::{Event, EventSubscriber, SubscriberFuture};
10
11/// Default broadcast channel capacity.
12const DEFAULT_CAPACITY: usize = 256;
13
14/// Broadcasts [`Event`]s to SSE clients via a [`tokio::sync::broadcast`] channel.
15///
16/// Register this as an [`EventSubscriber`] on the [`Engine`](ironflow_engine::engine::Engine)
17/// to forward all (or filtered) domain events to connected SSE clients.
18///
19/// # Examples
20///
21/// ```
22/// use ironflow_api::sse::SseBroadcaster;
23///
24/// let broadcaster = SseBroadcaster::new();
25/// assert_eq!(broadcaster.receiver_count(), 0);
26/// let _receiver = broadcaster.subscribe();
27/// assert_eq!(broadcaster.receiver_count(), 1);
28/// ```
29pub struct SseBroadcaster {
30    sender: broadcast::Sender<Event>,
31}
32
33impl SseBroadcaster {
34    /// Create a new broadcaster with the default capacity (256).
35    ///
36    /// # Examples
37    ///
38    /// ```
39    /// use ironflow_api::sse::SseBroadcaster;
40    ///
41    /// let broadcaster = SseBroadcaster::new();
42    /// ```
43    pub fn new() -> Self {
44        Self::with_capacity(DEFAULT_CAPACITY)
45    }
46
47    /// Create a new broadcaster with a custom channel capacity.
48    ///
49    /// # Examples
50    ///
51    /// ```
52    /// use ironflow_api::sse::SseBroadcaster;
53    ///
54    /// let broadcaster = SseBroadcaster::with_capacity(64);
55    /// ```
56    pub fn with_capacity(capacity: usize) -> Self {
57        let (sender, _) = broadcast::channel(capacity);
58        Self { sender }
59    }
60
61    /// Create a new receiver for the broadcast channel.
62    ///
63    /// Each SSE client connection calls this to get its own receiver.
64    pub fn subscribe(&self) -> broadcast::Receiver<Event> {
65        self.sender.subscribe()
66    }
67
68    /// Returns the number of active receivers (connected SSE clients).
69    pub fn receiver_count(&self) -> usize {
70        self.sender.receiver_count()
71    }
72
73    /// Returns a clone of the underlying sender.
74    ///
75    /// Stored in [`AppState`](crate::state::AppState) so that the SSE route
76    /// can create receivers without holding a reference to the broadcaster.
77    pub fn sender(&self) -> broadcast::Sender<Event> {
78        self.sender.clone()
79    }
80}
81
82impl Default for SseBroadcaster {
83    fn default() -> Self {
84        Self::new()
85    }
86}
87
88impl EventSubscriber for SseBroadcaster {
89    fn name(&self) -> &str {
90        "sse"
91    }
92
93    fn handle<'a>(&'a self, event: &'a Event) -> SubscriberFuture<'a> {
94        let event = event.clone();
95        Box::pin(async move {
96            // Ignore send errors: they mean no receivers are connected.
97            let _ = self.sender.send(event);
98        })
99    }
100}
101
102#[cfg(test)]
103mod tests {
104    use std::collections::HashMap;
105
106    use super::*;
107    use chrono::Utc;
108    use ironflow_store::models::RunStatus;
109    use rust_decimal::Decimal;
110    use uuid::Uuid;
111
112    fn sample_event() -> Event {
113        Event::RunStatusChanged {
114            run_id: Uuid::now_v7(),
115            workflow_name: "deploy".to_string(),
116            from: RunStatus::Running,
117            to: RunStatus::Completed,
118            error: None,
119            cost_usd: Decimal::ZERO,
120            duration_ms: 1000,
121            labels: HashMap::new(),
122            at: Utc::now(),
123        }
124    }
125
126    #[test]
127    fn new_creates_broadcaster() {
128        let broadcaster = SseBroadcaster::new();
129        assert_eq!(broadcaster.receiver_count(), 0);
130    }
131
132    #[test]
133    fn default_creates_broadcaster() {
134        let broadcaster = SseBroadcaster::default();
135        assert_eq!(broadcaster.receiver_count(), 0);
136    }
137
138    #[test]
139    fn subscribe_creates_receiver() {
140        let broadcaster = SseBroadcaster::new();
141        let _rx = broadcaster.subscribe();
142        assert_eq!(broadcaster.receiver_count(), 1);
143    }
144
145    #[test]
146    fn receiver_count_tracks_active_receivers() {
147        let broadcaster = SseBroadcaster::new();
148        let _rx1 = broadcaster.subscribe();
149        let _rx2 = broadcaster.subscribe();
150        assert_eq!(broadcaster.receiver_count(), 2);
151        drop(_rx1);
152        assert_eq!(broadcaster.receiver_count(), 1);
153    }
154
155    #[tokio::test]
156    async fn handle_sends_event_to_receivers() {
157        let broadcaster = SseBroadcaster::new();
158        let mut rx = broadcaster.subscribe();
159
160        let event = sample_event();
161        broadcaster.handle(&event).await;
162
163        let received = rx.recv().await.expect("should receive event");
164        assert_eq!(received.event_type(), "run_status_changed");
165    }
166
167    #[tokio::test]
168    async fn handle_no_receivers_does_not_panic() {
169        let broadcaster = SseBroadcaster::new();
170        let event = sample_event();
171        // No receivers -- should not panic
172        broadcaster.handle(&event).await;
173    }
174
175    #[test]
176    fn sender_returns_clone() {
177        let broadcaster = SseBroadcaster::new();
178        let sender = broadcaster.sender();
179        let _rx = sender.subscribe();
180        // The receiver from the cloned sender counts on the original channel
181        assert_eq!(broadcaster.receiver_count(), 1);
182    }
183
184    #[test]
185    fn name_returns_sse() {
186        let broadcaster = SseBroadcaster::new();
187        assert_eq!(broadcaster.name(), "sse");
188    }
189}
ironflow_api/sse.rs

ironflow_api/
sse.rs
