aster_forge_events/
transient_bus.rs

1use std::sync::Arc;
2
3use tokio::sync::broadcast;
4
5/// Process-local event broadcast paired with an optional shared transport.
6///
7/// The local channel is always available, including in single-process deployments and tests.
8/// Products decide how to encode events for `R`, how to publish through it, and how remote items
9/// are decoded back into the local channel.
10pub struct TransientEventBus<T, R = ()> {
11    local: broadcast::Sender<T>,
12    transport: Option<Arc<R>>,
13}
14
15impl<T, R> Clone for TransientEventBus<T, R> {
16    fn clone(&self) -> Self {
17        Self {
18            local: self.local.clone(),
19            transport: self.transport.clone(),
20        }
21    }
22}
23
24impl<T, R> TransientEventBus<T, R>
25where
26    T: Clone,
27{
28    /// Creates a process-local bus without a shared transport.
29    pub fn new(capacity: usize) -> Self {
30        Self::from_optional_transport(capacity, None)
31    }
32
33    /// Creates a bus with a shared transport.
34    pub fn with_transport(capacity: usize, transport: R) -> Self {
35        Self::from_optional_transport(capacity, Some(transport))
36    }
37
38    /// Creates a bus from an optional shared transport.
39    pub fn from_optional_transport(capacity: usize, transport: Option<R>) -> Self {
40        let (local, _) = broadcast::channel(capacity.max(1));
41        Self {
42            local,
43            transport: transport.map(Arc::new),
44        }
45    }
46
47    /// Publishes one event to process-local subscribers.
48    ///
49    /// # Errors
50    ///
51    /// Returns the event when the local broadcast channel has no active receivers.
52    pub fn publish_local(&self, event: T) -> Result<usize, broadcast::error::SendError<T>> {
53        self.local.send(event)
54    }
55
56    /// Subscribes to future process-local events.
57    #[must_use]
58    pub fn subscribe(&self) -> broadcast::Receiver<T> {
59        self.local.subscribe()
60    }
61
62    /// Returns the number of active process-local subscribers.
63    #[must_use]
64    pub fn local_subscriber_count(&self) -> usize {
65        self.local.receiver_count()
66    }
67
68    /// Returns whether a shared transport is configured.
69    #[must_use]
70    pub fn has_transport(&self) -> bool {
71        self.transport.is_some()
72    }
73
74    /// Returns the shared transport, if configured.
75    #[must_use]
76    pub fn transport(&self) -> Option<Arc<R>> {
77        self.transport.clone()
78    }
79}
80
81#[cfg(test)]
82mod tests {
83    use std::sync::Arc;
84
85    use super::TransientEventBus;
86    use tokio::sync::broadcast;
87
88    #[tokio::test]
89    async fn local_bus_delivers_to_all_current_subscribers() {
90        let bus = TransientEventBus::<String>::new(4);
91        let mut first = bus.subscribe();
92        let mut second = bus.subscribe();
93
94        assert_eq!(bus.local_subscriber_count(), 2);
95        assert_eq!(bus.publish_local("event".to_string()).expect("publish"), 2);
96        assert_eq!(first.recv().await, Ok("event".to_string()));
97        assert_eq!(second.recv().await, Ok("event".to_string()));
98    }
99
100    #[test]
101    fn zero_capacity_is_clamped_and_publish_without_subscribers_returns_event() {
102        let bus = TransientEventBus::<String>::new(0);
103
104        let error = bus
105            .publish_local("unobserved".to_string())
106            .expect_err("publish without subscribers should report the undelivered event");
107        assert_eq!(error.0, "unobserved");
108    }
109
110    #[tokio::test]
111    async fn bounded_local_channel_reports_lag() {
112        let bus = TransientEventBus::<u8>::new(1);
113        let mut receiver = bus.subscribe();
114
115        assert_eq!(bus.publish_local(1).expect("publish first"), 1);
116        assert_eq!(bus.publish_local(2).expect("publish second"), 1);
117        assert_eq!(
118            receiver.recv().await,
119            Err(broadcast::error::RecvError::Lagged(1))
120        );
121        assert_eq!(receiver.recv().await, Ok(2));
122    }
123
124    #[tokio::test]
125    async fn cloned_bus_shares_local_subscribers() {
126        let bus = TransientEventBus::<u8>::new(2);
127        let cloned = bus.clone();
128        let mut receiver = bus.subscribe();
129
130        assert_eq!(cloned.publish_local(7).expect("publish from clone"), 1);
131        assert_eq!(receiver.recv().await, Ok(7));
132    }
133
134    #[test]
135    fn optional_transport_is_shared_across_clones() {
136        let bus = TransientEventBus::<u8, String>::with_transport(2, "transport".to_string());
137        let cloned = bus.clone();
138
139        assert!(bus.has_transport());
140        assert!(Arc::ptr_eq(
141            &bus.transport().expect("transport"),
142            &cloned.transport().expect("cloned transport")
143        ));
144
145        let local = TransientEventBus::<u8, String>::new(2);
146        assert!(!local.has_transport());
147        assert!(local.transport().is_none());
148    }
149}
aster_forge_events/transient_bus.rs

aster_forge_events/
transient_bus.rs
