rmqtt_codec/v5/packet/
disconnect.rs

1use bytes::{Buf, BufMut, Bytes, BytesMut};
2use bytestring::ByteString;
3use serde::{Deserialize, Serialize};
4
5use crate::error::{DecodeError, EncodeError};
6use crate::utils::{self, Decode, Property};
7use crate::v5::{encode::*, property_type as pt, UserProperties, UserProperty};
8
9/// DISCONNECT message
10#[derive(Debug, PartialEq, Eq, Clone, Deserialize, Serialize)]
11pub struct Disconnect {
12    pub reason_code: DisconnectReasonCode,
13    pub session_expiry_interval_secs: Option<u32>,
14    pub server_reference: Option<ByteString>,
15    pub reason_string: Option<ByteString>,
16    pub user_properties: UserProperties,
17}
18
19pub trait ToReasonCode {
20    fn to_reason_code(&self) -> DisconnectReasonCode;
21}
22
23prim_enum! {
24    /// DISCONNECT reason codes
25    #[derive(Deserialize, Serialize)]
26    pub enum DisconnectReasonCode {
27        NormalDisconnection = 0,
28        DisconnectWithWillMessage = 4,
29        UnspecifiedError = 128,
30        MalformedPacket = 129,
31        ProtocolError = 130,
32        ImplementationSpecificError = 131,
33        NotAuthorized = 135,
34        ServerBusy = 137,
35        ServerShuttingDown = 139,
36        BadAuthenticationMethod = 140,
37        KeepAliveTimeout = 141,
38        SessionTakenOver = 142,
39        TopicFilterInvalid = 143,
40        TopicNameInvalid = 144,
41        ReceiveMaximumExceeded = 147,
42        TopicAliasInvalid = 148,
43        PacketTooLarge = 149,
44        MessageRateTooHigh = 150,
45        QuotaExceeded = 151,
46        AdministrativeAction = 152,
47        PayloadFormatInvalid = 153,
48        RetainNotSupported = 154,
49        QosNotSupported = 155,
50        UseAnotherServer = 156,
51        ServerMoved = 157,
52        SharedSubscriptionNotSupported = 158,
53        ConnectionRateExceeded = 159,
54        MaximumConnectTime = 160,
55        SubscriptionIdentifiersNotSupported = 161,
56        WildcardSubscriptionsNotSupported = 162
57    }
58}
59
60impl From<DisconnectReasonCode> for u8 {
61    fn from(v: DisconnectReasonCode) -> Self {
62        match v {
63            DisconnectReasonCode::NormalDisconnection => 0,
64            DisconnectReasonCode::DisconnectWithWillMessage => 4,
65            DisconnectReasonCode::UnspecifiedError => 128,
66            DisconnectReasonCode::MalformedPacket => 129,
67            DisconnectReasonCode::ProtocolError => 130,
68            DisconnectReasonCode::ImplementationSpecificError => 131,
69            DisconnectReasonCode::NotAuthorized => 135,
70            DisconnectReasonCode::ServerBusy => 137,
71            DisconnectReasonCode::ServerShuttingDown => 139,
72            DisconnectReasonCode::BadAuthenticationMethod => 140,
73            DisconnectReasonCode::KeepAliveTimeout => 141,
74            DisconnectReasonCode::SessionTakenOver => 142,
75            DisconnectReasonCode::TopicFilterInvalid => 143,
76            DisconnectReasonCode::TopicNameInvalid => 144,
77            DisconnectReasonCode::ReceiveMaximumExceeded => 147,
78            DisconnectReasonCode::TopicAliasInvalid => 148,
79            DisconnectReasonCode::PacketTooLarge => 149,
80            DisconnectReasonCode::MessageRateTooHigh => 150,
81            DisconnectReasonCode::QuotaExceeded => 151,
82            DisconnectReasonCode::AdministrativeAction => 152,
83            DisconnectReasonCode::PayloadFormatInvalid => 153,
84            DisconnectReasonCode::RetainNotSupported => 154,
85            DisconnectReasonCode::QosNotSupported => 155,
86            DisconnectReasonCode::UseAnotherServer => 156,
87            DisconnectReasonCode::ServerMoved => 157,
88            DisconnectReasonCode::SharedSubscriptionNotSupported => 158,
89            DisconnectReasonCode::ConnectionRateExceeded => 159,
90            DisconnectReasonCode::MaximumConnectTime => 160,
91            DisconnectReasonCode::SubscriptionIdentifiersNotSupported => 161,
92            DisconnectReasonCode::WildcardSubscriptionsNotSupported => 162,
93        }
94    }
95}
96
97impl Disconnect {
98    /// Create new instance of `Disconnect` with specified code
99    pub fn new(reason_code: DisconnectReasonCode) -> Self {
100        Self {
101            reason_code,
102            session_expiry_interval_secs: None,
103            server_reference: None,
104            reason_string: None,
105            user_properties: Vec::new(),
106        }
107    }
108
109    pub(crate) fn decode(src: &mut Bytes) -> Result<Self, DecodeError> {
110        let disconnect = if src.has_remaining() {
111            let reason_code = src.get_u8().try_into()?;
112
113            if src.has_remaining() {
114                let mut session_exp_secs = None;
115                let mut server_reference = None;
116                let mut reason_string = None;
117                let mut user_properties = Vec::new();
118
119                let prop_src = &mut utils::take_properties(src)?;
120                while prop_src.has_remaining() {
121                    match prop_src.get_u8() {
122                        pt::SESS_EXPIRY_INT => session_exp_secs.read_value(prop_src)?,
123                        pt::REASON_STRING => reason_string.read_value(prop_src)?,
124                        pt::USER => user_properties.push(UserProperty::decode(prop_src)?),
125                        pt::SERVER_REF => server_reference.read_value(prop_src)?,
126                        _ => return Err(DecodeError::MalformedPacket),
127                    }
128                }
129                ensure!(!src.has_remaining(), DecodeError::InvalidLength);
130
131                Self {
132                    reason_code,
133                    session_expiry_interval_secs: session_exp_secs,
134                    server_reference,
135                    reason_string,
136                    user_properties,
137                }
138            } else {
139                Self { reason_code, ..Default::default() }
140            }
141        } else {
142            Self::default()
143        };
144        Ok(disconnect)
145    }
146}
147
148impl Default for Disconnect {
149    fn default() -> Self {
150        Self {
151            reason_code: DisconnectReasonCode::NormalDisconnection,
152            session_expiry_interval_secs: None,
153            server_reference: None,
154            reason_string: None,
155            user_properties: Vec::new(),
156        }
157    }
158}
159
160impl EncodeLtd for Disconnect {
161    fn encoded_size(&self, limit: u32) -> usize {
162        const HEADER_LEN: usize = 1; // reason code
163
164        let mut prop_len = encoded_property_size(&self.session_expiry_interval_secs)
165            + encoded_property_size(&self.server_reference);
166        let diag_len = encoded_size_opt_props(
167            &self.user_properties,
168            &self.reason_string,
169            reduce_limit(limit, prop_len + HEADER_LEN + 4),
170        ); // exclude other props and max of 4 bytes for property length value
171        prop_len += diag_len;
172        HEADER_LEN + var_int_len(prop_len) as usize + prop_len
173    }
174
175    fn encode(&self, buf: &mut BytesMut, size: u32) -> Result<(), EncodeError> {
176        let start_len = buf.len();
177        buf.put_u8(self.reason_code.into());
178
179        let prop_len = var_int_len_from_size(size - 1);
180        utils::write_variable_length(prop_len, buf);
181        encode_property(&self.session_expiry_interval_secs, pt::SESS_EXPIRY_INT, buf)?;
182        encode_property(&self.server_reference, pt::SERVER_REF, buf)?;
183        encode_opt_props(
184            &self.user_properties,
185            &self.reason_string,
186            buf,
187            size - (buf.len() - start_len) as u32,
188        )
189    }
190}
rmqtt_codec/v5/packet/disconnect.rs

rmqtt_codec/v5/packet/
disconnect.rs
