pamoja_bus/
lib.rs

1//! An in-memory typed publish/subscribe event bus.
2//!
3//! [`BroadcastBus`] implements the core [`EventBus`] trait
4//! over a bounded broadcast channel: every event published is delivered to every
5//! current subscriber. Producers such as sensors and transports publish events,
6//! and consumers await them, all statically typed to one event type per bus.
7//!
8//! The bus is bounded, so a subscriber that falls far enough behind drops the
9//! events it missed and resumes from the most recent ones. This keeps a slow
10//! consumer from holding memory without bound, which matters on constrained
11//! devices.
12
13use pamoja_core::{EventBus, Result};
14use tokio::sync::broadcast;
15
16/// A typed publish/subscribe bus that broadcasts each event to all subscribers.
17///
18/// Every handle can both publish and receive. Use [`subscribe`](BroadcastBus::subscribe)
19/// to add an independent consumer; an event published after a handle subscribes
20/// is delivered to it. A subscriber only sees events published after it
21/// subscribed, mirroring a live pub/sub channel.
22///
23/// # Examples
24///
25/// ```
26/// use pamoja_core::EventBus;
27/// use pamoja_bus::BroadcastBus;
28///
29/// # async fn run() -> pamoja_core::Result<()> {
30/// let bus = BroadcastBus::new(16);
31/// let mut subscriber = bus.subscribe();
32/// bus.publish("reading").await?;
33/// assert_eq!(subscriber.next_event().await?, Some("reading"));
34/// # Ok(())
35/// # }
36/// ```
37pub struct BroadcastBus<E> {
38    sender: broadcast::Sender<E>,
39    receiver: broadcast::Receiver<E>,
40}
41
42impl<E: Clone> BroadcastBus<E> {
43    /// Creates a bus buffering up to `capacity` unread events per subscriber.
44    ///
45    /// # Arguments
46    ///
47    /// * `capacity` - the per-subscriber buffer depth; a subscriber further behind
48    ///   than this drops the events it missed. Values below one are raised to one.
49    ///
50    /// # Returns
51    ///
52    /// A bus with one handle that can publish and receive.
53    pub fn new(capacity: usize) -> Self {
54        let (sender, receiver) = broadcast::channel(capacity.max(1));
55        Self { sender, receiver }
56    }
57
58    /// Creates another handle to the same bus with its own independent subscription.
59    ///
60    /// # Returns
61    ///
62    /// A handle that receives events published after this call and can also publish.
63    pub fn subscribe(&self) -> Self {
64        Self {
65            sender: self.sender.clone(),
66            receiver: self.sender.subscribe(),
67        }
68    }
69}
70
71impl<E: Clone> EventBus for BroadcastBus<E> {
72    type Event = E;
73
74    async fn publish(&self, event: Self::Event) -> Result<()> {
75        // `send` errors only when there are no receivers; this handle holds its
76        // own, so a publish always succeeds.
77        let _ = self.sender.send(event);
78        Ok(())
79    }
80
81    async fn next_event(&mut self) -> Result<Option<Self::Event>> {
82        loop {
83            match self.receiver.recv().await {
84                Ok(event) => return Ok(Some(event)),
85                Err(broadcast::error::RecvError::Closed) => return Ok(None),
86                // The subscriber fell behind; skip the dropped events and resume.
87                Err(broadcast::error::RecvError::Lagged(_)) => continue,
88            }
89        }
90    }
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96
97    #[tokio::test]
98    async fn delivers_a_published_event() {
99        let mut bus = BroadcastBus::new(8);
100        bus.publish(1).await.expect("publish");
101        assert_eq!(bus.next_event().await.expect("next"), Some(1));
102    }
103
104    #[tokio::test]
105    async fn fans_out_to_every_subscriber() {
106        let bus = BroadcastBus::new(8);
107        let mut first = bus.subscribe();
108        let mut second = bus.subscribe();
109
110        bus.publish("event").await.expect("publish");
111
112        assert_eq!(first.next_event().await.expect("next"), Some("event"));
113        assert_eq!(second.next_event().await.expect("next"), Some("event"));
114    }
115
116    #[tokio::test]
117    async fn a_lagging_subscriber_skips_dropped_events_and_resumes() {
118        let bus = BroadcastBus::new(2);
119        let mut subscriber = bus.subscribe();
120
121        for value in 0..5 {
122            bus.publish(value).await.expect("publish");
123        }
124
125        // Capacity is two, so the three oldest events are dropped; the subscriber
126        // resumes with the two most recent, in order.
127        let mut seen = Vec::new();
128        seen.push(subscriber.next_event().await.expect("next").expect("event"));
129        seen.push(subscriber.next_event().await.expect("next").expect("event"));
130        assert_eq!(seen, vec![3, 4]);
131    }
132}
pamoja_bus/lib.rs

pamoja_bus/
lib.rs
