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_engine::notify::RunStatusChangedEvent;
109    use ironflow_store::models::RunStatus;
110    use rust_decimal::Decimal;
111    use uuid::Uuid;
112
113    fn sample_event() -> Event {
114        Event::RunStatusChanged(RunStatusChangedEvent {
115            run_id: Uuid::now_v7(),
116            workflow_name: "deploy".to_string(),
117            from: RunStatus::Running,
118            to: RunStatus::Completed,
119            error: None,
120            cost_usd: Decimal::ZERO,
121            duration_ms: 1000,
122            labels: HashMap::new(),
123            at: Utc::now(),
124        })
125    }
126
127    #[test]
128    fn new_creates_broadcaster() {
129        let broadcaster = SseBroadcaster::new();
130        assert_eq!(broadcaster.receiver_count(), 0);
131    }
132
133    #[test]
134    fn default_creates_broadcaster() {
135        let broadcaster = SseBroadcaster::default();
136        assert_eq!(broadcaster.receiver_count(), 0);
137    }
138
139    #[test]
140    fn subscribe_creates_receiver() {
141        let broadcaster = SseBroadcaster::new();
142        let _rx = broadcaster.subscribe();
143        assert_eq!(broadcaster.receiver_count(), 1);
144    }
145
146    #[test]
147    fn receiver_count_tracks_active_receivers() {
148        let broadcaster = SseBroadcaster::new();
149        let _rx1 = broadcaster.subscribe();
150        let _rx2 = broadcaster.subscribe();
151        assert_eq!(broadcaster.receiver_count(), 2);
152        drop(_rx1);
153        assert_eq!(broadcaster.receiver_count(), 1);
154    }
155
156    #[tokio::test]
157    async fn handle_sends_event_to_receivers() {
158        let broadcaster = SseBroadcaster::new();
159        let mut rx = broadcaster.subscribe();
160
161        let event = sample_event();
162        broadcaster.handle(&event).await;
163
164        let received = rx.recv().await.expect("should receive event");
165        assert_eq!(received.event_type(), "run_status_changed");
166    }
167
168    #[tokio::test]
169    async fn handle_no_receivers_does_not_panic() {
170        let broadcaster = SseBroadcaster::new();
171        let event = sample_event();
172        // No receivers -- should not panic
173        broadcaster.handle(&event).await;
174    }
175
176    #[test]
177    fn sender_returns_clone() {
178        let broadcaster = SseBroadcaster::new();
179        let sender = broadcaster.sender();
180        let _rx = sender.subscribe();
181        // The receiver from the cloned sender counts on the original channel
182        assert_eq!(broadcaster.receiver_count(), 1);
183    }
184
185    #[test]
186    fn name_returns_sse() {
187        let broadcaster = SseBroadcaster::new();
188        assert_eq!(broadcaster.name(), "sse");
189    }
190}
ironflow_api/sse.rs

ironflow_api/
sse.rs
