allow initialisation when reconnected

Cargo.lock

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

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mqtt-client"
-version = "3.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<S: std::marker::Send> {
-        fn new(client: Client, tx: Sender<MqttMessage>, config: S) -> 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,15 @@ pub mod mqtt_client {
         pub qos: QoS,
     }
 
+    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");
 
@@ -49,7 +56,7 @@ pub mod mqtt_client {
         let mqtt = thread::Builder::new()
             .name("mqtt".to_string())
             .spawn(move || {
-                handeler::<S, T>(connection, rx_sender);
+                handeler::<S, T>(connection, event_sender);
             });
         match mqtt {
             Err(_n) => println!("ERROR: mqtt client: failed to create mqtt thread"),
@@ -60,9 +67,8 @@ pub mod mqtt_client {
         let tool_runner = thread::Builder::new()
             .name("tool runner".to_string())
             .spawn(move || {
-                let mut tool = T::new(client, tx_sender, config);
+                let mut tool = T::new(tx_sender, config, client);
-                tool.run(rx_recever);
+                tool.run(event_recever);
-                println!("WARN : rool_runner: tool has ended");
             });
         match tool_runner {
             Err(_n) => println!("ERROR: mqtt client: failed to create publisher thread"),
@@ -94,7 +100,7 @@ pub mod mqtt_client {
         }
     }
 
-    pub(self) fn handeler<S: std::marker::Send,T: MqttTool<S>>(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 {
@@ -145,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()) {
@@ -160,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) => {}
                                     }