rust_elm/
bus.rs

1use crossbeam_channel::{bounded, Receiver, Sender, TrySendError};
2use std::sync::atomic::{AtomicU64, Ordering};
3
4/// Lock-free bounded MPMC message bus with explicit backpressure.
5#[derive(Debug)]
6pub struct Bus<M> {
7    sender: Sender<M>,
8    receiver: Receiver<M>,
9    capacity: usize,
10    dropped: AtomicU64,
11}
12
13impl<M: Send + 'static> Bus<M> {
14    pub fn new(capacity: usize) -> Self {
15        let (sender, receiver) = bounded(capacity.max(1));
16        Self {
17            sender,
18            receiver,
19            capacity: capacity.max(1),
20            dropped: AtomicU64::new(0),
21        }
22    }
23
24    pub fn sender(&self) -> BusSender<M> {
25        BusSender {
26            inner: self.sender.clone(),
27            dropped: &self.dropped,
28        }
29    }
30
31    pub fn receiver(&self) -> &Receiver<M> {
32        &self.receiver
33    }
34
35    pub fn capacity(&self) -> usize {
36        self.capacity
37    }
38
39    pub fn dropped_count(&self) -> u64 {
40        self.dropped.load(Ordering::Relaxed)
41    }
42}
43
44#[derive(Debug)]
45pub struct BusSender<M> {
46    inner: Sender<M>,
47    dropped: *const AtomicU64,
48}
49
50impl<M> Clone for BusSender<M> {
51    fn clone(&self) -> Self {
52        Self {
53            inner: self.inner.clone(),
54            dropped: self.dropped,
55        }
56    }
57}
58
59impl<M> BusSender<M> {
60    pub fn send(&self, msg: M) -> Result<(), M> {
61        match self.inner.try_send(msg) {
62            Ok(()) => Ok(()),
63            Err(TrySendError::Full(msg)) => {
64                // Backpressure: drop and count when full (caller may retry).
65                unsafe {
66                    (*self.dropped).fetch_add(1, Ordering::Relaxed);
67                }
68                Err(msg)
69            }
70            Err(TrySendError::Disconnected(msg)) => Err(msg),
71        }
72    }
73
74    pub fn send_blocking(&self, msg: M) -> Result<(), M> {
75        self.inner.send(msg).map_err(|e| e.into_inner())
76    }
77}
78
79unsafe impl<M> Send for BusSender<M> {}
80unsafe impl<M> Sync for BusSender<M> {}
81
82#[cfg(test)]
83mod tests {
84    use super::*;
85
86    #[test]
87    fn backpressure_counts_drops() {
88        let bus = Bus::<u32>::new(1);
89        let tx = bus.sender();
90        assert!(tx.send(1).is_ok());
91        let dropped = tx.send(2).unwrap_err();
92        assert_eq!(dropped, 2);
93        assert_eq!(bus.dropped_count(), 1);
94    }
95}
rust_elm/bus.rs

rust_elm/
bus.rs
