rmqtt_codec/v3/
packet.rs

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

rmqtt_codec/v3/
packet.rs
