diff --git a/src/client/client_config.rs b/src/client/client_config.rs index 88382ab..eaa02f6 100644 --- a/src/client/client_config.rs +++ b/src/client/client_config.rs @@ -22,7 +22,6 @@ * SOFTWARE. */ - use crate::packet::publish_packet::QualityOfService; use crate::utils::types::{BinaryData, EncodedString}; @@ -31,7 +30,7 @@ pub struct ClientConfig<'a> { pub username_flag: bool, pub username: EncodedString<'a>, pub password_flag: bool, - pub password: BinaryData<'a> + pub password: BinaryData<'a>, } impl ClientConfig<'a> { @@ -45,11 +44,11 @@ impl ClientConfig<'a> { } } - pub fn add_qos(& mut self, qos: QualityOfService) { + pub fn add_qos(&mut self, qos: QualityOfService) { self.qos = qos; } - pub fn add_username(& mut self, username: &'a str) { + pub fn add_username(&mut self, username: &'a str) { let mut username_s: EncodedString = EncodedString::new(); username_s.string = username; username_s.len = username.len() as u16; @@ -57,11 +56,11 @@ impl ClientConfig<'a> { self.username = username_s; } - pub fn add_password(& mut self, password: &'a str) { + pub fn add_password(&mut self, password: &'a str) { let mut password_s: BinaryData = BinaryData::new(); password_s.bin = password.as_bytes(); password_s.len = password_s.bin.len() as u16; self.password = password_s; self.password_flag = true; } -} \ No newline at end of file +} diff --git a/src/client/client_v5.rs b/src/client/client_v5.rs index 7c72640..acef2b3 100644 --- a/src/client/client_v5.rs +++ b/src/client/client_v5.rs @@ -1,19 +1,22 @@ -use rand_core::RngCore; use crate::client::client_config::ClientConfig; -use crate::network::network_trait::{Network}; +use crate::network::network_trait::Network; use crate::packet::connack_packet::ConnackPacket; use crate::packet::connect_packet::ConnectPacket; use crate::packet::disconnect_packet::DisconnectPacket; use crate::packet::mqtt_packet::Packet; +use crate::packet::pingreq_packet::PingreqPacket; +use crate::packet::pingresp_packet::PingrespPacket; use crate::packet::puback_packet::PubackPacket; -use crate::packet::publish_packet::{PublishPacket, QualityOfService}; use crate::packet::publish_packet::QualityOfService::QoS1; +use crate::packet::publish_packet::{PublishPacket, QualityOfService}; use crate::packet::reason_codes::ReasonCode; use crate::packet::suback_packet::SubackPacket; use crate::packet::subscription_packet::SubscriptionPacket; use crate::utils::buffer_reader::BuffReader; use crate::utils::rng_generator::CountingRng; use crate::utils::types::BufferError; +use heapless::Vec; +use rand_core::RngCore; pub struct MqttClientV5<'a, T, const MAX_PROPERTIES: usize> { network_driver: &'a mut T, @@ -29,7 +32,14 @@ impl<'a, T, const MAX_PROPERTIES: usize> MqttClientV5<'a, T, MAX_PROPERTIES> where T: Network, { - pub fn new(network_driver: &'a mut T, buffer: &'a mut [u8], buffer_len: usize, recv_buffer: &'a mut [u8], recv_buffer_len: usize, config: ClientConfig<'a>) -> Self { + pub fn new( + network_driver: &'a mut T, + buffer: &'a mut [u8], + buffer_len: usize, + recv_buffer: &'a mut [u8], + recv_buffer_len: usize, + config: ClientConfig<'a>, + ) -> Self { Self { network_driver, buffer, @@ -37,7 +47,7 @@ where recv_buffer, recv_buffer_len, rng: CountingRng(50), - config + config, } } @@ -45,10 +55,10 @@ where let len = { let mut connect = ConnectPacket::<'b, 3, 0>::clean(); if self.config.username_flag { - connect.add_username(& self.config.username); + connect.add_username(&self.config.username); } if self.config.password_flag { - connect.add_password(& self.config.password) + connect.add_password(&self.config.password) } connect.encode(self.buffer, self.buffer_len) }; @@ -57,7 +67,7 @@ where log::error!("[DECODE ERR]: {}", err); return Err(ReasonCode::BuffError); } - self.network_driver.send(self.buffer, len.unwrap()).await ?; + self.network_driver.send(self.buffer, len.unwrap()).await?; //connack let reason: Result = { @@ -80,7 +90,6 @@ where } else { Ok(()) } - } pub async fn disconnect<'b>(&'b mut self) -> Result<(), ReasonCode> { @@ -90,7 +99,7 @@ where log::error!("[DECODE ERR]: {}", err); return Err(ReasonCode::BuffError); } - self.network_driver.send(self.buffer, len.unwrap()).await ?; + self.network_driver.send(self.buffer, len.unwrap()).await?; Ok(()) } @@ -99,7 +108,6 @@ where topic_name: &'b str, message: &'b str, ) -> Result<(), ReasonCode> { - let identifier: u16 = self.rng.next_u32() as u16; let len = { let mut packet = PublishPacket::<'b, 5>::new(); @@ -115,15 +123,17 @@ where return Err(ReasonCode::BuffError); } - self.network_driver.send(self.buffer, len.unwrap()).await ?; - + self.network_driver.send(self.buffer, len.unwrap()).await?; //QoS1 - if >::into(self.config.qos ) == >::into(QoS1) { + if >::into(self.config.qos) + == >::into(QoS1) + { let reason: Result<[u16; 2], BufferError> = { - self.network_driver.receive(self.buffer).await ?; + self.network_driver.receive(self.buffer).await?; let mut packet = PubackPacket::<'b, 5>::new(); - if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len)) { + if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len)) + { Err(err) } else { Ok([packet.packet_identifier, packet.reason_code as u16]) @@ -147,9 +157,63 @@ where Ok(()) } - // TODO - multiple topic subscribe func + pub async fn subscribe_to_topics<'b, const TOPICS: usize>( + &'b mut self, + topic_names: &'b Vec<&'b str, TOPICS>, + ) -> Result<(), ReasonCode> { + let len = { + let mut subs = SubscriptionPacket::<'b, TOPICS, 1>::new(); + let mut i = 0; + loop { + if i == TOPICS { + break; + } + subs.add_new_filter(topic_names.get(i).unwrap(), self.config.qos); + i = i + 1; + } + subs.encode(self.buffer, self.buffer_len) + }; - pub async fn subscribe_to_topic<'b>(&'b mut self, topic_name: &'b str) -> Result<(), ReasonCode> { + if let Err(err) = len { + log::error!("[DECODE ERR]: {}", err); + return Err(ReasonCode::BuffError); + } + + self.network_driver.send(self.buffer, len.unwrap()).await?; + + let reason: Result, BufferError> = { + self.network_driver.receive(self.buffer).await?; + + let mut packet = SubackPacket::<'b, TOPICS, 5>::new(); + if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len)) { + Err(err) + } else { + Ok(packet.reason_codes) + } + }; + + if let Err(err) = reason { + log::error!("[DECODE ERR]: {}", err); + return Err(ReasonCode::BuffError); + } + let reasons = reason.unwrap(); + let mut i = 0; + loop { + if i == TOPICS { + break; + } + if *reasons.get(i).unwrap() != self.config.qos.into() { + return Err(ReasonCode::from(*reasons.get(i).unwrap())); + } + i = i + 1; + } + Ok(()) + } + + pub async fn subscribe_to_topic<'b>( + &'b mut self, + topic_name: &'b str, + ) -> Result<(), ReasonCode> { let len = { let mut subs = SubscriptionPacket::<'b, 1, 1>::new(); subs.add_new_filter(topic_name, self.config.qos); @@ -161,10 +225,10 @@ where return Err(ReasonCode::BuffError); } - self.network_driver.send(self.buffer, len.unwrap()).await ?; + self.network_driver.send(self.buffer, len.unwrap()).await?; let reason: Result = { - self.network_driver.receive(self.buffer).await ?; + self.network_driver.receive(self.buffer).await?; let mut packet = SubackPacket::<'b, 5, 5>::new(); if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len)) { @@ -185,18 +249,21 @@ where } else { Ok(()) } - } pub async fn receive_message<'b>(&'b mut self) -> Result<&'b [u8], ReasonCode> { - self.network_driver.receive(self.recv_buffer).await ?; + self.network_driver.receive(self.recv_buffer).await?; let mut packet = PublishPacket::<'b, 5>::new(); - if let Err(err) = packet.decode(&mut BuffReader::new(self.recv_buffer, self.recv_buffer_len)) { + if let Err(err) = + packet.decode(&mut BuffReader::new(self.recv_buffer, self.recv_buffer_len)) + { log::error!("[DECODE ERR]: {}", err); return Err(ReasonCode::BuffError); } - if (packet.fixed_header & 0x06) == >::into(QualityOfService::QoS1) { + if (packet.fixed_header & 0x06) + == >::into(QualityOfService::QoS1) + { let mut puback = PubackPacket::<'b, 5>::new(); puback.packet_identifier = packet.packet_identifier; puback.reason_code = 0x00; @@ -206,10 +273,33 @@ where log::error!("[DECODE ERR]: {}", err); return Err(ReasonCode::BuffError); } - self.network_driver.send(self.buffer, len.unwrap()).await ?; + self.network_driver.send(self.buffer, len.unwrap()).await?; } } return Ok(packet.message.unwrap()); } + + pub async fn send_ping<'b>(&'b mut self) -> Result<(), ReasonCode> { + let len = { + let mut packet = PingreqPacket::new(); + packet.encode(self.buffer, self.buffer_len) + }; + + if let Err(err) = len { + log::error!("[DECODE ERR]: {}", err); + return Err(ReasonCode::BuffError); + } + + self.network_driver.send(self.buffer, len.unwrap()).await?; + + self.network_driver.receive(self.buffer).await?; + let mut packet = PingrespPacket::new(); + if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len)) { + log::error!("[DECODE ERR]: {}", err); + return Err(ReasonCode::BuffError); + } else { + Ok(()) + } + } } diff --git a/src/client/mod.rs b/src/client/mod.rs index 8574d8d..6e6b344 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -22,6 +22,5 @@ * SOFTWARE. */ - -pub mod client_v5; pub mod client_config; +pub mod client_v5; diff --git a/src/encoding/mod.rs b/src/encoding/mod.rs index 0324c19..caef9ca 100644 --- a/src/encoding/mod.rs +++ b/src/encoding/mod.rs @@ -22,5 +22,4 @@ * SOFTWARE. */ - pub mod variable_byte_integer; diff --git a/src/main.rs b/src/main.rs index a6ee5b3..80c57aa 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,15 +1,17 @@ use std::time::Duration; -use tokio::{join, task}; +use heapless::Vec; + use tokio::time::sleep; +use tokio::{join, task}; use rust_mqtt::client::client_config::ClientConfig; use rust_mqtt::client::client_v5::MqttClientV5; use rust_mqtt::network::network_trait::{Network, NetworkError}; use rust_mqtt::packet::connect_packet::ConnectPacket; use rust_mqtt::packet::mqtt_packet::Packet; -use rust_mqtt::packet::publish_packet::{PublishPacket, QualityOfService}; use rust_mqtt::packet::publish_packet::QualityOfService::QoS1; +use rust_mqtt::packet::publish_packet::{PublishPacket, QualityOfService}; use rust_mqtt::packet::subscription_packet::SubscriptionPacket; use rust_mqtt::tokio_network::TokioNetwork; @@ -21,27 +23,49 @@ async fn receive() { let mut config = ClientConfig::new(); config.add_qos(QualityOfService::QoS1); config.add_username("test"); - config.add_password("testPass1"); + config.add_password("testPass"); let mut res2 = vec![0; 260]; let mut res3 = vec![0; 260]; - let mut client = MqttClientV5::::new(&mut tokio_network, &mut res2, 260, & mut res3, 260, config); + let mut client = MqttClientV5::::new( + &mut tokio_network, + &mut res2, + 260, + &mut res3, + 260, + config, + ); - let mut result = { - client.connect_to_broker().await - }; + let mut result = { client.connect_to_broker().await }; if let Err(r) = result { log::error!("[ERROR]: {}", r); return; } { - client.subscribe_to_topic("test/topic").await; + const TOPICS: usize = 2; + let t1 = "test/topic1"; + let t2 = "test/topic2"; + let mut names = Vec::<&str, TOPICS>::new(); + names.push(&t1); + names.push(&t2); + client.subscribe_to_topics::(&names).await; + //client.subscribe_to_topic("test/topic").await; }; + { + sleep(Duration::from_secs(10)); + client.send_ping().await; + } + let mut o = 0; + loop { + if o == 2 { + break; + } log::info!("Waiting for new message!"); let mes = client.receive_message().await.unwrap(); let x = String::from_utf8_lossy(mes); log::info!("Got new message: {}", x); + o = o + 1; } { client.disconnect().await; @@ -56,22 +80,30 @@ async fn publish(message: &str) { let config = ClientConfig::new(); let mut res2 = vec![0; 260]; let mut res3 = vec![0; 260]; - let mut client = MqttClientV5::::new(&mut tokio_network, &mut res2, 260, & mut res3, 260, config); + let mut client = MqttClientV5::::new( + &mut tokio_network, + &mut res2, + 260, + &mut res3, + 260, + config, + ); let mut result = { client.connect_to_broker().await }; log::info!("Waiting until send!"); sleep(Duration::from_secs(15)); - result= { + result = { log::info!("Sending new message!"); - client - .send_message("test/topic", message) - .await + /*client + .send_message("test/topic", message) + .await*/ + client.send_ping().await }; if let Err(e) = result { log::error!("Chyba!"); } - result = { + /*result = { log::info!("Sending new message!"); client .send_message("test/topic", "Dalsi zprava :)") @@ -79,7 +111,7 @@ async fn publish(message: &str) { }; if let Err(err) = result { log::error!("Chyba!"); - } + }*/ { client.disconnect().await; @@ -103,6 +135,7 @@ async fn main() { join!(recv, publ);*/ receive().await; + //publish("Ahoj 123").await; log::info!("Done"); } diff --git a/src/network/mod.rs b/src/network/mod.rs index fb22dd6..d1f654d 100644 --- a/src/network/mod.rs +++ b/src/network/mod.rs @@ -22,5 +22,4 @@ * SOFTWARE. */ - pub mod network_trait; diff --git a/src/packet/auth_packet.rs b/src/packet/auth_packet.rs index 9adf587..d45d9bf 100644 --- a/src/packet/auth_packet.rs +++ b/src/packet/auth_packet.rs @@ -22,7 +22,6 @@ * SOFTWARE. */ - use heapless::Vec; use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; @@ -69,7 +68,7 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for AuthPacket<'a, MAX_PROPERTI remain_len: 0, auth_reason: 0x00, property_len: 0, - properties: Vec::, MAX_PROPERTIES>::new() + properties: Vec::, MAX_PROPERTIES>::new(), } } @@ -77,23 +76,22 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for AuthPacket<'a, MAX_PROPERTI let mut buff_writer = BuffWriter::new(buffer, buff_len); let mut rm_ln = self.property_len; - let property_len_enc: [u8; 4] = - VariableByteIntegerEncoder::encode(self.property_len) ?; + let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len)?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); rm_ln = rm_ln + property_len_len as u32; rm_ln = rm_ln + 1; - buff_writer.write_u8(self.fixed_header) ?; - buff_writer.write_variable_byte_int(rm_ln) ?; - buff_writer.write_u8(self.auth_reason) ?; - buff_writer.write_variable_byte_int(self.property_len) ?; - buff_writer.encode_properties::(&self.properties) ?; + buff_writer.write_u8(self.fixed_header)?; + buff_writer.write_variable_byte_int(rm_ln)?; + buff_writer.write_u8(self.auth_reason)?; + buff_writer.write_variable_byte_int(self.property_len)?; + buff_writer.encode_properties::(&self.properties)?; Ok(buff_writer.position) } fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { - self.decode_fixed_header(buff_reader) ?; - self.auth_reason = buff_reader.read_u8() ?; + self.decode_fixed_header(buff_reader)?; + self.auth_reason = buff_reader.read_u8()?; return self.decode_properties(buff_reader); } diff --git a/src/packet/connack_packet.rs b/src/packet/connack_packet.rs index 8fea3f4..340984e 100644 --- a/src/packet/connack_packet.rs +++ b/src/packet/connack_packet.rs @@ -43,13 +43,16 @@ pub struct ConnackPacket<'a, const MAX_PROPERTIES: usize> { } impl<'a, const MAX_PROPERTIES: usize> ConnackPacket<'a, MAX_PROPERTIES> { - pub fn decode_connack_packet(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { - if self.decode_fixed_header(buff_reader) ? != (PacketType::Connack).into() { + pub fn decode_connack_packet( + &mut self, + buff_reader: &mut BuffReader<'a>, + ) -> Result<(), BufferError> { + if self.decode_fixed_header(buff_reader)? != (PacketType::Connack).into() { log::error!("Packet you are trying to decode is not CONNACK packet!"); return Err(BufferError::PacketTypeMismatch); } - self.ack_flags = buff_reader.read_u8() ?; - self.connect_reason_code = buff_reader.read_u8() ?; + self.ack_flags = buff_reader.read_u8()?; + self.connect_reason_code = buff_reader.read_u8()?; self.decode_properties(buff_reader) } } @@ -68,16 +71,16 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for ConnackPacket<'a, MAX_PROPE fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { let mut buff_writer = BuffWriter::new(buffer, buffer_len); - buff_writer.write_u8(self.fixed_header) ?; - let property_len_enc = VariableByteIntegerEncoder::encode(self.property_len) ?; + buff_writer.write_u8(self.fixed_header)?; + let property_len_enc = VariableByteIntegerEncoder::encode(self.property_len)?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); let rm_len: u32 = 2 + self.property_len + property_len_len as u32; - buff_writer.write_variable_byte_int(rm_len) ?; - buff_writer.write_u8(self.ack_flags) ?; - buff_writer.write_u8(self.connect_reason_code) ?; - buff_writer.write_variable_byte_int(self.property_len) ?; - buff_writer.encode_properties(&self.properties) ?; + buff_writer.write_variable_byte_int(rm_len)?; + buff_writer.write_u8(self.ack_flags)?; + buff_writer.write_u8(self.connect_reason_code)?; + buff_writer.write_variable_byte_int(self.property_len)?; + buff_writer.encode_properties(&self.properties)?; Ok(buff_writer.position) } diff --git a/src/packet/connect_packet.rs b/src/packet/connect_packet.rs index 649d501..06106f0 100644 --- a/src/packet/connect_packet.rs +++ b/src/packet/connect_packet.rs @@ -128,15 +128,13 @@ impl<'a, const MAX_PROPERTIES: usize, const MAX_WILL_PROPERTIES: usize> Packet<' let mut buff_writer = BuffWriter::new(buffer, buffer_len); let mut rm_ln = self.property_len; - let property_len_enc: [u8; 4] = - VariableByteIntegerEncoder::encode(self.property_len) ?; + let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len)?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); // Number 12 => protocol_name_len + protocol_name + protocol_version + connect_flags + keep_alive + client_id_len rm_ln = rm_ln + property_len_len as u32 + 12; if self.connect_flags & 0x04 != 0 { - let wil_prop_len_enc = - VariableByteIntegerEncoder::encode(self.will_property_len) ?; + let wil_prop_len_enc = VariableByteIntegerEncoder::encode(self.will_property_len)?; let wil_prop_len_len = VariableByteIntegerEncoder::len(wil_prop_len_enc); rm_ln = rm_ln + wil_prop_len_len as u32 @@ -152,31 +150,31 @@ impl<'a, const MAX_PROPERTIES: usize, const MAX_WILL_PROPERTIES: usize> Packet<' rm_ln = rm_ln + self.password.len as u32 + 2; } - buff_writer.write_u8(self.fixed_header) ?; - buff_writer.write_variable_byte_int(rm_ln) ?; + buff_writer.write_u8(self.fixed_header)?; + buff_writer.write_variable_byte_int(rm_ln)?; - buff_writer.write_u16(self.protocol_name_len) ?; - buff_writer.write_u32(self.protocol_name) ?; - buff_writer.write_u8(self.protocol_version) ?; - buff_writer.write_u8(self.connect_flags) ?; - buff_writer.write_u16(self.keep_alive) ?; - buff_writer.write_variable_byte_int(self.property_len) ?; - buff_writer.encode_properties::(&self.properties) ?; - buff_writer.write_string_ref(&self.client_id) ?; + buff_writer.write_u16(self.protocol_name_len)?; + buff_writer.write_u32(self.protocol_name)?; + buff_writer.write_u8(self.protocol_version)?; + buff_writer.write_u8(self.connect_flags)?; + buff_writer.write_u16(self.keep_alive)?; + buff_writer.write_variable_byte_int(self.property_len)?; + buff_writer.encode_properties::(&self.properties)?; + buff_writer.write_string_ref(&self.client_id)?; if self.connect_flags & 0x04 != 0 { - buff_writer.write_variable_byte_int(self.will_property_len) ?; - buff_writer.encode_properties(&self.will_properties) ?; - buff_writer.write_string_ref(&self.will_topic) ?; - buff_writer.write_binary_ref(&self.will_payload) ?; + buff_writer.write_variable_byte_int(self.will_property_len)?; + buff_writer.encode_properties(&self.will_properties)?; + buff_writer.write_string_ref(&self.will_topic)?; + buff_writer.write_binary_ref(&self.will_payload)?; } if self.connect_flags & 0x80 != 0 { - buff_writer.write_string_ref(&self.username) ?; + buff_writer.write_string_ref(&self.username)?; } if self.connect_flags & 0x40 != 0 { - buff_writer.write_binary_ref(&self.password) ?; + buff_writer.write_binary_ref(&self.password)?; } Ok(buff_writer.position) diff --git a/src/packet/disconnect_packet.rs b/src/packet/disconnect_packet.rs index 4ff1613..b93675a 100644 --- a/src/packet/disconnect_packet.rs +++ b/src/packet/disconnect_packet.rs @@ -47,7 +47,6 @@ pub struct DisconnectPacket<'a, const MAX_PROPERTIES: usize> { } impl<'a, const MAX_PROPERTIES: usize> DisconnectPacket<'a, MAX_PROPERTIES> { - fn add_reason(&mut self, reason: u8) { self.disconnect_reason = reason; } @@ -66,24 +65,24 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for DisconnectPacket<'a, MAX_PR fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { let mut buff_writer = BuffWriter::new(buffer, buffer_len); - buff_writer.write_u8(self.fixed_header) ?; - let property_len_enc = VariableByteIntegerEncoder::encode(self.property_len) ?; + buff_writer.write_u8(self.fixed_header)?; + let property_len_enc = VariableByteIntegerEncoder::encode(self.property_len)?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); let rm_len: u32 = 1 + self.property_len + property_len_len as u32; - buff_writer.write_variable_byte_int(rm_len) ?; - buff_writer.write_u8(self.disconnect_reason) ?; - buff_writer.write_variable_byte_int(self.property_len) ?; - buff_writer.encode_properties(&self.properties) ?; + buff_writer.write_variable_byte_int(rm_len)?; + buff_writer.write_u8(self.disconnect_reason)?; + buff_writer.write_variable_byte_int(self.property_len)?; + buff_writer.encode_properties(&self.properties)?; Ok(buff_writer.position) } fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { - if self.decode_fixed_header(buff_reader) ? != (PacketType::Pingresp).into() { + if self.decode_fixed_header(buff_reader)? != (PacketType::Pingresp).into() { log::error!("Packet you are trying to decode is not PUBACK packet!"); return Err(BufferError::WrongPacketToDecode); } - self.disconnect_reason = buff_reader.read_u8() ?; + self.disconnect_reason = buff_reader.read_u8()?; return self.decode_properties(buff_reader); } diff --git a/src/packet/mod.rs b/src/packet/mod.rs index 5ec0c97..486fe54 100644 --- a/src/packet/mod.rs +++ b/src/packet/mod.rs @@ -39,6 +39,6 @@ pub mod connect_packet; pub mod disconnect_packet; pub mod pingreq_packet; pub mod pingresp_packet; +pub mod reason_codes; pub mod suback_packet; pub mod unsuback_packet; -pub mod reason_codes; diff --git a/src/packet/mqtt_packet.rs b/src/packet/mqtt_packet.rs index c8130b3..b95cf41 100644 --- a/src/packet/mqtt_packet.rs +++ b/src/packet/mqtt_packet.rs @@ -57,7 +57,7 @@ pub trait Packet<'a> { let mut prop: Property; if self.get_property_len() != 0 { loop { - prop = Property::decode(buff_reader) ?; + prop = Property::decode(buff_reader)?; log::debug!("Parsed property {:?}", prop); x = x + prop.len() as u32 + 1; self.push_to_properties(prop); @@ -71,10 +71,13 @@ pub trait Packet<'a> { } /// Method is decoding packet header into fixed header part and remaining length - fn decode_fixed_header(&mut self, buff_reader: &mut BuffReader) -> Result { + fn decode_fixed_header( + &mut self, + buff_reader: &mut BuffReader, + ) -> Result { let first_byte: u8 = buff_reader.read_u8()?; self.set_fixed_header(first_byte); - self.set_remaining_len(buff_reader.read_variable_byte_int() ?); + self.set_remaining_len(buff_reader.read_variable_byte_int()?); return Ok(PacketType::from(first_byte)); } } diff --git a/src/packet/pingreq_packet.rs b/src/packet/pingreq_packet.rs index 652be7f..b1e2c37 100644 --- a/src/packet/pingreq_packet.rs +++ b/src/packet/pingreq_packet.rs @@ -41,14 +41,14 @@ impl<'a> Packet<'a> for PingreqPacket { fn new() -> Self { Self { fixed_header: PacketType::Pingreq.into(), - remain_len: 0 + remain_len: 0, } } fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { let mut buff_writer = BuffWriter::new(buffer, buffer_len); - buff_writer.write_u8(self.fixed_header) ?; - buff_writer.write_variable_byte_int(0 as u32) ?; + buff_writer.write_u8(self.fixed_header)?; + buff_writer.write_variable_byte_int(0 as u32)?; Ok(buff_writer.position) } diff --git a/src/packet/pingresp_packet.rs b/src/packet/pingresp_packet.rs index dec615d..d557cdd 100644 --- a/src/packet/pingresp_packet.rs +++ b/src/packet/pingresp_packet.rs @@ -35,26 +35,26 @@ pub struct PingrespPacket { pub remain_len: u32, } -impl<'a> PingrespPacket { -} +impl<'a> PingrespPacket {} impl<'a> Packet<'a> for PingrespPacket { fn new() -> Self { Self { fixed_header: PacketType::Pingresp.into(), - remain_len: 0 + remain_len: 0, } } fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { let mut buff_writer = BuffWriter::new(buffer, buffer_len); - buff_writer.write_u8(self.fixed_header) ?; - buff_writer.write_variable_byte_int(0 as u32) ?; + buff_writer.write_u8(self.fixed_header)?; + buff_writer.write_variable_byte_int(0 as u32)?; Ok(buff_writer.position) } fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { - if self.decode_fixed_header(buff_reader) ? != (PacketType::Pingresp).into() { + let x = self.decode_fixed_header(buff_reader)?; + if x != (PacketType::Pingresp).into() { log::error!("Packet you are trying to decode is not PINGRESP packet!"); return Err(BufferError::PacketTypeMismatch); } diff --git a/src/packet/puback_packet.rs b/src/packet/puback_packet.rs index 3bbc97d..4577990 100644 --- a/src/packet/puback_packet.rs +++ b/src/packet/puback_packet.rs @@ -42,9 +42,7 @@ pub struct PubackPacket<'a, const MAX_PROPERTIES: usize> { pub properties: Vec, MAX_PROPERTIES>, } -impl<'a, const MAX_PROPERTIES: usize> PubackPacket<'a, MAX_PROPERTIES> { - -} +impl<'a, const MAX_PROPERTIES: usize> PubackPacket<'a, MAX_PROPERTIES> {} impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubackPacket<'a, MAX_PROPERTIES> { fn new() -> Self { @@ -54,7 +52,7 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubackPacket<'a, MAX_PROPER packet_identifier: 0, reason_code: 0, property_len: 0, - properties: Vec::, MAX_PROPERTIES>::new() + properties: Vec::, MAX_PROPERTIES>::new(), } } @@ -62,33 +60,32 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubackPacket<'a, MAX_PROPER let mut buff_writer = BuffWriter::new(buffer, buffer_len); let mut rm_ln = self.property_len; - let property_len_enc: [u8; 4] = - VariableByteIntegerEncoder::encode(self.property_len) ?; + let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len)?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); rm_ln = rm_ln + property_len_len as u32 + 3; - buff_writer.write_u8(self.fixed_header) ?; - buff_writer.write_variable_byte_int(rm_ln) ?; - buff_writer.write_u16(self.packet_identifier) ?; - buff_writer.write_u8(self.reason_code) ?; - buff_writer.write_variable_byte_int(self.property_len) ?; - buff_writer.encode_properties::(&self.properties) ?; + buff_writer.write_u8(self.fixed_header)?; + buff_writer.write_variable_byte_int(rm_ln)?; + buff_writer.write_u16(self.packet_identifier)?; + buff_writer.write_u8(self.reason_code)?; + buff_writer.write_variable_byte_int(self.property_len)?; + buff_writer.encode_properties::(&self.properties)?; Ok(buff_writer.position) } fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { - if self.decode_fixed_header(buff_reader) ? != (PacketType::Puback).into() { + if self.decode_fixed_header(buff_reader)? != (PacketType::Puback).into() { log::error!("Packet you are trying to decode is not PUBACK packet!"); return Err(BufferError::PacketTypeMismatch); } - self.packet_identifier = buff_reader.read_u16() ?; + self.packet_identifier = buff_reader.read_u16()?; if self.remain_len != 2 { - self.reason_code = buff_reader.read_u8() ?; + self.reason_code = buff_reader.read_u8()?; } if self.remain_len < 4 { self.property_len = 0; } else { - self.decode_properties(buff_reader) ?; + self.decode_properties(buff_reader)?; } Ok(()) } diff --git a/src/packet/pubcomp_packet.rs b/src/packet/pubcomp_packet.rs index 9d65374..f51b297 100644 --- a/src/packet/pubcomp_packet.rs +++ b/src/packet/pubcomp_packet.rs @@ -42,8 +42,7 @@ pub struct PubcompPacket<'a, const MAX_PROPERTIES: usize> { pub properties: Vec, MAX_PROPERTIES>, } -impl<'a, const MAX_PROPERTIES: usize> PubcompPacket<'a, MAX_PROPERTIES> { -} +impl<'a, const MAX_PROPERTIES: usize> PubcompPacket<'a, MAX_PROPERTIES> {} impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubcompPacket<'a, MAX_PROPERTIES> { fn new() -> Self { @@ -53,7 +52,7 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubcompPacket<'a, MAX_PROPE packet_identifier: 0, reason_code: 0, property_len: 0, - properties: Vec::, MAX_PROPERTIES>::new() + properties: Vec::, MAX_PROPERTIES>::new(), } } @@ -61,28 +60,27 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubcompPacket<'a, MAX_PROPE let mut buff_writer = BuffWriter::new(buffer, buffer_len); let mut rm_ln = self.property_len; - let property_len_enc: [u8; 4] = - VariableByteIntegerEncoder::encode(self.property_len) ?; + let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len)?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); rm_ln = rm_ln + property_len_len as u32 + 3; - buff_writer.write_u8(self.fixed_header) ?; - buff_writer.write_variable_byte_int(rm_ln) ?; - buff_writer.write_u16(self.packet_identifier) ?; - buff_writer.write_u8(self.reason_code) ?; - buff_writer.write_variable_byte_int(self.property_len) ?; - buff_writer.encode_properties::(&self.properties) ?; + buff_writer.write_u8(self.fixed_header)?; + buff_writer.write_variable_byte_int(rm_ln)?; + buff_writer.write_u16(self.packet_identifier)?; + buff_writer.write_u8(self.reason_code)?; + buff_writer.write_variable_byte_int(self.property_len)?; + buff_writer.encode_properties::(&self.properties)?; Ok(buff_writer.position) } fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { - if self.decode_fixed_header(buff_reader) ? != (PacketType::Pubcomp).into() { + if self.decode_fixed_header(buff_reader)? != (PacketType::Pubcomp).into() { log::error!("Packet you are trying to decode is not PUBCOMP packet!"); return Err(BufferError::PacketTypeMismatch); } - self.packet_identifier = buff_reader.read_u16() ?; - self.reason_code = buff_reader.read_u8() ?; - self.decode_properties(buff_reader) ?; + self.packet_identifier = buff_reader.read_u16()?; + self.reason_code = buff_reader.read_u8()?; + self.decode_properties(buff_reader)?; Ok(()) } diff --git a/src/packet/publish_packet.rs b/src/packet/publish_packet.rs index cfc9c37..648ef17 100644 --- a/src/packet/publish_packet.rs +++ b/src/packet/publish_packet.rs @@ -26,7 +26,7 @@ use heapless::Vec; use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; -use crate::packet::publish_packet::QualityOfService::{INVALID, QoS0, QoS1, QoS2}; +use crate::packet::publish_packet::QualityOfService::{QoS0, QoS1, QoS2, INVALID}; use crate::utils::buffer_reader::BuffReader; use crate::utils::buffer_writer::BuffWriter; use crate::utils::types::{BufferError, EncodedString}; @@ -80,15 +80,15 @@ impl<'a, const MAX_PROPERTIES: usize> PublishPacket<'a, MAX_PROPERTIES> { self.topic_name.len = topic_name.len() as u16; } - pub fn add_message(& mut self, message: &'a [u8]) { + pub fn add_message(&mut self, message: &'a [u8]) { self.message = Some(message); } - pub fn add_qos(& mut self, qos: QualityOfService) { + pub fn add_qos(&mut self, qos: QualityOfService) { self.fixed_header = self.fixed_header | >::into(qos); } - pub fn add_identifier(& mut self, identifier: u16) { + pub fn add_identifier(&mut self, identifier: u16) { self.packet_identifier = identifier; } } @@ -102,7 +102,7 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PublishPacket<'a, MAX_PROPE packet_identifier: 1, property_len: 0, properties: Vec::, MAX_PROPERTIES>::new(), - message: None + message: None, } } @@ -110,45 +110,44 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PublishPacket<'a, MAX_PROPE let mut buff_writer = BuffWriter::new(buffer, buffer_len); let mut rm_ln = self.property_len; - let property_len_enc: [u8; 4] = - VariableByteIntegerEncoder::encode(self.property_len) ?; + let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len)?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); let msg_len = self.message.unwrap().len() as u32; rm_ln = rm_ln + property_len_len as u32 + msg_len + self.topic_name.len as u32 + 2; - buff_writer.write_u8(self.fixed_header) ?; + buff_writer.write_u8(self.fixed_header)?; let qos = self.fixed_header & 0x03; if qos != 0 { rm_ln = rm_ln + 2; } - buff_writer.write_variable_byte_int(rm_ln) ?; - buff_writer.write_string_ref(&self.topic_name) ?; + buff_writer.write_variable_byte_int(rm_ln)?; + buff_writer.write_string_ref(&self.topic_name)?; if qos != 0 { - buff_writer.write_u16(self.packet_identifier) ?; + buff_writer.write_u16(self.packet_identifier)?; } - buff_writer.write_variable_byte_int(self.property_len) ?; - buff_writer.encode_properties::(&self.properties) ?; - buff_writer.insert_ref(msg_len as usize, self.message.unwrap()) ?; + buff_writer.write_variable_byte_int(self.property_len)?; + buff_writer.encode_properties::(&self.properties)?; + buff_writer.insert_ref(msg_len as usize, self.message.unwrap())?; Ok(buff_writer.position) } fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { - if self.decode_fixed_header(buff_reader) ? != (PacketType::Publish).into() { + if self.decode_fixed_header(buff_reader)? != (PacketType::Publish).into() { log::error!("Packet you are trying to decode is not PUBLISH packet!"); return Err(BufferError::PacketTypeMismatch); } - self.topic_name = buff_reader.read_string() ?; + self.topic_name = buff_reader.read_string()?; let qos = self.fixed_header & 0x03; if qos != 0 { // Decode only for QoS 1 / 2 - self.packet_identifier = buff_reader.read_u16() ?; + self.packet_identifier = buff_reader.read_u16()?; } - self.decode_properties(buff_reader) ?; - let mut total_len = VariableByteIntegerEncoder::len( - VariableByteIntegerEncoder::encode(self.remain_len) ?); + self.decode_properties(buff_reader)?; + let mut total_len = + VariableByteIntegerEncoder::len(VariableByteIntegerEncoder::encode(self.remain_len)?); total_len = total_len + 1 + self.remain_len as usize; self.message = Some(buff_reader.read_message(total_len)); Ok(()) diff --git a/src/packet/pubrec_packet.rs b/src/packet/pubrec_packet.rs index f2570c7..10bb601 100644 --- a/src/packet/pubrec_packet.rs +++ b/src/packet/pubrec_packet.rs @@ -42,8 +42,7 @@ pub struct PubrecPacket<'a, const MAX_PROPERTIES: usize> { pub properties: Vec, MAX_PROPERTIES>, } -impl<'a, const MAX_PROPERTIES: usize> PubrecPacket<'a, MAX_PROPERTIES> { -} +impl<'a, const MAX_PROPERTIES: usize> PubrecPacket<'a, MAX_PROPERTIES> {} impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubrecPacket<'a, MAX_PROPERTIES> { fn new() -> Self { @@ -53,7 +52,7 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubrecPacket<'a, MAX_PROPER packet_identifier: 0, reason_code: 0, property_len: 0, - properties: Vec::, MAX_PROPERTIES>::new() + properties: Vec::, MAX_PROPERTIES>::new(), } } @@ -61,27 +60,26 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubrecPacket<'a, MAX_PROPER let mut buff_writer = BuffWriter::new(buffer, buffer_len); let mut rm_ln = self.property_len; - let property_len_enc: [u8; 4] = - VariableByteIntegerEncoder::encode(self.property_len) ?; + let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len)?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); rm_ln = rm_ln + property_len_len as u32 + 3; - buff_writer.write_u8(self.fixed_header) ?; - buff_writer.write_variable_byte_int(rm_ln) ?; - buff_writer.write_u16(self.packet_identifier) ?; - buff_writer.write_u8(self.reason_code) ?; - buff_writer.write_variable_byte_int(self.property_len) ?; - buff_writer.encode_properties::(&self.properties) ?; + buff_writer.write_u8(self.fixed_header)?; + buff_writer.write_variable_byte_int(rm_ln)?; + buff_writer.write_u16(self.packet_identifier)?; + buff_writer.write_u8(self.reason_code)?; + buff_writer.write_variable_byte_int(self.property_len)?; + buff_writer.encode_properties::(&self.properties)?; Ok(buff_writer.position) } fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { - if self.decode_fixed_header(buff_reader) ? != (PacketType::Pubrec).into() { + if self.decode_fixed_header(buff_reader)? != (PacketType::Pubrec).into() { log::error!("Packet you are trying to decode is not PUBREC packet!"); return Err(BufferError::PacketTypeMismatch); } - self.packet_identifier = buff_reader.read_u16() ?; - self.reason_code = buff_reader.read_u8() ?; + self.packet_identifier = buff_reader.read_u16()?; + self.reason_code = buff_reader.read_u8()?; return self.decode_properties(buff_reader); } diff --git a/src/packet/pubrel_packet.rs b/src/packet/pubrel_packet.rs index cab1a95..d7a0875 100644 --- a/src/packet/pubrel_packet.rs +++ b/src/packet/pubrel_packet.rs @@ -42,8 +42,7 @@ pub struct PubrelPacket<'a, const MAX_PROPERTIES: usize> { pub properties: Vec, MAX_PROPERTIES>, } -impl<'a, const MAX_PROPERTIES: usize> PubrelPacket<'a, MAX_PROPERTIES> { -} +impl<'a, const MAX_PROPERTIES: usize> PubrelPacket<'a, MAX_PROPERTIES> {} impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubrelPacket<'a, MAX_PROPERTIES> { fn new() -> Self { @@ -53,7 +52,7 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubrelPacket<'a, MAX_PROPER packet_identifier: 0, reason_code: 0, property_len: 0, - properties: Vec::, MAX_PROPERTIES>::new() + properties: Vec::, MAX_PROPERTIES>::new(), } } @@ -61,27 +60,26 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubrelPacket<'a, MAX_PROPER let mut buff_writer = BuffWriter::new(buffer, buffer_len); let mut rm_ln = self.property_len; - let property_len_enc: [u8; 4] = - VariableByteIntegerEncoder::encode(self.property_len) ?; + let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len)?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); rm_ln = rm_ln + property_len_len as u32 + 3; - buff_writer.write_u8(self.fixed_header) ?; - buff_writer.write_variable_byte_int(rm_ln) ?; - buff_writer.write_u16(self.packet_identifier) ?; - buff_writer.write_u8(self.reason_code) ?; - buff_writer.write_variable_byte_int(self.property_len) ?; - buff_writer.encode_properties::(&self.properties) ?; + buff_writer.write_u8(self.fixed_header)?; + buff_writer.write_variable_byte_int(rm_ln)?; + buff_writer.write_u16(self.packet_identifier)?; + buff_writer.write_u8(self.reason_code)?; + buff_writer.write_variable_byte_int(self.property_len)?; + buff_writer.encode_properties::(&self.properties)?; Ok(buff_writer.position) } fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { - if self.decode_fixed_header(buff_reader) ? != (PacketType::Pubrel).into() { + if self.decode_fixed_header(buff_reader)? != (PacketType::Pubrel).into() { log::error!("Packet you are trying to decode is not PUBREL packet!"); return Err(BufferError::PacketTypeMismatch); } - self.packet_identifier = buff_reader.read_u16() ?; - self.reason_code = buff_reader.read_u8() ?; + self.packet_identifier = buff_reader.read_u16()?; + self.reason_code = buff_reader.read_u8()?; return self.decode_properties(buff_reader); } diff --git a/src/packet/reason_codes.rs b/src/packet/reason_codes.rs index 1047e87..6a72e98 100644 --- a/src/packet/reason_codes.rs +++ b/src/packet/reason_codes.rs @@ -96,9 +96,8 @@ impl Into for ReasonCode { ReasonCode::SubscriptionIdentifiersNotSupported => 0xA1, ReasonCode::WildcardSubscriptionNotSupported => 0xA2, ReasonCode::BuffError => 0xFE, - ReasonCode::NetworkError => 0xFF - } - + ReasonCode::NetworkError => 0xFF, + }; } } @@ -148,8 +147,8 @@ impl From for ReasonCode { 0xA1 => ReasonCode::SubscriptionIdentifiersNotSupported, 0xA2 => ReasonCode::WildcardSubscriptionNotSupported, 0xFE => ReasonCode::BuffError, - _ => ReasonCode::NetworkError - } + _ => ReasonCode::NetworkError, + }; } } @@ -170,7 +169,9 @@ impl Display for ReasonCode { ReasonCode::ImplementationSpecificError => write!(f, "Implementation specific error!"), ReasonCode::UnsupportedProtocolVersion => write!(f, "Unsupported protocol version!"), ReasonCode::ClientIdNotValid => write!(f, "Client sent not valid identification"), - ReasonCode::BadUserNameOrPassword => write!(f, "Authentication error, username of password not valid!"), + ReasonCode::BadUserNameOrPassword => { + write!(f, "Authentication error, username of password not valid!") + } ReasonCode::NotAuthorized => write!(f, "Client not authorized!"), ReasonCode::ServerUnavailable => write!(f, "Server unavailable!"), ReasonCode::ServerBusy => write!(f, "Server is busy!"), @@ -194,13 +195,19 @@ impl Display for ReasonCode { ReasonCode::QoSNotSupported => write!(f, "Used QoS is not supported!"), ReasonCode::UseAnotherServer => write!(f, "Use another server!"), ReasonCode::ServerMoved => write!(f, "Server moved!"), - ReasonCode::SharedSubscriptionNotSupported => write!(f, "Shared subscription is not supported"), + ReasonCode::SharedSubscriptionNotSupported => { + write!(f, "Shared subscription is not supported") + } ReasonCode::ConnectionRateExceeded => write!(f, "Connection rate exceeded!"), ReasonCode::MaximumConnectTime => write!(f, "Maximum connect time exceeded!"), - ReasonCode::SubscriptionIdentifiersNotSupported => write!(f, "Subscription identifier not supported!"), - ReasonCode::WildcardSubscriptionNotSupported => write!(f, "Wildcard subscription not supported!"), - ReasonCode::BuffError => write!(f , "Error encountered during write / read from packet"), + ReasonCode::SubscriptionIdentifiersNotSupported => { + write!(f, "Subscription identifier not supported!") + } + ReasonCode::WildcardSubscriptionNotSupported => { + write!(f, "Wildcard subscription not supported!") + } + ReasonCode::BuffError => write!(f, "Error encountered during write / read from packet"), ReasonCode::NetworkError => write!(f, "Unknown error!"), } } -} \ No newline at end of file +} diff --git a/src/packet/suback_packet.rs b/src/packet/suback_packet.rs index b765888..ac7e93f 100644 --- a/src/packet/suback_packet.rs +++ b/src/packet/suback_packet.rs @@ -43,16 +43,19 @@ pub struct SubackPacket<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: usiz impl<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: usize> SubackPacket<'a, MAX_REASONS, MAX_PROPERTIES> { - pub fn read_reason_codes(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + pub fn read_reason_codes( + &mut self, + buff_reader: &mut BuffReader<'a>, + ) -> Result<(), BufferError> { let mut i = 0; loop { - self.reason_codes.push(buff_reader.read_u8() ?); + self.reason_codes.push(buff_reader.read_u8()?); i = i + 1; if i == MAX_REASONS { break; } } - return Ok(()) + return Ok(()); } } @@ -66,22 +69,22 @@ impl<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: usize> Packet<'a> packet_identifier: 0, property_len: 0, properties: Vec::, MAX_PROPERTIES>::new(), - reason_codes: Vec::::new() + reason_codes: Vec::::new(), } } fn encode(&mut self, _buffer: &mut [u8], _buffer_len: usize) -> Result { log::error!("SUBACK packet does not support encoding!"); - return Err(BufferError::WrongPacketToEncode) + return Err(BufferError::WrongPacketToEncode); } - fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError>{ - if self.decode_fixed_header(buff_reader) ? != (PacketType::Suback).into() { + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + if self.decode_fixed_header(buff_reader)? != (PacketType::Suback).into() { log::error!("Packet you are trying to decode is not SUBACK packet!"); return Err(BufferError::PacketTypeMismatch); } - self.packet_identifier = buff_reader.read_u16() ?; - self.decode_properties(buff_reader) ?; + self.packet_identifier = buff_reader.read_u16()?; + self.decode_properties(buff_reader)?; return self.read_reason_codes(buff_reader); } diff --git a/src/packet/subscription_packet.rs b/src/packet/subscription_packet.rs index 52bd6a2..4f57121 100644 --- a/src/packet/subscription_packet.rs +++ b/src/packet/subscription_packet.rs @@ -24,14 +24,14 @@ use heapless::Vec; +use super::packet_type::PacketType; +use super::property::Property; use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; use crate::packet::publish_packet::QualityOfService; use crate::utils::buffer_reader::BuffReader; use crate::utils::buffer_writer::BuffWriter; use crate::utils::types::{BufferError, TopicFilter}; -use super::packet_type::PacketType; -use super::property::Property; pub struct SubscriptionPacket<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> { pub fixed_header: u8, @@ -46,12 +46,13 @@ pub struct SubscriptionPacket<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> SubscriptionPacket<'a, MAX_FILTERS, MAX_PROPERTIES> { - pub fn add_new_filter(& mut self, topic_name: &'a str, qos: QualityOfService) { + pub fn add_new_filter(&mut self, topic_name: &'a str, qos: QualityOfService) { let len = topic_name.len(); let mut new_filter = TopicFilter::new(); new_filter.filter.string = topic_name; new_filter.filter.len = len as u16; - new_filter.sub_options = new_filter.sub_options | (>::into(qos) >> 1); + new_filter.sub_options = + new_filter.sub_options | (>::into(qos) >> 1); self.topic_filters.push(new_filter); self.topic_filter_len = self.topic_filter_len + 1; } @@ -60,7 +61,7 @@ impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> Packet<'a> for SubscriptionPacket<'a, MAX_FILTERS, MAX_PROPERTIES> { - fn new() -> Self { + fn new() -> Self { let x = Self { fixed_header: PacketType::Subscribe.into(), remain_len: 0, @@ -77,8 +78,7 @@ impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> Packet<'a> let mut buff_writer = BuffWriter::new(buffer, buffer_len); let mut rm_ln = self.property_len; - let property_len_enc: [u8; 4] = - VariableByteIntegerEncoder::encode(self.property_len) ?; + let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len)?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); let mut lt = 0; @@ -92,16 +92,16 @@ impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> Packet<'a> } rm_ln = rm_ln + property_len_len as u32 + 2 + filters_len as u32; - buff_writer.write_u8(self.fixed_header) ?; - buff_writer.write_variable_byte_int(rm_ln) ?; - buff_writer.write_u16(self.packet_identifier) ?; - buff_writer.write_variable_byte_int(self.property_len) ?; - buff_writer.encode_properties::(&self.properties) ?; + buff_writer.write_u8(self.fixed_header)?; + buff_writer.write_variable_byte_int(rm_ln)?; + buff_writer.write_u16(self.packet_identifier)?; + buff_writer.write_variable_byte_int(self.property_len)?; + buff_writer.encode_properties::(&self.properties)?; buff_writer.encode_topic_filters_ref( true, self.topic_filter_len as usize, &self.topic_filters, - ) ?; + )?; Ok(buff_writer.position) } diff --git a/src/packet/unsuback_packet.rs b/src/packet/unsuback_packet.rs index 79a8b05..ef0770f 100644 --- a/src/packet/unsuback_packet.rs +++ b/src/packet/unsuback_packet.rs @@ -43,10 +43,13 @@ pub struct UnsubackPacket<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: us impl<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: usize> UnsubackPacket<'a, MAX_REASONS, MAX_PROPERTIES> { - pub fn read_reason_codes(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + pub fn read_reason_codes( + &mut self, + buff_reader: &mut BuffReader<'a>, + ) -> Result<(), BufferError> { let mut i = 0; loop { - self.reason_codes.push(buff_reader.read_u8() ?); + self.reason_codes.push(buff_reader.read_u8()?); i = i + 1; if i == MAX_REASONS { break; @@ -66,7 +69,7 @@ impl<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: usize> Packet<'a> packet_identifier: 0, property_len: 0, properties: Vec::, MAX_PROPERTIES>::new(), - reason_codes: Vec::::new() + reason_codes: Vec::::new(), } } @@ -76,12 +79,12 @@ impl<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: usize> Packet<'a> } fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { - if self.decode_fixed_header(buff_reader) ? != (PacketType::Unsuback).into() { + if self.decode_fixed_header(buff_reader)? != (PacketType::Unsuback).into() { log::error!("Packet you are trying to decode is not UNSUBACK packet!"); return Err(BufferError::PacketTypeMismatch); } - self.packet_identifier = buff_reader.read_u16() ?; - self.decode_properties(buff_reader) ?; + self.packet_identifier = buff_reader.read_u16()?; + self.decode_properties(buff_reader)?; return self.read_reason_codes(buff_reader); } diff --git a/src/packet/unsubscription_packet.rs b/src/packet/unsubscription_packet.rs index f352010..6ba0796 100644 --- a/src/packet/unsubscription_packet.rs +++ b/src/packet/unsubscription_packet.rs @@ -58,7 +58,7 @@ impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> Packet<'a> property_len: 0, properties: Vec::, MAX_PROPERTIES>::new(), topic_filter_len: 0, - topic_filters: Vec::, MAX_FILTERS>::new() + topic_filters: Vec::, MAX_FILTERS>::new(), } } @@ -66,22 +66,21 @@ impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> Packet<'a> let mut buff_writer = BuffWriter::new(buffer, buffer_len); let mut rm_ln = self.property_len; - let property_len_enc: [u8; 4] = - VariableByteIntegerEncoder::encode(self.property_len) ?; + let property_len_enc: [u8; 4] = VariableByteIntegerEncoder::encode(self.property_len)?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); rm_ln = rm_ln + property_len_len as u32 + 4 + self.topic_filter_len as u32; - buff_writer.write_u8(self.fixed_header) ?; - buff_writer.write_variable_byte_int(rm_ln) ?; - buff_writer.write_u16(self.packet_identifier) ?; - buff_writer.write_variable_byte_int(self.property_len) ?; - buff_writer.encode_properties::(&self.properties) ?; - buff_writer.write_u16(self.topic_filter_len) ?; + buff_writer.write_u8(self.fixed_header)?; + buff_writer.write_variable_byte_int(rm_ln)?; + buff_writer.write_u16(self.packet_identifier)?; + buff_writer.write_variable_byte_int(self.property_len)?; + buff_writer.encode_properties::(&self.properties)?; + buff_writer.write_u16(self.topic_filter_len)?; buff_writer.encode_topic_filters_ref( false, self.topic_filter_len as usize, &self.topic_filters, - ) ?; + )?; Ok(buff_writer.position) } diff --git a/src/utils/buffer_reader.rs b/src/utils/buffer_reader.rs index 40db7bd..eaf600f 100644 --- a/src/utils/buffer_reader.rs +++ b/src/utils/buffer_reader.rs @@ -45,7 +45,7 @@ impl<'a> BuffReader<'a> { return BuffReader { buffer, position: 0, - len: buff_len + len: buff_len, }; } @@ -101,7 +101,7 @@ impl<'a> BuffReader<'a> { let len = self.read_u16(); match len { Err(err) => return Err(err), - _ => {}, + _ => {} } let len_res = len.unwrap(); let res_str = @@ -119,7 +119,7 @@ impl<'a> BuffReader<'a> { /// Read Binary data from buffer pub fn read_binary(&mut self) -> Result, BufferError> { - let len = self.read_u16() ?; + let len = self.read_u16()?; let res_bin = &(self.buffer[self.position..(self.position + len as usize)]); return Ok(BinaryData { bin: res_bin, @@ -147,7 +147,6 @@ impl<'a> BuffReader<'a> { /// Read payload message from buffer pub fn read_message(&mut self, total_len: usize) -> &'a [u8] { - if total_len > self.len { return &self.buffer[self.position..self.len]; } diff --git a/src/utils/buffer_writer.rs b/src/utils/buffer_writer.rs index 2a9fa3f..27734e5 100644 --- a/src/utils/buffer_writer.rs +++ b/src/utils/buffer_writer.rs @@ -26,12 +26,12 @@ use heapless::Vec; use crate::encoding::variable_byte_integer::{VariableByteInteger, VariableByteIntegerEncoder}; use crate::packet::property::Property; -use crate::utils::types::{BinaryData, EncodedString, StringPair, BufferError, TopicFilter}; +use crate::utils::types::{BinaryData, BufferError, EncodedString, StringPair, TopicFilter}; pub struct BuffWriter<'a> { buffer: &'a mut [u8], pub position: usize, - len: usize + len: usize, } impl<'a> BuffWriter<'a> { @@ -39,7 +39,7 @@ impl<'a> BuffWriter<'a> { return BuffWriter { buffer, position: 0, - len: buff_len + len: buff_len, }; } @@ -72,8 +72,7 @@ impl<'a> BuffWriter<'a> { self.buffer[self.position] = byte; self.increment_position(1); Ok(()) - } - + }; } pub fn write_u16(&mut self, two_bytes: u16) -> Result<(), BufferError> { @@ -87,56 +86,59 @@ impl<'a> BuffWriter<'a> { } pub fn write_string_ref(&mut self, str: &EncodedString<'a>) -> Result<(), BufferError> { - self.write_u16(str.len) ?; + self.write_u16(str.len)?; let bytes = str.string.as_bytes(); return self.insert_ref(str.len as usize, bytes); } pub fn write_string(&mut self, str: EncodedString<'a>) -> Result<(), BufferError> { - self.write_u16(str.len) ?; + self.write_u16(str.len)?; let bytes = str.string.as_bytes(); return self.insert_ref(str.len as usize, bytes); } pub fn write_binary_ref(&mut self, bin: &BinaryData<'a>) -> Result<(), BufferError> { - self.write_u16(bin.len) ?; + self.write_u16(bin.len)?; return self.insert_ref(bin.len as usize, bin.bin); } pub fn write_binary(&mut self, bin: BinaryData<'a>) -> Result<(), BufferError> { - self.write_u16(bin.len) ?; + self.write_u16(bin.len)?; return self.insert_ref(bin.len as usize, bin.bin); } pub fn write_string_pair_ref(&mut self, str_pair: &StringPair<'a>) -> Result<(), BufferError> { - self.write_string_ref(&str_pair.name) ?; + self.write_string_ref(&str_pair.name)?; return self.write_string_ref(&str_pair.value); } pub fn write_string_pair(&mut self, str_pair: StringPair<'a>) -> Result<(), BufferError> { - self.write_string(str_pair.name) ?; + self.write_string(str_pair.name)?; return self.write_string(str_pair.value); } pub fn write_variable_byte_int(&mut self, int: u32) -> Result<(), BufferError> { - let x: VariableByteInteger = VariableByteIntegerEncoder::encode(int) ?; + let x: VariableByteInteger = VariableByteIntegerEncoder::encode(int)?; let len = VariableByteIntegerEncoder::len(x); return self.insert_ref(len, &x); } pub fn encode_property(&mut self, property: &Property<'a>) -> Result<(), BufferError> { let x: u8 = property.into(); - self.write_u8(x) ?; + self.write_u8(x)?; return property.encode(self); } - pub fn encode_properties(&mut self, properties: &Vec, LEN>) -> Result<(), BufferError> { + pub fn encode_properties( + &mut self, + properties: &Vec, LEN>, + ) -> Result<(), BufferError> { let mut i = 0; let len = properties.len(); if len != 0 { loop { let prop: &Property = properties.get(i).unwrap_or(&Property::Reserved()); - self.encode_property(prop) ?; + self.encode_property(prop)?; i = i + 1; if i == len { break; @@ -146,12 +148,16 @@ impl<'a> BuffWriter<'a> { Ok(()) } - fn encode_topic_filter_ref(&mut self, sub: bool, topic_filter: &TopicFilter<'a>) -> Result<(), BufferError> { - self.write_string_ref(&topic_filter.filter) ?; + fn encode_topic_filter_ref( + &mut self, + sub: bool, + topic_filter: &TopicFilter<'a>, + ) -> Result<(), BufferError> { + self.write_string_ref(&topic_filter.filter)?; if sub { - self.write_u8(topic_filter.sub_options) ?; + self.write_u8(topic_filter.sub_options)?; } - return Ok(()) + return Ok(()); } pub fn encode_topic_filters_ref( @@ -163,7 +169,7 @@ impl<'a> BuffWriter<'a> { let mut i = 0; loop { let topic_filter: &TopicFilter<'a> = filters.get(i).unwrap(); - self.encode_topic_filter_ref(sub, topic_filter) ?; + self.encode_topic_filter_ref(sub, topic_filter)?; i = i + 1; if i == len { break; diff --git a/src/utils/rng_generator.rs b/src/utils/rng_generator.rs index b1c11e2..f703d2e 100644 --- a/src/utils/rng_generator.rs +++ b/src/utils/rng_generator.rs @@ -1,7 +1,7 @@ // This code is handed from Embedded Rust documentation and // is accessible from https://docs.rust-embedded.org/cortex-m-rt/0.6.0/rand/trait.RngCore.html -use rand_core::{Error, impls, RngCore}; +use rand_core::{impls, Error, RngCore}; pub struct CountingRng(pub u64); @@ -25,4 +25,4 @@ impl RngCore for CountingRng { fn try_fill_bytes(&mut self, dest: &mut [u8]) -> Result<(), Error> { Ok(self.fill_bytes(dest)) } -} \ No newline at end of file +} diff --git a/src/utils/types.rs b/src/utils/types.rs index 0c7a2b8..a4e9342 100644 --- a/src/utils/types.rs +++ b/src/utils/types.rs @@ -35,7 +35,7 @@ pub enum BufferError { PacketTypeMismatch, WrongPacketToDecode, WrongPacketToEncode, - PropertyNotFound + PropertyNotFound, } impl Display for BufferError { @@ -55,8 +55,7 @@ impl Display for BufferError { } } /// Encoded string provides structure representing UTF-8 encoded string in MQTTv5 packets -#[derive(Debug)] -#[derive(Clone)] +#[derive(Debug, Clone)] pub struct EncodedString<'a> { pub string: &'a str, pub len: u16, @@ -73,8 +72,7 @@ impl EncodedString<'_> { } /// Binary data represents `Binary data` in MQTTv5 protocol -#[derive(Debug)] -#[derive(Clone)] +#[derive(Debug, Clone)] pub struct BinaryData<'a> { pub bin: &'a [u8], pub len: u16, @@ -123,4 +121,4 @@ impl TopicFilter<'_> { pub fn len(&self) -> u16 { return self.filter.len + 3; } -} \ No newline at end of file +}