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, PropertyFrame, property_len};
9use super::property::{property_decode, property_decode_non_zero, property_encode};
10use crate::Error;
11use crate::codec::util::{
12    decode_byte, decode_bytes, decode_string, decode_variable_integer, encode_bytes, encode_string,
13    encode_variable_integer,
14};
15use crate::protocol::common::{ConnectFrame, WillFrame};
16use crate::protocol::common::{ConnectHeader, connect};
17use crate::protocol::util::len_bytes;
18use crate::protocol::{Credentials, Protocol, QoS};
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                        if value != 0 && value != 1 {
298                            return Err(Error::ProtocolError);
299                        }
300                    }
301                }
302                Property::MessageExpiryInterval => {
303                    property_decode!(&mut properties.message_expiry_interval, buf);
304                }
305                Property::ContentType => {
306                    property_decode!(&mut properties.content_type, buf);
307                }
308                Property::ResponseTopic => {
309                    property_decode!(&mut properties.response_topic, buf);
310                }
311                Property::CorrelationData => {
312                    property_decode!(&mut properties.correlation_data, buf);
313                }
314                Property::UserProp => {
315                    property_decode!(&mut properties.user_properties, buf);
316                }
317                _ => return Err(Error::PropertyMismatch),
318            }
319        }
320
321        Ok(Some(properties))
322    }
323}
324
325/// Represents a Will message in MQTT v5.
326///
327/// The Will message is published by the broker when the client disconnects
328/// unexpectedly. It includes the message content, delivery options, and properties.
329///
330/// # Example
331///
332/// ```rust
333/// use mqute_codec::protocol::v5::Will;
334/// use bytes::Bytes;
335/// use mqute_codec::protocol::QoS;
336///
337/// let will = Will::new(None, "tpoic", Bytes::new(), QoS::ExactlyOnce, false);
338/// ```
339#[derive(Debug, Clone, PartialEq, Eq)]
340pub struct Will {
341    /// Will message properties
342    pub properties: Option<WillProperties>,
343    /// Topic name to publish the Will message to
344    pub topic: String,
345    /// Will message payload
346    pub payload: Bytes,
347    /// Quality of Service level for the Will message
348    pub qos: QoS,
349    /// Whether the Will message should be retained
350    pub retain: bool,
351}
352
353impl Will {
354    /// Creates a new `Will` packet
355    pub fn new<T: Into<String>>(
356        properties: Option<WillProperties>,
357        topic: T,
358        payload: Bytes,
359        qos: QoS,
360        retain: bool,
361    ) -> Will {
362        Will {
363            properties,
364            topic: topic.into(),
365            payload,
366            qos,
367            retain,
368        }
369    }
370}
371
372impl WillFrame for Will {
373    /// Calculates the encoded length of the Will message
374    fn encoded_len(&self) -> usize {
375        let properties_len = self
376            .properties
377            .as_ref()
378            .map(|properties| properties.encoded_len())
379            .unwrap_or(0);
380
381        2 + self.topic.len() + 2 + self.payload.len() + len_bytes(properties_len) + properties_len
382    }
383
384    /// Updates the Connect packet flags based on Will message settings
385    fn update_flags(&self, flags: &mut u8) {
386        // Update the 'Will' flag
387        flags.set_bit(WILL_FLAG, true);
388
389        // Update 'Qos' flags
390        flags.set_bits(WILL_QOS, self.qos as u8);
391
392        // Update the 'Will Retain' flag
393        flags.set_bit(WILL_RETAIN, self.retain);
394    }
395
396    /// Encodes the Will message into a byte buffer
397    fn encode(&self, buf: &mut BytesMut) -> Result<(), Error> {
398        let properties_len = self
399            .properties
400            .as_ref()
401            .map(|properties| properties.encoded_len())
402            .unwrap_or(0) as u32;
403
404        encode_variable_integer(buf, properties_len)?;
405
406        if let Some(properties) = self.properties.as_ref() {
407            properties.encode(buf);
408        }
409
410        encode_string(buf, &self.topic);
411        encode_bytes(buf, &self.payload);
412        Ok(())
413    }
414
415    /// Decodes a Will message from a byte buffer
416    fn decode(buf: &mut Bytes, flags: u8) -> Result<Option<Self>, Error> {
417        if !flags.get_bit(WILL_FLAG) {
418            // No 'Will'
419            return Ok(None);
420        }
421
422        let properties_len = decode_variable_integer(buf)? as usize;
423        if buf.len() < properties_len + len_bytes(properties_len) {
424            return Err(Error::MalformedPacket);
425        }
426
427        // Skip properties len
428        buf.advance(len_bytes(properties_len));
429        let mut properties_buf = buf.split_to(properties_len);
430        let properties = WillProperties::decode(&mut properties_buf)?;
431        let qos = flags.get_bits(WILL_QOS).try_into()?;
432        let retain = flags.get_bit(WILL_RETAIN);
433
434        let topic = decode_string(buf)?;
435        let payload = decode_bytes(buf)?;
436
437        Ok(Some(Will {
438            properties,
439            topic,
440            payload,
441            qos,
442            retain,
443        }))
444    }
445}
446
447// Defines the `Connect` packet for MQTT V5
448connect!(Connect<ConnectProperties, Will>, Protocol::V5);
449
450impl Connect {
451    /// Creates a new Connect packet with properties
452    ///
453    /// # Panics
454    ///
455    /// Panics if the value of the "keep alive" parameter exceeds 65535
456    pub fn with_properties<S: Into<String>>(
457        client_id: S,
458        auth: Option<Credentials>,
459        will: Option<Will>,
460        properties: ConnectProperties,
461        keep_alive: Duration,
462        clean_session: bool,
463    ) -> Self {
464        Self::from_scratch(
465            client_id,
466            auth,
467            will,
468            Some(properties),
469            keep_alive,
470            clean_session,
471        )
472    }
473
474    /// Returns the Connect properties if present
475    pub fn properties(&self) -> Option<ConnectProperties> {
476        self.header.properties.clone()
477    }
478}
mqute_codec/protocol/v5/connect.rs

mqute_codec/protocol/v5/
connect.rs
