mqtt5_protocol/packet/
suback.rs

1use crate::error::{MqttError, Result};
2use crate::packet::{FixedHeader, MqttPacket, PacketType};
3use crate::protocol::v5::properties::Properties;
4use crate::types::ProtocolVersion;
5use crate::QoS;
6use bytes::{Buf, BufMut};
7
8/// MQTT SUBACK packet
9#[derive(Debug, Clone)]
10pub struct SubAckPacket {
11    /// Packet identifier
12    pub packet_id: u16,
13    /// Reason codes for each subscription
14    pub reason_codes: Vec<SubAckReasonCode>,
15    /// SUBACK properties (v5.0 only)
16    pub properties: Properties,
17    /// Protocol version (4 = v3.1.1, 5 = v5.0)
18    pub protocol_version: u8,
19}
20
21/// SUBACK reason codes
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23#[repr(u8)]
24pub enum SubAckReasonCode {
25    /// Maximum `QoS` 0
26    GrantedQoS0 = 0x00,
27    /// Maximum `QoS` 1
28    GrantedQoS1 = 0x01,
29    /// Maximum `QoS` 2
30    GrantedQoS2 = 0x02,
31    /// Unspecified error
32    UnspecifiedError = 0x80,
33    /// Implementation specific error
34    ImplementationSpecificError = 0x83,
35    /// Not authorized
36    NotAuthorized = 0x87,
37    /// Topic filter invalid
38    TopicFilterInvalid = 0x8F,
39    /// Packet identifier in use
40    PacketIdentifierInUse = 0x91,
41    /// Quota exceeded
42    QuotaExceeded = 0x97,
43    /// Shared subscriptions not supported
44    SharedSubscriptionsNotSupported = 0x9E,
45    /// Subscription identifiers not supported
46    SubscriptionIdentifiersNotSupported = 0xA1,
47    /// Wildcard subscriptions not supported
48    WildcardSubscriptionsNotSupported = 0xA2,
49}
50
51impl SubAckReasonCode {
52    /// Creates a reason code from a granted `QoS` level
53    #[must_use]
54    pub fn from_qos(qos: QoS) -> Self {
55        match qos {
56            QoS::AtMostOnce => Self::GrantedQoS0,
57            QoS::AtLeastOnce => Self::GrantedQoS1,
58            QoS::ExactlyOnce => Self::GrantedQoS2,
59        }
60    }
61
62    /// Converts a u8 to a `SubAckReasonCode`
63    #[must_use]
64    pub fn from_u8(value: u8) -> Option<Self> {
65        match value {
66            0x00 => Some(Self::GrantedQoS0),
67            0x01 => Some(Self::GrantedQoS1),
68            0x02 => Some(Self::GrantedQoS2),
69            0x80 => Some(Self::UnspecifiedError),
70            0x83 => Some(Self::ImplementationSpecificError),
71            0x87 => Some(Self::NotAuthorized),
72            0x8F => Some(Self::TopicFilterInvalid),
73            0x91 => Some(Self::PacketIdentifierInUse),
74            0x97 => Some(Self::QuotaExceeded),
75            0x9E => Some(Self::SharedSubscriptionsNotSupported),
76            0xA1 => Some(Self::SubscriptionIdentifiersNotSupported),
77            0xA2 => Some(Self::WildcardSubscriptionsNotSupported),
78            _ => None,
79        }
80    }
81
82    /// Returns true if this is a success code
83    #[must_use]
84    pub fn is_success(&self) -> bool {
85        matches!(
86            self,
87            Self::GrantedQoS0 | Self::GrantedQoS1 | Self::GrantedQoS2
88        )
89    }
90
91    /// Returns the granted `QoS` level if this is a success code
92    #[must_use]
93    pub fn granted_qos(&self) -> Option<QoS> {
94        match self {
95            Self::GrantedQoS0 => Some(QoS::AtMostOnce),
96            Self::GrantedQoS1 => Some(QoS::AtLeastOnce),
97            Self::GrantedQoS2 => Some(QoS::ExactlyOnce),
98            _ => None,
99        }
100    }
101}
102
103impl SubAckPacket {
104    /// Creates a new SUBACK packet (v5.0)
105    #[must_use]
106    pub fn new(packet_id: u16) -> Self {
107        Self {
108            packet_id,
109            reason_codes: Vec::new(),
110            properties: Properties::default(),
111            protocol_version: 5,
112        }
113    }
114
115    /// Creates a new SUBACK packet for v3.1.1
116    #[must_use]
117    pub fn new_v311(packet_id: u16) -> Self {
118        Self {
119            packet_id,
120            reason_codes: Vec::new(),
121            properties: Properties::default(),
122            protocol_version: 4,
123        }
124    }
125
126    fn reason_code_to_v311(code: SubAckReasonCode) -> u8 {
127        match code {
128            SubAckReasonCode::GrantedQoS0 => 0x00,
129            SubAckReasonCode::GrantedQoS1 => 0x01,
130            SubAckReasonCode::GrantedQoS2 => 0x02,
131            _ => 0x80,
132        }
133    }
134
135    /// Adds a reason code
136    #[must_use]
137    pub fn add_reason_code(mut self, code: SubAckReasonCode) -> Self {
138        self.reason_codes.push(code);
139        self
140    }
141
142    /// Adds a reason code for a granted `QoS`
143    #[must_use]
144    pub fn add_granted_qos(mut self, qos: QoS) -> Self {
145        self.reason_codes.push(SubAckReasonCode::from_qos(qos));
146        self
147    }
148
149    /// Sets the reason string
150    #[must_use]
151    pub fn with_reason_string(mut self, reason: String) -> Self {
152        self.properties.set_reason_string(reason);
153        self
154    }
155
156    /// Adds a user property
157    #[must_use]
158    pub fn with_user_property(mut self, key: String, value: String) -> Self {
159        self.properties.add_user_property(key, value);
160        self
161    }
162}
163
164impl MqttPacket for SubAckPacket {
165    fn packet_type(&self) -> PacketType {
166        PacketType::SubAck
167    }
168
169    fn encode_body<B: BufMut>(&self, buf: &mut B) -> Result<()> {
170        buf.put_u16(self.packet_id);
171
172        if self.protocol_version == 5 {
173            self.properties.encode(buf)?;
174        }
175
176        if self.reason_codes.is_empty() {
177            return Err(MqttError::MalformedPacket(
178                "SUBACK packet must contain at least one reason code".to_string(),
179            ));
180        }
181
182        if self.protocol_version == 5 {
183            for code in &self.reason_codes {
184                buf.put_u8(*code as u8);
185            }
186        } else {
187            for code in &self.reason_codes {
188                buf.put_u8(Self::reason_code_to_v311(*code));
189            }
190        }
191
192        Ok(())
193    }
194
195    fn decode_body<B: Buf>(buf: &mut B, fixed_header: &FixedHeader) -> Result<Self> {
196        Self::decode_body_with_version(buf, fixed_header, 5)
197    }
198}
199
200impl SubAckPacket {
201    /// Decodes the packet body with a specific protocol version
202    ///
203    /// # Errors
204    ///
205    /// Returns an error if decoding fails
206    pub fn decode_body_with_version<B: Buf>(
207        buf: &mut B,
208        _fixed_header: &FixedHeader,
209        protocol_version: u8,
210    ) -> Result<Self> {
211        ProtocolVersion::try_from(protocol_version)
212            .map_err(|()| MqttError::UnsupportedProtocolVersion)?;
213
214        if buf.remaining() < 2 {
215            return Err(MqttError::MalformedPacket(
216                "SUBACK missing packet identifier".to_string(),
217            ));
218        }
219        let packet_id = buf.get_u16();
220
221        let properties = if protocol_version == 5 {
222            Properties::decode(buf)?
223        } else {
224            Properties::default()
225        };
226
227        let mut reason_codes = Vec::new();
228
229        if !buf.has_remaining() {
230            return Err(MqttError::MalformedPacket(
231                "SUBACK packet must contain at least one reason code".to_string(),
232            ));
233        }
234
235        while buf.has_remaining() {
236            let code_byte = buf.get_u8();
237            let code = SubAckReasonCode::from_u8(code_byte).ok_or_else(|| {
238                MqttError::MalformedPacket(format!("Invalid SUBACK reason code: 0x{code_byte:02X}"))
239            })?;
240            reason_codes.push(code);
241        }
242
243        Ok(Self {
244            packet_id,
245            reason_codes,
246            properties,
247            protocol_version,
248        })
249    }
250}
251
252#[cfg(test)]
253mod tests {
254    use super::*;
255    use crate::protocol::v5::properties::PropertyId;
256    use bytes::BytesMut;
257
258    #[test]
259    fn test_suback_reason_code_from_qos() {
260        assert_eq!(
261            SubAckReasonCode::from_qos(QoS::AtMostOnce),
262            SubAckReasonCode::GrantedQoS0
263        );
264        assert_eq!(
265            SubAckReasonCode::from_qos(QoS::AtLeastOnce),
266            SubAckReasonCode::GrantedQoS1
267        );
268        assert_eq!(
269            SubAckReasonCode::from_qos(QoS::ExactlyOnce),
270            SubAckReasonCode::GrantedQoS2
271        );
272    }
273
274    #[test]
275    fn test_suback_reason_code_is_success() {
276        assert!(SubAckReasonCode::GrantedQoS0.is_success());
277        assert!(SubAckReasonCode::GrantedQoS1.is_success());
278        assert!(SubAckReasonCode::GrantedQoS2.is_success());
279        assert!(!SubAckReasonCode::NotAuthorized.is_success());
280        assert!(!SubAckReasonCode::TopicFilterInvalid.is_success());
281    }
282
283    #[test]
284    fn test_suback_basic() {
285        let packet = SubAckPacket::new(123)
286            .add_granted_qos(QoS::AtLeastOnce)
287            .add_granted_qos(QoS::ExactlyOnce)
288            .add_reason_code(SubAckReasonCode::NotAuthorized);
289
290        assert_eq!(packet.packet_id, 123);
291        assert_eq!(packet.reason_codes.len(), 3);
292        assert_eq!(packet.reason_codes[0], SubAckReasonCode::GrantedQoS1);
293        assert_eq!(packet.reason_codes[1], SubAckReasonCode::GrantedQoS2);
294        assert_eq!(packet.reason_codes[2], SubAckReasonCode::NotAuthorized);
295    }
296
297    #[test]
298    fn test_suback_encode_decode() {
299        let packet = SubAckPacket::new(789)
300            .add_granted_qos(QoS::AtMostOnce)
301            .add_granted_qos(QoS::AtLeastOnce)
302            .add_reason_code(SubAckReasonCode::TopicFilterInvalid)
303            .with_reason_string("Invalid wildcard usage".to_string());
304
305        let mut buf = BytesMut::new();
306        packet.encode(&mut buf).unwrap();
307
308        let fixed_header = FixedHeader::decode(&mut buf).unwrap();
309        assert_eq!(fixed_header.packet_type, PacketType::SubAck);
310
311        let decoded = SubAckPacket::decode_body(&mut buf, &fixed_header).unwrap();
312        assert_eq!(decoded.packet_id, 789);
313        assert_eq!(decoded.reason_codes.len(), 3);
314        assert_eq!(decoded.reason_codes[0], SubAckReasonCode::GrantedQoS0);
315        assert_eq!(decoded.reason_codes[1], SubAckReasonCode::GrantedQoS1);
316        assert_eq!(
317            decoded.reason_codes[2],
318            SubAckReasonCode::TopicFilterInvalid
319        );
320
321        let reason_str = decoded.properties.get(PropertyId::ReasonString);
322        assert!(reason_str.is_some());
323    }
324
325    #[test]
326    fn test_suback_empty_reason_codes() {
327        let packet = SubAckPacket::new(123);
328
329        let mut buf = BytesMut::new();
330        let result = packet.encode(&mut buf);
331        assert!(result.is_err());
332    }
333}
mqtt5_protocol/packet/suback.rs

mqtt5_protocol/packet/
suback.rs
