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

Clean warnings and delegate errors

Browse Source
This commit is contained in:
Ondrej Babec
2022-03-04 13:30:49 +01:00
parent 80c0e25eda
commit 57f0304d35
29 changed files with 696 additions and 675 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/client/client_config.rs
View File

@@ -24,7 +24,7 @@
use crate::packet::publish_packet::QualityOfService;
use crate::utils::buffer_reader::{BinaryData, EncodedString};
use crate::utils::types::{BinaryData, EncodedString};
pub struct ClientConfig<'a> {
pub qos: QualityOfService,

121
src/client/client_v5.rs
View File

@@ -1,26 +1,26 @@
use core::future::Future;
use embassy::traits::rng;
use rand_core::RngCore;
use crate::client::client_config::ClientConfig;
use crate::network::network_trait::{Network, NetworkError};
use crate::network::network_trait::{Network};
use crate::packet::connack_packet::ConnackPacket;
use crate::packet::connect_packet::ConnectPacket;
use crate::packet::disconnect_packet::DisconnectPacket;
use crate::packet::mqtt_packet::Packet;
use crate::packet::puback_packet::PubackPacket;
use crate::packet::publish_packet::QualityOfService::QoS1;
use crate::packet::publish_packet::{PublishPacket, QualityOfService};
use crate::packet::publish_packet::QualityOfService::QoS1;
use crate::packet::reason_codes::ReasonCode;
use crate::packet::suback_packet::SubackPacket;
use crate::packet::subscription_packet::SubscriptionPacket;
use crate::utils::buffer_reader::BuffReader;
use crate::utils::rng_generator::CountingRng;
use crate::utils::types::BufferError;
pub struct MqttClientV5<'a, T, const MAX_PROPERTIES: usize> {
network_driver: &'a mut T,
buffer: &'a mut [u8],
buffer_len: usize,
recv_buffer: &'a mut [u8],
recv_buffer_len: usize,
rng: CountingRng,
config: ClientConfig<'a>,
}
@@ -29,18 +29,20 @@ impl<'a, T, const MAX_PROPERTIES: usize> MqttClientV5<'a, T, MAX_PROPERTIES>
where
T: Network,
{
pub fn new(network_driver: &'a mut T, buffer: &'a mut [u8], recv_buffer: &'a mut [u8], config: ClientConfig<'a>) -> Self {
pub fn new(network_driver: &'a mut T, buffer: &'a mut [u8], buffer_len: usize, recv_buffer: &'a mut [u8], recv_buffer_len: usize, config: ClientConfig<'a>) -> Self {
Self {
network_driver,
buffer,
buffer_len,
recv_buffer,
recv_buffer_len,
rng: CountingRng(50),
config
}
}
pub async fn connect_to_broker<'b>(&'b mut self) -> Result<(), ReasonCode> {
let mut len = {
let len = {
let mut connect = ConnectPacket::<'b, 3, 0>::clean();
if self.config.username_flag {
connect.add_username(& self.config.username);
@@ -48,21 +50,33 @@ where
if self.config.password_flag {
connect.add_password(& self.config.password)
}
connect.encode(self.buffer)
connect.encode(self.buffer, self.buffer_len)
};
self.network_driver.send(self.buffer, len).await ?;
if let Err(err) = len {
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
self.network_driver.send(self.buffer, len.unwrap()).await ?;
//connack
let reason: u8 = {
let reason: Result<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(())
}
@@ -71,8 +85,12 @@ where
pub async fn disconnect<'b>(&'b mut self) -> Result<(), ReasonCode> {
let mut disconnect = DisconnectPacket::<'b, 5>::new();
let mut len = disconnect.encode(self.buffer);
self.network_driver.send(self.buffer, len).await?;
let len = disconnect.encode(self.buffer, self.buffer_len);
if let Err(err) = len {
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
self.network_driver.send(self.buffer, len.unwrap()).await ?;
Ok(())
}
@@ -89,27 +107,41 @@ where
packet.add_qos(self.config.qos);
packet.add_identifier(identifier);
packet.add_message(message.as_bytes());
packet.encode(self.buffer)
packet.encode(self.buffer, self.buffer_len)
};
self.network_driver.send(self.buffer, len).await ?;
if let Err(err) = len {
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
self.network_driver.send(self.buffer, len.unwrap()).await ?;
//QoS1
if <QualityOfService as Into<u8>>::into(self.config.qos ) == <QualityOfService as Into<u8>>::into(QoS1) {
let reason = {
let reason: Result<[u16; 2], BufferError> = {
self.network_driver.receive(self.buffer).await ?;
let mut packet = PubackPacket::<'b, 5>::new();
packet.decode(&mut BuffReader::new(self.buffer));
[packet.packet_identifier, packet.reason_code as u16]
if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len)) {
Err(err)
} else {
Ok([packet.packet_identifier, packet.reason_code as u16])
}
};
if identifier != reason[0] {
if let Err(err) = reason {
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
let res = reason.unwrap();
if identifier != res[0] {
return Err(ReasonCode::PacketIdentifierNotFound);
}
if reason[1] != 0 {
return Err(ReasonCode::from(reason[1] as u8));
if res[1] != 0 {
return Err(ReasonCode::from(res[1] as u8));
}
}
Ok(())
@@ -121,21 +153,35 @@ where
let len = {
let mut subs = SubscriptionPacket::<'b, 1, 1>::new();
subs.add_new_filter(topic_name, self.config.qos);
subs.encode(self.buffer)
subs.encode(self.buffer, self.buffer_len)
};
self.network_driver.send(self.buffer, len).await ?;
if let Err(err) = len {
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
let reason = {
self.network_driver.send(self.buffer, len.unwrap()).await ?;
let reason: Result<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(())
}
@@ -145,15 +191,22 @@ where
pub async fn receive_message<'b>(&'b mut self) -> Result<&'b [u8], ReasonCode> {
self.network_driver.receive(self.recv_buffer).await ?;
let mut packet = PublishPacket::<'b, 5>::new();
packet.decode(&mut BuffReader::new(self.recv_buffer));
if let Err(err) = packet.decode(&mut BuffReader::new(self.recv_buffer, self.recv_buffer_len)) {
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
if (packet.fixed_header & 0x06) == <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 ?;
}
}