mqute_codec/protocol/v5/
connect.rs

1//! # Connect Packet V5
2//!
3//! This module provides the complete implementation of the MQTT v5 Connect packet,
4//! including its properties, will message handling, and authentication support.
5//! The Connect packet is the first packet sent by a client to initiate a connection
6//! with an MQTT broker.
7
8use super::property::{property_decode, property_decode_non_zero, property_encode};
9use super::property::{property_len, Property, PropertyFrame};
10use crate::codec::util::{
11    decode_byte, decode_bytes, decode_string, decode_variable_integer, encode_bytes, encode_string,
12    encode_variable_integer,
13};
14use crate::protocol::common::{connect, ConnectHeader};
15use crate::protocol::common::{ConnectFrame, WillFrame};
16use crate::protocol::util::len_bytes;
17use crate::protocol::{util, Credentials, Protocol, QoS};
18use crate::Error;
19use bit_field::BitField;
20use bytes::{Buf, Bytes, BytesMut};
21use std::ops::RangeInclusive;
22use std::time::Duration;
23
24/// Bit flag positions for Connect packet flags
25const WILL_FLAG: usize = 2;
26const WILL_QOS: RangeInclusive<usize> = 3..=4;
27const WILL_RETAIN: usize = 5;
28
29/// Represents the properties of a Connect packet in MQTT v5.
30///
31/// These properties provide extended functionality beyond the basic connection
32/// parameters, including session management, flow control, and authentication.
33///
34/// # Example
35///
36/// ```rust
37/// use mqute_codec::protocol::v5::ConnectProperties;
38/// use std::time::Duration;
39///
40/// let connect_properties = ConnectProperties {
41///     session_expiry_interval: Some(Duration::from_secs(3600)),
42///     maximum_packet_size: Some(4096u32),
43///     ..Default::default()
44/// };
45/// ```
46#[derive(Debug, Default, Clone, PartialEq, Eq)]
47pub struct ConnectProperties {
48    /// Duration in seconds after which the session expires
49    pub session_expiry_interval: Option<Duration>,
50    /// Maximum number of QoS 1 and 2 publishes the client will process
51    pub receive_maximum: Option<u16>,
52    /// Maximum packet size the client will accept
53    pub maximum_packet_size: Option<u32>,
54    /// Highest value the client will accept as a topic alias
55    pub topic_alias_maximum: Option<u16>,
56    /// Whether the server should include response information
57    pub request_response_info: Option<bool>,
58    /// Whether the server should include reason strings
59    pub request_problem_info: Option<bool>,
60    /// User-defined key-value properties
61    pub user_properties: Vec<(String, String)>,
62    /// Authentication method name
63    pub auth_method: Option<String>,
64    /// Authentication data
65    pub auth_data: Option<Bytes>,
66}
67
68impl PropertyFrame for ConnectProperties {
69    /// Calculates the encoded length of the properties
70    fn encoded_len(&self) -> usize {
71        let mut len = 0;
72
73        len += property_len!(&self.session_expiry_interval);
74        len += property_len!(&self.receive_maximum);
75        len += property_len!(&self.maximum_packet_size);
76        len += property_len!(&self.topic_alias_maximum);
77        len += property_len!(&self.request_response_info);
78        len += property_len!(&self.request_problem_info);
79        len += property_len!(&self.user_properties);
80        len += property_len!(&self.auth_method);
81        len += property_len!(&self.auth_data);
82
83        len
84    }
85
86    /// Encodes the properties into a byte buffer
87    fn encode(&self, buf: &mut BytesMut) {
88        property_encode!(
89            &self.session_expiry_interval,
90            Property::SessionExpiryInterval,
91            buf
92        );
93        property_encode!(&self.receive_maximum, Property::ReceiveMaximum, buf);
94        property_encode!(&self.maximum_packet_size, Property::MaximumPacketSize, buf);
95        property_encode!(&self.topic_alias_maximum, Property::TopicAliasMaximum, buf);
96        property_encode!(
97            &self.request_response_info,
98            Property::RequestResponseInformation,
99            buf
100        );
101        property_encode!(
102            &self.request_problem_info,
103            Property::RequestProblemInformation,
104            buf
105        );
106        property_encode!(&self.user_properties, Property::UserProp, buf);
107        property_encode!(&self.auth_method, Property::AuthenticationMethod, buf);
108        property_encode!(&self.auth_data, Property::AuthenticationData, buf);
109    }
110
111    /// Decodes properties from a byte buffer
112    fn decode(buf: &mut Bytes) -> Result<Option<Self>, Error> {
113        if buf.is_empty() {
114            return Ok(None);
115        }
116
117        let mut properties = ConnectProperties::default();
118
119        while buf.has_remaining() {
120            let property: Property = decode_byte(buf)?.try_into()?;
121            match property {
122                Property::SessionExpiryInterval => {
123                    property_decode!(&mut properties.session_expiry_interval, buf);
124                }
125                Property::ReceiveMaximum => {
126                    property_decode_non_zero!(&mut properties.receive_maximum, buf);
127                }
128                Property::MaximumPacketSize => {
129                    property_decode_non_zero!(&mut properties.maximum_packet_size, buf);
130                }
131                Property::TopicAliasMaximum => {
132                    property_decode!(&mut properties.topic_alias_maximum, buf);
133                }
134                Property::RequestResponseInformation => {
135                    property_decode!(&mut properties.request_response_info, buf);
136                }
137                Property::RequestProblemInformation => {
138                    property_decode!(&mut properties.request_problem_info, buf);
139                }
140                Property::UserProp => {
141                    property_decode!(&mut properties.user_properties, buf);
142                }
143                Property::AuthenticationMethod => {
144                    property_decode!(&mut properties.auth_method, buf);
145                }
146                Property::AuthenticationData => {
147                    property_decode!(&mut properties.auth_data, buf);
148                }
149                _ => return Err(Error::PropertyMismatch),
150            };
151        }
152
153        if properties.auth_data.is_some() && properties.auth_method.is_none() {
154            return Err(Error::ProtocolError);
155        }
156
157        Ok(Some(properties))
158    }
159}
160
161impl ConnectFrame for ConnectHeader<ConnectProperties> {
162    /// Calculates the encoded length of the Connect header
163    fn encoded_len(&self) -> usize {
164        let properties_len = self
165            .properties
166            .as_ref()
167            .map(|properties| properties.encoded_len())
168            .unwrap_or(0);
169        properties_len + len_bytes(properties_len) + self.primary_encoded_len()
170    }
171
172    /// Encodes the Connect header into a byte buffer
173    fn encode(&self, buf: &mut BytesMut) -> Result<(), Error> {
174        self.primary_encode(buf);
175
176        let properties_len = self
177            .properties
178            .as_ref()
179            .map(|properties| properties.encoded_len())
180            .unwrap_or(0) as u32;
181
182        encode_variable_integer(buf, properties_len)?;
183
184        if let Some(properties) = self.properties.as_ref() {
185            properties.encode(buf);
186        }
187        Ok(())
188    }
189
190    /// Decodes a Connect header from a byte buffer
191    fn decode(buf: &mut Bytes) -> Result<Self, Error> {
192        let mut header = Self::primary_decode(buf)?;
193
194        let properties_len = decode_variable_integer(buf)? as usize;
195        if buf.len() < properties_len + len_bytes(properties_len) {
196            return Err(Error::MalformedPacket);
197        }
198
199        // Skip variable byte
200        buf.advance(len_bytes(properties_len));
201
202        let mut properties_buf = buf.split_to(properties_len);
203
204        // Deserialize properties
205        header.properties = ConnectProperties::decode(&mut properties_buf)?;
206
207        Ok(header)
208    }
209}
210
211/// Represents the properties of a Will message in MQTT v5.
212///
213/// These properties provide extended functionality for the last will and testament
214/// message, including delivery timing, content format, and correlation data.
215/// # Example
216///
217/// ```rust
218/// use std::time::Duration;
219/// use mqute_codec::protocol::v5::WillProperties;
220///
221/// let will_properties = WillProperties {
222///     delay_interval: Some(Duration::from_secs(10)),
223///     content_type: Some("json".to_string()),
224///     ..Default::default()
225/// };
226/// ```
227#[derive(Debug, Default, Clone, PartialEq, Eq)]
228pub struct WillProperties {
229    /// Delay before sending the Will message after connection loss
230    pub delay_interval: Option<Duration>,
231    /// Format of the Will message payload (0=bytes, 1=UTF-8)
232    pub payload_format_indicator: Option<u8>,
233    /// Lifetime of the Will message in seconds
234    pub message_expiry_interval: Option<Duration>,
235    /// Content type descriptor (MIME type)
236    pub content_type: Option<String>,
237    /// Topic name for the response message
238    pub response_topic: Option<String>,
239    /// Correlation data for the response message
240    pub correlation_data: Option<Bytes>,
241    /// User-defined key-value properties
242    pub user_properties: Vec<(String, String)>,
243}
244
245impl PropertyFrame for WillProperties {
246    /// Calculates the encoded length of the Will properties
247    fn encoded_len(&self) -> usize {
248        let mut len = 0;
249
250        len += property_len!(&self.delay_interval);
251        len += property_len!(&self.payload_format_indicator);
252        len += property_len!(&self.message_expiry_interval);
253        len += property_len!(&self.content_type);
254        len += property_len!(&self.response_topic);
255        len += property_len!(&self.correlation_data);
256        len += property_len!(&self.user_properties);
257
258        len
259    }
260
261    /// Encodes the Will properties into a byte buffer
262    fn encode(&self, buf: &mut BytesMut) {
263        property_encode!(&self.delay_interval, Property::WillDelayInterval, buf);
264        property_encode!(
265            &self.payload_format_indicator,
266            Property::PayloadFormatIndicator,
267            buf
268        );
269        property_encode!(
270            &self.message_expiry_interval,
271            Property::MessageExpiryInterval,
272            buf
273        );
274        property_encode!(&self.content_type, Property::ContentType, buf);
275        property_encode!(&self.response_topic, Property::ResponseTopic, buf);
276        property_encode!(&self.correlation_data, Property::CorrelationData, buf);
277        property_encode!(&self.user_properties, Property::UserProp, buf);
278    }
279
280    /// Decodes Will properties from a byte buffer
281    fn decode(buf: &mut Bytes) -> Result<Option<Self>, Error> {
282        if buf.is_empty() {
283            return Ok(None);
284        }
285
286        let mut properties = WillProperties::default();
287
288        while buf.has_remaining() {
289            let property: Property = decode_byte(buf)?.try_into()?;
290            match property {
291                Property::WillDelayInterval => {
292                    property_decode!(&mut properties.delay_interval, buf);
293                }
294                Property::PayloadFormatIndicator => {
295                    property_decode!(&mut properties.payload_format_indicator, buf);
296                    if let Some(value) = properties.payload_format_indicator
297                        && value != 0
298                        && value != 1
299                    {
300                        return Err(Error::ProtocolError);
301                    }
302                }
303                Property::MessageExpiryInterval => {
304                    property_decode!(&mut properties.message_expiry_interval, buf);
305                }
306                Property::ContentType => {
307                    property_decode!(&mut properties.content_type, buf);
308                }
309                Property::ResponseTopic => {
310                    property_decode!(&mut properties.response_topic, buf);
311                }
312                Property::CorrelationData => {
313                    property_decode!(&mut properties.correlation_data, buf);
314                }
315                Property::UserProp => {
316                    property_decode!(&mut properties.user_properties, buf);
317                }
318                _ => return Err(Error::PropertyMismatch),
319            }
320        }
321
322        Ok(Some(properties))
323    }
324}
325
326/// Represents a Last Will and Testament (LWT) message in MQTT v5.
327///
328/// The Will message is published by the broker when the client disconnects unexpectedly.
329/// It includes the message content, delivery options, and MQTT v5 properties that provide
330/// additional control over the will message delivery.
331///
332/// # Example
333///
334/// ```rust
335/// use mqute_codec::protocol::v5::Will;
336/// use bytes::Bytes;
337/// use mqute_codec::protocol::QoS;
338///
339/// let will = Will::new(None, "/topic", Bytes::new(), QoS::ExactlyOnce, false);
340/// ```
341#[derive(Debug, Clone, PartialEq, Eq)]
342pub struct Will {
343    /// Will message properties
344    pub properties: Option<WillProperties>,
345    /// Topic name to publish the Will message to
346    pub topic: String,
347    /// Will message payload
348    pub payload: Bytes,
349    /// Quality of Service level for the Will message
350    pub qos: QoS,
351    /// Whether the Will message should be retained
352    pub retain: bool,
353}
354
355impl Will {
356    /// Creates a new `Will` instance with the specified parameters.
357    ///
358    /// # Panics
359    ///
360    /// Panics if the topic name is invalid according to MQTT topic naming rules.
361    pub fn new<T: Into<String>>(
362        properties: Option<WillProperties>,
363        topic: T,
364        payload: Bytes,
365        qos: QoS,
366        retain: bool,
367    ) -> Self {
368        let topic = topic.into();
369
370        if !util::is_valid_topic_name(&topic) {
371            panic!("Invalid topic name: '{}'", topic);
372        }
373
374        Will {
375            properties,
376            topic,
377            payload,
378            qos,
379            retain,
380        }
381    }
382}
383
384impl WillFrame for Will {
385    /// Calculates the encoded length of the Will message
386    fn encoded_len(&self) -> usize {
387        let properties_len = self
388            .properties
389            .as_ref()
390            .map(|properties| properties.encoded_len())
391            .unwrap_or(0);
392
393        2 + self.topic.len() + 2 + self.payload.len() + len_bytes(properties_len) + properties_len
394    }
395
396    /// Updates the Connect packet flags based on Will message settings
397    fn update_flags(&self, flags: &mut u8) {
398        // Update the 'Will' flag
399        flags.set_bit(WILL_FLAG, true);
400
401        // Update 'Qos' flags
402        flags.set_bits(WILL_QOS, self.qos as u8);
403
404        // Update the 'Will Retain' flag
405        flags.set_bit(WILL_RETAIN, self.retain);
406    }
407
408    /// Encodes the Will message into a byte buffer
409    fn encode(&self, buf: &mut BytesMut) -> Result<(), Error> {
410        let properties_len = self
411            .properties
412            .as_ref()
413            .map(|properties| properties.encoded_len())
414            .unwrap_or(0) as u32;
415
416        encode_variable_integer(buf, properties_len)?;
417
418        if let Some(properties) = self.properties.as_ref() {
419            properties.encode(buf);
420        }
421
422        encode_string(buf, &self.topic);
423        encode_bytes(buf, &self.payload);
424        Ok(())
425    }
426
427    /// Decodes a Will message from a byte buffer
428    fn decode(buf: &mut Bytes, flags: u8) -> Result<Option<Self>, Error> {
429        if !flags.get_bit(WILL_FLAG) {
430            // No 'Will'
431            return Ok(None);
432        }
433
434        let properties_len = decode_variable_integer(buf)? as usize;
435        if buf.len() < properties_len + len_bytes(properties_len) {
436            return Err(Error::MalformedPacket);
437        }
438
439        // Skip properties len
440        buf.advance(len_bytes(properties_len));
441        let mut properties_buf = buf.split_to(properties_len);
442        let properties = WillProperties::decode(&mut properties_buf)?;
443        let qos = flags.get_bits(WILL_QOS).try_into()?;
444        let retain = flags.get_bit(WILL_RETAIN);
445
446        let topic = decode_string(buf)?;
447
448        if !util::is_valid_topic_name(&topic) {
449            return Err(Error::InvalidTopicName(topic));
450        }
451
452        let payload = decode_bytes(buf)?;
453
454        Ok(Some(Will {
455            properties,
456            topic,
457            payload,
458            qos,
459            retain,
460        }))
461    }
462}
463
464// Defines the `Connect` packet for MQTT V5
465connect!(Connect<ConnectProperties, Will>, Protocol::V5);
466
467impl Connect {
468    /// Creates a new Connect packet with properties
469    ///
470    /// # Panics
471    ///
472    /// Panics if the value of the "keep alive" parameter exceeds 65535
473    pub fn with_properties<S: Into<String>>(
474        client_id: S,
475        auth: Option<Credentials>,
476        will: Option<Will>,
477        properties: ConnectProperties,
478        keep_alive: Duration,
479        clean_session: bool,
480    ) -> Self {
481        Self::from_scratch(
482            client_id,
483            auth,
484            will,
485            Some(properties),
486            keep_alive,
487            clean_session,
488        )
489    }
490
491    /// Returns the Connect properties if present
492    pub fn properties(&self) -> Option<ConnectProperties> {
493        self.header.properties.clone()
494    }
495}
mqute_codec/protocol/v5/connect.rs

mqute_codec/protocol/v5/
connect.rs
