Using Arc Mutex seems more appropriate

Author: anselanza

Cargo.lock

@@ -1,7 +1,7 @@
 [package]
 name = "tether-agent"
 description = "Standardised use of MQTT and MessagePack for inter-process communication"
-version = "0.14.1"
+version = "0.14.2"
 edition = "2021"
 license = "MIT"
 repository = "https://github.com/RandomStudio/tether"

base_agent/rs/Cargo.toml

@@ -4,6 +4,7 @@ use rmp_serde::to_vec_named;
 use rumqttc::tokio_rustls::rustls::ClientConfig;
 use rumqttc::{Client, Event, MqttOptions, Packet, QoS, Transport};
 use serde::Serialize;
+use std::sync::{Arc, Mutex};
 use std::{sync::mpsc, thread, time::Duration};
 use uuid::Uuid;
 
@@ -29,6 +30,7 @@ pub struct TetherAgent {
     pub(crate) client: Option<Client>,
     message_sender: mpsc::Sender<(TetherOrCustomTopic, Vec<u8>)>,
     message_receiver: mpsc::Receiver<(TetherOrCustomTopic, Vec<u8>)>,
+    is_connected: Arc<Mutex<bool>>,
 }
 
 #[derive(Clone)]
@@ -148,6 +150,7 @@ impl TetherAgentOptionsBuilder {
             message_sender,
             message_receiver,
             mqtt_client_id: self.mqtt_client_id,
+            is_connected: Arc::new(Mutex::new(false)),
         };
 
         if self.auto_connect {
@@ -283,19 +286,18 @@ impl TetherAgent {
 
         let message_tx = self.message_sender.clone();
 
-        let (connected_tx, connected_rx) = mpsc::channel();
+        let connection_state = Arc::clone(&self.is_connected);
 
         thread::spawn(move || {
-            let send_connected = connected_tx.clone();
             for event in connection.iter() {
                 match event {
                     Ok(e) => match e {
                         Event::Incoming(incoming) => match incoming {
                             Packet::ConnAck(_) => {
                                 info!("(Connected) ConnAck received!");
-                                send_connected
-                                    .send(true)
-                                    .expect("failed to push connected status form thread");
+                                let mut is_c =
+                                    connection_state.lock().expect("failed to lock mutex");
+                                *is_c = true;
                             }
                             Packet::Publish(p) => {
                                 debug!("Incoming Publish packet (message received), {:?}", &p);
@@ -334,11 +336,15 @@ impl TetherAgent {
         let mut is_ready = false;
 
         while !is_ready {
-            std::thread::sleep(Duration::from_millis(100));
+            debug!("Check whether connected...");
+            std::thread::sleep(Duration::from_millis(1));
             trace!("Is ready? {}", is_ready);
-            if let Ok(is_connected) = connected_rx.try_recv() {
-                is_ready = is_connected;
-                trace!("Is connected? {}", is_connected);
+            let get_state = *self.is_connected.lock().expect("failed to lock mutex");
+            if get_state {
+                info!("Connection status confirmed");
+                is_ready = true;
+            } else {
+                debug!("Not connected yet...");
             }
         }
 

base_agent/rs/src/agent/mod.rs

@@ -1,7 +1,7 @@
 [package]
 name = "tether-utils"
 description = "Utilities for Tether Systems"
-version = "0.11.3"
+version = "0.11.4"
 edition = "2021"
 license = "MIT"
 repository = "https://github.com/RandomStudio/tether"
@@ -14,8 +14,8 @@ name = "tether"
 
 
 [dependencies]
-# tether-agent = { path = "../../base_agent/rs" }
-tether-agent = "0.14.1"
+tether-agent = { path = "../../base_agent/rs" }
+# tether-agent = "0.14.1"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0.91"
 rmp-serde = "1.1.1"

utilities/tether-utils/Cargo.toml

@@ -99,6 +99,10 @@ fn main() {
         Commands::Send(options) => {
             tether_send::send(options, &mut tether_agent)
                 .unwrap_or_else(|e| error!("Failed to send: {}", e));
+            // This silliness (below) is only needed because the command might
+            // finish before the message is even sent!
+            // Less of an issue with long-running applications, and should
+            // be properly solved with an async version.
             std::thread::sleep(std::time::Duration::from_millis(1000));
         }
         Commands::Topics(options) => {