mqute_codec/protocol/v5/
connack.rs

1//! # ConnAck Packet V5
2//!
3//! This module defines the `ConnAck` packet, which is sent by the server in response to a
4//! `Connect` packet from a client in the MQTT v5 protocol. The `ConnAck` packet indicates
5//! the result of the connection attempt and includes session status and optional properties
6//! for session configuration and server capabilities.
7
8use std::time::Duration;
9
10use super::property::{
11    Property, PropertyFrame, property_decode, property_decode_non_zero, property_encode,
12    property_len,
13};
14use crate::Error;
15use crate::codec::util::{decode_byte, decode_variable_integer, encode_variable_integer};
16use crate::codec::{Decode, Encode, RawPacket};
17use crate::protocol::util::len_bytes;
18use crate::protocol::v5::reason::ReasonCode;
19use crate::protocol::{FixedHeader, PacketType, QoS};
20use bit_field::BitField;
21use bytes::{Buf, BufMut, Bytes, BytesMut};
22
23/// Represents the properties of the `ConnAck` packet.
24///
25/// These properties provide additional connection-related information from the server
26/// to the client, including session configuration and server capabilities.
27///
28/// # Example
29///
30/// ```rust
31///
32/// use mqute_codec::protocol::v5::ConnAckProperties;
33/// use std::time::Duration;
34///
35/// let connack_properties = ConnAckProperties {
36///     session_expiry_interval: Some(Duration::from_secs(3600)),
37///     retain_available: Some(true),
38///     ..Default::default()
39/// };
40/// ```
41#[derive(Debug, Default, Clone, PartialEq, Eq)]
42pub struct ConnAckProperties {
43    /// Duration in seconds the session will be kept after disconnection
44    pub session_expiry_interval: Option<Duration>,
45    /// Maximum number of QoS 1 and 2 messages the server will process concurrently
46    pub receive_maximum: Option<u16>,
47    /// Maximum QoS level the server supports
48    pub maximum_qos: Option<QoS>,
49    /// Whether the server supports retained messages
50    pub retain_available: Option<bool>,
51    /// Maximum packet size the server will accept
52    pub maximum_packet_size: Option<u32>,
53    /// Client identifier assigned by the server
54    pub assigned_client_id: Option<String>,
55    /// Maximum number of topic aliases the server will accept
56    pub topic_alias_maximum: Option<u16>,
57    /// Human-readable reason string for the connection result
58    pub reason: Option<String>,
59    /// User-defined properties for extensibility
60    pub user_properties: Vec<(String, String)>,
61    /// Whether wildcard subscriptions are supported
62    pub wildcard_subscription_available: Option<bool>,
63    /// Whether subscription identifiers are supported
64    pub subscription_id_available: Option<bool>,
65    /// Whether shared subscriptions are supported
66    pub shared_subscription_available: Option<bool>,
67    /// Keep alive time suggested by the server
68    pub server_keep_alive: Option<u16>,
69    /// Response information for authentication
70    pub response_info: Option<String>,
71    /// Server reference for redirection
72    pub server_reference: Option<String>,
73    /// Authentication method
74    pub auth_method: Option<String>,
75    /// Authentication data
76    pub auth_data: Option<Bytes>,
77}
78
79impl PropertyFrame for ConnAckProperties {
80    /// Returns the encoded length of the `ConnAckProperties`.
81    fn encoded_len(&self) -> usize {
82        let mut len = 0;
83
84        len += property_len!(&self.session_expiry_interval);
85        len += property_len!(&self.receive_maximum);
86        len += property_len!(&self.maximum_qos);
87        len += property_len!(&self.retain_available);
88        len += property_len!(&self.maximum_packet_size);
89        len += property_len!(&self.assigned_client_id);
90        len += property_len!(&self.topic_alias_maximum);
91        len += property_len!(&self.reason);
92        len += property_len!(&self.user_properties);
93        len += property_len!(&self.wildcard_subscription_available);
94        len += property_len!(&self.subscription_id_available);
95        len += property_len!(&self.shared_subscription_available);
96        len += property_len!(&self.server_keep_alive);
97        len += property_len!(&self.response_info);
98        len += property_len!(&self.server_reference);
99        len += property_len!(&self.auth_method);
100        len += property_len!(&self.auth_data);
101
102        len
103    }
104
105    /// Encodes the `ConnAckProperties` into a byte buffer.
106    fn encode(&self, buf: &mut BytesMut) {
107        property_encode!(
108            &self.session_expiry_interval,
109            Property::SessionExpiryInterval,
110            buf
111        );
112        property_encode!(&self.receive_maximum, Property::ReceiveMaximum, buf);
113        property_encode!(&self.maximum_qos, Property::MaximumQoS, buf);
114        property_encode!(&self.retain_available, Property::RetainAvailable, buf);
115        property_encode!(&self.maximum_packet_size, Property::MaximumPacketSize, buf);
116        property_encode!(
117            &self.assigned_client_id,
118            Property::AssignedClientIdentifier,
119            buf
120        );
121        property_encode!(&self.topic_alias_maximum, Property::TopicAliasMaximum, buf);
122        property_encode!(&self.reason, Property::ReasonString, buf);
123        property_encode!(&self.user_properties, Property::UserProp, buf);
124        property_encode!(
125            &self.wildcard_subscription_available,
126            Property::WildcardSubscriptionAvailable,
127            buf
128        );
129        property_encode!(
130            &self.subscription_id_available,
131            Property::SubscriptionIdentifierAvailable,
132            buf
133        );
134        property_encode!(
135            &self.shared_subscription_available,
136            Property::SharedSubscriptionAvailable,
137            buf
138        );
139        property_encode!(&self.server_keep_alive, Property::ServerKeepAlive, buf);
140        property_encode!(&self.response_info, Property::ResponseInformation, buf);
141        property_encode!(&self.server_reference, Property::ServerReference, buf);
142        property_encode!(&self.auth_method, Property::AuthenticationMethod, buf);
143        property_encode!(&self.auth_data, Property::AuthenticationData, buf);
144    }
145
146    /// Decodes the `ConnAckProperties` from a byte buffer.
147    fn decode(buf: &mut Bytes) -> Result<Option<Self>, Error> {
148        if buf.is_empty() {
149            return Ok(None);
150        }
151
152        let mut properties = ConnAckProperties::default();
153
154        while buf.has_remaining() {
155            let property: Property = decode_byte(buf)?.try_into()?;
156            match property {
157                Property::SessionExpiryInterval => {
158                    property_decode!(&mut properties.session_expiry_interval, buf);
159                }
160                Property::ReceiveMaximum => {
161                    property_decode_non_zero!(&mut properties.receive_maximum, buf);
162                }
163                Property::MaximumQoS => {
164                    property_decode!(&mut properties.maximum_qos, buf);
165                }
166                Property::RetainAvailable => {
167                    property_decode!(&mut properties.retain_available, buf);
168                }
169                Property::MaximumPacketSize => {
170                    property_decode_non_zero!(&mut properties.maximum_packet_size, buf);
171                }
172                Property::AssignedClientIdentifier => {
173                    property_decode!(&mut properties.assigned_client_id, buf);
174                }
175                Property::TopicAliasMaximum => {
176                    property_decode!(&mut properties.topic_alias_maximum, buf);
177                }
178                Property::ReasonString => {
179                    property_decode!(&mut properties.reason, buf);
180                }
181                Property::UserProp => {
182                    property_decode!(&mut properties.user_properties, buf);
183                }
184                Property::WildcardSubscriptionAvailable => {
185                    property_decode!(&mut properties.wildcard_subscription_available, buf);
186                }
187                Property::SubscriptionIdentifierAvailable => {
188                    property_decode!(&mut properties.subscription_id_available, buf);
189                }
190                Property::SharedSubscriptionAvailable => {
191                    property_decode!(&mut properties.shared_subscription_available, buf);
192                }
193                Property::ServerKeepAlive => {
194                    property_decode!(&mut properties.server_keep_alive, buf);
195                }
196                Property::ResponseInformation => {
197                    property_decode!(&mut properties.response_info, buf);
198                }
199                Property::ServerReference => {
200                    property_decode!(&mut properties.server_reference, buf);
201                }
202                Property::AuthenticationMethod => {
203                    property_decode!(&mut properties.auth_method, buf);
204                }
205                Property::AuthenticationData => {
206                    property_decode!(&mut properties.auth_data, buf);
207                }
208                _ => return Err(Error::PropertyMismatch),
209            }
210        }
211
212        Ok(Some(properties))
213    }
214}
215
216/// Validates that the reason code is appropriate for a ConnAck packet
217fn validate_connack_reason_code(code: ReasonCode) -> bool {
218    matches!(code.into(), 0 | 128..=138 | 140 | 144 | 149 | 151 | 153..=157 | 159)
219}
220
221#[derive(Debug, Clone, PartialEq, Eq)]
222struct ConnAckHeader {
223    code: ReasonCode,
224    session_present: bool,
225    properties: Option<ConnAckProperties>,
226}
227
228impl ConnAckHeader {
229    fn new(code: ReasonCode, session_present: bool, properties: Option<ConnAckProperties>) -> Self {
230        if !validate_connack_reason_code(code) {
231            panic!("Invalid reason code {code}");
232        }
233        ConnAckHeader {
234            code,
235            session_present,
236            properties,
237        }
238    }
239
240    fn encoded_len(&self) -> usize {
241        let properties_len = self
242            .properties
243            .as_ref()
244            .map(|properties| properties.encoded_len())
245            .unwrap_or(0);
246        1 + 1 + len_bytes(properties_len) + properties_len
247    }
248
249    fn encode(&self, buf: &mut BytesMut) -> Result<(), Error> {
250        let mut flags = 0u8;
251        flags.set_bit(0, self.session_present);
252
253        // Write a session present flag
254        buf.put_u8(flags);
255
256        // Write a connect return code
257        buf.put_u8(self.code.into());
258
259        let properties_len = self
260            .properties
261            .as_ref()
262            .map(|properties| properties.encoded_len())
263            .unwrap_or(0) as u32;
264
265        encode_variable_integer(buf, properties_len)?;
266
267        if let Some(properties) = self.properties.as_ref() {
268            properties.encode(buf);
269        }
270
271        Ok(())
272    }
273
274    fn decode(payload: &mut Bytes) -> Result<Self, Error> {
275        let conn_ack_flag = decode_byte(payload)?;
276        let code = decode_byte(payload)?.try_into()?;
277
278        if !validate_connack_reason_code(code) {
279            return Err(Error::InvalidReasonCode(code.into()));
280        }
281
282        let session_present = conn_ack_flag.get_bit(0);
283
284        let properties_len = decode_variable_integer(payload)? as usize;
285        if payload.len() < properties_len + len_bytes(properties_len) {
286            return Err(Error::MalformedPacket);
287        }
288
289        // Skip properties len
290        payload.advance(len_bytes(properties_len));
291
292        let mut frame = payload.split_to(properties_len);
293        let properties = ConnAckProperties::decode(&mut frame)?;
294
295        Ok(ConnAckHeader {
296            code,
297            session_present,
298            properties,
299        })
300    }
301}
302
303/// Represents an MQTT `CONNACK` packet.
304///
305/// The `ConnAck` packet is sent by the server in response to a `Connect` packet
306/// from a client. It indicates whether the connection was accepted and provides
307/// session status and optional properties.
308///
309/// # Example
310///
311/// ```rust
312/// use mqute_codec::protocol::v5::{ConnAck, ConnAckProperties, ReasonCode};
313/// use std::time::Duration;
314///
315/// let properties = ConnAckProperties {
316///     session_expiry_interval: Some(Duration::from_secs(3600)),
317///     receive_maximum: Some(10),
318///     ..Default::default()
319/// };
320/// let connack = ConnAck::new(
321///     ReasonCode::Success,
322///     true,
323///     Some(properties)
324/// );
325///
326/// assert_eq!(connack.code(), ReasonCode::Success);
327/// assert!(connack.session_present());
328/// ```
329#[derive(Debug, Clone, PartialEq, Eq)]
330pub struct ConnAck {
331    header: ConnAckHeader,
332}
333
334impl ConnAck {
335    /// Creates a new `ConnAck` packet.
336    pub fn new(
337        code: ReasonCode,
338        session_present: bool,
339        properties: Option<ConnAckProperties>,
340    ) -> Self {
341        ConnAck {
342            header: ConnAckHeader::new(code, session_present, properties),
343        }
344    }
345
346    /// Returns the reason code indicating the connection result.
347    pub fn code(&self) -> ReasonCode {
348        self.header.code
349    }
350
351    /// Returns whether the server has a previous session for this client.
352    pub fn session_present(&self) -> bool {
353        self.header.session_present
354    }
355
356    /// Returns the optional properties of the `ConnAck` packet.
357    pub fn properties(&self) -> Option<ConnAckProperties> {
358        self.header.properties.clone()
359    }
360}
361
362impl Encode for ConnAck {
363    /// Encodes the `ConnAck` packet into a byte buffer.
364    fn encode(&self, buf: &mut BytesMut) -> Result<(), Error> {
365        let header = FixedHeader::new(PacketType::ConnAck, self.payload_len());
366        header.encode(buf)?;
367        self.header.encode(buf)
368    }
369
370    /// Returns the length of the `ConnAck` packet payload.
371    fn payload_len(&self) -> usize {
372        self.header.encoded_len()
373    }
374}
375
376impl Decode for ConnAck {
377    /// Decodes a `ConnAck` packet from a raw MQTT packet.
378    fn decode(mut packet: RawPacket) -> Result<Self, Error> {
379        if packet.header.packet_type() != PacketType::ConnAck || !packet.header.flags().is_default()
380        {
381            return Err(Error::MalformedPacket);
382        }
383
384        let header = ConnAckHeader::decode(&mut packet.payload)?;
385        Ok(ConnAck { header })
386    }
387}
mqute_codec/protocol/v5/connack.rs

mqute_codec/protocol/v5/
connack.rs
