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

Add subscription to multiple topics

Browse Source
This commit is contained in:
Ondrej Babec
2022-03-04 15:32:11 +01:00
parent 57f0304d35
commit 3c00a7b81d
28 changed files with 399 additions and 274 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/client/client_config.rs
View File

@@ -22,7 +22,6 @@
* SOFTWARE.
*/
use crate::packet::publish_packet::QualityOfService;
use crate::utils::types::{BinaryData, EncodedString};
@@ -31,7 +30,7 @@ pub struct ClientConfig<'a> {
pub username_flag: bool,
pub username: EncodedString<'a>,
pub password_flag: bool,
pub password: BinaryData<'a>
pub password: BinaryData<'a>,
}
impl ClientConfig<'a> {
@@ -45,11 +44,11 @@ impl ClientConfig<'a> {
}
}
pub fn add_qos(& mut self, qos: QualityOfService) {
pub fn add_qos(&mut self, qos: QualityOfService) {
self.qos = qos;
}
pub fn add_username(& mut self, username: &'a str) {
pub fn add_username(&mut self, username: &'a str) {
let mut username_s: EncodedString = EncodedString::new();
username_s.string = username;
username_s.len = username.len() as u16;
@@ -57,11 +56,11 @@ impl ClientConfig<'a> {
self.username = username_s;
}
pub fn add_password(& mut self, password: &'a str) {
pub fn add_password(&mut self, password: &'a str) {
let mut password_s: BinaryData = BinaryData::new();
password_s.bin = password.as_bytes();
password_s.len = password_s.bin.len() as u16;
self.password = password_s;
self.password_flag = true;
}
}
}

140
src/client/client_v5.rs
View File

@@ -1,19 +1,22 @@
use rand_core::RngCore;
use crate::client::client_config::ClientConfig;
use crate::network::network_trait::{Network};
use crate::network::network_trait::Network;
use crate::packet::connack_packet::ConnackPacket;
use crate::packet::connect_packet::ConnectPacket;
use crate::packet::disconnect_packet::DisconnectPacket;
use crate::packet::mqtt_packet::Packet;
use crate::packet::pingreq_packet::PingreqPacket;
use crate::packet::pingresp_packet::PingrespPacket;
use crate::packet::puback_packet::PubackPacket;
use crate::packet::publish_packet::{PublishPacket, QualityOfService};
use crate::packet::publish_packet::QualityOfService::QoS1;
use crate::packet::publish_packet::{PublishPacket, QualityOfService};
use crate::packet::reason_codes::ReasonCode;
use crate::packet::suback_packet::SubackPacket;
use crate::packet::subscription_packet::SubscriptionPacket;
use crate::utils::buffer_reader::BuffReader;
use crate::utils::rng_generator::CountingRng;
use crate::utils::types::BufferError;
use heapless::Vec;
use rand_core::RngCore;
pub struct MqttClientV5<'a, T, const MAX_PROPERTIES: usize> {
network_driver: &'a mut T,
@@ -29,7 +32,14 @@ impl<'a, T, const MAX_PROPERTIES: usize> MqttClientV5<'a, T, MAX_PROPERTIES>
where
T: Network,
{
pub fn new(network_driver: &'a mut T, buffer: &'a mut [u8], buffer_len: usize, recv_buffer: &'a mut [u8], recv_buffer_len: usize, config: ClientConfig<'a>) -> Self {
pub fn new(
network_driver: &'a mut T,
buffer: &'a mut [u8],
buffer_len: usize,
recv_buffer: &'a mut [u8],
recv_buffer_len: usize,
config: ClientConfig<'a>,
) -> Self {
Self {
network_driver,
buffer,
@@ -37,7 +47,7 @@ where
recv_buffer,
recv_buffer_len,
rng: CountingRng(50),
config
config,
}
}
@@ -45,10 +55,10 @@ where
let len = {
let mut connect = ConnectPacket::<'b, 3, 0>::clean();
if self.config.username_flag {
connect.add_username(& self.config.username);
connect.add_username(&self.config.username);
}
if self.config.password_flag {
connect.add_password(& self.config.password)
connect.add_password(&self.config.password)
}
connect.encode(self.buffer, self.buffer_len)
};
@@ -57,7 +67,7 @@ where
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
self.network_driver.send(self.buffer, len.unwrap()).await ?;
self.network_driver.send(self.buffer, len.unwrap()).await?;
//connack
let reason: Result<u8, BufferError> = {
@@ -80,7 +90,6 @@ where
} else {
Ok(())
}
}
pub async fn disconnect<'b>(&'b mut self) -> Result<(), ReasonCode> {
@@ -90,7 +99,7 @@ where
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
self.network_driver.send(self.buffer, len.unwrap()).await ?;
self.network_driver.send(self.buffer, len.unwrap()).await?;
Ok(())
}
@@ -99,7 +108,6 @@ where
topic_name: &'b str,
message: &'b str,
) -> Result<(), ReasonCode> {
let identifier: u16 = self.rng.next_u32() as u16;
let len = {
let mut packet = PublishPacket::<'b, 5>::new();
@@ -115,15 +123,17 @@ where
return Err(ReasonCode::BuffError);
}
self.network_driver.send(self.buffer, len.unwrap()).await ?;
self.network_driver.send(self.buffer, len.unwrap()).await?;
//QoS1
if <QualityOfService as Into<u8>>::into(self.config.qos ) == <QualityOfService as Into<u8>>::into(QoS1) {
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 ?;
self.network_driver.receive(self.buffer).await?;
let mut packet = PubackPacket::<'b, 5>::new();
if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len)) {
if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len))
{
Err(err)
} else {
Ok([packet.packet_identifier, packet.reason_code as u16])
@@ -147,9 +157,63 @@ where
Ok(())
}
// TODO - multiple topic subscribe func
pub async fn subscribe_to_topics<'b, const TOPICS: usize>(
&'b mut self,
topic_names: &'b Vec<&'b str, TOPICS>,
) -> Result<(), ReasonCode> {
let len = {
let mut subs = SubscriptionPacket::<'b, TOPICS, 1>::new();
let mut i = 0;
loop {
if i == TOPICS {
break;
}
subs.add_new_filter(topic_names.get(i).unwrap(), self.config.qos);
i = i + 1;
}
subs.encode(self.buffer, self.buffer_len)
};
pub async fn subscribe_to_topic<'b>(&'b mut self, topic_name: &'b str) -> Result<(), ReasonCode> {
if let Err(err) = len {
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
self.network_driver.send(self.buffer, len.unwrap()).await?;
let reason: Result<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);
@@ -161,10 +225,10 @@ where
return Err(ReasonCode::BuffError);
}
self.network_driver.send(self.buffer, len.unwrap()).await ?;
self.network_driver.send(self.buffer, len.unwrap()).await?;
let reason: Result<u8, BufferError> = {
self.network_driver.receive(self.buffer).await ?;
self.network_driver.receive(self.buffer).await?;
let mut packet = SubackPacket::<'b, 5, 5>::new();
if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len)) {
@@ -185,18 +249,21 @@ where
} else {
Ok(())
}
}
pub async fn receive_message<'b>(&'b mut self) -> Result<&'b [u8], ReasonCode> {
self.network_driver.receive(self.recv_buffer).await ?;
self.network_driver.receive(self.recv_buffer).await?;
let mut packet = PublishPacket::<'b, 5>::new();
if let Err(err) = packet.decode(&mut BuffReader::new(self.recv_buffer, self.recv_buffer_len)) {
if let Err(err) =
packet.decode(&mut BuffReader::new(self.recv_buffer, self.recv_buffer_len))
{
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
if (packet.fixed_header & 0x06) == <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;
@@ -206,10 +273,33 @@ where
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
self.network_driver.send(self.buffer, len.unwrap()).await ?;
self.network_driver.send(self.buffer, len.unwrap()).await?;
}
}
return Ok(packet.message.unwrap());
}
pub async fn send_ping<'b>(&'b mut self) -> Result<(), ReasonCode> {
let len = {
let mut packet = PingreqPacket::new();
packet.encode(self.buffer, self.buffer_len)
};
if let Err(err) = len {
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
}
self.network_driver.send(self.buffer, len.unwrap()).await?;
self.network_driver.receive(self.buffer).await?;
let mut packet = PingrespPacket::new();
if let Err(err) = packet.decode(&mut BuffReader::new(self.buffer, self.buffer_len)) {
log::error!("[DECODE ERR]: {}", err);
return Err(ReasonCode::BuffError);
} else {
Ok(())
}
}
}

3
src/client/mod.rs
View File

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