refactor

kairo-common/src/mqtt.rs

@@ -62,16 +62,28 @@ pub mod for_sync {
     use paho_mqtt as mqtt;
     use std::{process, time::Duration};
 
-    pub fn get_mqtt_cli() -> mqtt::Client {
+    pub struct MqttClient {
+        cli: mqtt::Client,
+    }
+
+    impl MqttClient {
+        pub fn new(client_id: Option<&str>) -> MqttClient {
             let host = dotenv::var("MQTT_BROKER").unwrap_or_else(|_| {
             println! {"MQTT_BROKER not found in .evn file, using default: tcp://localhost:1883"};
             "tcp://localhost:1883".to_string()
         });
 
-        let mut cli = mqtt::Client::new(host).unwrap_or_else(|e| {
+            let mut cli = if let Some(client_id) = client_id {
+                mqtt::Client::new((host, String::from(client_id))).unwrap_or_else(|e| {
                     println!("Error creating the client: {:?}", e);
                     process::exit(1);
-        });
+                })
+            } else {
+                mqtt::Client::new(host).unwrap_or_else(|e| {
+                    println!("Error creating the client: {:?}", e);
+                    process::exit(1);
+                })
+            };
 
             // Use 5sec timeouts for sync calls.
             cli.set_timeout(Duration::from_secs(5));
@@ -81,20 +93,20 @@ pub mod for_sync {
                 println!("Unable to connect: {:?}", e);
                 process::exit(1);
             }
-        cli
-    }
 
-    pub fn mqtt_pub(
-        client: &mqtt::Client,
-        topic: &str,
-        payload: &str,
-    ) -> Result<(), paho_mqtt::Error> {
-        let msg = mqtt::MessageBuilder::new()
+            MqttClient { cli }
+        }
+        pub fn publish(&self, topic: &str, payload: Option<&str>) -> Result<(), paho_mqtt::Error> {
+            let msg = if let Some(payload) = payload {
+                mqtt::MessageBuilder::new()
                     .topic(topic)
-            .payload(payload)
                     .qos(0)
-            .finalize();
-
-        client.publish(msg)
+                    .payload(payload)
+                    .finalize()
+            } else {
+                mqtt::MessageBuilder::new().topic(topic).qos(0).finalize()
+            };
+            self.cli.publish(msg)
+        }
     }
 }

simulation-tools/src/nav_dev/main.rs

@@ -3,7 +3,7 @@ use std::{thread, time};
 
 mod error_report;
 
-use kairo_common::mqtt::for_sync::{get_mqtt_cli, mqtt_pub};
+use kairo_common::mqtt::for_sync::MqttClient;
 use kairo_common::{Antenna, BeaconMeasure, DeviceReport, Point};
 
 #[derive(Clone)]
@@ -29,7 +29,7 @@ async fn main() {
     //     });
     // }
 
-    let client = get_mqtt_cli();
+    let client = MqttClient::new(None);
 
     let mut position = Point::new(config.radius, 0.0);
 
@@ -57,7 +57,9 @@ async fn main() {
                 .push(BeaconMeasure::new(&config.id, &ant.id, rssi + noise));
         }
         let payload = serde_json::to_string(&report).unwrap_or_else(|_| "".to_string());
-        mqtt_pub(&client, topic.as_str(), payload.as_str()).expect("Pub error");
+        client
+            .publish(topic.as_str(), Some(payload.as_str()))
+            .expect("Pub error");
 
         // if config.real {
         //     let _r = KnownPosition::new(position).write_for("real").await;