pamoja_loopback/
broker.rs

1//! The shared in-memory broker that routes loopback messages.
2
3use std::sync::{Arc, Mutex};
4
5use tokio::sync::mpsc::UnboundedSender;
6
7use crate::transport::Message;
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#[derive(Clone, Default)]
16pub struct LoopbackBroker {
17    subscriptions: Arc<Mutex<Vec<Subscription>>>,
18}
19
20/// One connected transport's topic filters and delivery channel.
21struct Subscription {
22    filters: Arc<Mutex<Vec<String>>>,
23    sender: UnboundedSender<Message>,
24}
25
26impl LoopbackBroker {
27    /// Creates an empty broker.
28    ///
29    /// # Returns
30    ///
31    /// A broker with no registered transports.
32    pub fn new() -> Self {
33        Self::default()
34    }
35
36    /// Registers a transport's filters and delivery channel.
37    pub(crate) fn register(
38        &self,
39        filters: Arc<Mutex<Vec<String>>>,
40        sender: UnboundedSender<Message>,
41    ) {
42        self.subscriptions
43            .lock()
44            .expect("broker lock")
45            .push(Subscription { filters, sender });
46    }
47
48    /// Delivers a message to every subscription whose filters match its topic,
49    /// pruning channels whose receiver has been dropped.
50    pub(crate) fn publish(&self, message: &Message) {
51        let mut subscriptions = self.subscriptions.lock().expect("broker lock");
52        subscriptions.retain(|subscription| {
53            if subscription.sender.is_closed() {
54                return false;
55            }
56            let matched = subscription
57                .filters
58                .lock()
59                .expect("filters lock")
60                .iter()
61                .any(|filter| topic_matches(filter, &message.topic));
62            if matched {
63                let _ = subscription.sender.send(message.clone());
64            }
65            true
66        });
67    }
68}
69
70/// Returns whether an MQTT-style topic `filter` matches a concrete `topic`.
71///
72/// `+` matches exactly one level, and `#` matches the remaining levels including
73/// none. Following the MQTT specification, a filter that begins with a wildcard does
74/// not match a topic that begins with `$`, which is reserved for system topics.
75fn topic_matches(filter: &str, topic: &str) -> bool {
76    if topic.starts_with('$') {
77        if let Some(first) = filter.split('/').next() {
78            if first == "#" || first == "+" {
79                return false;
80            }
81        }
82    }
83    let mut filter_levels = filter.split('/');
84    let mut topic_levels = topic.split('/');
85    loop {
86        match (filter_levels.next(), topic_levels.next()) {
87            (Some("#"), _) => return true,
88            (Some("+"), Some(_)) => {}
89            (Some(filter_level), Some(topic_level)) if filter_level == topic_level => {}
90            (None, None) => return true,
91            _ => return false,
92        }
93    }
94}
95
96#[cfg(test)]
97mod tests {
98    use super::topic_matches;
99
100    #[test]
101    fn exact_topics_match() {
102        assert!(topic_matches("a/b/c", "a/b/c"));
103        assert!(!topic_matches("a/b/c", "a/b/d"));
104        assert!(!topic_matches("a/b", "a/b/c"));
105        assert!(!topic_matches("a/b/c", "a/b"));
106    }
107
108    #[test]
109    fn single_level_wildcard_matches_one_level() {
110        assert!(topic_matches("a/+/c", "a/b/c"));
111        assert!(topic_matches(
112            "sensors/+/temperature",
113            "sensors/1/temperature"
114        ));
115        assert!(!topic_matches("a/+/c", "a/b/c/d"));
116        assert!(!topic_matches("a/+", "a"));
117    }
118
119    #[test]
120    fn multi_level_wildcard_matches_the_rest() {
121        assert!(topic_matches("a/#", "a/b/c"));
122        assert!(topic_matches("a/#", "a"));
123        assert!(topic_matches("#", "a/b/c"));
124        assert!(!topic_matches("a/#", "b/c"));
125    }
126
127    #[test]
128    fn leading_wildcards_do_not_match_dollar_topics() {
129        // MQTT reserves $-prefixed topics from wildcard subscriptions.
130        assert!(!topic_matches("#", "$SYS/broker/uptime"));
131        assert!(!topic_matches("+/broker", "$SYS/broker"));
132        // An explicit filter still reaches a $ topic.
133        assert!(topic_matches("$SYS/#", "$SYS/broker/uptime"));
134        assert!(topic_matches("$SYS/+", "$SYS/uptime"));
135    }
136}
pamoja_loopback/broker.rs

pamoja_loopback/
broker.rs
