Run in loop

Cargo.toml

@@ -13,6 +13,8 @@ rumqttc = "0.24.0"
 serde = { version = "1.0.215", features = ["derive"] }
 serde_json = "1.0.133"
 tokio = { version = "1.41.1", features = ["full"] }
+tracing = "0.1.40"
+tracing-subscriber = "0.3.18"
 url = { version = "2.5.4", features = ["serde"] }
 
 [dev-dependencies]

config/default.toml

@@ -1,11 +1,12 @@
+update_interval_minutes = 60
+
 [mqtt]
 client_id = "affaldvarme"
 host = "10.0.1.3"
 port = 1883
 username = ""
 password = ""
 
-
 [affaldvarme]
 base_url = "https://portal-api.kredslob.dk"
 

src/main.rs

@@ -1,18 +1,37 @@
 use ha_mitaffald::settings::Settings;
 use ha_mitaffald::sync_data;
+use tracing::{error, info, Level};
+use tracing_subscriber::FmtSubscriber;
 
 #[tokio::main]
 async fn main() {
-    println!("Starting data synchronization");
+    let subscriber = FmtSubscriber::builder()
+        .with_max_level(Level::INFO)
+        .finish();
 
-    let settings = Settings::new().expect("Failed to read settings");
-    let report = sync_data(settings).await;
+    tracing::subscriber::set_global_default(subscriber).expect("setting default subscriber failed");
+    loop {
+        info!("Starting data synchronization");
 
-    match report {
-        Ok(_) => println!("Data synchronization completed"),
-        Err(x) => eprintln!(
-            "Data synchronization failed (some entities may have been updated), error: {}",
-            x
-        ),
+        let settings = Settings::new().expect("Failed to read settings");
+        let update_interval =
+            tokio::time::Duration::from_secs(settings.update_interval_minutes * 60);
+
+        let report = sync_data(settings).await;
+
+        match report {
+            Ok(_) => info!("Data synchronization completed"),
+            Err(x) => error!(
+                "Data synchronization failed (some entities may have been updated), error: {}",
+                x
+            ),
+        }
+
+        info!(
+            "Next synchronization scheduled at {}",
+            (chrono::Local::now() + update_interval).format("%Y-%m-%d %H:%M:%S")
+        );
+
+        tokio::time::sleep(update_interval).await;
     }
 }

src/mitaffald/mod.rs

@@ -3,6 +3,7 @@ pub mod settings;
 use chrono::{DateTime, Utc};
 use serde::Deserialize;
 use settings::{Address, AffaldVarmeConfig};
+use tracing::info;
 use url::Url;
 
 use self::settings::{AddressId, TraditionalAddress};
@@ -25,7 +26,7 @@ pub async fn get_containers(config: AffaldVarmeConfig) -> Result<Vec<Container>,
                 .next()
                 .ok_or_else(|| "No data found".to_string())
                 .map(|response| {
-                    println!("Received information for stand: {}", response.stand_name);
+                    info!("Received information for stand: {}", response.stand_name);
                     response.into()
                 })
         })

src/settings.rs

@@ -7,6 +7,7 @@ use serde::Deserialize;
 pub struct Settings {
     pub mqtt: MQTTConfig,
     pub affaldvarme: AffaldVarmeConfig,
+    pub update_interval_minutes: u64,
 }
 
 impl Settings {

tests/full_flow_insta.rs

@@ -46,6 +46,7 @@ async fn smoke_test_insta() {
         .await;
 
     let settings = Settings {
+        update_interval_minutes: 60,
         affaldvarme: AffaldVarmeConfig {
             address: Address::Id(AddressId { id: address_id }),
             base_url: mit_affald_server_url,

tests/mqtt/mod.rs

@@ -6,6 +6,7 @@ use std::{
 };
 
 use rumqttc::{Client, Event, Packet, Publish, QoS};
+use tracing::info;
 
 pub struct CollectingClient {
     received_messages: std::sync::Arc<Mutex<Vec<Publish>>>,
@@ -37,7 +38,7 @@ impl CollectingClient {
 
             loop {
                 let message = connection.recv_timeout(Duration::from_secs(1));
-                println!("Received message: {:?}", &message);
+                info!("Received message: {:?}", &message);
                 match message {
                     Ok(Ok(Event::Incoming(Packet::SubAck(_)))) => {
                         tx.send(()).expect("Cannot report ready to main thread")
@@ -49,7 +50,7 @@ impl CollectingClient {
                 }
 
                 if stopping_flag.load(std::sync::atomic::Ordering::Relaxed) {
-                    println!("Thread is terminating");
+                    info!("Thread is terminating");
                     break;
                 }
             }
@@ -83,7 +84,7 @@ impl CollectingClient {
             std::thread::sleep(std::time::Duration::from_millis(500));
         }
 
-        println!("Joining worker thread...");
+        info!("Joining worker thread...");
         if let Some(handle) = self.join_handle {
             handle.join().unwrap();
         }