ruststream_rumqttc/
filter.rs

1//! [`MqttTopic`]: the subscription descriptor.
2//!
3//! Wildcards are the protocol's own (`+` per level, `#` terminal); `shared` wraps the filter
4//! into an MQTT 5 shared subscription (`$share/<group>/<filter>`), which is how competing
5//! consumers are expressed at all.
6
7use rumqttc::v5::mqttbytes::valid_filter;
8use ruststream::SubscriptionSource;
9
10use crate::broker::ConnectedMqttBroker;
11use crate::error::MqttError;
12use crate::subscriber::MqttSubscriber;
13
14/// Delivery quality of service for a subscription or a publish policy.
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
16pub enum Qos {
17    /// Fire and forget; deliveries carry no acknowledgement
18    /// ([`AckError::Unsupported`](ruststream::AckError::Unsupported)).
19    AtMostOnce,
20    /// Acknowledged delivery. The default.
21    #[default]
22    AtLeastOnce,
23    /// Exactly-once handshake (the client completes the second leg automatically).
24    ExactlyOnce,
25}
26
27impl Qos {
28    pub(crate) fn to_client(self) -> rumqttc::v5::mqttbytes::QoS {
29        match self {
30            Self::AtMostOnce => rumqttc::v5::mqttbytes::QoS::AtMostOnce,
31            Self::AtLeastOnce => rumqttc::v5::mqttbytes::QoS::AtLeastOnce,
32            Self::ExactlyOnce => rumqttc::v5::mqttbytes::QoS::ExactlyOnce,
33        }
34    }
35}
36
37/// A subscription descriptor for one MQTT topic filter.
38///
39/// Implements [`SubscriptionSource`], so it can sit inline in the `#[subscriber(..)]`
40/// decorator:
41///
42/// ```
43/// use ruststream_rumqttc::{MqttTopic, Qos};
44///
45/// let source = MqttTopic::new("devices/+/telemetry")
46///     .qos(Qos::AtLeastOnce)
47///     .shared("workers");
48/// # let _ = source;
49/// ```
50#[derive(Debug, Clone, PartialEq, Eq)]
51#[must_use]
52pub struct MqttTopic {
53    filter: String,
54    qos: Qos,
55    shared: Option<String>,
56}
57
58impl MqttTopic {
59    /// Names the topic filter, with wildcards as the protocol defines them.
60    pub fn new(filter: impl Into<String>) -> Self {
61        Self {
62            filter: filter.into(),
63            qos: Qos::default(),
64            shared: None,
65        }
66    }
67
68    /// Sets the delivery quality of service. Defaults to [`Qos::AtLeastOnce`].
69    pub fn qos(mut self, qos: Qos) -> Self {
70        self.qos = qos;
71        self
72    }
73
74    /// Makes this an MQTT 5 shared subscription in `group`: the broker distributes matching
75    /// messages across the group's consumers instead of fanning out to each.
76    pub fn shared(mut self, group: impl Into<String>) -> Self {
77        self.shared = Some(group.into());
78        self
79    }
80
81    /// The plain topic filter (without any share group).
82    #[must_use]
83    pub fn filter(&self) -> &str {
84        &self.filter
85    }
86
87    pub(crate) fn qos_value(&self) -> Qos {
88        self.qos
89    }
90
91    /// The filter as subscribed on the wire (`$share/<group>/<filter>` when shared).
92    pub(crate) fn wire_filter(&self) -> String {
93        self.shared.as_ref().map_or_else(
94            || self.filter.clone(),
95            |group| format!("$share/{group}/{}", self.filter),
96        )
97    }
98
99    /// Rejects descriptors that cannot form a subscription, before any I/O. The client's own
100    /// send-path error cannot say why a request failed, so validation happens here.
101    pub(crate) fn validate(&self) -> Result<(), MqttError> {
102        if !valid_filter(&self.filter) {
103            return Err(MqttError::Invalid(format!(
104                "'{}' is not a valid MQTT topic filter",
105                self.filter
106            )));
107        }
108        if let Some(group) = &self.shared {
109            if group.is_empty() || group.contains(['/', '+', '#']) {
110                return Err(MqttError::Invalid(format!(
111                    "'{group}' is not a valid share group name"
112                )));
113            }
114        }
115        Ok(())
116    }
117}
118
119impl SubscriptionSource<ConnectedMqttBroker> for MqttTopic {
120    type Subscriber = MqttSubscriber;
121
122    fn name(&self) -> &str {
123        self.filter()
124    }
125
126    async fn subscribe(self, connected: &ConnectedMqttBroker) -> Result<MqttSubscriber, MqttError> {
127        connected.subscribe_topic(self).await
128    }
129}
130
131#[cfg(test)]
132mod tests {
133    use super::*;
134
135    #[test]
136    fn invalid_filters_are_rejected_before_io() {
137        assert!(MqttTopic::new("a/#/b").validate().is_err());
138        assert!(MqttTopic::new("").validate().is_err());
139    }
140
141    #[test]
142    fn invalid_share_groups_are_rejected_before_io() {
143        assert!(MqttTopic::new("a").shared("g/1").validate().is_err());
144        assert!(MqttTopic::new("a").shared("").validate().is_err());
145    }
146
147    #[test]
148    fn shared_filters_wrap_on_the_wire_only() {
149        let topic = MqttTopic::new("orders/+").shared("workers");
150        assert_eq!(topic.filter(), "orders/+");
151        assert_eq!(topic.wire_filter(), "$share/workers/orders/+");
152    }
153}
ruststream_rumqttc/filter.rs

ruststream_rumqttc/
filter.rs
