From c924081133beb158fa71edc34d44ab254641293a Mon Sep 17 00:00:00 2001 From: LailaTheElf Date: Thu, 30 Jan 2025 11:31:06 +0100 Subject: [PATCH] add retain and qos to message and use crossbeam channel for subscriptions --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/lib.rs | 67 +++++++++++++++++++++++++++++++++++++----------------- 3 files changed, 48 insertions(+), 23 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 13cbb44..88ef188 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -246,7 +246,7 @@ dependencies = [ [[package]] name = "mqtt-client" -version = "1.0.0" +version = "2.0.0" dependencies = [ "crossbeam", "rumqttc", diff --git a/Cargo.toml b/Cargo.toml index ac17df2..387e0d3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mqtt-client" -version = "1.0.0" +version = "2.0.0" edition = "2021" [dependencies] diff --git a/src/lib.rs b/src/lib.rs index 9aa24bb..3f46891 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,21 +3,27 @@ pub mod mqtt_client { use std::time::Duration; use std::thread; - use rumqttc::{Client, Connection, Event, MqttOptions, Outgoing, Packet, QoS}; - use crossbeam::channel::{unbounded, Receiver, Sender}; + use rumqttc::{Client, Connection, Event, MqttOptions, Outgoing, Packet}; + use crossbeam::channel::unbounded; + + pub use crossbeam::channel::{Receiver, Sender}; + pub use rumqttc::QoS; pub trait MqttTool { fn new(client: Client, tx: Sender) -> Self; - fn rx(&mut self, messaage: MqttMessage); + fn run(&mut self, rx: Receiver); } pub struct MqttMessage { pub topic: String, - pub payload: String + pub payload: String, + pub retain: bool, + pub qos: QoS, } pub fn run(host: String, port: u16, client: String, user: String, pass: String) { - let (tx, rx) = unbounded::(); + let (tx_sender, tx_recever) = unbounded::(); + let (rx_sender, rx_recever) = unbounded::(); let mut mqttoptions = MqttOptions::new(client, host, port); mqttoptions.set_keep_alive(Duration::from_secs(5)); @@ -30,10 +36,22 @@ pub mod mqtt_client { let publisher = thread::Builder::new() .name("publisher".to_string()) .spawn(move || { - publisher(rx, client_publisher); + publisher(tx_recever, client_publisher); }); match publisher { - Err(_n) => println!("ERROR: main: fait to create publisher thread"), + Err(_n) => println!("ERROR: main: failed to create publisher thread"), + Ok(_n) => {} + } + + // thread tool runner + let tool_runner = thread::Builder::new() + .name("tool runner".to_string()) + .spawn(move || { + let mut tool = T::new(client, tx_sender); + tool.run(rx_recever); + }); + match tool_runner { + Err(_n) => println!("ERROR: main: failed to create publisher thread"), Ok(_n) => {} } @@ -41,13 +59,12 @@ pub mod mqtt_client { let mqtt = thread::Builder::new() .name("mqtt".to_string()) .spawn(move || { - let tool = T::new(client, tx); - handeler::(connection, tool); + handeler::(connection, rx_sender); }); match mqtt { - Err(_n) => println!("ERROR: main: fait to create mqtt thread"), + Err(_n) => println!("ERROR: main: failed to create mqtt thread"), Ok(join) => match join.join() { - Err(_) => println!("ERROR: main: failt to join mqtt thread"), + Err(_) => println!("ERROR: main: failed to join mqtt thread"), Ok(_) => {} } } @@ -57,11 +74,11 @@ pub mod mqtt_client { loop { let message = rx.recv(); match message { - Err(e) => println!("ERROR: publisher: faild to receve an message ({})", e), + Err(e) => println!("ERROR: publisher: failed to receve an message ({})", e), Ok(msg) => { println!("INFO : publisher: topic={}; payload={}", msg.topic, msg.payload); match client.publish(msg.topic, QoS::AtMostOnce, false, msg.payload) { - Err(_n) => println!("ERROR: publisher: faild to publish"), + Err(_n) => println!("ERROR: publisher: failed to publish"), Ok(_n) => {} } } @@ -69,14 +86,14 @@ pub mod mqtt_client { } } - pub(self) fn handeler(mut connection: Connection, mut tool: T) { + pub(self) fn handeler(mut connection: Connection, rx: Sender) { for (_i, notification) in connection.iter().enumerate() { let mut delay: bool = false; match notification { Err(e) => match e { rumqttc::ConnectionError::MqttState(state) => match state { rumqttc::StateError::Io(e) => { - println!("ERROR: mqtt: Io ({}) {}", e.kind(), e); + println!("ERROR: mqtt: State io {}: {}", e.kind(), e); delay = true; }, rumqttc::StateError::InvalidState => println!("ERROR: mqtt: InvalidState"), @@ -91,7 +108,10 @@ pub mod mqtt_client { }, rumqttc::ConnectionError::NetworkTimeout => println!("ERROR: mqtt: NetworkTimeout"), rumqttc::ConnectionError::FlushTimeout => println!("ERROR: mqtt: FlushTimeout"), - rumqttc::ConnectionError::Io(e) => println!("ERROR: mqtt: Io ({}) {}", e.kind(), e.to_string()), + rumqttc::ConnectionError::Io(e) => { + println!("ERROR: mqtt: Io ({}) {}", e.kind(), e.to_string()); + delay = true; + }, rumqttc::ConnectionError::ConnectionRefused(code) => { println!("ERROR: mqtt: ConnectionRefused {:?}", code); delay = true; @@ -122,16 +142,21 @@ pub mod mqtt_client { Packet::Publish(msg) => { let mut payload: String = String::from(""); match String::from_utf8(msg.payload.to_vec()) { - Err(e) => println!("ERROR: pqtt_recive: faild to decode payload ({})", e), + Err(e) => println!("ERROR: pqtt_recive: failed to decode payload ({})", e), Ok(v) => payload = v }; if payload.len() > 0 { - let message: MqttMessage = { MqttMessage { + let message = MqttMessage { topic: msg.topic, - payload: payload - }}; + payload: payload, + retain: msg.retain, + qos: msg.qos, + }; - tool.rx(message); + match rx.send(message) { + Err(n) => println!("ERROR: faild to send incomming message to tool ({:?})", n), + Ok(_n) => {} + } } }, Packet::PubAck(_) => {}, //println!("INFO : mqtt_recive: in PubAck"),