Dreaded_X/rust-mqtt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #3 from obabec/error-delegation

Browse Source 
Error delegation
This commit is contained in:
obabec 2022-03-04 15:40:57 +01:00 committed by GitHub
commit ff287dc451
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
37 changed files with 982 additions and 809 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -746,6 +746,7 @@ dependencies = [
"env_logger",
"heapless",
"log",
"rand_core",
"tokio",
]

1
Cargo.toml
View File

@ -14,6 +14,7 @@ env_logger = "0.9.0"
log = "0.4.14"
heapless = "0.7.10"
tokio = { version = "1", features = ["full"] }
rand_core = "0.6.0"
[patch.crates-io]
embassy = { git = "https://github.com/embassy-rs/embassy.git", rev = "d76cd5ceaf5140c48ef97180beae156c0c0e07c8" }

15
src/client/client_config.rs
View File

@ -22,16 +22,15 @@
* SOFTWARE.
*/
use crate::packet::publish_packet::QualityOfService;
use crate::utils::buffer_reader::{BinaryData, EncodedString};
use crate::packet::v5::publish_packet::QualityOfService;
use crate::utils::types::{BinaryData, EncodedString};
pub struct ClientConfig<'a> {
pub qos: QualityOfService,
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;
}
}
}

257
src/client/client_v5.rs
View File

@ -1,26 +1,29 @@
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::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::reason_codes::ReasonCode;
use crate::packet::suback_packet::SubackPacket;
use crate::packet::subscription_packet::SubscriptionPacket;
use crate::network::network_trait::Network;
use crate::packet::v5::connack_packet::ConnackPacket;
use crate::packet::v5::connect_packet::ConnectPacket;
use crate::packet::v5::disconnect_packet::DisconnectPacket;
use crate::packet::v5::mqtt_packet::Packet;
use crate::packet::v5::pingreq_packet::PingreqPacket;
use crate::packet::v5::pingresp_packet::PingrespPacket;
use crate::packet::v5::puback_packet::PubackPacket;
use crate::packet::v5::publish_packet::QualityOfService::QoS1;
use crate::packet::v5::publish_packet::{PublishPacket, QualityOfService};
use crate::packet::v5::reason_codes::ReasonCode;
use crate::packet::v5::suback_packet::SubackPacket;
use crate::packet::v5::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,
buffer: &'a mut [u8],
buffer_len: usize,
recv_buffer: &'a mut [u8],
recv_buffer_len: usize,
rng: CountingRng,
config: ClientConfig<'a>,
}
@ -29,50 +32,74 @@ 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
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);
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)
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<u8, BufferError> = {
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(())
}
}
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(())
}
@ -81,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();
@ -89,74 +115,191 @@ 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 <QualityOfService as Into<u8>>::into(self.config.qos ) == <QualityOfService as Into<u8>>::into(QoS1) {
let reason = {
self.network_driver.receive(self.buffer).await ?;
if <QualityOfService as Into<u8>>::into(self.config.qos)
== <QualityOfService as Into<u8>>::into(QoS1)
{
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(())
}
// 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<Vec<u8, TOPICS>, 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);
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.receive(self.buffer).await ?;
self.network_driver.send(self.buffer, len.unwrap()).await?;
let reason: Result<u8, BufferError> = {
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 != (<QualityOfService as Into<u8>>::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 != (<QualityOfService as Into<u8>>::into(self.config.qos) >> 1) {
Err(ReasonCode::from(res))
} 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();
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) == <QualityOfService as Into<u8>>::into(QualityOfService::QoS1) {
if (packet.fixed_header & 0x06)
== <QualityOfService as Into<u8>>::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?;
}
}
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(())
}
}
}

3
src/client/mod.rs
View File

@ -22,6 +22,5 @@
* SOFTWARE.
*/
pub mod client_v5;
pub mod client_config;
pub mod client_v5;

1
src/encoding/mod.rs
View File

@ -22,5 +22,4 @@
* SOFTWARE.
*/
pub mod variable_byte_integer;

14
src/encoding/variable_byte_integer.rs
View File

