mqute_codec/protocol/v5/
subscribe.rs

1//! # Subscribe Packet - MQTT v5
2//!
3//! This module implements the MQTT v5 `Subscribe` packet, which is sent by clients to
4//! request subscription to one or more topics. The packet includes detailed subscription
5//! options and properties for each topic filter.
6
7use crate::Error;
8use crate::codec::util::{
9    decode_byte, decode_string, decode_variable_integer, encode_string, encode_variable_integer,
10};
11use crate::codec::{Decode, Encode, RawPacket};
12use crate::protocol::util::len_bytes;
13use crate::protocol::v5::property::{
14    Property, PropertyFrame, property_decode, property_encode, property_len,
15};
16use crate::protocol::v5::util::id_header;
17use crate::protocol::{FixedHeader, Flags, PacketType, QoS, traits, util};
18use bytes::{Buf, BufMut, Bytes, BytesMut};
19use std::borrow::Borrow;
20use std::ops::{Index, IndexMut};
21
22/// Properties specific to `Subscribe` packets
23///
24/// In MQTT v5, `Subscribe` packets can include:
25/// - Subscription Identifier (for shared subscriptions)
26/// - User Properties (key-value pairs for extended metadata)
27///
28/// # Example
29///
30/// ```rust
31/// use mqute_codec::protocol::v5::SubscribeProperties;
32///
33/// let properties = SubscribeProperties {
34///     subscription_id: Some(42),  // Shared subscription ID
35///     user_properties: vec![("client".into(), "rust".into())],
36/// };
37/// ```
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub struct SubscribeProperties {
40    /// Identifier for shared subscriptions
41    pub subscription_id: Option<u32>,
42    /// User-defined key-value properties
43    pub user_properties: Vec<(String, String)>,
44}
45
46impl PropertyFrame for SubscribeProperties {
47    /// Calculates the encoded length of the properties
48    fn encoded_len(&self) -> usize {
49        let mut len = 0usize;
50
51        if let Some(value) = self.subscription_id {
52            len += 1 + len_bytes(value as usize);
53        }
54        len += property_len!(&self.user_properties);
55
56        len
57    }
58
59    /// Encodes the properties into a byte buffer
60    fn encode(&self, buf: &mut BytesMut) {
61        if let Some(value) = self.subscription_id {
62            buf.put_u8(Property::SubscriptionIdentifier.into());
63            encode_variable_integer(buf, value).expect("");
64        }
65
66        property_encode!(&self.user_properties, Property::UserProp, buf);
67    }
68
69    /// Decodes properties from a byte buffer
70    fn decode(buf: &mut Bytes) -> Result<Option<Self>, Error>
71    where
72        Self: Sized,
73    {
74        if buf.is_empty() {
75            return Ok(None);
76        }
77
78        let mut subscription_id: Option<u32> = None;
79        let mut user_properties: Vec<(String, String)> = Vec::new();
80
81        while buf.has_remaining() {
82            let property: Property = decode_byte(buf)?.try_into()?;
83            match property {
84                Property::SubscriptionIdentifier => {
85                    if subscription_id.is_some() {
86                        return Err(Error::ProtocolError);
87                    }
88                    let value = decode_variable_integer(buf)?;
89                    buf.advance(len_bytes(value as usize));
90                    subscription_id = Some(value);
91                }
92                Property::UserProp => {
93                    property_decode!(&mut user_properties, buf);
94                }
95                _ => return Err(Error::PropertyMismatch),
96            }
97        }
98
99        Ok(Some(SubscribeProperties {
100            subscription_id,
101            user_properties,
102        }))
103    }
104}
105
106/// Controls how retained messages are handled for subscriptions
107#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd)]
108pub enum RetainHandling {
109    /// Send retained messages at the time of subscribe (default)
110    Send = 0,
111    /// Send retained messages only if subscription is new
112    SendForNewSub = 1,
113    /// Never send retained messages
114    DoNotSend = 2,
115}
116
117impl TryFrom<u8> for RetainHandling {
118    type Error = Error;
119
120    fn try_from(value: u8) -> Result<Self, Self::Error> {
121        match value {
122            0 => Ok(RetainHandling::Send),
123            1 => Ok(RetainHandling::SendForNewSub),
124            2 => Ok(RetainHandling::DoNotSend),
125            n => Err(Error::InvalidRetainHandling(n)),
126        }
127    }
128}
129
130impl From<RetainHandling> for u8 {
131    fn from(value: RetainHandling) -> Self {
132        value as u8
133    }
134}
135
136/// Represents a single topic filter with subscription options
137///
138/// # Example
139///
140/// ```rust
141/// use mqute_codec::protocol::v5::{TopicOptionFilter, RetainHandling};
142/// use mqute_codec::protocol::QoS;
143///
144/// let filter = TopicOptionFilter::new("topic1", QoS::AtLeastOnce, false, true, RetainHandling::DoNotSend);
145/// ```
146#[derive(Debug, Clone, PartialEq, Eq)]
147pub struct TopicOptionFilter {
148    /// The topic filter to subscribe to
149    pub topic: String,
150    /// Requested QoS level
151    pub qos: QoS,
152    /// If true, messages published by this client won't be received
153    pub no_local: bool,
154    /// If true, retain flag on published messages is kept as-is
155    pub retain_as_published: bool,
156    /// Controls how retained messages are handled
157    pub retain_handling: RetainHandling,
158}
159
160impl TopicOptionFilter {
161    /// Creates a new topic filter with options
162    ///
163    /// # Panics
164    ///
165    /// Panics if the iterator is empty, as at least one topic filter is required.
166    pub fn new<S: Into<String>>(
167        topic: S,
168        qos: QoS,
169        no_local: bool,
170        retain_as_published: bool,
171        retain_handling: RetainHandling,
172    ) -> Self {
173        let topic = topic.into();
174
175        if !util::is_valid_topic_filter(&topic) {
176            panic!("Invalid topic filter: '{}'", topic);
177        }
178
179        TopicOptionFilter {
180            topic,
181            qos,
182            no_local,
183            retain_as_published,
184            retain_handling,
185        }
186    }
187}
188
189/// Collection of topic filters for a subscription
190///
191/// # Example
192///
193/// ```rust
194/// use mqute_codec::protocol::v5::{Subscribe, TopicOptionFilters, TopicOptionFilter, RetainHandling};
195/// use mqute_codec::protocol::QoS;
196///
197/// let filters = vec![
198///     TopicOptionFilter::new("topic1", QoS::AtLeastOnce, false, true, RetainHandling::DoNotSend),
199///     TopicOptionFilter::new("topic2", QoS::ExactlyOnce, true, true, RetainHandling::SendForNewSub),
200/// ];
201/// let topic_filters = TopicOptionFilters::new(filters);
202/// assert_eq!(topic_filters.len(), 2);
203/// ```
204#[derive(Debug, Clone, PartialEq, Eq)]
205pub struct TopicOptionFilters(Vec<TopicOptionFilter>);
206
207#[allow(clippy::len_without_is_empty)]
208impl TopicOptionFilters {
209    /// Creates a new collection of topic filters
210    ///
211    /// # Panics
212    ///
213    /// Panics if:
214    /// - No filters are provided.
215    /// - The topic filters are invalid according to MQTT topic naming rules.
216    pub fn new<T: IntoIterator<Item = TopicOptionFilter>>(filters: T) -> Self {
217        let values: Vec<TopicOptionFilter> = filters.into_iter().collect();
218
219        if values.is_empty() {
220            panic!("At least one topic filter is required");
221        }
222
223        TopicOptionFilters(values)
224    }
225
226    /// Returns the number of topic filters in the collection.
227    pub fn len(&self) -> usize {
228        self.0.len()
229    }
230
231    /// Decodes topic filters from payload
232    pub(crate) fn decode(payload: &mut Bytes) -> Result<Self, Error> {
233        let mut filters = Vec::with_capacity(1);
234
235        while payload.has_remaining() {
236            let topic = decode_string(payload)?;
237
238            if !util::is_valid_topic_filter(&topic) {
239                return Err(Error::InvalidTopicFilter(topic));
240            }
241
242            let flags = decode_byte(payload)?;
243
244            // The upper 2 bits of the requested option byte must be zero
245            if flags & 0b1100_0000 > 0 {
246                return Err(Error::MalformedPacket);
247            }
248
249            let qos = (flags & 0x03).try_into()?;
250            let no_local = flags & 0x04 != 0;
251            let retain_as_published = flags & 0x08 != 0;
252            let retain_handling = ((flags >> 4) & 0x03).try_into()?;
253
254            filters.push(TopicOptionFilter::new(
255                topic,
256                qos,
257                no_local,
258                retain_as_published,
259                retain_handling,
260            ));
261        }
262
263        if filters.is_empty() {
264            return Err(Error::NoTopic);
265        }
266
267        Ok(TopicOptionFilters(filters))
268    }
269
270    /// Encodes topic filters into buffer
271    pub(crate) fn encode(&self, buf: &mut BytesMut) {
272        self.0.iter().for_each(|f| {
273            let qos: u8 = f.qos.into();
274            let retain_handling: u8 = f.retain_handling.into();
275
276            let options: u8 = retain_handling << 4
277                | (f.retain_as_published as u8) << 3
278                | (f.no_local as u8) << 2
279                | qos;
280
281            encode_string(buf, &f.topic);
282            buf.put_u8(options);
283        });
284    }
285
286    pub(crate) fn encoded_len(&self) -> usize {
287        self.0.iter().fold(0, |acc, f| acc + 2 + f.topic.len() + 1)
288    }
289}
290
291// Various trait implementations for TopicOptionFilters
292impl AsRef<Vec<TopicOptionFilter>> for TopicOptionFilters {
293    #[inline]
294    fn as_ref(&self) -> &Vec<TopicOptionFilter> {
295        &self.0
296    }
297}
298
299impl Borrow<Vec<TopicOptionFilter>> for TopicOptionFilters {
300    fn borrow(&self) -> &Vec<TopicOptionFilter> {
301        &self.0
302    }
303}
304
305impl IntoIterator for TopicOptionFilters {
306    type Item = TopicOptionFilter;
307    type IntoIter = std::vec::IntoIter<TopicOptionFilter>;
308
309    fn into_iter(self) -> Self::IntoIter {
310        self.0.into_iter()
311    }
312}
313
314impl FromIterator<TopicOptionFilter> for TopicOptionFilters {
315    fn from_iter<T: IntoIterator<Item = TopicOptionFilter>>(iter: T) -> Self {
316        TopicOptionFilters(Vec::from_iter(iter))
317    }
318}
319
320impl From<TopicOptionFilters> for Vec<TopicOptionFilter> {
321    #[inline]
322    fn from(value: TopicOptionFilters) -> Self {
323        value.0
324    }
325}
326
327impl From<Vec<TopicOptionFilter>> for TopicOptionFilters {
328    #[inline]
329    fn from(value: Vec<TopicOptionFilter>) -> Self {
330        TopicOptionFilters(value)
331    }
332}
333
334impl Index<usize> for TopicOptionFilters {
335    type Output = TopicOptionFilter;
336
337    fn index(&self, index: usize) -> &Self::Output {
338        self.0.index(index)
339    }
340}
341
342impl IndexMut<usize> for TopicOptionFilters {
343    fn index_mut(&mut self, index: usize) -> &mut Self::Output {
344        self.0.index_mut(index)
345    }
346}
347
348// Internal header structure for `Subscribe` packets
349id_header!(SubscribeHeader, SubscribeProperties);
350
351/// Represents an MQTT v5 `Subscribe` packet
352///
353/// Used to request subscription to one or more topics with various options:
354/// - QoS levels
355/// - Retain handling preferences
356/// - Local message filtering
357///
358/// # Example
359///
360/// ```rust
361/// use mqute_codec::protocol::v5::{Subscribe, TopicOptionFilter, RetainHandling};
362/// use mqute_codec::protocol::QoS;
363///
364/// let subscribe = Subscribe::new(
365///     1234,
366///     None,
367///     vec![
368///         TopicOptionFilter::new(
369///             "sensors/temperature",
370///             QoS::AtLeastOnce,
371///             false,
372///             true,
373///             RetainHandling::Send
374///         ),
375///         TopicOptionFilter::new(
376///             "control/#",
377///             QoS::ExactlyOnce,
378///             true,
379///             false,
380///             RetainHandling::SendForNewSub
381///         )
382///     ]
383/// );
384///
385/// let filters = subscribe.filters();
386/// assert_eq!(filters[0],
387///            TopicOptionFilter::new(
388///                             "sensors/temperature",
389///                             QoS::AtLeastOnce,
390///                             false,
391///                             true,
392///                             RetainHandling::Send
393///                         ));
394/// ```
395#[derive(Debug, Clone, PartialEq, Eq)]
396pub struct Subscribe {
397    header: SubscribeHeader,
398    filters: TopicOptionFilters,
399}
400
401impl Subscribe {
402    /// Creates a new `Subscribe` packet
403    pub fn new<T: IntoIterator<Item = TopicOptionFilter>>(
404        packet_id: u16,
405        properties: Option<SubscribeProperties>,
406        filters: T,
407    ) -> Self {
408        let header = SubscribeHeader::new(packet_id, properties);
409        let filters = TopicOptionFilters::new(filters);
410
411        Subscribe { header, filters }
412    }
413
414    /// Returns the packet identifier
415    pub fn packet_id(&self) -> u16 {
416        self.header.packet_id
417    }
418
419    /// Returns the subscription properties
420    pub fn properties(&self) -> Option<SubscribeProperties> {
421        self.header.properties.clone()
422    }
423
424    /// Returns the collection of topic filters
425    pub fn filters(&self) -> TopicOptionFilters {
426        self.filters.clone()
427    }
428}
429
430impl Encode for Subscribe {
431    /// Encodes the `Subscribe` packet into a byte buffer
432    fn encode(&self, buf: &mut BytesMut) -> Result<(), Error> {
433        let header = FixedHeader::with_flags(
434            PacketType::Subscribe,
435            Flags::new(QoS::AtLeastOnce),
436            self.payload_len(),
437        );
438        header.encode(buf)?;
439
440        self.header.encode(buf)?;
441        self.filters.encode(buf);
442
443        Ok(())
444    }
445
446    /// Calculates the total packet length
447    fn payload_len(&self) -> usize {
448        self.header.encoded_len() + self.filters.encoded_len()
449    }
450}
451
452impl Decode for Subscribe {
453    /// Decodes a `Subscribe` packet from raw bytes
454    fn decode(mut packet: RawPacket) -> Result<Self, Error> {
455        // Validate header flags
456        if packet.header.packet_type() != PacketType::Subscribe
457            || packet.header.flags() != Flags::new(QoS::AtLeastOnce)
458        {
459            return Err(Error::MalformedPacket);
460        }
461
462        let header = SubscribeHeader::decode(&mut packet.payload)?;
463        let filters = TopicOptionFilters::decode(&mut packet.payload)?;
464
465        Ok(Subscribe::new(header.packet_id, header.properties, filters))
466    }
467}
468
469impl traits::Subscribe for Subscribe {}
470
471#[cfg(test)]
472mod tests {
473    use super::*;
474    use crate::codec::PacketCodec;
475    use tokio_util::codec::Decoder;
476
477    #[test]
478    fn subscribe_properties_decode_advances_past_subscription_identifier() {
479        // Regression test: `decode_variable_integer` only inspects bytes, it
480        // doesn't consume them. Previously the SubscriptionIdentifier branch
481        // forgot to advance the buffer afterwards, so the following UserProp
482        // property would be misread as part of the identifier's own bytes.
483        let mut buf = BytesMut::new();
484
485        buf.put_u8(Property::SubscriptionIdentifier.into());
486        encode_variable_integer(&mut buf, 42).unwrap();
487
488        buf.put_u8(Property::UserProp.into());
489        encode_string(&mut buf, "client");
490        encode_string(&mut buf, "rust");
491
492        let mut buf = buf.freeze();
493        let properties = SubscribeProperties::decode(&mut buf).unwrap().unwrap();
494
495        assert_eq!(properties.subscription_id, Some(42));
496        assert_eq!(
497            properties.user_properties,
498            vec![("client".to_string(), "rust".to_string())]
499        );
500        assert!(buf.is_empty(), "buffer should be fully consumed");
501    }
502
503    #[test]
504    fn subscribe_properties_decode_rejects_duplicate_subscription_identifier() {
505        let mut buf = BytesMut::new();
506        buf.put_u8(Property::SubscriptionIdentifier.into());
507        encode_variable_integer(&mut buf, 1).unwrap();
508        buf.put_u8(Property::SubscriptionIdentifier.into());
509        encode_variable_integer(&mut buf, 2).unwrap();
510
511        let mut buf = buf.freeze();
512        let result = SubscribeProperties::decode(&mut buf);
513        assert!(matches!(result, Err(Error::ProtocolError)));
514    }
515
516    #[test]
517    fn subscribe_decode_full_packet_with_subscription_identifier() {
518        let mut codec = PacketCodec::new(None, None);
519
520        // Properties: Subscription Identifier = 7
521        let mut properties_buf = BytesMut::new();
522        properties_buf.put_u8(Property::SubscriptionIdentifier.into());
523        encode_variable_integer(&mut properties_buf, 7).unwrap();
524
525        // Variable header: packet id + properties length + properties
526        let mut payload = BytesMut::new();
527        payload.put_u16(0x1234);
528        encode_variable_integer(&mut payload, properties_buf.len() as u32).unwrap();
529        payload.extend_from_slice(&properties_buf);
530
531        // Payload: one topic filter "sensors/#" with default options
532        encode_string(&mut payload, "sensors/#");
533        payload.put_u8(0x00);
534
535        let mut stream = BytesMut::new();
536        stream.put_u8(((PacketType::Subscribe as u8) << 4) | 0x02);
537        encode_variable_integer(&mut stream, payload.len() as u32).unwrap();
538        stream.extend_from_slice(&payload);
539
540        let raw_packet = codec.decode(&mut stream).unwrap().unwrap();
541        let packet = Subscribe::decode(raw_packet).unwrap();
542
543        assert_eq!(packet.packet_id(), 0x1234);
544        assert_eq!(packet.properties().unwrap().subscription_id, Some(7));
545        assert_eq!(packet.filters().len(), 1);
546        assert_eq!(packet.filters()[0].topic, "sensors/#");
547    }
548}
mqute_codec/protocol/v5/subscribe.rs

mqute_codec/protocol/v5/
subscribe.rs
