rmqtt_codec/v5/packet/
mod.rs

1use bytes::{Buf, BufMut, Bytes, BytesMut};
2use bytestring::ByteString;
3
4pub use crate::types::{ConnectAckFlags, ConnectFlags, QoS};
5
6use super::{encode::*, property_type as pt, UserProperties};
7use crate::error::{DecodeError, EncodeError};
8use crate::types::packet_type;
9use crate::utils::{take_properties, write_variable_length, Decode, Property};
10
11mod auth;
12mod connack;
13mod connect;
14mod disconnect;
15mod pubacks;
16mod publish;
17mod subscribe;
18
19pub use auth::*;
20pub use connack::*;
21pub use connect::*;
22pub use disconnect::*;
23pub use pubacks::*;
24pub use publish::*;
25pub use subscribe::*;
26
27#[derive(Debug, PartialEq, Eq, Clone)]
28/// MQTT Control Packets
29pub enum Packet {
30    /// Client request to connect to Server
31    Connect(Box<Connect>),
32    /// Connect acknowledgment
33    ConnectAck(Box<ConnectAck>),
34    /// Publish message
35    Publish(Box<Publish>),
36    /// Publish acknowledgment
37    PublishAck(PublishAck),
38    /// Publish received (assured delivery part 1)
39    PublishReceived(PublishAck),
40    /// Publish release (assured delivery part 2)
41    PublishRelease(PublishAck2),
42    /// Publish complete (assured delivery part 3)
43    PublishComplete(PublishAck2),
44    /// Client subscribe request
45    Subscribe(Subscribe),
46    /// Subscribe acknowledgment
47    SubscribeAck(SubscribeAck),
48    /// Unsubscribe request
49    Unsubscribe(Unsubscribe),
50    /// Unsubscribe acknowledgment
51    UnsubscribeAck(UnsubscribeAck),
52    /// PING request
53    PingRequest,
54    /// PING response
55    PingResponse,
56    /// Disconnection is advertised
57    Disconnect(Disconnect),
58    /// Auth exchange
59    Auth(Auth),
60}
61
62impl Packet {
63    /// Returns the MQTT packet type byte for this packet
64    pub fn packet_type(&self) -> u8 {
65        match self {
66            Packet::Connect(_) => packet_type::CONNECT,
67            Packet::ConnectAck(_) => packet_type::CONNACK,
68            Packet::Publish(_) => packet_type::PUBLISH_START,
69            Packet::PublishAck(_) => packet_type::PUBACK,
70            Packet::PublishReceived(_) => packet_type::PUBREC,
71            Packet::PublishRelease(_) => packet_type::PUBREL,
72            Packet::PublishComplete(_) => packet_type::PUBCOMP,
73            Packet::Subscribe(_) => packet_type::SUBSCRIBE,
74            Packet::SubscribeAck(_) => packet_type::SUBACK,
75            Packet::Unsubscribe(_) => packet_type::UNSUBSCRIBE,
76            Packet::UnsubscribeAck(_) => packet_type::UNSUBACK,
77            Packet::PingRequest => packet_type::PINGREQ,
78            Packet::PingResponse => packet_type::PINGRESP,
79            Packet::Disconnect(_) => packet_type::DISCONNECT,
80            Packet::Auth(_) => packet_type::AUTH,
81        }
82    }
83}
84
85impl From<Connect> for Packet {
86    fn from(pkt: Connect) -> Self {
87        Self::Connect(Box::new(pkt))
88    }
89}
90
91impl From<Box<Connect>> for Packet {
92    fn from(pkt: Box<Connect>) -> Self {
93        Self::Connect(pkt)
94    }
95}
96
97impl From<ConnectAck> for Packet {
98    fn from(pkt: ConnectAck) -> Self {
99        Self::ConnectAck(Box::new(pkt))
100    }
101}
102
103impl From<Box<ConnectAck>> for Packet {
104    fn from(pkt: Box<ConnectAck>) -> Self {
105        Self::ConnectAck(pkt)
106    }
107}
108
109impl From<Publish> for Packet {
110    fn from(pkt: Publish) -> Self {
111        Self::Publish(Box::new(pkt))
112    }
113}
114
115impl From<PublishAck> for Packet {
116    fn from(pkt: PublishAck) -> Self {
117        Self::PublishAck(pkt)
118    }
119}
120
121impl From<Subscribe> for Packet {
122    fn from(pkt: Subscribe) -> Self {
123        Self::Subscribe(pkt)
124    }
125}
126
127impl From<SubscribeAck> for Packet {
128    fn from(pkt: SubscribeAck) -> Self {
129        Self::SubscribeAck(pkt)
130    }
131}
132
133impl From<Unsubscribe> for Packet {
134    fn from(pkt: Unsubscribe) -> Self {
135        Self::Unsubscribe(pkt)
136    }
137}
138
139impl From<UnsubscribeAck> for Packet {
140    fn from(pkt: UnsubscribeAck) -> Self {
141        Self::UnsubscribeAck(pkt)
142    }
143}
144
145impl From<Disconnect> for Packet {
146    fn from(pkt: Disconnect) -> Self {
147        Self::Disconnect(pkt)
148    }
149}
150
151impl From<Auth> for Packet {
152    fn from(pkt: Auth) -> Self {
153        Self::Auth(pkt)
154    }
155}
156
157/// MQTT v5 property type identifiers
158///
159/// Defines byte codes for all property types defined in the MQTT v5.0 specification.
160pub(super) mod property_type {
161    pub(crate) const UTF8_PAYLOAD: u8 = 0x01;
162    pub(crate) const MSG_EXPIRY_INT: u8 = 0x02;
163    pub(crate) const CONTENT_TYPE: u8 = 0x03;
164    pub(crate) const RESP_TOPIC: u8 = 0x08;
165    pub(crate) const CORR_DATA: u8 = 0x09;
166    pub(crate) const SUB_ID: u8 = 0x0B;
167    pub(crate) const SESS_EXPIRY_INT: u8 = 0x11;
168    pub(crate) const ASSND_CLIENT_ID: u8 = 0x12;
169    pub(crate) const SERVER_KA: u8 = 0x13;
170    pub(crate) const AUTH_METHOD: u8 = 0x15;
171    pub(crate) const AUTH_DATA: u8 = 0x16;
172    pub(crate) const REQ_PROB_INFO: u8 = 0x17;
173    pub(crate) const WILL_DELAY_INT: u8 = 0x18;
174    pub(crate) const REQ_RESP_INFO: u8 = 0x19;
175    pub(crate) const RESP_INFO: u8 = 0x1A;
176    pub(crate) const SERVER_REF: u8 = 0x1C;
177    pub(crate) const REASON_STRING: u8 = 0x1F;
178    pub(crate) const RECEIVE_MAX: u8 = 0x21;
179    pub(crate) const TOPIC_ALIAS_MAX: u8 = 0x22;
180    pub(crate) const TOPIC_ALIAS: u8 = 0x23;
181    pub(crate) const MAX_QOS: u8 = 0x24;
182    pub(crate) const RETAIN_AVAIL: u8 = 0x25;
183    pub(crate) const USER: u8 = 0x26;
184    pub(crate) const MAX_PACKET_SIZE: u8 = 0x27;
185    pub(crate) const WILDCARD_SUB_AVAIL: u8 = 0x28;
186    pub(crate) const SUB_IDS_AVAIL: u8 = 0x29;
187    pub(crate) const SHARED_SUB_AVAIL: u8 = 0x2A;
188}
189
190mod ack_props {
191    use super::*;
192    use crate::v5::UserProperty;
193
194    pub(crate) fn encoded_size(
195        properties: &[UserProperty],
196        reason_string: &Option<ByteString>,
197        limit: u32,
198    ) -> usize {
199        if limit < 4 {
200            // todo: not really needed in practice
201            return 1; // 1 byte to encode property length = 0
202        }
203
204        let len = encoded_size_opt_props(properties, reason_string, limit - 4);
205        var_int_len(len) as usize + len
206    }
207
208    pub(crate) fn encode(
209        properties: &[UserProperty],
210        reason_string: &Option<ByteString>,
211        buf: &mut BytesMut,
212        size: u32,
213    ) -> Result<(), EncodeError> {
214        debug_assert!(size > 0); // formalize in signature?
215
216        if size == 1 {
217            // empty properties
218            buf.put_u8(0);
219            return Ok(());
220        }
221
222        let size = var_int_len_from_size(size);
223        write_variable_length(size, buf);
224        encode_opt_props(properties, reason_string, buf, size)
225    }
226
227    /// Parses ACK properties (User and Reason String properties) from `src`
228    pub(crate) fn decode(src: &mut Bytes) -> Result<(UserProperties, Option<ByteString>), DecodeError> {
229        let prop_src = &mut take_properties(src)?;
230        let mut reason_string = None;
231        let mut user_props = Vec::new();
232        while prop_src.has_remaining() {
233            let prop_id = prop_src.get_u8();
234            match prop_id {
235                pt::REASON_STRING => reason_string.read_value(prop_src)?,
236                pt::USER => user_props.push(<(ByteString, ByteString)>::decode(prop_src)?),
237                _ => return Err(DecodeError::MalformedPacket),
238            }
239        }
240
241        Ok((user_props, reason_string))
242    }
243}
rmqtt_codec/v5/packet/mod.rs

rmqtt_codec/v5/packet/
mod.rs