@ -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<VariableByteInteger, ParseError> {
pub fn encode(mut target: u32) -> Result<VariableByteInteger, BufferError> {
// 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<u32, ParseError> {
pub fn decode(encoded: VariableByteInteger) -> Result<u32, BufferError> {
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 {

77
src/main.rs
View File

@ -1,15 +1,19 @@
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::subscription_packet::SubscriptionPacket;
use rust_mqtt::tokio_network::TokioNetwork;
use std::time::Duration;
use heapless::Vec;
use tokio::time::sleep;
use tokio::{join, task};
use rust_mqtt::client::client_config::ClientConfig;
use rust_mqtt::packet::publish_packet::QualityOfService::QoS1;
use rust_mqtt::client::client_v5::MqttClientV5;
use rust_mqtt::network::network_trait::{Network, NetworkError};
use rust_mqtt::packet::v5::connect_packet::ConnectPacket;
use rust_mqtt::packet::v5::mqtt_packet::Packet;
use rust_mqtt::packet::v5::publish_packet::QualityOfService::QoS1;
use rust_mqtt::packet::v5::publish_packet::{PublishPacket, QualityOfService};
use rust_mqtt::packet::v5::subscription_packet::SubscriptionPacket;
use rust_mqtt::tokio_network::TokioNetwork;
async fn receive() {
let mut ip: [u8; 4] = [37, 205, 11, 180];
@ -19,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::<TokioNetwork, 5>::new(&mut tokio_network, &mut res2, & mut res3, config);
let mut client = MqttClientV5::<TokioNetwork, 5>::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::<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;
@ -54,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::<TokioNetwork, 5>::new(&mut tokio_network, &mut res2, & mut res3, config);
let mut client = MqttClientV5::<TokioNetwork, 5>::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 :)")
@ -77,7 +111,7 @@ async fn publish(message: &str) {
};
if let Err(err) = result {
log::error!("Chyba!");
}
}*/
{
client.disconnect().await;
@ -101,6 +135,7 @@ async fn main() {
join!(recv, publ);*/
receive().await;
//publish("Ahoj 123").await;
log::info!("Done");
}

1
src/network/mod.rs
View File

@ -22,5 +22,4 @@
* SOFTWARE.
*/
pub mod network_trait;

4
src/network/network_trait.rs
View File

@ -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::v5::reason_codes::ReasonCode;
#[derive(Debug)]
pub enum NetworkError {

21
src/packet/mod.rs
View File

@ -22,23 +22,4 @@
* SOFTWARE.
*/
pub mod auth_packet;
pub mod connack_packet;
pub mod mqtt_packet;
pub mod packet_type;
pub mod property;
pub mod puback_packet;
pub mod pubcomp_packet;
pub mod publish_packet;
pub mod pubrec_packet;
pub mod pubrel_packet;
pub mod subscription_packet;
pub mod unsubscription_packet;
pub mod connect_packet;
pub mod disconnect_packet;
pub mod pingreq_packet;
pub mod pingresp_packet;
pub mod suback_packet;
pub mod unsuback_packet;
pub mod reason_codes;
pub mod v5;

38
src/packet/auth_packet.rs → src/packet/v5/auth_packet.rs
View File

@ -22,13 +22,13 @@
* SOFTWARE.
*/
use heapless::Vec;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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 +44,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!");
@ -75,30 +68,31 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for AuthPacket<'a, MAX_PROPERTI
remain_len: 0,
auth_reason: 0x00,
property_len: 0,
properties: Vec::<Property<'a>, MAX_PROPERTIES>::new()
properties: Vec::<Property<'a>, 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], buff_len: usize) -> Result<usize, BufferError> {
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();
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::<MAX_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::<MAX_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) {

45
src/packet/connack_packet.rs → src/packet/v5/connack_packet.rs
View File

@ -25,17 +25,16 @@
use heapless::Vec;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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,17 @@ 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 +69,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<usize, BufferError> {
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) {

136
src/packet/connect_packet.rs → src/packet/v5/connect_packet.rs
View File

@ -25,35 +25,26 @@
use heapless::Vec;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::mqtt_packet::Packet;
use crate::utils::buffer_reader::BinaryData;
use crate::packet::v5::mqtt_packet::Packet;
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<Property<'a>, MAX_PROPERTIES>,
//payload
pub client_id: EncodedString<'a>,
// property len
pub will_property_len: u32,
pub will_properties: Vec<Property<'a>, 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 | <PacketType as Into<u8>>::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,39 @@ 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::<Property<'a>, MAX_PROPERTIES>::new(),
client_id: EncodedString::new(),
will_property_len: 0,
will_properties: Vec::<Property<'a>, 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<usize, BufferError> {
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();
let property_len_enc: [u8; 4] = 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();
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
@ -160,7 +142,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 +150,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::<MAX_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::<MAX_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 +204,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<Property, ParseError>;
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;
}
}
}
}

39
src/packet/disconnect_packet.rs → src/packet/v5/disconnect_packet.rs
View File

@ -25,9 +25,10 @@
use heapless::Vec;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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,15 +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 +63,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<usize, BufferError> {
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) {

44
src/packet/v5/mod.rs Normal file
View File

@ -0,0 +1,44 @@
/*
* MIT License
*
* Copyright (c) [2022] [Ondrej Babec <ond.babec@gmail.com>]
*
* 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.
*/
pub mod auth_packet;
pub mod connack_packet;
pub mod mqtt_packet;
pub mod packet_type;
pub mod property;
pub mod puback_packet;
pub mod pubcomp_packet;
pub mod publish_packet;
pub mod pubrec_packet;
pub mod pubrel_packet;
pub mod subscription_packet;
pub mod unsubscription_packet;
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;

38
src/packet/mqtt_packet.rs → src/packet/v5/mqtt_packet.rs
View File

@ -22,19 +22,20 @@
* SOFTWARE.
*/
use crate::packet::packet_type::PacketType;
use crate::packet::v5::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<usize, BufferError>;
/// 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,33 @@ 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<Property, ParseError>;
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<PacketType, BufferError> {
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));
}
}

