allow initialisation when reconnected

Cargo.lock

@@ -246,7 +246,7 @@ dependencies = [
 
 [[package]]
 name = "mqtt-client"
-version = "2.0.0"
+version = "4.0.0"
 dependencies = [
  "crossbeam",
  "rumqttc",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mqtt-client"
-version = "2.0.0"
+version = "4.0.0"
 edition = "2021"
 
 [dependencies]

src/lib.rs

@@ -3,15 +3,16 @@ pub mod mqtt_client {
     use std::time::Duration;
     use std::thread;
 
-    use rumqttc::{Client, Connection, Event, MqttOptions, Outgoing, Packet};
+    use rumqttc::{Connection, Event, MqttOptions, Outgoing, Packet};
     use crossbeam::channel::unbounded;
-
+    
+    pub use rumqttc::Client;
     pub use crossbeam::channel::{Receiver, Sender};
     pub use rumqttc::QoS;
 
-    pub trait MqttTool {
+    pub trait MqttTool<S: std::marker::Send> {
-        fn new(client: Client, tx: Sender<MqttMessage>) -> Self;
+        fn new(tx: Sender<MqttMessage>, config: S, client: Client) -> Self;
-        fn run(&mut self, rx: Receiver<MqttMessage>);
+        fn run(&mut self, event_recever: Receiver<MqttEvent>);
     }
 
     pub struct MqttMessage {
@@ -21,9 +22,17 @@ pub mod mqtt_client {
         pub qos: QoS,
     }
 
-    pub fn run<T: MqttTool>(host: String, port: u16, client: String, user: String, pass: String) {
+    pub enum MqttEvent {
+        Connected,
+        Disconnected,
+        Message(MqttMessage)
+    }
+
+    pub fn run<S: std::marker::Send, T: MqttTool<S>>(host: String, port: u16, client: String, user: String, pass: String, config: S) where S: 'static {
         let (tx_sender, tx_recever) = unbounded::<MqttMessage>();
-        let (rx_sender, rx_recever) = unbounded::<MqttMessage>();
+        let (event_sender, event_recever) = unbounded::<MqttEvent>();
+
+        println!("INFO : mqtt client: run");
 
         let mut mqttoptions = MqttOptions::new(client, host, port);
         mqttoptions.set_keep_alive(Duration::from_secs(5));
@@ -39,19 +48,7 @@ pub mod mqtt_client {
                 publisher(tx_recever, client_publisher);
             });
         match publisher {
-            Err(_n) => println!("ERROR: main: failed to create publisher thread"),
+            Err(_n) => println!("ERROR: mqtt client: 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) => {}
         }
 
@@ -59,25 +56,42 @@ pub mod mqtt_client {
         let mqtt = thread::Builder::new()
             .name("mqtt".to_string())
             .spawn(move || {
-                handeler::<T>(connection, rx_sender);
+                handeler::<S, T>(connection, event_sender);
             });
         match mqtt {
-            Err(_n) => println!("ERROR: main: failed to create mqtt thread"),
+            Err(_n) => println!("ERROR: mqtt client: failed to create mqtt thread"),
-            Ok(join) => match join.join() {
+            Ok(_) => {}
-                Err(_) => println!("ERROR: main: failed to join mqtt thread"),
+        }
+
+        // thread tool runner
+        let tool_runner = thread::Builder::new()
+            .name("tool runner".to_string())
+            .spawn(move || {
+                let mut tool = T::new(tx_sender, config, client);
+                tool.run(event_recever);
+            });
+        match tool_runner {
+            Err(_n) => println!("ERROR: mqtt client: failed to create publisher thread"),
+            Ok(runner) => match runner.join() {
+                Err(_) => println!("ERROR: mqtt client: failed to join tool runner thread"),
                 Ok(_) => {}
             }
         }
+        println!("INFO : mqtt client: exit");
     }
 
     pub(self) fn publisher(rx: Receiver<MqttMessage>, client: rumqttc::Client) {
         loop {
             let message = rx.recv();
             match message {
-                Err(e) => println!("ERROR: publisher: failed to receve an message ({})", e),
+                Err(e) => {
+                    println!("WARN : publisher: {}", e);
+                    println!("INFO : publisher: exit, no one can send messages");
+                    return
+                },
                 Ok(msg) => {
                     println!("INFO : publisher: topic={}; payload={}", msg.topic, msg.payload);
-                    match client.publish(msg.topic, QoS::AtMostOnce, false, msg.payload) {
+                    match client.publish(msg.topic, msg.qos, msg.retain, msg.payload) {
                         Err(_n) => println!("ERROR: publisher: failed to publish"),
                         Ok(_n) => {}
                     }
@@ -86,7 +100,7 @@ pub mod mqtt_client {
         }
     }
 
-    pub(self) fn handeler<T: MqttTool>(mut connection: Connection, rx: Sender<MqttMessage>) {
+    pub(self) fn handeler<S: std::marker::Send,T: MqttTool<S>>(mut connection: Connection, rx: Sender<MqttEvent>) {
         for (_i, notification) in connection.iter().enumerate() {
             let mut delay: bool = false;
             match notification {
@@ -123,7 +137,6 @@ pub mod mqtt_client {
                 Ok(event) => {
                     match event {
                         Event::Outgoing(n) => match n {
-
                             Outgoing::Publish(_n) => {}, //println!("INFO : mqtt_recive: out Publish"),
                             Outgoing::Subscribe(_n) => {}, //println!("INFO : mqtt_recive: out Subscribe"),
                             Outgoing::Unsubscribe(_n) => {}, //println!("INFO : mqtt_recive: out Unsubscribe"),
@@ -138,7 +151,10 @@ pub mod mqtt_client {
                         },
                         Event::Incoming(n) => match n {
                             Packet::Connect(_) => println!("INFO : mqtt: connected"),
-                            Packet::ConnAck(_) => println!("INFO : mqtt: connaction acknolaged"),
+                            Packet::ConnAck(_) => {
+                                println!("INFO : mqtt: connaction acknolaged");
+                                let _ = rx.send(MqttEvent::Connected);
+                            },
                             Packet::Publish(msg) => {
                                 let mut payload: String = String::from("");
                                 match String::from_utf8(msg.payload.to_vec()) {
@@ -153,7 +169,7 @@ pub mod mqtt_client {
                                         qos: msg.qos,
                                     };
                     
-                                    match rx.send(message) {
+                                    match rx.send(MqttEvent::Message(message)) {
                                         Err(n) => println!("ERROR: faild to send incomming message to tool ({:?})", n),
                                         Ok(_n) => {}
                                     }