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    pub fn publish_local(&self, event: T) -> Result<usize, broadcast::error::SendError<T>> {
49        self.local.send(event)
50    }
51
52    /// Subscribes to future process-local events.
53    pub fn subscribe(&self) -> broadcast::Receiver<T> {
54        self.local.subscribe()
55    }
56
57    /// Returns the number of active process-local subscribers.
58    pub fn local_subscriber_count(&self) -> usize {
59        self.local.receiver_count()
60    }
61
62    /// Returns whether a shared transport is configured.
63    pub fn has_transport(&self) -> bool {
64        self.transport.is_some()
65    }
66
67    /// Returns the shared transport, if configured.
68    pub fn transport(&self) -> Option<Arc<R>> {
69        self.transport.clone()
70    }
71}
72
73#[cfg(test)]
74mod tests {
75    use std::sync::Arc;
76
77    use super::TransientEventBus;
78    use tokio::sync::broadcast;
79
80    #[tokio::test]
81    async fn local_bus_delivers_to_all_current_subscribers() {
82        let bus = TransientEventBus::<String>::new(4);
83        let mut first = bus.subscribe();
84        let mut second = bus.subscribe();
85
86        assert_eq!(bus.local_subscriber_count(), 2);
87        assert_eq!(bus.publish_local("event".to_string()).expect("publish"), 2);
88        assert_eq!(first.recv().await, Ok("event".to_string()));
89        assert_eq!(second.recv().await, Ok("event".to_string()));
90    }
91
92    #[test]
93    fn zero_capacity_is_clamped_and_publish_without_subscribers_returns_event() {
94        let bus = TransientEventBus::<String>::new(0);
95
96        let error = bus
97            .publish_local("unobserved".to_string())
98            .expect_err("publish without subscribers should report the undelivered event");
99        assert_eq!(error.0, "unobserved");
100    }
101
102    #[tokio::test]
103    async fn bounded_local_channel_reports_lag() {
104        let bus = TransientEventBus::<u8>::new(1);
105        let mut receiver = bus.subscribe();
106
107        assert_eq!(bus.publish_local(1).expect("publish first"), 1);
108        assert_eq!(bus.publish_local(2).expect("publish second"), 1);
109        assert_eq!(
110            receiver.recv().await,
111            Err(broadcast::error::RecvError::Lagged(1))
112        );
113        assert_eq!(receiver.recv().await, Ok(2));
114    }
115
116    #[tokio::test]
117    async fn cloned_bus_shares_local_subscribers() {
118        let bus = TransientEventBus::<u8>::new(2);
119        let cloned = bus.clone();
120        let mut receiver = bus.subscribe();
121
122        assert_eq!(cloned.publish_local(7).expect("publish from clone"), 1);
123        assert_eq!(receiver.recv().await, Ok(7));
124    }
125
126    #[test]
127    fn optional_transport_is_shared_across_clones() {
128        let bus = TransientEventBus::<u8, String>::with_transport(2, "transport".to_string());
129        let cloned = bus.clone();
130
131        assert!(bus.has_transport());
132        assert!(Arc::ptr_eq(
133            &bus.transport().expect("transport"),
134            &cloned.transport().expect("cloned transport")
135        ));
136
137        let local = TransientEventBus::<u8, String>::new(2);
138        assert!(!local.has_transport());
139        assert!(local.transport().is_none());
140    }
141}
aster_forge_events/transient_bus.rs

aster_forge_events/
transient_bus.rs
