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.
74fn topic_matches(filter: &str, topic: &str) -> bool {
75    let mut filter_levels = filter.split('/');
76    let mut topic_levels = topic.split('/');
77    loop {
78        match (filter_levels.next(), topic_levels.next()) {
79            (Some("#"), _) => return true,
80            (Some("+"), Some(_)) => {}
81            (Some(filter_level), Some(topic_level)) if filter_level == topic_level => {}
82            (None, None) => return true,
83            _ => return false,
84        }
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use super::topic_matches;
91
92    #[test]
93    fn exact_topics_match() {
94        assert!(topic_matches("a/b/c", "a/b/c"));
95        assert!(!topic_matches("a/b/c", "a/b/d"));
96        assert!(!topic_matches("a/b", "a/b/c"));
97        assert!(!topic_matches("a/b/c", "a/b"));
98    }
99
100    #[test]
101    fn single_level_wildcard_matches_one_level() {
102        assert!(topic_matches("a/+/c", "a/b/c"));
103        assert!(topic_matches(
104            "sensors/+/temperature",
105            "sensors/1/temperature"
106        ));
107        assert!(!topic_matches("a/+/c", "a/b/c/d"));
108        assert!(!topic_matches("a/+", "a"));
109    }
110
111    #[test]
112    fn multi_level_wildcard_matches_the_rest() {
113        assert!(topic_matches("a/#", "a/b/c"));
114        assert!(topic_matches("a/#", "a"));
115        assert!(topic_matches("#", "a/b/c"));
116        assert!(!topic_matches("a/#", "b/c"));
117    }
118}
pamoja_loopback/broker.rs

pamoja_loopback/
broker.rs
