diff --git a/src/client/client_config.rs b/src/client/client_config.rs index 8430e6e..88382ab 100644 --- a/src/client/client_config.rs +++ b/src/client/client_config.rs @@ -24,7 +24,7 @@ use crate::packet::publish_packet::QualityOfService; -use crate::utils::buffer_reader::{BinaryData, EncodedString}; +use crate::utils::types::{BinaryData, EncodedString}; pub struct ClientConfig<'a> { pub qos: QualityOfService, diff --git a/src/client/client_v5.rs b/src/client/client_v5.rs index 23638d9..7c72640 100644 --- a/src/client/client_v5.rs +++ b/src/client/client_v5.rs @@ -1,26 +1,26 @@ -use core::future::Future; -use embassy::traits::rng; use rand_core::RngCore; use crate::client::client_config::ClientConfig; - -use crate::network::network_trait::{Network, NetworkError}; +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::puback_packet::PubackPacket; -use crate::packet::publish_packet::QualityOfService::QoS1; use crate::packet::publish_packet::{PublishPacket, QualityOfService}; +use crate::packet::publish_packet::QualityOfService::QoS1; 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; pub struct MqttClientV5<'a, T, const MAX_PROPERTIES: usize> { network_driver: &'a mut T, buffer: &'a mut [u8], + buffer_len: usize, recv_buffer: &'a mut [u8], + recv_buffer_len: usize, rng: CountingRng, config: ClientConfig<'a>, } @@ -29,18 +29,20 @@ 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], recv_buffer: &'a mut [u8], 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, + buffer_len, recv_buffer, + recv_buffer_len, rng: CountingRng(50), config } } pub async fn connect_to_broker<'b>(&'b mut self) -> Result<(), ReasonCode> { - let mut len = { + let len = { let mut connect = ConnectPacket::<'b, 3, 0>::clean(); if self.config.username_flag { connect.add_username(& self.config.username); @@ -48,21 +50,33 @@ where if self.config.password_flag { connect.add_password(& self.config.password) } - connect.encode(self.buffer) + connect.encode(self.buffer, self.buffer_len) }; - self.network_driver.send(self.buffer, len).await ?; + if let Err(err) = len { + log::error!("[DECODE ERR]: {}", err); + return Err(ReasonCode::BuffError); + } + self.network_driver.send(self.buffer, len.unwrap()).await ?; //connack - let reason: u8 = { + let reason: Result = { self.network_driver.receive(self.buffer).await?; let mut packet = ConnackPacket::<'b, 5>::new(); - packet.decode(&mut BuffReader::new(self.buffer)); - packet.connect_reason_code + if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len)) { + Err(err) + } else { + Ok(packet.connect_reason_code) + } }; - if reason != 0x00 { - return Err(ReasonCode::from(reason)); + if let Err(err) = reason { + log::error!("[DECODE ERR]: {}", err); + return Err(ReasonCode::BuffError); + } + let res = reason.unwrap(); + if res != 0x00 { + return Err(ReasonCode::from(res)); } else { Ok(()) } @@ -71,8 +85,12 @@ where pub async fn disconnect<'b>(&'b mut self) -> Result<(), ReasonCode> { let mut disconnect = DisconnectPacket::<'b, 5>::new(); - let mut len = disconnect.encode(self.buffer); - self.network_driver.send(self.buffer, len).await?; + let len = disconnect.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 ?; Ok(()) } @@ -89,27 +107,41 @@ where packet.add_qos(self.config.qos); packet.add_identifier(identifier); packet.add_message(message.as_bytes()); - packet.encode(self.buffer) + packet.encode(self.buffer, self.buffer_len) }; - self.network_driver.send(self.buffer, len).await ?; + if let Err(err) = len { + log::error!("[DECODE ERR]: {}", err); + return Err(ReasonCode::BuffError); + } + + self.network_driver.send(self.buffer, len.unwrap()).await ?; //QoS1 if >::into(self.config.qos ) == >::into(QoS1) { - let reason = { + let reason: Result<[u16; 2], BufferError> = { self.network_driver.receive(self.buffer).await ?; let mut packet = PubackPacket::<'b, 5>::new(); - packet.decode(&mut BuffReader::new(self.buffer)); - [packet.packet_identifier, packet.reason_code as u16] + 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]) + } }; - if identifier != reason[0] { + if let Err(err) = reason { + log::error!("[DECODE ERR]: {}", err); + return Err(ReasonCode::BuffError); + } + + let res = reason.unwrap(); + if identifier != res[0] { return Err(ReasonCode::PacketIdentifierNotFound); } - if reason[1] != 0 { - return Err(ReasonCode::from(reason[1] as u8)); + if res[1] != 0 { + return Err(ReasonCode::from(res[1] as u8)); } } Ok(()) @@ -121,21 +153,35 @@ where let len = { let mut subs = SubscriptionPacket::<'b, 1, 1>::new(); subs.add_new_filter(topic_name, self.config.qos); - subs.encode(self.buffer) + subs.encode(self.buffer, self.buffer_len) }; - self.network_driver.send(self.buffer, len).await ?; + if let Err(err) = len { + log::error!("[DECODE ERR]: {}", err); + return Err(ReasonCode::BuffError); + } - let reason = { + self.network_driver.send(self.buffer, len.unwrap()).await ?; + + let reason: Result = { self.network_driver.receive(self.buffer).await ?; let mut packet = SubackPacket::<'b, 5, 5>::new(); - packet.decode(&mut BuffReader::new(self.buffer)); - *packet.reason_codes.get(0).unwrap() + if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len)) { + Err(err) + } else { + Ok(*packet.reason_codes.get(0).unwrap()) + } }; - if reason != (>::into(self.config.qos) >> 1) { - Err(ReasonCode::from(reason)) + if let Err(err) = reason { + log::error!("[DECODE ERR]: {}", err); + return Err(ReasonCode::BuffError); + } + + let res = reason.unwrap(); + if res != (>::into(self.config.qos) >> 1) { + Err(ReasonCode::from(res)) } else { Ok(()) } @@ -145,15 +191,22 @@ where pub async fn receive_message<'b>(&'b mut self) -> Result<&'b [u8], ReasonCode> { self.network_driver.receive(self.recv_buffer).await ?; let mut packet = PublishPacket::<'b, 5>::new(); - packet.decode(&mut BuffReader::new(self.recv_buffer)); + 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) { let mut puback = PubackPacket::<'b, 5>::new(); puback.packet_identifier = packet.packet_identifier; puback.reason_code = 0x00; { - let len = puback.encode(self.buffer); - self.network_driver.send(self.buffer, len).await ?; + let len = puback.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 ?; } } diff --git a/src/encoding/variable_byte_integer.rs b/src/encoding/variable_byte_integer.rs index afb561f..ccfa82b 100644 --- a/src/encoding/variable_byte_integer.rs +++ b/src/encoding/variable_byte_integer.rs @@ -22,9 +22,7 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. */ -#![crate_name = "doc"] - -use crate::utils::buffer_reader::ParseError; +use crate::utils::types::BufferError; /// VariableByteIntegerEncoder and VariableByteIntegerDecoder are implemented based on /// pseudo code which is introduced in MQTT version 5.0 OASIS standard accesible from @@ -45,13 +43,13 @@ impl VariableByteIntegerEncoder { /// this integer into maximal 4 Bytes. MSb of each Byte is controll bit. /// This bit is saying if there is continuing Byte in stream or not, this way /// we can effectively use 1 to 4 Bytes based in integer len. - pub fn encode(mut target: u32) -> Result { + pub fn encode(mut target: u32) -> Result { // General known informations from OASIS const MAX_ENCODABLE: u32 = 268435455; const MOD: u32 = 128; if target > MAX_ENCODABLE { log::error!("Maximal value of integer for encoding was exceeded"); - return Err(ParseError::EncodingError); + return Err(BufferError::EncodingError); } let mut res: [u8; 4] = [0; 4]; @@ -76,7 +74,7 @@ impl VariableByteIntegerEncoder { pub fn len(var_int: VariableByteInteger) -> usize { let mut i: usize = 0; loop { - let mut encoded_byte: u8; + let encoded_byte: u8; encoded_byte = var_int[i]; i = i + 1; if (encoded_byte & 128) == 0 { @@ -95,7 +93,7 @@ impl VariableByteIntegerDecoder { /// Decode function takes as paramater encoded integer represented /// as array of 4 unsigned numbers of exactly 1 Byte each -> 4 Bytes maximal /// same as maximal amount of bytes for variable byte encoding in MQTT. - pub fn decode(encoded: VariableByteInteger) -> Result { + pub fn decode(encoded: VariableByteInteger) -> Result { let mut multiplier: u32 = 1; let mut ret: u32 = 0; @@ -107,7 +105,7 @@ impl VariableByteIntegerDecoder { i = i + 1; ret = ret + ((encoded_byte & 127) as u32 * multiplier) as u32; if multiplier > 128 * 128 * 128 { - return Err(ParseError::DecodingError); + return Err(BufferError::DecodingError); } multiplier = multiplier * 128; if (encoded_byte & 128) == 0 { diff --git a/src/main.rs b/src/main.rs index c6c200d..a6ee5b3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,15 +1,17 @@ +use std::time::Duration; + +use tokio::{join, task}; +use tokio::time::sleep; + +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::subscription_packet::SubscriptionPacket; use rust_mqtt::tokio_network::TokioNetwork; -use std::time::Duration; -use tokio::time::sleep; -use tokio::{join, task}; -use rust_mqtt::client::client_config::ClientConfig; -use rust_mqtt::packet::publish_packet::QualityOfService::QoS1; async fn receive() { let mut ip: [u8; 4] = [37, 205, 11, 180]; @@ -22,7 +24,7 @@ async fn receive() { config.add_password("testPass1"); let mut res2 = vec![0; 260]; let mut res3 = vec![0; 260]; - let mut client = MqttClientV5::::new(&mut tokio_network, &mut res2, & mut res3, config); + let mut client = MqttClientV5::::new(&mut tokio_network, &mut res2, 260, & mut res3, 260, config); let mut result = { client.connect_to_broker().await @@ -54,7 +56,7 @@ 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, & mut res3, 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!"); diff --git a/src/network/network_trait.rs b/src/network/network_trait.rs index bfe623f..4db9313 100644 --- a/src/network/network_trait.rs +++ b/src/network/network_trait.rs @@ -1,8 +1,6 @@ -use core::fmt::Error; use core::future::Future; -use crate::packet::reason_codes::ReasonCode; -use crate::packet::mqtt_packet::Packet; +use crate::packet::reason_codes::ReasonCode; #[derive(Debug)] pub enum NetworkError { diff --git a/src/packet/auth_packet.rs b/src/packet/auth_packet.rs index 032d5a0..9adf587 100644 --- a/src/packet/auth_packet.rs +++ b/src/packet/auth_packet.rs @@ -29,6 +29,7 @@ use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::BufferError; use super::packet_type::PacketType; use super::property::Property; @@ -44,13 +45,6 @@ pub struct AuthPacket<'a, const MAX_PROPERTIES: usize> { } impl<'a, const MAX_PROPERTIES: usize> AuthPacket<'a, MAX_PROPERTIES> { - /// Function is decoding auth packet from Byte array (buffer). - pub fn decode_auth_packet(&mut self, buff_reader: &mut BuffReader<'a>) { - self.decode_fixed_header(buff_reader); - self.auth_reason = buff_reader.read_u8().unwrap(); - self.decode_properties(buff_reader); - } - pub fn add_reason_code(&mut self, code: u8) { if code != 0 && code != 24 && code != 25 { log::error!("Provided reason code is not supported!"); @@ -79,26 +73,28 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for AuthPacket<'a, MAX_PROPERTI } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); + fn encode(&mut self, buffer: &mut [u8], buff_len: usize) -> Result { + 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).unwrap(); + 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); - return buff_writer.position; + 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>) { - self.decode_auth_packet(buff_reader); + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + self.decode_fixed_header(buff_reader) ?; + self.auth_reason = buff_reader.read_u8() ?; + return self.decode_properties(buff_reader); } fn set_property_len(&mut self, value: u32) { diff --git a/src/packet/connack_packet.rs b/src/packet/connack_packet.rs index 98d005e..8fea3f4 100644 --- a/src/packet/connack_packet.rs +++ b/src/packet/connack_packet.rs @@ -28,14 +28,13 @@ use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::BufferError; use super::packet_type::PacketType; use super::property::Property; pub struct ConnackPacket<'a, const MAX_PROPERTIES: usize> { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, pub ack_flags: u8, pub connect_reason_code: u8, @@ -44,14 +43,14 @@ 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>) { - 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; + return Err(BufferError::PacketTypeMismatch); } - self.ack_flags = buff_reader.read_u8().unwrap(); - self.connect_reason_code = buff_reader.read_u8().unwrap(); - self.decode_properties(buff_reader); + self.ack_flags = buff_reader.read_u8() ?; + self.connect_reason_code = buff_reader.read_u8() ?; + self.decode_properties(buff_reader) } } @@ -67,22 +66,23 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for ConnackPacket<'a, MAX_PROPE } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); - buff_writer.write_u8(self.fixed_header); - let mut property_len_enc = VariableByteIntegerEncoder::encode(self.property_len).unwrap(); + 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) ?; 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); - return buff_writer.position; + 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) } - fn decode(&mut self, buff_reader: &mut BuffReader<'a>) { - self.decode_connack_packet(buff_reader); + + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + self.decode_connack_packet(buff_reader) } fn set_property_len(&mut self, value: u32) { diff --git a/src/packet/connect_packet.rs b/src/packet/connect_packet.rs index d3d2893..649d501 100644 --- a/src/packet/connect_packet.rs +++ b/src/packet/connect_packet.rs @@ -26,34 +26,25 @@ use heapless::Vec; use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; -use crate::utils::buffer_reader::BinaryData; use crate::utils::buffer_reader::BuffReader; -use crate::utils::buffer_reader::EncodedString; -use crate::utils::buffer_reader::ParseError; + use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::{BinaryData, BufferError, EncodedString}; use super::packet_type::PacketType; use super::property::Property; pub struct ConnectPacket<'a, const MAX_PROPERTIES: usize, const MAX_WILL_PROPERTIES: usize> { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, pub protocol_name_len: u16, pub protocol_name: u32, pub protocol_version: u8, pub connect_flags: u8, pub keep_alive: u16, - // property len pub property_len: u32, - - // properties pub properties: Vec, MAX_PROPERTIES>, - - //payload pub client_id: EncodedString<'a>, - // property len pub will_property_len: u32, pub will_properties: Vec, MAX_WILL_PROPERTIES>, pub will_topic: EncodedString<'a>, @@ -91,37 +82,11 @@ impl<'a, const MAX_PROPERTIES: usize, const MAX_WILL_PROPERTIES: usize> return x; } - pub fn get_reason_code(&self) { - log::info!("Getting reason code!"); - } - pub fn add_packet_type(&mut self, new_packet_type: PacketType) { self.fixed_header = self.fixed_header & 0x0F; self.fixed_header = self.fixed_header | >::into(new_packet_type); } - pub fn add_flags(&mut self, dup: bool, qos: u8, retain: bool) { - let cur_type: u8 = self.fixed_header & 0xF0; - if cur_type != 0x30 { - log::error!("Cannot add flags into packet with other than PUBLISH type"); - return; - } - let mut flags: u8 = 0x00; - if dup { - flags = flags | 0x08; - } - if qos == 1 { - flags = flags | 0x02; - } - if qos == 2 { - flags = flags | 0x04; - } - if retain { - flags = flags | 0x01; - } - self.fixed_header = cur_type | flags; - } - pub fn add_username(&mut self, username: &EncodedString<'a>) { self.username = (*username).clone(); self.connect_flags = self.connect_flags | 0x80; @@ -137,22 +102,41 @@ impl<'a, const MAX_PROPERTIES: usize, const MAX_WILL_PROPERTIES: usize> Packet<' for ConnectPacket<'a, MAX_PROPERTIES, MAX_WILL_PROPERTIES> { fn new() -> Self { - todo!() + Self { + fixed_header: PacketType::Connect.into(), + remain_len: 0, + protocol_name_len: 4, + // MQTT + protocol_name: 0x4d515454, + protocol_version: 5, + // Clean start flag + connect_flags: 0x02, + keep_alive: 180, + property_len: 0, + properties: Vec::, MAX_PROPERTIES>::new(), + client_id: EncodedString::new(), + will_property_len: 0, + will_properties: Vec::, MAX_WILL_PROPERTIES>::new(), + will_topic: EncodedString::new(), + will_payload: BinaryData::new(), + username: EncodedString::new(), + password: BinaryData::new(), + } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); + fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { + 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).unwrap(); + VariableByteIntegerEncoder::encode(self.property_len) ?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); - // 12 = protocol_name_len + protocol_name + protocol_version + connect_flags + keep_alive + client_id_len + // 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).unwrap(); + 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 @@ -160,7 +144,6 @@ impl<'a, const MAX_PROPERTIES: usize, const MAX_WILL_PROPERTIES: usize> Packet<' + self.will_topic.len as u32 + self.will_payload.len as u32; } - let x = self.connect_flags & 0x80; if (self.connect_flags & 0x80) != 0 { rm_ln = rm_ln + self.username.len as u32 + 2; } @@ -169,38 +152,39 @@ 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) ?; } - return buff_writer.position; + Ok(buff_writer.position) } - fn decode(&mut self, buff_reader: &mut BuffReader<'a>) { - log::error!("Decode function is not available for control packet!") + fn decode(&mut self, _buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + log::error!("Decode function is not available for control packet!"); + Err(BufferError::WrongPacketToDecode) } fn set_property_len(&mut self, value: u32) { @@ -222,26 +206,4 @@ impl<'a, const MAX_PROPERTIES: usize, const MAX_WILL_PROPERTIES: usize> Packet<' fn set_remaining_len(&mut self, remaining_len: u32) { self.remain_len = remaining_len; } - - fn decode_properties(&mut self, buff_reader: &mut BuffReader<'a>) { - self.property_len = buff_reader.read_variable_byte_int().unwrap(); - let mut x: u32 = 0; - let mut prop: Result; - loop { - let mut res: Property; - prop = Property::decode(buff_reader); - if let Ok(res) = prop { - log::info!("Parsed property {:?}", res); - x = x + res.len() as u32 + 1; - self.properties.push(res); - } else { - // error handlo - log::error!("Problem during property decoding"); - } - - if x == self.property_len { - break; - } - } - } } diff --git a/src/packet/disconnect_packet.rs b/src/packet/disconnect_packet.rs index 6507240..4ff1613 100644 --- a/src/packet/disconnect_packet.rs +++ b/src/packet/disconnect_packet.rs @@ -28,6 +28,7 @@ use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::BufferError; use super::packet_type::PacketType; use super::property::Property; @@ -46,14 +47,6 @@ pub struct DisconnectPacket<'a, const MAX_PROPERTIES: usize> { } impl<'a, const MAX_PROPERTIES: usize> DisconnectPacket<'a, MAX_PROPERTIES> { - pub fn decode_auth_packet(&mut self, buff_reader: &mut BuffReader<'a>) { - if self.decode_fixed_header(buff_reader) != (PacketType::Pingresp).into() { - log::error!("Packet you are trying to decode is not PUBACK packet!"); - return; - } - self.disconnect_reason = buff_reader.read_u8().unwrap(); - self.decode_properties(buff_reader); - } fn add_reason(&mut self, reason: u8) { self.disconnect_reason = reason; @@ -71,22 +64,27 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for DisconnectPacket<'a, MAX_PR } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); - buff_writer.write_u8(self.fixed_header); - let mut property_len_enc = VariableByteIntegerEncoder::encode(self.property_len).unwrap(); + 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) ?; 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); - return buff_writer.position; + 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>) { - self.decode_auth_packet(buff_reader); + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + 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() ?; + return self.decode_properties(buff_reader); } fn set_property_len(&mut self, value: u32) { diff --git a/src/packet/mqtt_packet.rs b/src/packet/mqtt_packet.rs index e7ea635..c8130b3 100644 --- a/src/packet/mqtt_packet.rs +++ b/src/packet/mqtt_packet.rs @@ -24,17 +24,18 @@ use crate::packet::packet_type::PacketType; use crate::utils::buffer_reader::BuffReader; -use crate::utils::buffer_reader::ParseError; +use crate::utils::types::BufferError; use super::property::Property; + /// This trait provide interface for mapping MQTTv5 packets to human readable structures /// which can be later modified and used for communication purposes. pub trait Packet<'a> { fn new() -> Self; /// Method encode provide way how to transfer Packet struct into Byte array (buffer) - fn encode(&mut self, buffer: &mut [u8]) -> usize; + fn encode(&mut self, buffer: &mut [u8], buff_len: usize) -> Result; /// Decode method is opposite of encode - decoding Byte array and mapping it into corresponding Packet struct - fn decode(&mut self, buff_reader: &mut BuffReader<'a>); + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError>; /// Setter method for packet properties len - not all Packet types support this fn set_property_len(&mut self, value: u32); @@ -50,34 +51,30 @@ pub trait Packet<'a> { /// Method is decoding Byte array pointing to properties into heapless Vec /// in packet. If decoding goes wrong method is returning Error - fn decode_properties(&mut self, buff_reader: &mut BuffReader<'a>) { + fn decode_properties(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { self.set_property_len(buff_reader.read_variable_byte_int().unwrap()); let mut x: u32 = 0; - let mut prop: Result; + let mut prop: Property; if self.get_property_len() != 0 { loop { - let mut res: Property; - prop = Property::decode(buff_reader); - if let Ok(res) = prop { - log::debug!("Parsed property {:?}", res); - x = x + res.len() as u32 + 1; - self.push_to_properties(res); - } else { - log::error!("Problem during property decoding"); - } + prop = Property::decode(buff_reader) ?; + log::debug!("Parsed property {:?}", prop); + x = x + prop.len() as u32 + 1; + self.push_to_properties(prop); if x == self.get_property_len() { break; } } } + Ok(()) } /// Method is decoding packet header into fixed header part and remaining length - fn decode_fixed_header(&mut self, buff_reader: &mut BuffReader) -> PacketType { - let first_byte: u8 = buff_reader.read_u8().unwrap(); + 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().unwrap()); - return PacketType::from(first_byte); + 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 f7f3d6f..652be7f 100644 --- a/src/packet/pingreq_packet.rs +++ b/src/packet/pingreq_packet.rs @@ -25,14 +25,13 @@ use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::BufferError; use super::packet_type::PacketType; use super::property::Property; pub struct PingreqPacket { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, } @@ -40,21 +39,25 @@ impl PingreqPacket {} impl<'a> Packet<'a> for PingreqPacket { fn new() -> Self { - todo!() + Self { + fixed_header: PacketType::Pingreq.into(), + remain_len: 0 + } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); - buff_writer.write_u8(self.fixed_header); - buff_writer.write_variable_byte_int(0 as u32); - return buff_writer.position; + 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) ?; + Ok(buff_writer.position) } - fn decode(&mut self, buff_reader: &mut BuffReader<'a>) { - log::error!("PingreqPacket packet does not support decode funtion on client!"); + fn decode(&mut self, _buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + log::error!("Pingreq Packet packet does not support decode funtion on client!"); + Err(BufferError::WrongPacketToDecode) } - fn set_property_len(&mut self, value: u32) { + fn set_property_len(&mut self, _value: u32) { log::error!("PINGREQ packet does not contain any properties!"); } @@ -63,7 +66,7 @@ impl<'a> Packet<'a> for PingreqPacket { return 0; } - fn push_to_properties(&mut self, property: Property<'a>) { + fn push_to_properties(&mut self, _property: Property<'a>) { log::error!("PINGREQ packet does not contain any properties!"); } diff --git a/src/packet/pingresp_packet.rs b/src/packet/pingresp_packet.rs index ef140e1..dec615d 100644 --- a/src/packet/pingresp_packet.rs +++ b/src/packet/pingresp_packet.rs @@ -25,43 +25,47 @@ use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::BufferError; use super::packet_type::PacketType; use super::property::Property; pub struct PingrespPacket { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, } impl<'a> PingrespPacket { - pub fn decode_pingresp_packet(&mut self, buff_reader: &mut BuffReader<'a>) { - if self.decode_fixed_header(buff_reader) != (PacketType::Pingresp).into() { - log::error!("Packet you are trying to decode is not PUBACK packet!"); - return; - } - } } impl<'a> Packet<'a> for PingrespPacket { fn new() -> Self { - todo!() + Self { + fixed_header: PacketType::Pingresp.into(), + remain_len: 0 + } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); - buff_writer.write_u8(self.fixed_header); - buff_writer.write_variable_byte_int(0 as u32); - return buff_writer.position; + 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) ?; + Ok(buff_writer.position) } - fn decode(&mut self, buff_reader: &mut BuffReader<'a>) { - self.decode_pingresp_packet(buff_reader); + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + if self.decode_fixed_header(buff_reader) ? != (PacketType::Pingresp).into() { + log::error!("Packet you are trying to decode is not PINGRESP packet!"); + return Err(BufferError::PacketTypeMismatch); + } + if self.remain_len != 0 { + log::error!("PINGRESP packet does not have 0 lenght!"); + return Err(BufferError::PacketTypeMismatch); + } + Ok(()) } - fn set_property_len(&mut self, value: u32) { + fn set_property_len(&mut self, _value: u32) { log::error!("PINGRESP packet does not contain any properties!"); } @@ -70,7 +74,7 @@ impl<'a> Packet<'a> for PingrespPacket { return 0; } - fn push_to_properties(&mut self, property: Property<'a>) { + fn push_to_properties(&mut self, _property: Property<'a>) { log::error!("PINGRESP packet does not contain any properties!"); } diff --git a/src/packet/property.rs b/src/packet/property.rs index 03aca0b..e729e66 100644 --- a/src/packet/property.rs +++ b/src/packet/property.rs @@ -22,13 +22,9 @@ * SOFTWARE. */ -use crate::encoding::variable_byte_integer::{VariableByteInteger, VariableByteIntegerEncoder}; -use crate::utils::buffer_reader::BinaryData; use crate::utils::buffer_reader::BuffReader; -use crate::utils::buffer_reader::EncodedString; -use crate::utils::buffer_reader::ParseError; -use crate::utils::buffer_reader::StringPair; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::{BinaryData, BufferError, EncodedString, StringPair}; #[derive(Debug)] pub enum Property<'a> { @@ -106,7 +102,7 @@ impl<'a> Property<'a> { }; } - pub fn encode(&self, buff_writer: &mut BuffWriter<'a>) { + pub fn encode(&self, buff_writer: &mut BuffWriter<'a>) -> Result<(), BufferError> { return match self { Property::PayloadFormat(u) => buff_writer.write_u8(*u), Property::MessageExpiryInterval(u) => buff_writer.write_u32(*u), @@ -135,11 +131,11 @@ impl<'a> Property<'a> { Property::WildcardSubscriptionAvailable(u) => buff_writer.write_u8(*u), Property::SubscriptionIdentifierAvailable(u) => buff_writer.write_u8(*u), Property::SharedSubscriptionAvailable(u) => buff_writer.write_u8(*u), - _ => (), + _ => Err(BufferError::PropertyNotFound), }; } - pub fn decode(buff_reader: &mut BuffReader<'a>) -> Result, ParseError> { + pub fn decode(buff_reader: &mut BuffReader<'a>) -> Result, BufferError> { let property_identifier = buff_reader.read_u8(); return match property_identifier { Ok(0x01) => Ok(Property::PayloadFormat(buff_reader.read_u8()?)), @@ -179,7 +175,7 @@ impl<'a> Property<'a> { buff_reader.read_u8()?, )), Err(err) => Err(err), - _ => Err(ParseError::IdNotFound), + _ => Err(BufferError::IdNotFound), }; } } diff --git a/src/packet/puback_packet.rs b/src/packet/puback_packet.rs index 630e2fb..3bbc97d 100644 --- a/src/packet/puback_packet.rs +++ b/src/packet/puback_packet.rs @@ -28,42 +28,22 @@ use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::BufferError; use super::packet_type::PacketType; use super::property::Property; pub struct PubackPacket<'a, const MAX_PROPERTIES: usize> { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, - pub packet_identifier: u16, pub reason_code: u8, - pub property_len: u32, - - // properties pub properties: Vec, MAX_PROPERTIES>, } impl<'a, const MAX_PROPERTIES: usize> PubackPacket<'a, MAX_PROPERTIES> { - pub fn decode_puback_packet(&mut self, buff_reader: &mut BuffReader<'a>) { - if self.decode_fixed_header(buff_reader) != (PacketType::Puback).into() { - log::error!("Packet you are trying to decode is not PUBACK packet!"); - return; - } - self.packet_identifier = buff_reader.read_u16().unwrap(); - if self.remain_len != 2 { - self.reason_code = buff_reader.read_u8().unwrap(); - } - if self.remain_len < 4 { - self.property_len = 0; - } else { - self.decode_properties(buff_reader); - } - } } impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubackPacket<'a, MAX_PROPERTIES> { @@ -78,26 +58,39 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubackPacket<'a, MAX_PROPER } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); + fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { + 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).unwrap(); + 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); - return buff_writer.position; + 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>) { - self.decode_puback_packet(buff_reader); + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + 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() ?; + if self.remain_len != 2 { + self.reason_code = buff_reader.read_u8() ?; + } + if self.remain_len < 4 { + self.property_len = 0; + } else { + self.decode_properties(buff_reader) ?; + } + Ok(()) } fn set_property_len(&mut self, value: u32) { diff --git a/src/packet/pubcomp_packet.rs b/src/packet/pubcomp_packet.rs index 7510a9b..9d65374 100644 --- a/src/packet/pubcomp_packet.rs +++ b/src/packet/pubcomp_packet.rs @@ -28,62 +28,62 @@ use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::BufferError; use super::packet_type::PacketType; use super::property::Property; pub struct PubcompPacket<'a, const MAX_PROPERTIES: usize> { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, - pub packet_identifier: u16, pub reason_code: u8, - pub property_len: u32, - - // properties pub properties: Vec, MAX_PROPERTIES>, } impl<'a, const MAX_PROPERTIES: usize> PubcompPacket<'a, MAX_PROPERTIES> { - pub fn decode_puback_packet(&mut self, buff_reader: &mut BuffReader<'a>) { - if self.decode_fixed_header(buff_reader) != (PacketType::Pubcomp).into() { - log::error!("Packet you are trying to decode is not PUBCOMP packet!"); - return; - } - self.packet_identifier = buff_reader.read_u16().unwrap(); - self.reason_code = buff_reader.read_u8().unwrap(); - self.decode_properties(buff_reader); - } } impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubcompPacket<'a, MAX_PROPERTIES> { fn new() -> Self { - todo!() + Self { + fixed_header: PacketType::Pubcomp.into(), + remain_len: 0, + packet_identifier: 0, + reason_code: 0, + property_len: 0, + properties: Vec::, MAX_PROPERTIES>::new() + } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); + fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { + 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).unwrap(); + 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); - return buff_writer.position; + 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>) { - self.decode_puback_packet(buff_reader); + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + 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) ?; + Ok(()) } fn set_property_len(&mut self, value: u32) { diff --git a/src/packet/publish_packet.rs b/src/packet/publish_packet.rs index d22fce4..cfc9c37 100644 --- a/src/packet/publish_packet.rs +++ b/src/packet/publish_packet.rs @@ -22,15 +22,14 @@ * SOFTWARE. */ -use core::ptr::null; use heapless::Vec; use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; -use crate::packet::publish_packet::QualityOfService::{QoS0, QoS1, QoS2, INVALID}; +use crate::packet::publish_packet::QualityOfService::{INVALID, QoS0, QoS1, QoS2}; use crate::utils::buffer_reader::BuffReader; -use crate::utils::buffer_reader::EncodedString; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::{BufferError, EncodedString}; use super::packet_type::PacketType; use super::property::Property; @@ -66,19 +65,12 @@ impl Into for QualityOfService { } pub struct PublishPacket<'a, const MAX_PROPERTIES: usize> { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, - pub topic_name: EncodedString<'a>, pub packet_identifier: u16, - pub property_len: u32, - - // properties pub properties: Vec, MAX_PROPERTIES>, - pub message: Option<&'a [u8]>, } @@ -99,24 +91,6 @@ impl<'a, const MAX_PROPERTIES: usize> PublishPacket<'a, MAX_PROPERTIES> { pub fn add_identifier(& mut self, identifier: u16) { self.packet_identifier = identifier; } - - pub fn decode_publish_packet(&mut self, buff_reader: &mut BuffReader<'a>) { - if self.decode_fixed_header(buff_reader) != (PacketType::Publish).into() { - log::error!("Packet you are trying to decode is not PUBLISH packet!"); - return; - } - self.topic_name = buff_reader.read_string().unwrap(); - let qos = self.fixed_header & 0x03; - if qos != 0 { - // je potreba dekodovat jenom pro QoS 1 / 2 - self.packet_identifier = buff_reader.read_u16().unwrap(); - } - self.decode_properties(buff_reader); - let mut total_len = VariableByteIntegerEncoder::len( - VariableByteIntegerEncoder::encode(self.remain_len).unwrap()); - total_len = total_len + 1 + self.remain_len as usize; - self.message = Some(buff_reader.read_message(total_len)); - } } impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PublishPacket<'a, MAX_PROPERTIES> { @@ -132,37 +106,52 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PublishPacket<'a, MAX_PROPE } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); + fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { + 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).unwrap(); + VariableByteIntegerEncoder::encode(self.property_len) ?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); - let mut msg_len = self.message.unwrap().len() as u32; + 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()); - return buff_writer.position; + 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>) { - self.decode_publish_packet(buff_reader); + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + 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() ?; + let qos = self.fixed_header & 0x03; + if qos != 0 { + // Decode only for QoS 1 / 2 + self.packet_identifier = buff_reader.read_u16() ?; + } + 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(()) } fn set_property_len(&mut self, value: u32) { diff --git a/src/packet/pubrec_packet.rs b/src/packet/pubrec_packet.rs index 08c7e5c..f2570c7 100644 --- a/src/packet/pubrec_packet.rs +++ b/src/packet/pubrec_packet.rs @@ -28,62 +28,61 @@ use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::BufferError; use super::packet_type::PacketType; use super::property::Property; pub struct PubrecPacket<'a, const MAX_PROPERTIES: usize> { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, - pub packet_identifier: u16, pub reason_code: u8, - pub property_len: u32, - - // properties pub properties: Vec, MAX_PROPERTIES>, } impl<'a, const MAX_PROPERTIES: usize> PubrecPacket<'a, MAX_PROPERTIES> { - pub fn decode_pubrec_packet(&mut self, buff_reader: &mut BuffReader<'a>) { - if self.decode_fixed_header(buff_reader) != (PacketType::Pubrec).into() { - log::error!("Packet you are trying to decode is not PUBREC packet!"); - return; - } - self.packet_identifier = buff_reader.read_u16().unwrap(); - self.reason_code = buff_reader.read_u8().unwrap(); - self.decode_properties(buff_reader); - } } impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubrecPacket<'a, MAX_PROPERTIES> { fn new() -> Self { - todo!() + Self { + fixed_header: PacketType::Pubrec.into(), + remain_len: 0, + packet_identifier: 0, + reason_code: 0, + property_len: 0, + properties: Vec::, MAX_PROPERTIES>::new() + } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); + fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { + 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).unwrap(); + 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); - return buff_writer.position; + 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>) { - self.decode_pubrec_packet(buff_reader); + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + 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() ?; + return self.decode_properties(buff_reader); } fn set_property_len(&mut self, value: u32) { diff --git a/src/packet/pubrel_packet.rs b/src/packet/pubrel_packet.rs index 0d25638..cab1a95 100644 --- a/src/packet/pubrel_packet.rs +++ b/src/packet/pubrel_packet.rs @@ -28,62 +28,61 @@ use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::BufferError; use super::packet_type::PacketType; use super::property::Property; pub struct PubrelPacket<'a, const MAX_PROPERTIES: usize> { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, - pub packet_identifier: u16, pub reason_code: u8, - pub property_len: u32, - - // properties pub properties: Vec, MAX_PROPERTIES>, } impl<'a, const MAX_PROPERTIES: usize> PubrelPacket<'a, MAX_PROPERTIES> { - pub fn decode_puback_packet(&mut self, buff_reader: &mut BuffReader<'a>) { - if self.decode_fixed_header(buff_reader) != (PacketType::Pubrel).into() { - log::error!("Packet you are trying to decode is not PUBREL packet!"); - return; - } - self.packet_identifier = buff_reader.read_u16().unwrap(); - self.reason_code = buff_reader.read_u8().unwrap(); - self.decode_properties(buff_reader); - } } impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubrelPacket<'a, MAX_PROPERTIES> { fn new() -> Self { - todo!() + Self { + fixed_header: 0, + remain_len: 0, + packet_identifier: 0, + reason_code: 0, + property_len: 0, + properties: Vec::, MAX_PROPERTIES>::new() + } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); + fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { + 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).unwrap(); + 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); - return buff_writer.position; + 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>) { - self.decode_puback_packet(buff_reader); + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + 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() ?; + return self.decode_properties(buff_reader); } fn set_property_len(&mut self, value: u32) { diff --git a/src/packet/reason_codes.rs b/src/packet/reason_codes.rs index de5fb8f..1047e87 100644 --- a/src/packet/reason_codes.rs +++ b/src/packet/reason_codes.rs @@ -1,5 +1,4 @@ -use core::fmt::{Display, Formatter, write}; -use crate::packet::reason_codes::ReasonCode::ServerMoved; +use core::fmt::{Display, Formatter}; #[derive(Debug)] pub enum ReasonCode { @@ -46,6 +45,7 @@ pub enum ReasonCode { MaximumConnectTime, SubscriptionIdentifiersNotSupported, WildcardSubscriptionNotSupported, + BuffError, NetworkError, } @@ -95,6 +95,7 @@ impl Into for ReasonCode { ReasonCode::MaximumConnectTime => 0xA0, ReasonCode::SubscriptionIdentifiersNotSupported => 0xA1, ReasonCode::WildcardSubscriptionNotSupported => 0xA2, + ReasonCode::BuffError => 0xFE, ReasonCode::NetworkError => 0xFF } @@ -146,6 +147,7 @@ impl From for ReasonCode { 0xA0 => ReasonCode::MaximumConnectTime, 0xA1 => ReasonCode::SubscriptionIdentifiersNotSupported, 0xA2 => ReasonCode::WildcardSubscriptionNotSupported, + 0xFE => ReasonCode::BuffError, _ => ReasonCode::NetworkError } } @@ -197,6 +199,7 @@ impl Display for ReasonCode { 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::NetworkError => write!(f, "Unknown error!"), } } diff --git a/src/packet/suback_packet.rs b/src/packet/suback_packet.rs index bcd7878..b765888 100644 --- a/src/packet/suback_packet.rs +++ b/src/packet/suback_packet.rs @@ -24,52 +24,35 @@ use heapless::Vec; -use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; -use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::BufferError; use super::packet_type::PacketType; use super::property::Property; pub struct SubackPacket<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: usize> { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, - pub packet_identifier: u16, - pub property_len: u32, - - // properties pub properties: Vec, MAX_PROPERTIES>, - pub reason_codes: Vec, } 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>) { + 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().unwrap()); + self.reason_codes.push(buff_reader.read_u8() ?); i = i + 1; if i == MAX_REASONS { break; } } - } - - pub fn decode_suback_packet(&mut self, buff_reader: &mut BuffReader<'a>) { - if self.decode_fixed_header(buff_reader) != (PacketType::Suback).into() { - log::error!("Packet you are trying to decode is not SUBACK packet!"); - return; - } - self.packet_identifier = buff_reader.read_u16().unwrap(); - self.decode_properties(buff_reader); - self.read_reason_codes(buff_reader); + return Ok(()) } } @@ -87,13 +70,19 @@ impl<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: usize> Packet<'a> } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { + fn encode(&mut self, _buffer: &mut [u8], _buffer_len: usize) -> Result { log::error!("SUBACK packet does not support encoding!"); - return 0; + return Err(BufferError::WrongPacketToEncode) } - fn decode(&mut self, buff_reader: &mut BuffReader<'a>) { - self.decode_suback_packet(buff_reader); + 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) ?; + return self.read_reason_codes(buff_reader); } fn set_property_len(&mut self, value: u32) { diff --git a/src/packet/subscription_packet.rs b/src/packet/subscription_packet.rs index e32ea8d..52bd6a2 100644 --- a/src/packet/subscription_packet.rs +++ b/src/packet/subscription_packet.rs @@ -28,29 +28,18 @@ 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_reader::TopicFilter; 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> { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, - pub packet_identifier: u16, - pub property_len: u32, - - // properties pub properties: Vec, MAX_PROPERTIES>, - - // topic filter len pub topic_filter_len: u16, - - // payload pub topic_filters: Vec, MAX_FILTERS>, } @@ -72,7 +61,7 @@ impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> Packet<'a> for SubscriptionPacket<'a, MAX_FILTERS, MAX_PROPERTIES> { fn new() -> Self { - let mut x = Self { + let x = Self { fixed_header: PacketType::Subscribe.into(), remain_len: 0, packet_identifier: 1, @@ -84,12 +73,12 @@ impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> Packet<'a> return x; } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); + fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { + 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).unwrap(); + VariableByteIntegerEncoder::encode(self.property_len) ?; let property_len_len = VariableByteIntegerEncoder::len(property_len_enc); let mut lt = 0; @@ -103,21 +92,22 @@ 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, - ); - return buff_writer.position; + ) ?; + Ok(buff_writer.position) } - fn decode(&mut self, buff_reader: &mut BuffReader<'a>) { + fn decode(&mut self, _buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { log::error!("Subscribe packet does not support decode funtion on client!"); + Err(BufferError::WrongPacketToDecode) } fn set_property_len(&mut self, value: u32) { self.property_len = value; diff --git a/src/packet/unsuback_packet.rs b/src/packet/unsuback_packet.rs index b834a33..79a8b05 100644 --- a/src/packet/unsuback_packet.rs +++ b/src/packet/unsuback_packet.rs @@ -26,48 +26,33 @@ use heapless::Vec; use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; +use crate::utils::types::BufferError; use super::packet_type::PacketType; use super::property::Property; pub struct UnsubackPacket<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: usize> { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, - pub packet_identifier: u16, - pub property_len: u32, - - // properties pub properties: Vec, MAX_PROPERTIES>, - pub reason_codes: Vec, } 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>) { + 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().unwrap()); + self.reason_codes.push(buff_reader.read_u8() ?); i = i + 1; if i == MAX_REASONS { break; } } - } - - pub fn decode_suback_packet(&mut self, buff_reader: &mut BuffReader<'a>) { - if self.decode_fixed_header(buff_reader) != (PacketType::Suback).into() { - log::error!("Packet you are trying to decode is not UNSUBACK packet!"); - return; - } - self.packet_identifier = buff_reader.read_u16().unwrap(); - self.decode_properties(buff_reader); - self.read_reason_codes(buff_reader); + Ok(()) } } @@ -75,16 +60,29 @@ impl<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: usize> Packet<'a> for UnsubackPacket<'a, MAX_REASONS, MAX_PROPERTIES> { fn new() -> Self { - todo!() + Self { + fixed_header: PacketType::Unsuback.into(), + remain_len: 0, + packet_identifier: 0, + property_len: 0, + properties: Vec::, MAX_PROPERTIES>::new(), + reason_codes: Vec::::new() + } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { + fn encode(&mut self, _buffer: &mut [u8], _buffer_len: usize) -> Result { log::error!("UNSUBACK packet does not support encoding!"); - return 0; + Err(BufferError::WrongPacketToEncode) } - fn decode(&mut self, buff_reader: &mut BuffReader<'a>) { - self.decode_suback_packet(buff_reader); + fn decode(&mut self, buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { + 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) ?; + return self.read_reason_codes(buff_reader); } fn set_property_len(&mut self, value: u32) { diff --git a/src/packet/unsubscription_packet.rs b/src/packet/unsubscription_packet.rs index 1d374d7..f352010 100644 --- a/src/packet/unsubscription_packet.rs +++ b/src/packet/unsubscription_packet.rs @@ -27,29 +27,18 @@ use heapless::Vec; use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder; use crate::packet::mqtt_packet::Packet; use crate::utils::buffer_reader::BuffReader; -use crate::utils::buffer_reader::TopicFilter; use crate::utils::buffer_writer::BuffWriter; +use crate::utils::types::{BufferError, TopicFilter}; -use super::packet_type::PacketType; use super::property::Property; pub struct UnsubscriptionPacket<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> { - // 7 - 4 mqtt control packet type, 3-0 flagy pub fixed_header: u8, - // 1 - 4 B lenght of variable header + len of payload pub remain_len: u32, - pub packet_identifier: u16, - pub property_len: u32, - - // properties pub properties: Vec, MAX_PROPERTIES>, - - // topic filter len pub topic_filter_len: u16, - - // payload pub topic_filters: Vec, MAX_FILTERS>, } @@ -62,34 +51,43 @@ impl<'a, const MAX_FILTERS: usize, const MAX_PROPERTIES: usize> Packet<'a> for UnsubscriptionPacket<'a, MAX_FILTERS, MAX_PROPERTIES> { fn new() -> Self { - todo!() + Self { + fixed_header: 0, + remain_len: 0, + packet_identifier: 0, + property_len: 0, + properties: Vec::, MAX_PROPERTIES>::new(), + topic_filter_len: 0, + topic_filters: Vec::, MAX_FILTERS>::new() + } } - fn encode(&mut self, buffer: &mut [u8]) -> usize { - let mut buff_writer = BuffWriter::new(buffer); + fn encode(&mut self, buffer: &mut [u8], buffer_len: usize) -> Result { + 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).unwrap(); + 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, - ); - return buff_writer.position; + ) ?; + Ok(buff_writer.position) } - fn decode(&mut self, buff_reader: &mut BuffReader<'a>) { + fn decode(&mut self, _buff_reader: &mut BuffReader<'a>) -> Result<(), BufferError> { log::error!("Unsubscribe packet does not support decode funtion on client!"); + Err(BufferError::WrongPacketToDecode) } fn set_property_len(&mut self, value: u32) { diff --git a/src/tokio_network.rs b/src/tokio_network.rs index 5471f44..2beb87f 100644 --- a/src/tokio_network.rs +++ b/src/tokio_network.rs @@ -1,15 +1,11 @@ use alloc::format; use alloc::string::String; -use core::borrow::BorrowMut; -use core::fmt::Error; use core::future::Future; -use core::ptr::null; use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::net::{TcpListener, TcpStream}; +use tokio::net::TcpStream; -use crate::network::network_trait::{Network}; -use crate::packet::mqtt_packet::Packet; +use crate::network::network_trait::Network; use crate::packet::reason_codes::ReasonCode; pub struct TokioNetwork { diff --git a/src/utils/buffer_reader.rs b/src/utils/buffer_reader.rs index 9ecf40e..40db7bd 100644 --- a/src/utils/buffer_reader.rs +++ b/src/utils/buffer_reader.rs @@ -26,93 +26,14 @@ use core::mem; use core::str; use crate::encoding::variable_byte_integer::VariableByteIntegerDecoder; - -/// Encoded string provides structure representing UTF-8 encoded string in MQTTv5 packets -#[derive(Debug)] -#[derive(Clone)] -pub struct EncodedString<'a> { - pub string: &'a str, - pub len: u16, -} - -impl EncodedString<'_> { - pub fn new() -> Self { - Self { string: "", len: 0 } - } - /// Return length of string - pub fn len(&self) -> u16 { - return self.len + 2; - } -} - -/// Binary data represents `Binary data` in MQTTv5 protocol -#[derive(Debug)] -#[derive(Clone)] -pub struct BinaryData<'a> { - pub bin: &'a [u8], - pub len: u16, -} - -impl BinaryData<'_> { - pub fn new() -> Self { - Self { bin: &[0], len: 0 } - } - /// Returns length of Byte array - pub fn len(&self) -> u16 { - return self.len + 2; - } -} - -/// String pair struct represents `String pair` in MQTTv5 (2 UTF-8 encoded strings name-value) -#[derive(Debug)] -pub struct StringPair<'a> { - pub name: EncodedString<'a>, - pub value: EncodedString<'a>, -} - -impl StringPair<'_> { - /// Returns length which is equal to sum of the lenghts of UTF-8 encoded strings in pair - pub fn len(&self) -> u16 { - let ln = self.name.len() + self.value.len(); - return ln; - } -} - -/// Topic filter serves as bound for topic selection and subscription options for `SUBSCRIPTION` packet -#[derive(Debug)] -pub struct TopicFilter<'a> { - pub filter: EncodedString<'a>, - pub sub_options: u8, -} - -impl TopicFilter<'_> { - pub fn new() -> Self { - Self { - filter: EncodedString::new(), - sub_options: 0, - } - } - - pub fn len(&self) -> u16 { - return self.filter.len + 3; - } -} - -#[derive(core::fmt::Debug, Clone)] -pub enum ParseError { - Utf8Error, - IndexOutOfBounce, - VariableByteIntegerError, - IdNotFound, - EncodingError, - DecodingError, -} +use crate::utils::types::{BinaryData, BufferError, EncodedString, StringPair}; /// Buff reader is reading corresponding types from buffer (Byte array) and stores current position /// (later as cursor) pub struct BuffReader<'a> { buffer: &'a [u8], pub position: usize, + len: usize, } impl<'a> BuffReader<'a> { @@ -120,16 +41,17 @@ impl<'a> BuffReader<'a> { self.position = self.position + increment; } - pub fn new(buffer: &'a [u8]) -> Self { + pub fn new(buffer: &'a [u8], buff_len: usize) -> Self { return BuffReader { - buffer: buffer, + buffer, position: 0, + len: buff_len }; } /// Variable byte integer can be 1-4 Bytes long. Buffer reader takes all 4 Bytes at first and /// than check what is true length of varbyteint and increment cursor by that - pub fn read_variable_byte_int(&mut self) -> Result { + pub fn read_variable_byte_int(&mut self) -> Result { let variable_byte_integer: [u8; 4] = [ self.buffer[self.position], self.buffer[self.position + 1], @@ -137,7 +59,7 @@ impl<'a> BuffReader<'a> { self.buffer[self.position + 3], ]; let mut len: usize = 1; - /// Everytime checking first bit of Byte which determines whenever there is continous Byte + // Everytime checking first bit of Byte which determines whenever there is continous Byte if variable_byte_integer[0] & 0x80 == 1 { len = len + 1; if variable_byte_integer[1] & 0x80 == 1 { @@ -152,30 +74,30 @@ impl<'a> BuffReader<'a> { } /// Reading u32 from buffer as `Big endian` - pub fn read_u32(&mut self) -> Result { - let (int_bytes, rest) = self.buffer[self.position..].split_at(mem::size_of::()); + pub fn read_u32(&mut self) -> Result { + let (int_bytes, _rest) = self.buffer[self.position..].split_at(mem::size_of::()); let ret: u32 = u32::from_be_bytes(int_bytes.try_into().unwrap()); self.increment_position(4); return Ok(ret); } /// Reading u16 from buffer as `Big endinan` - pub fn read_u16(&mut self) -> Result { - let (int_bytes, rest) = self.buffer[self.position..].split_at(mem::size_of::()); + pub fn read_u16(&mut self) -> Result { + let (int_bytes, _rest) = self.buffer[self.position..].split_at(mem::size_of::()); let ret: u16 = u16::from_be_bytes(int_bytes.try_into().unwrap()); self.increment_position(2); return Ok(ret); } /// Reading one byte from buffer as `Big endian` - pub fn read_u8(&mut self) -> Result { + pub fn read_u8(&mut self) -> Result { let ret: u8 = self.buffer[self.position]; self.increment_position(1); return Ok(ret); } /// Reading UTF-8 encoded string from buffer - pub fn read_string(&mut self) -> Result, ParseError> { + pub fn read_string(&mut self) -> Result, BufferError> { let len = self.read_u16(); match len { Err(err) => return Err(err), @@ -186,7 +108,7 @@ impl<'a> BuffReader<'a> { str::from_utf8(&(self.buffer[self.position..(self.position + len_res as usize)])); if res_str.is_err() { log::error!("Could not parse utf-8 string"); - return Err(ParseError::Utf8Error); + return Err(BufferError::Utf8Error); } self.increment_position(len_res as usize); return Ok(EncodedString { @@ -195,24 +117,18 @@ impl<'a> BuffReader<'a> { }); } - //TODO: Index out of bounce err !!!!! /// Read Binary data from buffer - pub fn read_binary(&mut self) -> Result, ParseError> { - let len = self.read_u16(); - match len { - Err(err) => return Err(err), - _ => log::debug!("[parseBinary] let not parsed"), - } - let len_res = len.unwrap(); - let res_bin = &(self.buffer[self.position..(self.position + len_res as usize)]); + pub fn read_binary(&mut self) -> Result, BufferError> { + let len = self.read_u16() ?; + let res_bin = &(self.buffer[self.position..(self.position + len as usize)]); return Ok(BinaryData { bin: res_bin, - len: len_res, + len: len, }); } /// Read string pair from buffer - pub fn read_string_pair(&mut self) -> Result, ParseError> { + pub fn read_string_pair(&mut self) -> Result, BufferError> { let name = self.read_string(); match name { Err(err) => return Err(err), @@ -231,6 +147,10 @@ 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]; + } return &self.buffer[self.position..total_len]; } } diff --git a/src/utils/buffer_writer.rs b/src/utils/buffer_writer.rs index 2f2f8af..2a9fa3f 100644 --- a/src/utils/buffer_writer.rs +++ b/src/utils/buffer_writer.rs @@ -22,22 +22,36 @@ * SOFTWARE. */ -use core::str; - use heapless::Vec; use crate::encoding::variable_byte_integer::{VariableByteInteger, VariableByteIntegerEncoder}; use crate::packet::property::Property; -use crate::utils::buffer_reader::{BinaryData, EncodedString, StringPair, TopicFilter}; +use crate::utils::types::{BinaryData, EncodedString, StringPair, BufferError, TopicFilter}; pub struct BuffWriter<'a> { buffer: &'a mut [u8], pub position: usize, + len: usize } impl<'a> BuffWriter<'a> { - pub fn insert_ref(&mut self, len: usize, array: &[u8]) { + pub fn new(buffer: &'a mut [u8], buff_len: usize) -> Self { + return BuffWriter { + buffer, + position: 0, + len: buff_len + }; + } + + fn increment_position(&mut self, increment: usize) { + self.position = self.position + increment; + } + + pub fn insert_ref(&mut self, len: usize, array: &[u8]) -> Result<(), BufferError> { let mut x: usize = 0; + if self.position + 3 >= self.len { + return Err(BufferError::InsufficientBufferSize); + } if len != 0 { loop { self.buffer[self.position] = array[x]; @@ -48,98 +62,96 @@ impl<'a> BuffWriter<'a> { } } } + return Ok(()); } - pub fn new(buffer: &'a mut [u8]) -> Self { - return BuffWriter { - buffer, - position: 0, - }; + pub fn write_u8(&mut self, byte: u8) -> Result<(), BufferError> { + return if self.position >= self.len { + Err(BufferError::InsufficientBufferSize) + } else { + self.buffer[self.position] = byte; + self.increment_position(1); + Ok(()) + } + } - fn increment_position(&mut self, increment: usize) { - self.position = self.position + increment; - } - - pub fn write_u8(&mut self, byte: u8) { - self.buffer[self.position] = byte; - self.increment_position(1); - } - - pub fn write_u16(&mut self, two_bytes: u16) { + pub fn write_u16(&mut self, two_bytes: u16) -> Result<(), BufferError> { let bytes: [u8; 2] = two_bytes.to_be_bytes(); - self.insert_ref(2, &bytes); + return self.insert_ref(2, &bytes); } - pub fn write_u32(&mut self, four_bytes: u32) { + pub fn write_u32(&mut self, four_bytes: u32) -> Result<(), BufferError> { let bytes: [u8; 4] = four_bytes.to_be_bytes(); - self.insert_ref(4, &bytes); + return self.insert_ref(4, &bytes); } - pub fn write_string_ref(&mut self, str: &EncodedString<'a>) { - self.write_u16(str.len); + pub fn write_string_ref(&mut self, str: &EncodedString<'a>) -> Result<(), BufferError> { + self.write_u16(str.len) ?; let bytes = str.string.as_bytes(); - self.insert_ref(str.len as usize, bytes); + return self.insert_ref(str.len as usize, bytes); } - pub fn write_string(&mut self, str: EncodedString<'a>) { - self.write_u16(str.len); + pub fn write_string(&mut self, str: EncodedString<'a>) -> Result<(), BufferError> { + self.write_u16(str.len) ?; let bytes = str.string.as_bytes(); - self.insert_ref(str.len as usize, bytes); + return self.insert_ref(str.len as usize, bytes); } - pub fn write_binary_ref(&mut self, bin: &BinaryData<'a>) { - self.write_u16(bin.len); - self.insert_ref(bin.len as usize, bin.bin); + pub fn write_binary_ref(&mut self, bin: &BinaryData<'a>) -> Result<(), BufferError> { + self.write_u16(bin.len) ?; + return self.insert_ref(bin.len as usize, bin.bin); } - pub fn write_binary(&mut self, bin: BinaryData<'a>) { - self.write_u16(bin.len); - 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) ?; + return self.insert_ref(bin.len as usize, bin.bin); } - pub fn write_string_pair_ref(&mut self, str_pair: &StringPair<'a>) { - self.write_string_ref(&str_pair.name); - self.write_string_ref(&str_pair.value); + pub fn write_string_pair_ref(&mut self, str_pair: &StringPair<'a>) -> Result<(), BufferError> { + 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>) { - self.write_string(str_pair.name); - self.write_string(str_pair.value); + pub fn write_string_pair(&mut self, str_pair: StringPair<'a>) -> Result<(), BufferError> { + self.write_string(str_pair.name) ?; + return self.write_string(str_pair.value); } - pub fn write_variable_byte_int(&mut self, int: u32) { - let x: VariableByteInteger = VariableByteIntegerEncoder::encode(int).unwrap(); + pub fn write_variable_byte_int(&mut self, int: u32) -> Result<(), BufferError> { + let x: VariableByteInteger = VariableByteIntegerEncoder::encode(int) ?; let len = VariableByteIntegerEncoder::len(x); - self.insert_ref(len, &x); + return self.insert_ref(len, &x); } - pub fn encode_property(&mut self, property: &Property<'a>) { + pub fn encode_property(&mut self, property: &Property<'a>) -> Result<(), BufferError> { let x: u8 = property.into(); - self.write_u8(x); - property.encode(self); + self.write_u8(x) ?; + return property.encode(self); } - pub fn encode_properties(&mut self, properties: &Vec, LEN>) { + 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(); - self.encode_property(prop); + let prop: &Property = properties.get(i).unwrap_or(&Property::Reserved()); + self.encode_property(prop) ?; i = i + 1; if i == len { break; } } } + Ok(()) } - fn encode_topic_filter_ref(&mut self, sub: bool, topic_filter: &TopicFilter<'a>) { - 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(()) } pub fn encode_topic_filters_ref( @@ -147,15 +159,16 @@ impl<'a> BuffWriter<'a> { sub: bool, len: usize, filters: &Vec, MAX>, - ) { + ) -> Result<(), BufferError> { 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; } } + Ok(()) } } diff --git a/src/utils/mod.rs b/src/utils/mod.rs index de04e4c..82a72fe 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -25,3 +25,4 @@ pub mod buffer_reader; pub mod buffer_writer; pub mod rng_generator; +pub mod types; diff --git a/src/utils/rng_generator.rs b/src/utils/rng_generator.rs index 2e96e02..b1c11e2 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::{RngCore, Error, impls}; +use rand_core::{Error, impls, RngCore}; pub struct CountingRng(pub u64); diff --git a/src/utils/types.rs b/src/utils/types.rs new file mode 100644 index 0000000..0c7a2b8 --- /dev/null +++ b/src/utils/types.rs @@ -0,0 +1,126 @@ +/* + * MIT License + * + * Copyright (c) [2022] [Ondrej Babec ] + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +use core::fmt::{Display, Formatter}; + +#[derive(core::fmt::Debug, Clone)] +pub enum BufferError { + Utf8Error, + InsufficientBufferSize, + VariableByteIntegerError, + IdNotFound, + EncodingError, + DecodingError, + PacketTypeMismatch, + WrongPacketToDecode, + WrongPacketToEncode, + PropertyNotFound +} + +impl Display for BufferError { + fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result { + match *self { + BufferError::Utf8Error => write!(f, "Error encountered during UTF8 decoding!"), + BufferError::InsufficientBufferSize => write!(f, "Buffer size is not sufficient for packet!"), + BufferError::VariableByteIntegerError => write!(f, "Error encountered during variable byte integer decoding / encoding!"), + BufferError::IdNotFound => write!(f, "Packet identifier not found!"), + BufferError::EncodingError => write!(f, "Error encountered during packet encoding!"), + BufferError::DecodingError => write!(f, "Error encountered during packet decoding!"), + BufferError::PacketTypeMismatch => write!(f, "Packet type not matched during decoding (Received different packet type than encode type)!"), + BufferError::WrongPacketToDecode => write!(f, "Not able to decode packet, this packet is used just for sending to broker, not receiving by client!"), + BufferError::WrongPacketToEncode => write!(f, "Not able to encode packet, this packet is used only from server to client not the opposite way!"), + BufferError::PropertyNotFound => write!(f, "Property with ID not found!") + } + } +} +/// Encoded string provides structure representing UTF-8 encoded string in MQTTv5 packets +#[derive(Debug)] +#[derive(Clone)] +pub struct EncodedString<'a> { + pub string: &'a str, + pub len: u16, +} + +impl EncodedString<'_> { + pub fn new() -> Self { + Self { string: "", len: 0 } + } + /// Return length of string + pub fn len(&self) -> u16 { + return self.len + 2; + } +} + +/// Binary data represents `Binary data` in MQTTv5 protocol +#[derive(Debug)] +#[derive(Clone)] +pub struct BinaryData<'a> { + pub bin: &'a [u8], + pub len: u16, +} + +impl BinaryData<'_> { + pub fn new() -> Self { + Self { bin: &[0], len: 0 } + } + /// Returns length of Byte array + pub fn len(&self) -> u16 { + return self.len + 2; + } +} + +/// String pair struct represents `String pair` in MQTTv5 (2 UTF-8 encoded strings name-value) +#[derive(Debug)] +pub struct StringPair<'a> { + pub name: EncodedString<'a>, + pub value: EncodedString<'a>, +} + +impl StringPair<'_> { + /// Returns length which is equal to sum of the lenghts of UTF-8 encoded strings in pair + pub fn len(&self) -> u16 { + let ln = self.name.len() + self.value.len(); + return ln; + } +} + +/// Topic filter serves as bound for topic selection and subscription options for `SUBSCRIPTION` packet +#[derive(Debug)] +pub struct TopicFilter<'a> { + pub filter: EncodedString<'a>, + pub sub_options: u8, +} + +impl TopicFilter<'_> { + pub fn new() -> Self { + Self { + filter: EncodedString::new(), + sub_options: 0, + } + } + + pub fn len(&self) -> u16 { + return self.filter.len + 3; + } +} \ No newline at end of file