finish with rumqtt

src/main.rs

@@ -1,25 +1,26 @@
+use std::env;
+
 mod mqtt_pub;
 mod system_values;
 mod device_var;
 
-use std::env;
-
-extern crate paho_mqtt as mqtt;
-
 fn main() {
 
     device_var::read_env_var();
 
-    let mut cli = mqtt_pub::init_cli();
+    let mut count = 3;
+    let (mut cli, connection) = mqtt_pub::init_cli();
 
-    cli = mqtt_pub::publish_message(cli, "temperature", &system_values::get_temperature().to_string());
-    cli = mqtt_pub::publish_message(cli, "storage", &system_values::get_storage().to_string());
-    cli = mqtt_pub::publish_message(cli, "mem", &system_values::get_mem().to_string());
+    cli = mqtt_pub::publish_message(cli, "temperature", system_values::get_temperature().to_string());
+    cli = mqtt_pub::publish_message(cli, "storage", system_values::get_storage().to_string());
+    cli = mqtt_pub::publish_message(cli, "mem", system_values::get_mem().to_string());
 
     let battery = env::var("BATTERY").unwrap().to_string();
     if battery == "true" {
-        cli = mqtt_pub::publish_message(cli, "battery", &system_values::get_battery().to_string());
+        count = 4;
+        mqtt_pub::publish_message(cli, "battery", system_values::get_battery().to_string());
     }
 
-    mqtt_pub::disconnect(cli);
+    mqtt_pub::disconnect(connection, count);
 }
+

src/mqtt_pub.rs

@@ -1,65 +1,38 @@
 use std::{
     env,
-    process,
     time::Duration
 };
+use rumqttc::{Client, Connection, MqttOptions, QoS};
 
-// Define the qos.
-const QOS:i32 = 1;
+pub fn init_cli() -> (Client, Connection) {
+    let mut mqttoptions = MqttOptions::new("test-1", env::var("BROKER_ADDRESS").unwrap().to_string(), 1883);
+    mqttoptions.set_keep_alive(Duration::from_secs(1));
+    let (client, connection) = Client::new(mqttoptions, 10);
 
-pub fn init_cli() -> mqtt::Client {
-
-    let host = env::args().nth(1).unwrap_or_else(||
-        env::var("BROKER_ADDRESS").unwrap().to_string()
-    );
-
-    // Define the set of options for the create.
-    // Use an ID for a persistent session.
-    let create_opts = mqtt::CreateOptionsBuilder::new()
-        .server_uri(host)
-        .client_id(env::var("BROKER_CLIENT").unwrap().to_string())
-        .finalize();
-
-    // Create a client.
-    let cli = mqtt::Client::new(create_opts).unwrap_or_else(|err| {
-        println!("Error creating the client: {:?}", err);
-        process::exit(1);
-    });
-
-    // Define the set of options for the connection.
-    let conn_opts = mqtt::ConnectOptionsBuilder::new()
-        .keep_alive_interval(Duration::from_secs(20))
-        .clean_session(true)
-        .finalize();
-
-    // Connect and wait for it to complete or fail.
-    if let Err(e) = cli.connect(conn_opts) {
-        println!("Unable to connect:\n\t{:?}", e);
-        process::exit(1);
-    }
-
-    return cli;
+    (client, connection)
 }
 
-pub fn publish_message(cli: mqtt::Client, topic: &str, message: &str) -> mqtt::Client {
-    // let topic = get_topic_name(topic);
-    let msg = mqtt::Message::new(get_topic_name(topic), message, QOS);
-    let tok = cli.publish(msg);
-
-    if let Err(e) = tok {
-        println!("Error sending message: {:?}", e);
-    }
-
-    println!("{} : {}", topic, message);
-
-    return cli;
+pub fn publish_message(mut client: Client, topic: &str, payload: String) -> Client {
+    client.publish(get_topic_name(topic), QoS::AtLeastOnce, true, payload).unwrap();
+    client
 }
 
-pub fn disconnect(cli: mqtt::Client) {
+pub fn disconnect(mut connection: Connection, count: usize) {
     // Disconnect from the broker.
-    let tok = cli.disconnect(None);
-    println!("Disconnect from the broker");
-    tok.unwrap();
+    for (i, notification) in connection.iter().enumerate() {
+        match notification {
+            Ok(notif) => {
+                println!("{i}. Notification = {notif:?}");
+                if i > count {
+                    return;
+                }
+            }
+            Err(error) => {
+                println!("{i}. Notification ERROR = {error:?}");
+                return;
+            }
+        }
+    }
 }
 
 pub fn get_topic_name(types: &str) -> String {

src/system_values.rs

@@ -65,7 +65,7 @@ pub fn get_battery() -> i8 {
 
     if system == "ubuntu" {
         let battery_command = Command::new("cat")
-            .arg("/sys/class/power_supply/BAT0/capacity")
+            .arg("/sys/class/power_supply/BAT1/capacity")
             .output()
             .unwrap_or_else(|e| panic!("failed to execute process: {}", e));