ironflow_engine/notify/
publisher.rs

1//! [`EventPublisher`] -- broadcasts events to filtered subscribers.
2
3use std::sync::Arc;
4
5use tokio::spawn;
6
7use super::{Event, EventSubscriber};
8
9/// A subscriber paired with its event type filter.
10#[derive(Clone)]
11struct Subscription {
12    subscriber: Arc<dyn EventSubscriber>,
13    event_types: Vec<&'static str>,
14}
15
16impl Subscription {
17    /// Returns `true` if this subscription accepts the given event.
18    fn accepts(&self, event: &Event) -> bool {
19        self.event_types.contains(&event.event_type())
20    }
21}
22
23/// Broadcasts [`Event`]s to registered [`EventSubscriber`]s.
24///
25/// Each subscriber is paired with an event type filter at subscription
26/// time. Only matching events are dispatched. Each call runs in a
27/// spawned task so that slow subscribers do not block the engine.
28///
29/// # Examples
30///
31/// ```no_run
32/// use ironflow_engine::notify::{EventPublisher, WebhookSubscriber, Event};
33///
34/// let mut publisher = EventPublisher::new();
35/// publisher.subscribe(
36///     WebhookSubscriber::new("https://hooks.example.com/events"),
37///     &[Event::RUN_STATUS_CHANGED, Event::STEP_FAILED],
38/// );
39/// ```
40#[derive(Clone)]
41pub struct EventPublisher {
42    subscriptions: Vec<Subscription>,
43}
44
45impl EventPublisher {
46    /// Create an empty publisher with no subscribers.
47    ///
48    /// # Examples
49    ///
50    /// ```
51    /// use ironflow_engine::notify::EventPublisher;
52    ///
53    /// let publisher = EventPublisher::new();
54    /// assert_eq!(publisher.subscriber_count(), 0);
55    /// ```
56    pub fn new() -> Self {
57        Self {
58            subscriptions: Vec::new(),
59        }
60    }
61
62    /// Register a subscriber with an event type filter.
63    ///
64    /// The subscriber is called only for events whose
65    /// [`event_type()`](Event::event_type) is in `event_types`.
66    /// Pass [`Event::ALL`] to receive every event.
67    ///
68    /// Use the `Event::*` constants for the filter values.
69    ///
70    /// # Examples
71    ///
72    /// ```no_run
73    /// use ironflow_engine::notify::{EventPublisher, WebhookSubscriber, Event};
74    ///
75    /// let mut publisher = EventPublisher::new();
76    ///
77    /// // Only on specific event types:
78    /// publisher.subscribe(
79    ///     WebhookSubscriber::new("https://example.com/hook"),
80    ///     &[Event::RUN_STATUS_CHANGED, Event::STEP_FAILED],
81    /// );
82    ///
83    /// // On all events:
84    /// publisher.subscribe(
85    ///     WebhookSubscriber::new("https://example.com/all"),
86    ///     Event::ALL,
87    /// );
88    /// ```
89    pub fn subscribe(
90        &mut self,
91        subscriber: impl EventSubscriber + 'static,
92        event_types: &[&'static str],
93    ) {
94        self.subscriptions.push(Subscription {
95            subscriber: Arc::new(subscriber),
96            event_types: event_types.to_vec(),
97        });
98    }
99
100    /// Number of registered subscribers.
101    pub fn subscriber_count(&self) -> usize {
102        self.subscriptions.len()
103    }
104
105    /// Broadcast an event to all matching subscribers.
106    ///
107    /// Each matching subscriber runs in its own spawned task. This
108    /// method returns immediately and never blocks.
109    pub fn publish(&self, event: Event) {
110        for subscription in &self.subscriptions {
111            if !subscription.accepts(&event) {
112                continue;
113            }
114            let subscriber = subscription.subscriber.clone();
115            let event = event.clone();
116            spawn(async move {
117                subscriber.handle(&event).await;
118            });
119        }
120    }
121}
122
123impl Default for EventPublisher {
124    fn default() -> Self {
125        Self::new()
126    }
127}
128
129#[cfg(test)]
130mod tests {
131    use std::collections::HashMap;
132    use std::sync::atomic::{AtomicU32, Ordering};
133    use std::time::Duration;
134
135    use super::*;
136    use crate::notify::{
137        RunStatusChangedEvent, SubscriberFuture, UserSignedInEvent, WebhookSubscriber,
138    };
139    use rust_decimal::Decimal;
140    use tokio::time::sleep;
141
142    use chrono::Utc;
143    use ironflow_store::models::RunStatus;
144    use uuid::Uuid;
145
146    fn sample_run_status_changed() -> Event {
147        Event::RunStatusChanged(RunStatusChangedEvent {
148            run_id: Uuid::now_v7(),
149            workflow_name: "deploy".to_string(),
150            from: RunStatus::Running,
151            to: RunStatus::Completed,
152            error: None,
153            cost_usd: Decimal::new(42, 2),
154            duration_ms: 5000,
155            labels: HashMap::new(),
156            at: Utc::now(),
157        })
158    }
159
160    fn sample_user_signed_in() -> Event {
161        Event::UserSignedIn(UserSignedInEvent {
162            user_id: Uuid::now_v7(),
163            username: "alice".to_string(),
164            at: Utc::now(),
165        })
166    }
167
168    #[test]
169    fn starts_empty() {
170        let publisher = EventPublisher::new();
171        assert_eq!(publisher.subscriber_count(), 0);
172    }
173
174    #[test]
175    fn subscribe_increments_count() {
176        let mut publisher = EventPublisher::new();
177        publisher.subscribe(
178            WebhookSubscriber::new("https://example.com"),
179            &[Event::RUN_STATUS_CHANGED],
180        );
181        assert_eq!(publisher.subscriber_count(), 1);
182    }
183
184    #[test]
185    fn publish_with_no_subscribers_is_noop() {
186        let publisher = EventPublisher::new();
187        publisher.publish(sample_run_status_changed());
188    }
189
190    #[test]
191    fn default_is_empty() {
192        let publisher = EventPublisher::default();
193        assert_eq!(publisher.subscriber_count(), 0);
194    }
195
196    struct CountingSubscriber {
197        count: AtomicU32,
198    }
199
200    impl CountingSubscriber {
201        fn new() -> Self {
202            Self {
203                count: AtomicU32::new(0),
204            }
205        }
206
207        fn count(&self) -> u32 {
208            self.count.load(Ordering::SeqCst)
209        }
210    }
211
212    impl EventSubscriber for CountingSubscriber {
213        fn name(&self) -> &str {
214            "counting"
215        }
216
217        fn handle<'a>(&'a self, _event: &'a Event) -> SubscriberFuture<'a> {
218            Box::pin(async move {
219                self.count.fetch_add(1, Ordering::SeqCst);
220            })
221        }
222    }
223
224    #[tokio::test]
225    async fn subscriber_receives_matching_events() {
226        let subscriber = Arc::new(CountingSubscriber::new());
227        let mut publisher = EventPublisher::new();
228
229        struct ArcSub(Arc<CountingSubscriber>);
230        impl EventSubscriber for ArcSub {
231            fn name(&self) -> &str {
232                self.0.name()
233            }
234            fn handle<'a>(&'a self, event: &'a Event) -> SubscriberFuture<'a> {
235                self.0.handle(event)
236            }
237        }
238
239        publisher.subscribe(ArcSub(subscriber.clone()), &[Event::RUN_STATUS_CHANGED]);
240
241        publisher.publish(sample_run_status_changed()); // matches
242        publisher.publish(sample_user_signed_in()); // filtered out
243
244        sleep(Duration::from_millis(50)).await;
245
246        assert_eq!(subscriber.count(), 1);
247    }
248
249    #[tokio::test]
250    async fn all_filter_matches_everything() {
251        let subscriber = Arc::new(CountingSubscriber::new());
252        let mut publisher = EventPublisher::new();
253
254        struct ArcSub(Arc<CountingSubscriber>);
255        impl EventSubscriber for ArcSub {
256            fn name(&self) -> &str {
257                self.0.name()
258            }
259            fn handle<'a>(&'a self, event: &'a Event) -> SubscriberFuture<'a> {
260                self.0.handle(event)
261            }
262        }
263
264        publisher.subscribe(ArcSub(subscriber.clone()), Event::ALL);
265
266        publisher.publish(sample_run_status_changed());
267        publisher.publish(sample_user_signed_in());
268
269        sleep(Duration::from_millis(50)).await;
270
271        assert_eq!(subscriber.count(), 2);
272    }
273
274    #[tokio::test]
275    async fn empty_filter_matches_nothing() {
276        let subscriber = Arc::new(CountingSubscriber::new());
277        let mut publisher = EventPublisher::new();
278
279        struct ArcSub(Arc<CountingSubscriber>);
280        impl EventSubscriber for ArcSub {
281            fn name(&self) -> &str {
282                self.0.name()
283            }
284            fn handle<'a>(&'a self, event: &'a Event) -> SubscriberFuture<'a> {
285                self.0.handle(event)
286            }
287        }
288
289        publisher.subscribe(ArcSub(subscriber.clone()), &[]);
290
291        publisher.publish(sample_run_status_changed());
292        publisher.publish(sample_user_signed_in());
293
294        sleep(Duration::from_millis(50)).await;
295
296        assert_eq!(subscriber.count(), 0);
297    }
298}
ironflow_engine/notify/publisher.rs

ironflow_engine/notify/
publisher.rs
