rmqtt_codec/v5/packet/
subscribe.rs

1use std::num::{NonZeroU16, NonZeroU32};
2
3use bytes::{Buf, BufMut, Bytes, BytesMut};
4use bytestring::ByteString;
5use serde::{Deserialize, Serialize};
6
7use super::ack_props;
8use crate::error::{DecodeError, EncodeError};
9use crate::types::QoS;
10use crate::utils::{self, write_variable_length, Decode, Encode};
11use crate::v5::{encode::*, property_type as pt, UserProperties, UserProperty};
12
13/// Represents SUBSCRIBE packet
14#[derive(Debug, PartialEq, Eq, Clone)]
15pub struct Subscribe {
16    /// Packet Identifier
17    pub packet_id: NonZeroU16,
18    /// Subscription Identifier
19    pub id: Option<NonZeroU32>,
20    pub user_properties: UserProperties,
21    /// the list of Topic Filters and QoS to which the Client wants to subscribe.
22    pub topic_filters: Vec<(ByteString, SubscriptionOptions)>,
23}
24
25/// Subscription options for MQTT v5 SUBSCRIBE packets
26///
27/// Contains per-subscription options including QoS, no-local, retain-as-published,
28/// and retain handling as defined in the MQTT v5.0 specification.
29#[derive(Debug, PartialEq, Eq, Copy, Clone)]
30pub struct SubscriptionOptions {
31    pub qos: QoS,
32    pub no_local: bool,
33    pub retain_as_published: bool,
34    pub retain_handling: RetainHandling,
35}
36
37impl Default for SubscriptionOptions {
38    fn default() -> Self {
39        Self {
40            qos: QoS::AtMostOnce,
41            no_local: false,
42            retain_as_published: false,
43            retain_handling: RetainHandling::AtSubscribe,
44        }
45    }
46}
47
48prim_enum! {
49    /// Retain handling options for MQTT v5 subscriptions
50    ///
51    /// Controls how retained messages are sent when a subscription is established.
52    pub enum RetainHandling {
53        AtSubscribe = 0,
54        AtSubscribeNew = 1,
55        NoAtSubscribe = 2
56    }
57}
58
59impl From<RetainHandling> for u8 {
60    fn from(v: RetainHandling) -> Self {
61        match v {
62            RetainHandling::AtSubscribe => 0,
63            RetainHandling::AtSubscribeNew => 1,
64            RetainHandling::NoAtSubscribe => 2,
65        }
66    }
67}
68
69/// Represents SUBACK packet
70#[derive(Debug, PartialEq, Eq, Clone, Deserialize, Serialize)]
71pub struct SubscribeAck {
72    pub packet_id: NonZeroU16,
73    pub properties: UserProperties,
74    pub reason_string: Option<ByteString>,
75    /// corresponds to a Topic Filter in the SUBSCRIBE Packet being acknowledged.
76    pub status: Vec<SubscribeAckReason>,
77}
78
79/// Represents UNSUBSCRIBE packet
80#[derive(Debug, PartialEq, Eq, Clone)]
81pub struct Unsubscribe {
82    /// Packet Identifier
83    pub packet_id: NonZeroU16,
84    pub user_properties: UserProperties,
85    /// the list of Topic Filters that the Client wishes to unsubscribe from.
86    pub topic_filters: Vec<ByteString>,
87}
88
89/// Represents UNSUBACK packet
90#[derive(Debug, PartialEq, Eq, Clone, Deserialize, Serialize)]
91pub struct UnsubscribeAck {
92    /// Packet Identifier
93    pub packet_id: NonZeroU16,
94    pub properties: UserProperties,
95    pub reason_string: Option<ByteString>,
96    pub status: Vec<UnsubscribeAckReason>,
97}
98
99prim_enum! {
100    /// SUBACK reason codes
101    #[derive(Deserialize, Serialize)]
102    pub enum SubscribeAckReason {
103        GrantedQos0 = 0,
104        GrantedQos1 = 1,
105        GrantedQos2 = 2,
106        UnspecifiedError = 128,
107        ImplementationSpecificError = 131,
108        NotAuthorized = 135,
109        TopicFilterInvalid = 143,
110        PacketIdentifierInUse = 145,
111        QuotaExceeded = 151,
112        SharedSubscriptionNotSupported = 158,
113        SubscriptionIdentifiersNotSupported = 161,
114        WildcardSubscriptionsNotSupported = 162
115    }
116}
117
118impl From<SubscribeAckReason> for u8 {
119    fn from(v: SubscribeAckReason) -> Self {
120        match v {
121            SubscribeAckReason::GrantedQos0 => 0,
122            SubscribeAckReason::GrantedQos1 => 1,
123            SubscribeAckReason::GrantedQos2 => 2,
124            SubscribeAckReason::UnspecifiedError => 128,
125            SubscribeAckReason::ImplementationSpecificError => 131,
126            SubscribeAckReason::NotAuthorized => 135,
127            SubscribeAckReason::TopicFilterInvalid => 143,
128            SubscribeAckReason::PacketIdentifierInUse => 145,
129            SubscribeAckReason::QuotaExceeded => 151,
130            SubscribeAckReason::SharedSubscriptionNotSupported => 158,
131            SubscribeAckReason::SubscriptionIdentifiersNotSupported => 161,
132            SubscribeAckReason::WildcardSubscriptionsNotSupported => 162,
133        }
134    }
135}
136
137prim_enum! {
138    /// UNSUBACK reason codes
139    #[derive(Deserialize, Serialize)]
140    pub enum UnsubscribeAckReason {
141        Success = 0,
142        NoSubscriptionExisted = 17,
143        UnspecifiedError = 128,
144        ImplementationSpecificError = 131,
145        NotAuthorized = 135,
146        TopicFilterInvalid = 143,
147        PacketIdentifierInUse = 145
148    }
149}
150
151impl From<UnsubscribeAckReason> for u8 {
152    fn from(v: UnsubscribeAckReason) -> Self {
153        match v {
154            UnsubscribeAckReason::Success => 0,
155            UnsubscribeAckReason::NoSubscriptionExisted => 17,
156            UnsubscribeAckReason::UnspecifiedError => 128,
157            UnsubscribeAckReason::ImplementationSpecificError => 131,
158            UnsubscribeAckReason::NotAuthorized => 135,
159            UnsubscribeAckReason::TopicFilterInvalid => 143,
160            UnsubscribeAckReason::PacketIdentifierInUse => 145,
161        }
162    }
163}
164
165impl Subscribe {
166    pub(crate) fn decode(src: &mut Bytes) -> Result<Self, DecodeError> {
167        let packet_id = NonZeroU16::decode(src)?;
168        let prop_src = &mut utils::take_properties(src)?;
169        let mut sub_id = None;
170        let mut user_properties = Vec::new();
171        while prop_src.has_remaining() {
172            let prop_id = prop_src.get_u8();
173            match prop_id {
174                pt::SUB_ID => {
175                    ensure!(sub_id.is_none(), DecodeError::MalformedPacket); // can't appear twice
176                    let val = utils::decode_variable_length_cursor(prop_src)?;
177                    sub_id = Some(NonZeroU32::new(val).ok_or(DecodeError::MalformedPacket)?);
178                }
179                pt::USER => user_properties.push(UserProperty::decode(prop_src)?),
180                _ => return Err(DecodeError::MalformedPacket),
181            }
182        }
183
184        let mut topic_filters = Vec::new();
185        while src.has_remaining() {
186            let topic = ByteString::decode(src)?;
187            let opts = SubscriptionOptions::decode(src)?;
188            topic_filters.push((topic, opts));
189        }
190
191        Ok(Self { packet_id, id: sub_id, user_properties, topic_filters })
192    }
193}
194
195impl SubscribeAck {
196    pub(crate) fn decode(src: &mut Bytes) -> Result<Self, DecodeError> {
197        let packet_id = NonZeroU16::decode(src)?;
198        let (properties, reason_string) = ack_props::decode(src)?;
199        let mut status = Vec::with_capacity(src.remaining());
200        for code in src.as_ref().iter().copied() {
201            status.push(code.try_into()?);
202        }
203        Ok(Self { packet_id, properties, reason_string, status })
204    }
205}
206
207impl Unsubscribe {
208    pub(crate) fn decode(src: &mut Bytes) -> Result<Self, DecodeError> {
209        let packet_id = NonZeroU16::decode(src)?;
210
211        let prop_src = &mut utils::take_properties(src)?;
212        let mut user_properties = Vec::new();
213        while prop_src.has_remaining() {
214            let prop_id = prop_src.get_u8();
215            match prop_id {
216                pt::USER => user_properties.push(UserProperty::decode(prop_src)?),
217                _ => return Err(DecodeError::MalformedPacket),
218            }
219        }
220
221        let mut topic_filters = Vec::new();
222        while src.remaining() > 0 {
223            topic_filters.push(ByteString::decode(src)?);
224        }
225
226        Ok(Self { packet_id, user_properties, topic_filters })
227    }
228}
229
230impl UnsubscribeAck {
231    pub(crate) fn decode(src: &mut Bytes) -> Result<Self, DecodeError> {
232        let packet_id = NonZeroU16::decode(src)?;
233        let (properties, reason_string) = ack_props::decode(src)?;
234        let mut status = Vec::with_capacity(src.remaining());
235        for code in src.as_ref().iter().copied() {
236            status.push(code.try_into()?);
237        }
238        Ok(Self { packet_id, properties, reason_string, status })
239    }
240}
241
242impl EncodeLtd for Subscribe {
243    fn encoded_size(&self, _limit: u32) -> usize {
244        let prop_len = self.id.map_or(0, |v| 1 + var_int_len(v.get() as usize) as usize) // +1 to account for property type byte
245            + self.user_properties.encoded_size();
246        let payload_len =
247            self.topic_filters.iter().fold(0, |acc, (filter, _opts)| acc + filter.encoded_size() + 1);
248        self.packet_id.encoded_size() + var_int_len(prop_len) as usize + prop_len + payload_len
249    }
250
251    fn encode(&self, buf: &mut BytesMut, _: u32) -> Result<(), EncodeError> {
252        self.packet_id.encode(buf)?;
253
254        // encode properties
255        let prop_len = self.id.map_or(0, |v| 1 + var_int_len(v.get() as usize))
256            + self.user_properties.encoded_size() as u32; // safe: size was already checked against maximum
257        utils::write_variable_length(prop_len, buf);
258
259        if let Some(id) = self.id {
260            buf.put_u8(pt::SUB_ID);
261            write_variable_length(id.get(), buf);
262        }
263
264        self.user_properties.encode(buf)?;
265
266        // payload
267        for (filter, opts) in self.topic_filters.iter() {
268            filter.encode(buf)?;
269            opts.encode(buf)?;
270        }
271
272        Ok(())
273    }
274}
275
276impl Decode for SubscriptionOptions {
277    fn decode(src: &mut Bytes) -> Result<Self, DecodeError> {
278        ensure!(src.has_remaining(), DecodeError::InvalidLength);
279        let val = src.get_u8();
280        let qos = (val & 0b0000_0011).try_into()?;
281        let retain_handling = ((val & 0b0011_0000) >> 4).try_into()?;
282        Ok(SubscriptionOptions {
283            qos,
284            no_local: val & 0b0000_0100 != 0,
285            retain_as_published: val & 0b0000_1000 != 0,
286            retain_handling,
287        })
288    }
289}
290
291impl Encode for SubscriptionOptions {
292    fn encoded_size(&self) -> usize {
293        1
294    }
295    fn encode(&self, buf: &mut BytesMut) -> Result<(), EncodeError> {
296        buf.put_u8(
297            u8::from(self.qos)
298                | ((self.no_local as u8) << 2)
299                | ((self.retain_as_published as u8) << 3)
300                | (u8::from(self.retain_handling) << 4),
301        );
302        Ok(())
303    }
304}
305
306impl EncodeLtd for SubscribeAck {
307    fn encoded_size(&self, limit: u32) -> usize {
308        let len = self.status.len();
309        if len > (u32::MAX - 2) as usize {
310            return usize::MAX; // bail to avoid overflow
311        }
312
313        2 + ack_props::encoded_size(&self.properties, &self.reason_string, limit - 2 - len as u32) + len
314    }
315
316    fn encode(&self, buf: &mut BytesMut, size: u32) -> Result<(), EncodeError> {
317        self.packet_id.encode(buf)?;
318        let len = self.status.len() as u32; // safe: max size checked already
319        ack_props::encode(&self.properties, &self.reason_string, buf, size - 2 - len)?;
320        for &reason in self.status.iter() {
321            buf.put_u8(reason.into());
322        }
323        Ok(())
324    }
325}
326
327impl EncodeLtd for Unsubscribe {
328    fn encoded_size(&self, _limit: u32) -> usize {
329        let prop_len = self.user_properties.encoded_size();
330        2 + var_int_len(prop_len) as usize
331            + prop_len
332            + self.topic_filters.iter().fold(0, |acc, filter| acc + 2 + filter.len())
333    }
334
335    fn encode(&self, buf: &mut BytesMut, _size: u32) -> Result<(), EncodeError> {
336        self.packet_id.encode(buf)?;
337
338        // properties
339        let prop_len = self.user_properties.encoded_size();
340        utils::write_variable_length(prop_len as u32, buf); // safe: max size check is done already
341        self.user_properties.encode(buf)?;
342
343        // payload
344        for filter in self.topic_filters.iter() {
345            filter.encode(buf)?;
346        }
347        Ok(())
348    }
349}
350
351impl EncodeLtd for UnsubscribeAck {
352    // todo: almost identical to SUBACK
353    fn encoded_size(&self, limit: u32) -> usize {
354        let len = self.status.len();
355        2 + len + ack_props::encoded_size(&self.properties, &self.reason_string, reduce_limit(limit, 2 + len))
356    }
357
358    fn encode(&self, buf: &mut BytesMut, size: u32) -> Result<(), EncodeError> {
359        self.packet_id.encode(buf)?;
360        let len = self.status.len() as u32;
361
362        ack_props::encode(&self.properties, &self.reason_string, buf, size - 2 - len)?;
363        for &reason in self.status.iter() {
364            buf.put_u8(reason.into());
365        }
366        Ok(())
367    }
368}
369
370#[cfg(test)]
371mod tests {
372    use super::*;
373    use crate::v5::{Codec, Packet};
374    use tokio_util::codec::Decoder;
375    use tokio_util::codec::Encoder;
376
377    #[test]
378    fn test_sub() {
379        let pkt = Subscribe {
380            packet_id: 12.try_into().unwrap(),
381            id: Some(10.try_into().unwrap()),
382            user_properties: vec![("a".into(), "1".into())],
383            topic_filters: vec![("test".into(), SubscriptionOptions::default())],
384        };
385
386        let size = pkt.encoded_size(99999);
387        let mut buf = BytesMut::with_capacity(size);
388        pkt.encode(&mut buf, size as u32).unwrap();
389        assert_eq!(buf.len(), size);
390        assert_eq!(pkt, Subscribe::decode(&mut buf.freeze()).unwrap());
391
392        let pkt = Unsubscribe {
393            packet_id: 12.try_into().unwrap(),
394            user_properties: vec![("a".into(), "1".into())],
395            topic_filters: vec!["test".into()],
396        };
397
398        let size = pkt.encoded_size(99999);
399        let mut buf = BytesMut::with_capacity(size);
400        pkt.encode(&mut buf, size as u32).unwrap();
401        assert_eq!(buf.len(), size);
402        assert_eq!(pkt, Unsubscribe::decode(&mut buf.freeze()).unwrap());
403    }
404
405    #[test]
406    fn test_sub_pkt() {
407        let pkt = Packet::Subscribe(Subscribe {
408            packet_id: 12.try_into().unwrap(),
409            id: None,
410            user_properties: vec![("a".into(), "1".into())],
411            topic_filters: vec![("test".into(), SubscriptionOptions::default())],
412        });
413        let mut codec = Codec::default();
414
415        let mut buf = BytesMut::new();
416        codec.encode(pkt.clone(), &mut buf).unwrap();
417
418        assert_eq!(pkt, codec.decode(&mut buf).unwrap().unwrap().0);
419    }
420
421    #[test]
422    fn test_sub_ack() {
423        let ack = SubscribeAck {
424            packet_id: NonZeroU16::new(1).unwrap(),
425            properties: Vec::new(),
426            reason_string: Some("some reason".into()),
427            status: Vec::new(),
428        };
429
430        let size = ack.encoded_size(99999);
431        let mut buf = BytesMut::with_capacity(size);
432        ack.encode(&mut buf, size as u32).unwrap();
433        assert_eq!(ack, SubscribeAck::decode(&mut buf.freeze()).unwrap());
434
435        let ack = SubscribeAck {
436            packet_id: NonZeroU16::new(1).unwrap(),
437            properties: vec![("prop1".into(), "val1".into()), ("prop2".into(), "val2".into())],
438            reason_string: None,
439            status: vec![SubscribeAckReason::GrantedQos0],
440        };
441        let size = ack.encoded_size(99999);
442        let mut buf = BytesMut::with_capacity(size);
443        ack.encode(&mut buf, size as u32).unwrap();
444        assert_eq!(ack, SubscribeAck::decode(&mut buf.freeze()).unwrap());
445
446        let ack = UnsubscribeAck {
447            packet_id: NonZeroU16::new(1).unwrap(),
448            properties: Vec::new(),
449            reason_string: Some("some reason".into()),
450            status: Vec::new(),
451        };
452        let mut buf = BytesMut::new();
453        let size = ack.encoded_size(99999);
454        ack.encode(&mut buf, size as u32).unwrap();
455        assert_eq!(ack, UnsubscribeAck::decode(&mut buf.freeze()).unwrap());
456
457        let ack = UnsubscribeAck {
458            packet_id: NonZeroU16::new(1).unwrap(),
459            properties: vec![("prop1".into(), "val1".into()), ("prop2".into(), "val2".into())],
460            reason_string: None,
461            status: vec![UnsubscribeAckReason::Success],
462        };
463        let size = ack.encoded_size(99999);
464        let mut buf = BytesMut::with_capacity(size);
465        ack.encode(&mut buf, size as u32).unwrap();
466        assert_eq!(ack, UnsubscribeAck::decode(&mut buf.freeze()).unwrap());
467    }
468}
rmqtt_codec/v5/packet/subscribe.rs

rmqtt_codec/v5/packet/
subscribe.rs
