Cargo.lock

@@ -340,22 +340,12 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
-[[package]]
-name = "mqtt-client"
-version = "4.0.0"
-source = "git+https://gitea.finnvanreenen.nl/LailaTheElf/mqttClient.git?tag=v4.0.0#49e8adf2eb768fcd147f0d6508a2f32eed86a641"
-dependencies = [
- "crossbeam",
- "rumqttc",
-]
-
 [[package]]
 name = "mqttClock"
-version = "1.3.0"
+version = "1.0.0"
 dependencies = [
  "chrono",
  "crossbeam",
- "mqtt-client",
  "rumqttc",
  "serde",
  "serde_yaml",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mqttClock"
-version = "1.3.0"
+version = "1.0.0"
 edition = "2021"
 
 [dependencies]
@@ -9,4 +9,3 @@ crossbeam = "0.8.4"
 rumqttc = "0.24.0"
 serde = { version = "1.0.217", features = ["derive"] }
 serde_yaml = "0.9.34"
-mqtt-client = { tag = "v4.0.0", git = "https://gitea.finnvanreenen.nl/LailaTheElf/mqttClient.git" }

build.sh

@@ -1,7 +0,0 @@
-#!/bin/sh
-
-cross build --target aarch64-unknown-linux-gnu --release
-cargo build --release
-
-cp target/aarch64-unknown-linux-gnu/release/mqttClock mqttClock-aarch64
-cp target/release/mqttClock mqttClock-x86_64

mqttClock.yml

@@ -1,6 +1,5 @@
 mqtt:
   host: "localhost"
   port: 1883
-  client: "mqttClock"
   user: "mqttClock"
   pass: "password"

src/main.rs

@@ -1,175 +1,101 @@
-use std::{fs, time::Duration};
+use std::time::Duration;
+use std::thread;
+use std::fs;
 
-use serde::Deserialize;
+use rumqttc::{Client, Event, MqttOptions, Outgoing, Packet, QoS};
 use chrono::{Local, Timelike, Datelike};
+use crossbeam::channel::{unbounded, Sender};
+use serde::Deserialize;
 
-use mqtt_client::{mqtt_client, MqttMessage, MqttEvent, Sender, Receiver, QoS, Client, MqttTool};
+struct MqttMessage {
+    topic: String,
+    payload: String 
+}
 
-fn get_u16_from_i32(value: i32) -> Option<u16> {
+fn send_publish(publish: &Sender<MqttMessage>, message: MqttMessage) {
-    match u16::try_from(value) {
+    match publish.send(message) {
-        Ok(n) => Some(n),
+        Err(n) => println!("ERROR: faild to send publish ({:?})", n),
-        Err(_) => None
+        Ok(_n) => {}
     }
 }
 
-fn get_u16_from_u32(value: u32) -> Option<u16> {
+fn mqtt_clock(publish: Sender<MqttMessage>) {
-    match u16::try_from(value) {
+    // init last values with invalid values
-        Ok(n) => Some(n),
+    let mut last_year: i32 = 65535;
-        Err(_) => None
+    let mut last_month: u32 = 65535;
-    }
+    let mut last_dom: u32 = 65535;
-}
+    let mut last_dow: u32 = 65535;
+    let mut last_iso_week: u32 = 65535;
+    let mut last_iso_year: i32 = 65535;
+    let mut last_hour: u32 = 65535;
+    let mut last_minute: u32 = 65535;
+    let mut last_second: u32 = 65535;
 
-struct Clock {
+    loop {
-    tx: Sender<MqttMessage>,
-    // client: Client,
-    connected: bool,
-
-    last_year: Option<u16>,
-    last_month: Option<u16>,
-    last_dom: Option<u16>,
-    last_dow: Option<u16>,
-    last_iso_week: Option<u16>,
-    last_iso_year: Option<u16>,
-    last_hour: Option<u16>,
-    last_minute: Option<u16>,
-    last_second: Option<u16>
-}
-impl Clock {
-    fn tx(&self, topic: String, value: Option<u16>, retain: bool) {
-        match value {
-            None => {},
-            Some(v) => {
-                let message = MqttMessage {
-                    topic,
-                    payload: v.to_string(),
-                    retain: retain,
-                    qos: QoS::AtMostOnce,
-                };
-                match self.tx.send(message) {
-                    Err(n) => println!("ERROR: faild to send publish ({:?})", n),
-                    Ok(_) => {}
-                }
-            }
-        }
-    }
-    fn tx_time(&self) {
-        let second = self.last_second;
-        let minute = self.last_minute;
-        let hour = self.last_hour;
-        if let (Some(second), Some(minute), Some(hour)) = (second, minute, hour) {
-            let payload = 
-                format!("{{\"second\":{},\"minute\":{},\"hour\":{},\"time\":\"{}:{:02}:{:02}\"}}",
-                    second, minute, hour, hour, minute, second);
-
-            let message = MqttMessage {
-                topic: String::from("clock/time/time"),
-                payload,
-                retain: false,
-                qos: QoS::AtMostOnce,
-            };
-            match self.tx.send(message) {
-                Err(n) => 
-                    println!("ERROR: faild to send publish ({:?})", n),
-                Ok(_) => {}
-            }
-        }
-    }
-
-    fn time(&mut self) {
         let datetime = Local::now();
-        if self.last_second != get_u16_from_u32(datetime.second()) {
+        if last_second != datetime.second() {
-            self.last_second = get_u16_from_u32(datetime.second());
+            last_second = datetime.second();
-            if self.last_minute != get_u16_from_u32(datetime.minute()) {
+            send_publish(&publish, { MqttMessage {
-                self.last_minute = get_u16_from_u32(datetime.minute());
+                topic: String::from("clock/time/second"), 
-                if self.last_hour != get_u16_from_u32(datetime.hour()) {
+                payload: last_second.to_string()
-                    self.last_hour = get_u16_from_u32(datetime.hour());
+            }});
-                    self.tx(String::from("clock/time/hour"), self.last_hour, true);
+            if last_minute != datetime.minute() {
-                    self.date(datetime);
+                last_minute = datetime.minute();
-                }
+                send_publish(&publish, { MqttMessage {
-                self.tx(String::from("clock/time/minute"), self.last_minute, false);
+                    topic: String::from("clock/time/minute"), 
-            }
+                    payload: last_minute.to_string()
-            self.tx(String::from("clock/time/second"), self.last_second, false);
+                }});
-            self.tx_time();
+                if last_hour != datetime.hour() {
-        }
+                    last_hour = datetime.hour();
-    }
+                    send_publish(&publish, { MqttMessage {
-
+                        topic: String::from("clock/time/hour"), 
-    fn date(&mut self, datetime: chrono::DateTime<Local>) {
+                        payload: last_hour.to_string()
-        if self.last_dom != get_u16_from_u32(datetime.day()) {
+                    }});
-            self.last_dom = get_u16_from_u32(datetime.day());
+                    if last_dom != datetime.day() {
-            if self.last_iso_week != get_u16_from_u32(datetime.iso_week().week()) {
+                        last_dom = datetime.day();
-                self.last_iso_week = get_u16_from_u32(datetime.iso_week().week());
+                        send_publish(&publish, { MqttMessage {topic: String::from("clock/date/dom"), 
-                if self.last_iso_year != get_u16_from_i32(datetime.iso_week().year()) {
+                            payload: last_dom.to_string()
-                    self.last_iso_year = get_u16_from_i32(datetime.iso_week().year());
+                        }});
-                    self.tx(String::from("clock/date/isoYear"), self.last_iso_year, true);
+                        if last_iso_week != datetime.iso_week().week() {
-                }
+                            last_iso_week = datetime.iso_week().week();
-                self.tx(String::from("clock/date/isoWeek"), self.last_iso_week, true);
+                            send_publish(&publish, { MqttMessage {
-            }
+                                topic: String::from("clock/date/isoWeek"), 
-            self.tx(String::from("clock/date/dom"), self.last_dom, true);
+                                payload: last_iso_week.to_string()
-
+                            }});
-            if self.last_month != get_u16_from_u32(datetime.month()) {
+                            if last_iso_year != datetime.iso_week().year() {
-                self.last_month = get_u16_from_u32(datetime.month());
+                                last_iso_year = datetime.iso_week().year();
-                if self.last_year != get_u16_from_i32(datetime.year()) {
+                                send_publish(&publish, { MqttMessage {
-                    self.last_year = get_u16_from_i32(datetime.year());
+                                    topic: String::from("clock/date/isoYear"), 
-                    self.tx(String::from("clock/date/year"), self.last_year, true);
+                                    payload: last_iso_year.to_string()
-                }
+                                }});
-                self.tx(String::from("clock/date/month"), self.last_month, true);
+                            }
-            }
+                        }
-
+                        if last_dow != datetime.weekday() as u32 {
-            self.last_dow = Some(datetime.weekday() as u16);
+                            last_dow = datetime.weekday() as u32;
-            self.tx(String::from("clock/date/dow"), self.last_dow, true);
+                            send_publish(&publish, { MqttMessage {
-        }
+                                topic: String::from("clock/date/dow"), 
-    }
+                                payload: last_dow.to_string()
-
+                            }});
-    fn init(&mut self) {
+                        }
-        self.last_year = None;
+                        if last_month != datetime.month() {
-        self.last_month = None;
+                            last_month = datetime.month();
-        self.last_dom = None;
+                            send_publish(&publish, { MqttMessage {
-        self.last_dow = None;
+                                topic: String::from("clock/date/month"), 
-        self.last_iso_week = None;
+                                payload: last_month.to_string()
-        self.last_iso_year = None;
+                            }});
-        self.last_hour = None;
+                            if last_year != datetime.year() {
-    }
+                                last_year = datetime.year();
-
+                                send_publish(&publish, { MqttMessage {
-}
+                                    topic: String::from("clock/date/year"), 
-impl MqttTool<i8> for Clock {
+                                    payload: last_year.to_string()
-    fn new(tx: Sender<MqttMessage>, _config: i8, _client: Client) -> Clock {
+                                }});
-        Clock {
+                            }
-            tx,
+                        }
-            // client,
-            connected: false,
-            last_year: None,
-            last_month: None,
-            last_dom: None,
-            last_dow: None,
-            last_iso_week: None,
-            last_iso_year: None,
-            last_hour: None,
-            last_minute: None,
-            last_second: None
-        }
-    }
-
-    fn run(&mut self, event: Receiver<MqttEvent>) {
-        loop {
-            match event.recv_timeout(Duration::from_millis(500)) {
-                Ok(e) => {
-                    match e {
-                        MqttEvent::Connected => {
-                            self.init();
-                            self.connected = true;
-                        },
-                        MqttEvent::Disconnected => {
-                            self.connected = false;
-                        },
-                        MqttEvent::Message(_) => {/* nothing to do, no subscriptions */}
                     }
-                },
+                }
-                Err(_) => {},
-            }
-            if self.connected {
-                self.time();
             }
         }
+        thread::sleep(Duration::from_millis(500));
     }
 }
 
@@ -178,7 +104,6 @@ impl MqttTool<i8> for Clock {
 struct SettingsMQTT {
     host: String,
     port: u16,
-    client: String,
     user: String,
     pass: String
 }
@@ -226,20 +151,103 @@ fn read_config() -> Result<Settings,SettingsError> {
 }
 
 fn main() {
+    let (mqtt_publish, mqtt_publish_rx) = unbounded::<MqttMessage>();
+
     // get setting
+    let conf_ok: bool;
+    let conf: Settings;
     match read_config() {
-        Ok(conf) =>
+        Ok(n) => {
-            mqtt_client::run::<i8, Clock>(
+            conf = n;
-                conf.mqtt.host,
+            conf_ok = true;
-                conf.mqtt.port,
+        },
-                conf.mqtt.client,
+        Err(_) => {
-                conf.mqtt.user,
+            conf = Settings {
-                conf.mqtt.pass,
+                mqtt: SettingsMQTT {host:String::new(),port:0,user:String::new(),pass:String::new()}
-                0
+            };
-            ),
+            conf_ok = false;
-        Err(_) => {}
+        }
     }
 
-    println!("INFO : main: exit");
+    if conf_ok {
-}
+        let mut mqttoptions = MqttOptions::new("rumqtt-sync", conf.mqtt.host, conf.mqtt.port);
+        mqttoptions.set_keep_alive(Duration::from_secs(5));
+        mqttoptions.set_credentials(conf.mqtt.user, conf.mqtt.pass);
+        let (client, mut connection) = Client::new(mqttoptions, 10);
+    
+        // thread publisher
+        let publisher = thread::Builder::new()
+            .name("publisher".to_string())
+            .spawn(move || {
+                loop {
+                    let message = mqtt_publish_rx.recv();
+                    match message {
+                        Err(e) => println!("ERROR: publisher: faild to receve an message ({})", e),
+                        Ok(msg) => {
+                            // println!("DEBUG: publisher: topic={}; payload={}", msg.topic, msg.payload);
+                            match client.publish(msg.topic, QoS::AtMostOnce, false, msg.payload) {
+                                Err(_n) => println!("ERROR: publisher: faild to publish"),
+                                Ok(_n) => {}
+                            }
+                        }
+                    }
+                }
+            });
+        match publisher {
+            Err(_n) => println!("ERROR: main: fait to create publisher thread"),
+            Ok(_n) => {}
+        }
+    
+        // treath mqtt clock
+        let mqtt_clock_stat = thread::Builder::new()
+            .name("mqtt_clock".to_string())
+            .spawn(move || {
+                mqtt_clock(mqtt_publish);
+            });
+        match mqtt_clock_stat {
+            Err(_n) => println!("ERROR: main: faild to start mqtt clock thread"),
+            Ok(_n) => {}
+        }
 
+        for (_i, notification) in connection.iter().enumerate() {
+            match notification {
+                Err(e) => println!("ERROR: mqtt: {}", e),
+                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"),
+                            Outgoing::PubAck(_n) => {}, //println!("INFO : mqtt_recive: out PubAck"),
+                            Outgoing::PubRec(_n) => {}, //println!("INFO : mqtt_recive: out PubRec"),
+                            Outgoing::PubRel(_n) => {}, //println!("INFO : mqtt_recive: out PubRel"),
+                            Outgoing::PubComp(_n) => {}, //println!("INFO : mqtt_recive: out PubComp"),
+                            Outgoing::PingReq => {}, //println!("INFO : mqtt_recive: out PingReq"),
+                            Outgoing::PingResp => {}, //println!("INFO : mqtt_recive: out PingResp"),
+                            Outgoing::Disconnect => {}, //println!("INFO : mqtt_recive: out Disconnect"),
+                            Outgoing::AwaitAck(_n) => {} //println!("INFO : mqtt_recive: out AwaitAck")
+                        },
+                        Event::Incoming(n) => match n {
+                            Packet::Connect(_) => println!("INFO : mqtt: connected"), //println!("INFO : mqtt_recive: in Connect"),
+                            Packet::ConnAck(_) => println!("INFO : mqtt: conn ack"), //println!("INFO : mqtt_recive: in ConnAck"),
+                            Packet::Publish(_) => {},
+                            Packet::PubAck(_) => {}, //println!("INFO : mqtt_recive: in PubAck"),
+                            Packet::PubRec(_) => {}, //println!("INFO : mqtt_recive: in PubRec"),
+                            Packet::PubRel(_) => {}, //println!("INFO : mqtt_recive: in PubRel"),
+                            Packet::PubComp(_) => {}, //println!("INFO : mqtt_recive: in PubComp"),
+                            Packet::Subscribe(_) => {}, //println!("INFO : mqtt_recive: in Subscribe"),
+                            Packet::SubAck(_) => {}, //println!("INFO : mqtt_recive: in SubAck"),
+                            Packet::Unsubscribe(_) => {}, //println!("INFO : mqtt_recive: in Unsubscribe"),
+                            Packet::UnsubAck(_) => {}, //println!("INFO : mqtt_recive: in UnsubAck"),
+                            Packet::PingReq => {}, //println!("INFO : mqtt_recive: in PingReq"),
+                            Packet::PingResp => {}, //println!("INFO : mqtt_recive: in PingResp"),
+                            Packet::Disconnect => println!("INFO : mqtt: disconected"),
+                        }
+                    }
+                }
+            }
+        }
+        println!("INFO : main: exit");
+    }
+}