pamoja_loopback/
broker.rs

1//! The shared in-memory broker that routes loopback messages.
2
3use std::sync::atomic::{AtomicBool, Ordering};
4use std::sync::{Arc, Mutex};
5
6use pamoja_core::{topic_matches, Error, Message, Result};
7use tokio::sync::mpsc::UnboundedSender;
8
9/// A shared, in-process router for [`LoopbackTransport`](crate::LoopbackTransport)s.
10///
11/// Clone a single broker into every transport that should share a namespace; a
12/// publish on one transport is delivered to every transport whose subscriptions
13/// match the topic. The broker is cheap to clone, and all clones share one
14/// routing table.
15///
16/// A broker stands for the network its links share, so taking it out of reach
17/// with [`set_reachable`](LoopbackBroker::set_reachable) is how a test puts every
18/// link on it out of range at once, including links a ladder now owns.
19#[derive(Clone)]
20pub struct LoopbackBroker {
21    subscriptions: Arc<Mutex<Vec<Subscription>>>,
22    reachable: Arc<AtomicBool>,
23}
24
25/// One connected transport's topic filters and delivery channel.
26struct Subscription {
27    filters: Arc<Mutex<Vec<String>>>,
28    sender: UnboundedSender<Message>,
29}
30
31impl Default for LoopbackBroker {
32    fn default() -> Self {
33        Self {
34            subscriptions: Arc::default(),
35            reachable: Arc::new(AtomicBool::new(true)),
36        }
37    }
38}
39
40impl LoopbackBroker {
41    /// Creates an empty broker.
42    ///
43    /// # Returns
44    ///
45    /// A broker with no registered transports, in reach.
46    pub fn new() -> Self {
47        Self::default()
48    }
49
50    /// Takes the broker out of reach, or brings it back.
51    ///
52    /// Out of reach, every link on the broker fails to connect, send, or
53    /// subscribe with [`Error::Transport`], the error an out-of-range radio or a
54    /// dropped network raises, and nothing is delivered. A link keeps its
55    /// connection and its filters through the outage, so it carries traffic again
56    /// as soon as the broker is back.
57    ///
58    /// # Arguments
59    ///
60    /// * `reachable` - `false` to take the broker out of reach, `true` to bring it
61    ///   back.
62    pub fn set_reachable(&self, reachable: bool) {
63        self.reachable.store(reachable, Ordering::SeqCst);
64    }
65
66    /// Reports whether the broker is in reach.
67    ///
68    /// # Returns
69    ///
70    /// `true` unless [`set_reachable`](LoopbackBroker::set_reachable) took it out.
71    pub fn is_reachable(&self) -> bool {
72        self.reachable.load(Ordering::SeqCst)
73    }
74
75    /// Fails with the error a link on an unreachable broker reports.
76    pub(crate) fn in_reach(&self) -> Result<()> {
77        if self.is_reachable() {
78            Ok(())
79        } else {
80            Err(Error::Transport("the broker is out of reach".to_owned()))
81        }
82    }
83
84    /// Registers a transport's filters and delivery channel.
85    pub(crate) fn register(
86        &self,
87        filters: Arc<Mutex<Vec<String>>>,
88        sender: UnboundedSender<Message>,
89    ) {
90        self.subscriptions
91            .lock()
92            .expect("broker lock")
93            .push(Subscription { filters, sender });
94    }
95
96    /// Delivers a message to every subscription whose filters match its topic,
97    /// pruning channels whose receiver has been dropped.
98    pub(crate) fn publish(&self, message: &Message) {
99        let mut subscriptions = self.subscriptions.lock().expect("broker lock");
100        subscriptions.retain(|subscription| {
101            if subscription.sender.is_closed() {
102                return false;
103            }
104            let matched = subscription
105                .filters
106                .lock()
107                .expect("filters lock")
108                .iter()
109                .any(|filter| topic_matches(filter, &message.topic));
110            if matched {
111                let _ = subscription.sender.send(message.clone());
112            }
113            true
114        });
115    }
116}
pamoja_loopback/broker.rs

pamoja_loopback/
broker.rs