0
src/packet/packet_type.rs → src/packet/v5/packet_type.rs
View File

29
src/packet/pingreq_packet.rs → src/packet/v5/pingreq_packet.rs
View File

@ -22,17 +22,16 @@
* SOFTWARE.
*/
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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<usize, BufferError> {
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!");
}

46
src/packet/pingresp_packet.rs → src/packet/v5/pingresp_packet.rs
View File

@ -22,46 +22,50 @@
* SOFTWARE.
*/
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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> PingrespPacket {}
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<usize, BufferError> {
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> {
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);
}
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!");
}

14
src/packet/property.rs → src/packet/v5/property.rs
View File

@ -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<Property<'a>, ParseError> {
pub fn decode(buff_reader: &mut BuffReader<'a>) -> Result<Property<'a>, 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),
};
}
}

68
src/packet/puback_packet.rs → src/packet/v5/puback_packet.rs
View File

@ -25,46 +25,24 @@
use heapless::Vec;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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<Property<'a>, 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> PubackPacket<'a, MAX_PROPERTIES> {}
impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubackPacket<'a, MAX_PROPERTIES> {
fn new() -> Self {
@ -74,30 +52,42 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PubackPacket<'a, MAX_PROPER
packet_identifier: 0,
reason_code: 0,
property_len: 0,
properties: Vec::<Property<'a>, MAX_PROPERTIES>::new()
properties: Vec::<Property<'a>, 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<usize, BufferError> {
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();
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::<MAX_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::<MAX_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) {

62
src/packet/pubcomp_packet.rs → src/packet/v5/pubcomp_packet.rs
View File

@ -25,65 +25,63 @@
use heapless::Vec;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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<Property<'a>, 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> PubcompPacket<'a, MAX_PROPERTIES> {}
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::<Property<'a>, 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<usize, BufferError> {
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();
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::<MAX_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::<MAX_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) {

84
src/packet/publish_packet.rs → src/packet/v5/publish_packet.rs
View File

@ -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::v5::mqtt_packet::Packet;
use crate::packet::v5::publish_packet::QualityOfService::{QoS0, QoS1, QoS2, INVALID};
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<u8> 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<Property<'a>, MAX_PROPERTIES>,
pub message: Option<&'a [u8]>,
}
@ -88,35 +80,17 @@ 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 | <QualityOfService as Into<u8>>::into(qos);
}
pub fn add_identifier(& mut self, identifier: u16) {
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> {
@ -128,41 +102,55 @@ impl<'a, const MAX_PROPERTIES: usize> Packet<'a> for PublishPacket<'a, MAX_PROPE
packet_identifier: 1,
property_len: 0,
properties: Vec::<Property<'a>, MAX_PROPERTIES>::new(),
message: None
message: None,
}
}
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<usize, BufferError> {
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();
let property_len_enc: [u8; 4] = 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::<MAX_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::<MAX_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) {

61
src/packet/pubrec_packet.rs → src/packet/v5/pubrec_packet.rs
View File

@ -25,65 +25,62 @@
use heapless::Vec;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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<Property<'a>, 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> PubrecPacket<'a, MAX_PROPERTIES> {}
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::<Property<'a>, 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<usize, BufferError> {
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();
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::<MAX_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::<MAX_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) {

61
src/packet/pubrel_packet.rs → src/packet/v5/pubrel_packet.rs
View File

@ -25,65 +25,62 @@
use heapless::Vec;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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<Property<'a>, 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> PubrelPacket<'a, MAX_PROPERTIES> {}
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::<Property<'a>, 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<usize, BufferError> {
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();
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::<MAX_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::<MAX_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) {

34
src/packet/reason_codes.rs → src/packet/v5/reason_codes.rs
View File

@ -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,9 +95,9 @@ impl Into<u8> for ReasonCode {
ReasonCode::MaximumConnectTime => 0xA0,
ReasonCode::SubscriptionIdentifiersNotSupported => 0xA1,
ReasonCode::WildcardSubscriptionNotSupported => 0xA2,
ReasonCode::NetworkError => 0xFF
}
ReasonCode::BuffError => 0xFE,
ReasonCode::NetworkError => 0xFF,
};
}
}
@ -146,8 +146,9 @@ impl From<u8> for ReasonCode {
0xA0 => ReasonCode::MaximumConnectTime,
0xA1 => ReasonCode::SubscriptionIdentifiersNotSupported,
0xA2 => ReasonCode::WildcardSubscriptionNotSupported,
_ => ReasonCode::NetworkError
}
0xFE => ReasonCode::BuffError,
_ => ReasonCode::NetworkError,
};
}
}
@ -168,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!"),
@ -192,12 +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::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!"),
}
}
}
}

