rmqtt_codec/v3/
packet.rs

1use std::num::NonZeroU16;
2
3use crate::cert::CertInfo;
4use crate::types::{packet_type, Protocol, QoS};
5use bytes::Bytes;
6use bytestring::ByteString;
7use serde::{Deserialize, Serialize};
8
9prim_enum! {
10    /// Connect Return Code
11    #[derive(Deserialize, Serialize)]
12    pub enum ConnectAckReason {
13        /// Connection accepted
14        ConnectionAccepted = 0,
15        /// Connection Refused, unacceptable protocol version
16        UnacceptableProtocolVersion = 1,
17        /// Connection Refused, identifier rejected
18        IdentifierRejected = 2,
19        /// Connection Refused, Server unavailable
20        ServiceUnavailable = 3,
21        /// Connection Refused, bad user name or password
22        BadUserNameOrPassword = 4,
23        /// Connection Refused, not authorized
24        NotAuthorized = 5,
25        /// Reserved
26        Reserved = 6
27    }
28}
29
30impl From<ConnectAckReason> for u8 {
31    fn from(v: ConnectAckReason) -> Self {
32        match v {
33            ConnectAckReason::ConnectionAccepted => 0,
34            ConnectAckReason::UnacceptableProtocolVersion => 1,
35            ConnectAckReason::IdentifierRejected => 2,
36            ConnectAckReason::ServiceUnavailable => 3,
37            ConnectAckReason::BadUserNameOrPassword => 4,
38            ConnectAckReason::NotAuthorized => 5,
39            ConnectAckReason::Reserved => 6,
40        }
41    }
42}
43
44impl ConnectAckReason {
45    /// Returns a human-readable description of the reason code
46    pub fn reason(self) -> &'static str {
47        match self {
48            ConnectAckReason::ConnectionAccepted => "Connection Accepted",
49            ConnectAckReason::UnacceptableProtocolVersion => {
50                "Connection Refused, unacceptable protocol version"
51            }
52            ConnectAckReason::IdentifierRejected => "Connection Refused, identifier rejected",
53            ConnectAckReason::ServiceUnavailable => "Connection Refused, Server unavailable",
54            ConnectAckReason::BadUserNameOrPassword => "Connection Refused, bad user name or password",
55            ConnectAckReason::NotAuthorized => "Connection Refused, not authorized",
56            _ => "Connection Refused",
57        }
58    }
59}
60
61#[derive(Debug, PartialEq, Eq, Clone, Deserialize, Serialize)]
62/// Connection Will
63pub struct LastWill {
64    /// the QoS level to be used when publishing the Will Message.
65    pub qos: QoS,
66    /// the Will Message is to be Retained when it is published.
67    pub retain: bool,
68    /// the Will Topic
69    pub topic: ByteString,
70    /// defines the Application Message that is to be published to the Will Topic
71    pub message: Bytes,
72}
73
74#[derive(Default, Debug, PartialEq, Eq, Clone, Deserialize, Serialize)]
75/// Connect packet content
76pub struct Connect {
77    /// mqtt protocol version
78    pub protocol: Protocol,
79    /// the handling of the Session state.
80    pub clean_session: bool,
81    /// a time interval measured in seconds.
82    pub keep_alive: u16,
83    /// Will Message be stored on the Server and associated with the Network Connection.
84    pub last_will: Option<LastWill>,
85    /// identifies the Client to the Server.
86    pub client_id: ByteString,
87    /// username can be used by the Server for authentication and authorization.
88    pub username: Option<ByteString>,
89    /// password can be used by the Server for authentication and authorization.
90    pub password: Option<Bytes>,
91    /// certificate information
92    pub cert: Option<CertInfo>,
93}
94
95impl Connect {
96    /// Set client_id value
97    pub fn client_id<T>(mut self, client_id: T) -> Self
98    where
99        ByteString: From<T>,
100    {
101        self.client_id = client_id.into();
102        self
103    }
104}
105
106pub(crate) type Publish = crate::types::Publish;
107
108#[derive(Debug, PartialEq, Eq, Copy, Clone, Deserialize, Serialize)]
109/// ConnectAck message
110pub struct ConnectAck {
111    pub return_code: ConnectAckReason,
112    /// enables a Client to establish whether the Client and Server have a consistent view
113    /// about whether there is already stored Session state.
114    pub session_present: bool,
115}
116
117#[derive(Debug, PartialEq, Eq, Copy, Clone, Deserialize, Serialize)]
118/// Subscribe Return Code
119pub enum SubscribeReturnCode {
120    Success(QoS),
121    Failure,
122}
123
124#[derive(Debug, PartialEq, Eq, Clone)]
125/// MQTT Control Packets
126pub enum Packet {
127    /// Client request to connect to Server
128    Connect(Box<Connect>),
129
130    /// Connect acknowledgment
131    ConnectAck(ConnectAck),
132
133    /// Publish message
134    Publish(Box<Publish>),
135
136    /// Publish acknowledgment
137    PublishAck {
138        /// Packet Identifier
139        packet_id: NonZeroU16,
140    },
141    /// Publish received (assured delivery part 1)
142    PublishReceived {
143        /// Packet Identifier
144        packet_id: NonZeroU16,
145    },
146    /// Publish release (assured delivery part 2)
147    PublishRelease {
148        /// Packet Identifier
149        packet_id: NonZeroU16,
150    },
151    /// Publish complete (assured delivery part 3)
152    PublishComplete {
153        /// Packet Identifier
154        packet_id: NonZeroU16,
155    },
156
157    /// Client subscribe request
158    Subscribe {
159        /// Packet Identifier
160        packet_id: NonZeroU16,
161        /// the list of Topic Filters and QoS to which the Client wants to subscribe.
162        topic_filters: Vec<(ByteString, QoS)>,
163    },
164    /// Subscribe acknowledgment
165    SubscribeAck {
166        packet_id: NonZeroU16,
167        /// corresponds to a Topic Filter in the SUBSCRIBE Packet being acknowledged.
168        status: Vec<SubscribeReturnCode>,
169    },
170
171    /// Unsubscribe request
172    Unsubscribe {
173        /// Packet Identifier
174        packet_id: NonZeroU16,
175        /// the list of Topic Filters that the Client wishes to unsubscribe from.
176        topic_filters: Vec<ByteString>,
177    },
178    /// Unsubscribe acknowledgment
179    UnsubscribeAck {
180        /// Packet Identifier
181        packet_id: NonZeroU16,
182    },
183
184    /// PING request
185    PingRequest,
186    /// PING response
187    PingResponse,
188    /// Client is disconnecting
189    Disconnect,
190}
191
192impl From<Connect> for Packet {
193    fn from(val: Connect) -> Packet {
194        Packet::Connect(Box::new(val))
195    }
196}
197
198impl From<Publish> for Packet {
199    fn from(val: Publish) -> Packet {
200        Packet::Publish(Box::new(val))
201    }
202}
203
204impl Packet {
205    /// Returns the MQTT packet type byte for this packet
206    pub fn packet_type(&self) -> u8 {
207        match self {
208            Packet::Connect(_) => packet_type::CONNECT,
209            Packet::ConnectAck { .. } => packet_type::CONNACK,
210            Packet::Publish(_) => packet_type::PUBLISH_START,
211            Packet::PublishAck { .. } => packet_type::PUBACK,
212            Packet::PublishReceived { .. } => packet_type::PUBREC,
213            Packet::PublishRelease { .. } => packet_type::PUBREL,
214            Packet::PublishComplete { .. } => packet_type::PUBCOMP,
215            Packet::Subscribe { .. } => packet_type::SUBSCRIBE,
216            Packet::SubscribeAck { .. } => packet_type::SUBACK,
217            Packet::Unsubscribe { .. } => packet_type::UNSUBSCRIBE,
218            Packet::UnsubscribeAck { .. } => packet_type::UNSUBACK,
219            Packet::PingRequest => packet_type::PINGREQ,
220            Packet::PingResponse => packet_type::PINGRESP,
221            Packet::Disconnect => packet_type::DISCONNECT,
222        }
223    }
224}
225
226#[cfg(test)]
227mod tests {
228    use super::*;
229
230    #[test]
231    fn test_ack_reason() {
232        assert_eq!(ConnectAckReason::ConnectionAccepted.reason(), "Connection Accepted");
233        assert_eq!(
234            ConnectAckReason::UnacceptableProtocolVersion.reason(),
235            "Connection Refused, unacceptable protocol version"
236        );
237        assert_eq!(ConnectAckReason::IdentifierRejected.reason(), "Connection Refused, identifier rejected");
238        assert_eq!(ConnectAckReason::ServiceUnavailable.reason(), "Connection Refused, Server unavailable");
239        assert_eq!(
240            ConnectAckReason::BadUserNameOrPassword.reason(),
241            "Connection Refused, bad user name or password"
242        );
243        assert_eq!(ConnectAckReason::NotAuthorized.reason(), "Connection Refused, not authorized");
244    }
245}
rmqtt_codec/v3/packet.rs

rmqtt_codec/v3/
packet.rs
