use std::convert::{TryFrom, TryInto};
use bytes::{BufMut, Bytes, BytesMut};
use crate::MqttString;
use super::{
len_len, length, read_mqtt_string, read_u32, read_u8, write_mqtt_string,
write_remaining_length, Buf, Debug, Error, FixedHeader, PacketType,
};
use super::{property, PropertyType};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum DisconnectReasonCode {
NormalDisconnection = 0x00,
DisconnectWithWillMessage = 0x04,
UnspecifiedError = 0x80,
MalformedPacket = 0x81,
ProtocolError = 0x82,
ImplementationSpecificError = 0x83,
NotAuthorized = 0x87,
ServerBusy = 0x89,
ServerShuttingDown = 0x8B,
KeepAliveTimeout = 0x8D,
SessionTakenOver = 0x8E,
TopicFilterInvalid = 0x8F,
TopicNameInvalid = 0x90,
ReceiveMaximumExceeded = 0x93,
TopicAliasInvalid = 0x94,
PacketTooLarge = 0x95,
MessageRateTooHigh = 0x96,
QuotaExceeded = 0x97,
AdministrativeAction = 0x98,
PayloadFormatInvalid = 0x99,
RetainNotSupported = 0x9A,
QoSNotSupported = 0x9B,
UseAnotherServer = 0x9C,
ServerMoved = 0x9D,
SharedSubscriptionNotSupported = 0x9E,
ConnectionRateExceeded = 0x9F,
MaximumConnectTime = 0xA0,
SubscriptionIdentifiersNotSupported = 0xA1,
WildcardSubscriptionsNotSupported = 0xA2,
}
impl TryFrom<u8> for DisconnectReasonCode {
type Error = Error;
fn try_from(value: u8) -> Result<Self, Self::Error> {
let rc = match value {
0x00 => Self::NormalDisconnection,
0x04 => Self::DisconnectWithWillMessage,
0x80 => Self::UnspecifiedError,
0x81 => Self::MalformedPacket,
0x82 => Self::ProtocolError,
0x83 => Self::ImplementationSpecificError,
0x87 => Self::NotAuthorized,
0x89 => Self::ServerBusy,
0x8B => Self::ServerShuttingDown,
0x8D => Self::KeepAliveTimeout,
0x8E => Self::SessionTakenOver,
0x8F => Self::TopicFilterInvalid,
0x90 => Self::TopicNameInvalid,
0x93 => Self::ReceiveMaximumExceeded,
0x94 => Self::TopicAliasInvalid,
0x95 => Self::PacketTooLarge,
0x96 => Self::MessageRateTooHigh,
0x97 => Self::QuotaExceeded,
0x98 => Self::AdministrativeAction,
0x99 => Self::PayloadFormatInvalid,
0x9A => Self::RetainNotSupported,
0x9B => Self::QoSNotSupported,
0x9C => Self::UseAnotherServer,
0x9D => Self::ServerMoved,
0x9E => Self::SharedSubscriptionNotSupported,
0x9F => Self::ConnectionRateExceeded,
0xA0 => Self::MaximumConnectTime,
0xA1 => Self::SubscriptionIdentifiersNotSupported,
0xA2 => Self::WildcardSubscriptionsNotSupported,
other => return Err(Error::InvalidConnectReturnCode(other)),
};
Ok(rc)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DisconnectProperties {
pub session_expiry_interval: Option<u32>,
pub reason_string: Option<MqttString>,
pub user_properties: Vec<(MqttString, MqttString)>,
pub server_reference: Option<MqttString>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Disconnect {
pub reason_code: DisconnectReasonCode,
pub properties: Option<DisconnectProperties>,
}
impl DisconnectProperties {
fn len(&self) -> usize {
let mut length = 0;
if self.session_expiry_interval.is_some() {
length += 1 + 4;
}
if let Some(reason) = &self.reason_string {
length += 1 + 2 + reason.len();
}
for (key, value) in &self.user_properties {
length += 1 + 2 + key.len() + 2 + value.len();
}
if let Some(server_reference) = &self.server_reference {
length += 1 + 2 + server_reference.len();
}
length
}
pub fn extract(bytes: &mut Bytes) -> Result<Option<Self>, Error> {
let (properties_len_len, properties_len) = length(bytes.iter())?;
bytes.advance(properties_len_len);
if properties_len == 0 {
return Ok(None);
}
let mut session_expiry_interval = None;
let mut reason_string = None;
let mut user_properties = Vec::new();
let mut server_reference = None;
let mut cursor = 0;
while cursor < properties_len {
let prop = read_u8(bytes)?;
cursor += 1;
match property(prop)? {
PropertyType::SessionExpiryInterval => {
session_expiry_interval = Some(read_u32(bytes)?);
cursor += 4;
}
PropertyType::ReasonString => {
let reason = read_mqtt_string(bytes)?;
cursor += 2 + reason.len();
reason_string = Some(reason);
}
PropertyType::UserProperty => {
let key = read_mqtt_string(bytes)?;
let value = read_mqtt_string(bytes)?;
cursor += 2 + key.len() + 2 + value.len();
user_properties.push((key, value));
}
PropertyType::ServerReference => {
let reference = read_mqtt_string(bytes)?;
cursor += 2 + reference.len();
server_reference = Some(reference);
}
_ => return Err(Error::InvalidPropertyType(prop)),
}
}
let properties = Self {
session_expiry_interval,
reason_string,
user_properties,
server_reference,
};
Ok(Some(properties))
}
fn write(&self, buffer: &mut BytesMut) -> Result<(), Error> {
let length = self.len();
write_remaining_length(buffer, length)?;
if let Some(session_expiry_interval) = self.session_expiry_interval {
buffer.put_u8(PropertyType::SessionExpiryInterval as u8);
buffer.put_u32(session_expiry_interval);
}
if let Some(reason) = &self.reason_string {
buffer.put_u8(PropertyType::ReasonString as u8);
write_mqtt_string(buffer, reason)?;
}
for (key, value) in &self.user_properties {
buffer.put_u8(PropertyType::UserProperty as u8);
write_mqtt_string(buffer, key)?;
write_mqtt_string(buffer, value)?;
}
if let Some(reference) = &self.server_reference {
buffer.put_u8(PropertyType::ServerReference as u8);
write_mqtt_string(buffer, reference)?;
}
Ok(())
}
}
impl Disconnect {
#[must_use]
pub fn new(reason: DisconnectReasonCode) -> Self {
Self {
reason_code: reason,
properties: None,
}
}
fn len(&self) -> usize {
if self.reason_code == DisconnectReasonCode::NormalDisconnection
&& self.properties.is_none()
{
return 2; }
let mut length = 0;
if let Some(properties) = &self.properties {
length += 1; let properties_len = properties.len();
let properties_len_len = len_len(properties_len);
length += properties_len_len + properties_len;
} else {
length += 1;
}
length
}
#[must_use]
pub fn size(&self) -> usize {
let len = self.len();
if len == 2 {
return len;
}
let remaining_len_size = len_len(len);
1 + remaining_len_size + len
}
pub fn read(fixed_header: FixedHeader, mut bytes: Bytes) -> Result<Self, Error> {
let packet_type = fixed_header.byte1 >> 4;
let flags = fixed_header.byte1 & 0b0000_1111;
bytes.advance(fixed_header.fixed_header_len);
if packet_type != PacketType::Disconnect as u8 {
return Err(Error::InvalidPacketType(packet_type));
};
if flags != 0x00 {
return Err(Error::MalformedPacket);
};
if fixed_header.remaining_len == 0 {
return Ok(Self::new(DisconnectReasonCode::NormalDisconnection));
}
let reason_code = read_u8(&mut bytes)?;
let disconnect = Self {
reason_code: reason_code.try_into()?,
properties: DisconnectProperties::extract(&mut bytes)?,
};
Ok(disconnect)
}
pub fn write(&self, buffer: &mut BytesMut) -> Result<usize, Error> {
buffer.put_u8(0xE0);
let length = self.len();
if length == 2 {
buffer.put_u8(0x00);
return Ok(length);
}
let len_len = write_remaining_length(buffer, length)?;
buffer.put_u8(self.reason_code as u8);
if let Some(properties) = &self.properties {
properties.write(buffer)?;
} else {
write_remaining_length(buffer, 0)?;
}
Ok(1 + len_len + length)
}
}
#[cfg(test)]
mod test {
use super::{Disconnect, DisconnectProperties, DisconnectReasonCode};
use crate::parse_fixed_header;
use crate::test::read_write_packets;
use crate::Packet;
use bytes::BytesMut;
#[test]
fn disconnect1_parsing_works() {
let mut buffer = bytes::BytesMut::new();
let packet_bytes = [
0xE0, 0x00, ];
let expected = Disconnect::new(DisconnectReasonCode::NormalDisconnection);
buffer.extend_from_slice(&packet_bytes[..]);
let fixed_header = parse_fixed_header(buffer.iter()).unwrap();
let disconnect_bytes = buffer.split_to(fixed_header.frame_length()).freeze();
let disconnect = Disconnect::read(fixed_header, disconnect_bytes).unwrap();
assert_eq!(disconnect, expected);
}
#[test]
fn disconnect1_encoding_works() {
let mut buffer = BytesMut::new();
let disconnect = Disconnect::new(DisconnectReasonCode::NormalDisconnection);
let expected = [
0xE0, 0x00, ];
disconnect.write(&mut buffer).unwrap();
assert_eq!(&buffer[..], &expected);
}
fn sample2() -> Disconnect {
let properties = DisconnectProperties {
session_expiry_interval: Some(1234),
reason_string: Some("test".into()),
user_properties: vec![("test".into(), "test".into())],
server_reference: Some("test".into()),
};
Disconnect {
reason_code: DisconnectReasonCode::UnspecifiedError,
properties: Some(properties),
}
}
fn sample_bytes2() -> Vec<u8> {
vec![
0xE0, 0x22, 0x80, 0x20, 0x11, 0x00, 0x00, 0x04, 0xd2, 0x1F, 0x00, 0x04, 0x74, 0x65, 0x73, 0x74, 0x26, 0x00, 0x04, 0x74, 0x65, 0x73, 0x74, 0x00, 0x04, 0x74, 0x65, 0x73,
0x74, 0x1C, 0x00, 0x04, 0x74, 0x65, 0x73, 0x74, ]
}
#[test]
fn disconnect2_parsing_works() {
let mut buffer = bytes::BytesMut::new();
let packet_bytes = sample_bytes2();
let expected = sample2();
buffer.extend_from_slice(&packet_bytes[..]);
let fixed_header = parse_fixed_header(buffer.iter()).unwrap();
let disconnect_bytes = buffer.split_to(fixed_header.frame_length()).freeze();
let disconnect = Disconnect::read(fixed_header, disconnect_bytes).unwrap();
assert_eq!(disconnect, expected);
}
#[test]
fn disconnect2_encoding_works() {
let mut buffer = BytesMut::new();
let disconnect = sample2();
let expected = sample_bytes2();
disconnect.write(&mut buffer).unwrap();
assert_eq!(&buffer[..], &expected);
}
use super::super::test::{USER_PROP_KEY, USER_PROP_VAL};
use pretty_assertions::assert_eq;
#[test]
fn length_calculation() {
let mut dummy_bytes = BytesMut::new();
let disconn_props = DisconnectProperties {
session_expiry_interval: None,
reason_string: None,
user_properties: vec![(USER_PROP_KEY.into(), USER_PROP_VAL.into())],
server_reference: None,
};
let mut disconn_pkt = Disconnect::new(DisconnectReasonCode::NormalDisconnection);
disconn_pkt.properties = Some(disconn_props);
let size_from_size = disconn_pkt.size();
let size_from_write = disconn_pkt.write(&mut dummy_bytes).unwrap();
let size_from_bytes = dummy_bytes.len();
assert_eq!(size_from_write, size_from_bytes);
assert_eq!(size_from_size, size_from_bytes);
}
#[test]
fn test_write_read() {
read_write_packets(write_read_provider());
}
fn write_read_provider() -> Vec<Packet> {
vec![
Packet::Disconnect(Disconnect::new(DisconnectReasonCode::NormalDisconnection)),
Packet::Disconnect(Disconnect {
reason_code: DisconnectReasonCode::UnspecifiedError,
properties: Some(DisconnectProperties {
session_expiry_interval: Some(1234),
reason_string: Some("test".into()),
user_properties: vec![("test".into(), "test".into())],
server_reference: Some("test".into()),
}),
}),
]
}
}