46
src/packet/suback_packet.rs → src/packet/v5/suback_packet.rs
View File

@ -24,52 +24,38 @@
use heapless::Vec;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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<Property<'a>, MAX_PROPERTIES>,
pub reason_codes: Vec<u8, MAX_REASONS>,
}
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(());
}
}
@ -83,17 +69,23 @@ impl<'a, const MAX_REASONS: usize, const MAX_PROPERTIES: usize> Packet<'a>
packet_identifier: 0,
property_len: 0,
properties: Vec::<Property<'a>, MAX_PROPERTIES>::new(),
reason_codes: Vec::<u8, MAX_REASONS>::new()
reason_codes: Vec::<u8, MAX_REASONS>::new(),
}
}
fn encode(&mut self, buffer: &mut [u8]) -> usize {
fn encode(&mut self, _buffer: &mut [u8], _buffer_len: usize) -> Result<usize, BufferError> {
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) {

56
src/packet/subscription_packet.rs → src/packet/v5/subscription_packet.rs
View File

@ -24,45 +24,35 @@
use heapless::Vec;
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 super::packet_type::PacketType;
use super::property::Property;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::v5::mqtt_packet::Packet;
use crate::packet::v5::publish_packet::QualityOfService;
use crate::utils::buffer_reader::BuffReader;
use crate::utils::buffer_writer::BuffWriter;
use crate::utils::types::{BufferError, TopicFilter};
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<Property<'a>, MAX_PROPERTIES>,
// topic filter len
pub topic_filter_len: u16,
// payload
pub topic_filters: Vec<TopicFilter<'a>, MAX_FILTERS>,
}
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 | (<QualityOfService as Into<u8>>::into(qos) >> 1);
new_filter.sub_options =
new_filter.sub_options | (<QualityOfService as Into<u8>>::into(qos) >> 1);
self.topic_filters.push(new_filter);
self.topic_filter_len = self.topic_filter_len + 1;
}
@ -71,8 +61,8 @@ 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 {
let mut x = Self {
fn new() -> Self {
let x = Self {
fixed_header: PacketType::Subscribe.into(),
remain_len: 0,
packet_identifier: 1,
@ -84,12 +74,11 @@ 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<usize, BufferError> {
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();
let property_len_enc: [u8; 4] = 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::<MAX_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::<MAX_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;

51
src/packet/unsuback_packet.rs → src/packet/v5/unsuback_packet.rs
View File

@ -24,50 +24,38 @@
use heapless::Vec;
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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<Property<'a>, MAX_PROPERTIES>,
pub reason_codes: Vec<u8, MAX_REASONS>,
}
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 +63,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::<Property<'a>, MAX_PROPERTIES>::new(),
reason_codes: Vec::<u8, MAX_REASONS>::new(),
}
}
fn encode(&mut self, buffer: &mut [u8]) -> usize {
fn encode(&mut self, _buffer: &mut [u8], _buffer_len: usize) -> Result<usize, BufferError> {
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) {

51
src/packet/unsubscription_packet.rs → src/packet/v5/unsubscription_packet.rs
View File

@ -25,31 +25,20 @@
use heapless::Vec;
use crate::encoding::variable_byte_integer::VariableByteIntegerEncoder;
use crate::packet::mqtt_packet::Packet;
use crate::packet::v5::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<Property<'a>, MAX_PROPERTIES>,
// topic filter len
pub topic_filter_len: u16,
// payload
pub topic_filters: Vec<TopicFilter<'a>, MAX_FILTERS>,
}
@ -62,34 +51,42 @@ 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::<Property<'a>, MAX_PROPERTIES>::new(),
topic_filter_len: 0,
topic_filters: Vec::<TopicFilter<'a>, 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<usize, BufferError> {
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();
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::<MAX_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::<MAX_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) {

10
src/tokio_network.rs
View File

@ -1,16 +1,12 @@
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::packet::reason_codes::ReasonCode;
use crate::network::network_trait::Network;
use crate::packet::v5::reason_codes::ReasonCode;
pub struct TokioNetwork {
ip: [u8; 4],

127
src/utils/buffer_reader.rs
View File

@ -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<u32, ParseError> {
pub fn read_variable_byte_int(&mut self) -> Result<u32, BufferError> {
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,41 +74,41 @@ impl<'a> BuffReader<'a> {
}
/// Reading u32 from buffer as `Big endian`
pub fn read_u32(&mut self) -> Result<u32, ParseError> {
let (int_bytes, rest) = self.buffer[self.position..].split_at(mem::size_of::<u32>());
pub fn read_u32(&mut self) -> Result<u32, BufferError> {
let (int_bytes, _rest) = self.buffer[self.position..].split_at(mem::size_of::<u32>());
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<u16, ParseError> {
let (int_bytes, rest) = self.buffer[self.position..].split_at(mem::size_of::<u16>());
pub fn read_u16(&mut self) -> Result<u16, BufferError> {
let (int_bytes, _rest) = self.buffer[self.position..].split_at(mem::size_of::<u16>());
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<u8, ParseError> {
pub fn read_u8(&mut self) -> Result<u8, BufferError> {
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<EncodedString<'a>, ParseError> {
pub fn read_string(&mut self) -> Result<EncodedString<'a>, BufferError> {
let len = self.read_u16();
match len {
Err(err) => return Err(err),
_ => {},
_ => {}
}
let len_res = len.unwrap();
let res_str =
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<BinaryData<'a>, 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<BinaryData<'a>, 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<StringPair<'a>, ParseError> {
pub fn read_string_pair(&mut self) -> Result<StringPair<'a>, BufferError> {
let name = self.read_string();
match name {
Err(err) => return Err(err),
@ -231,6 +147,9 @@ 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];
}
}

127
src/utils/buffer_writer.rs
View File

@ -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::packet::v5::property::Property;
use crate::utils::types::{BinaryData, BufferError, EncodedString, StringPair, 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,102 @@ 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<const LEN: usize>(&mut self, properties: &Vec<Property<'a>, LEN>) {
pub fn encode_properties<const LEN: usize>(
&mut self,
properties: &Vec<Property<'a>, 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<const MAX: usize>(
@ -147,15 +165,16 @@ impl<'a> BuffWriter<'a> {
sub: bool,
len: usize,
filters: &Vec<TopicFilter<'a>, 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(())
}
}

1
src/utils/mod.rs
View File

@ -25,3 +25,4 @@
pub mod buffer_reader;
pub mod buffer_writer;
pub mod rng_generator;
pub mod types;

4
src/utils/rng_generator.rs
View File

@ -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::{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))
}
}
}

124
src/utils/types.rs Normal file
View File

@ -0,0 +1,124 @@
/*
* MIT License
*
* Copyright (c) [2022] [Ondrej Babec <ond.babec@gmail.com>]
*
* 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, 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, 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;
}
}