Werte in die InfluxDB übermitteln

src/influx.rs

@@ -0,0 +1,112 @@
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::Duration;
+use crate::config::{AppConfig, ModbusValueMaps};
+use influxdb::{Client, WriteQuery, Timestamp};
+use chrono::Utc;
+use tokio::runtime::Runtime;
+
+pub fn start_influx_thread(config: Arc<Mutex<AppConfig>>, values: Arc<Mutex<ModbusValueMaps>>) {
+    let influx_config = {
+        let cfg = config.lock().unwrap();
+        cfg.influxdb.clone()
+    };
+    let url = influx_config.url.clone();
+    let bucket = influx_config.bucket.clone();
+    let _org = influx_config.org.clone();
+    let token = influx_config.token.clone();
+    let measurement = influx_config.measurement.clone().unwrap_or_else(|| "modbus".to_string());
+    let location = influx_config.location.clone().unwrap_or_else(|| "default".to_string());
+
+    thread::spawn(move || {
+        let client = Client::new(url, bucket).with_token(token);
+        let rt = Runtime::new().unwrap();
+        loop {
+            let mut points = Vec::new();
+            {
+                let values = values.lock().unwrap();
+                // Input Register
+                for (name, val) in &values.modbus_input_register_values {
+                    if let Some(v) = val {
+                        if should_write(&config, name, "input_register") {
+                            let point = WriteQuery::new(Timestamp::from(Utc::now()), measurement.clone())
+                                .add_field(name, *v)
+                                .add_tag("type", "input_register")
+                                .add_tag("location", location.as_str())
+                                .to_owned();
+                            points.push(point);
+                        }
+                    }
+                }
+                // Holding Register
+                for (name, val) in &values.modbus_holding_register_values {
+                    if let Some(v) = val {
+                        if should_write(&config, name, "holding_register") {
+                            let point = WriteQuery::new(Timestamp::from(Utc::now()), measurement.clone())
+                                .add_field(name, *v)
+                                .add_tag("type", "holding_register")
+                                .add_tag("location", location.as_str())
+                                .to_owned();
+                            points.push(point);
+                        }
+                    }
+                }
+                // Coils
+                for (name, val) in &values.modbus_coils_values {
+                    if let Some(v) = val {
+                        if should_write(&config, name, "coils") {
+                            let point = WriteQuery::new(Timestamp::from(Utc::now()), measurement.clone())
+                                .add_field(name, *v)
+                                .add_tag("type", "coils")
+                                .add_tag("location", location.as_str())
+                                .to_owned();
+                            points.push(point);
+                        }
+                    }
+                }
+            }
+            if !points.is_empty() {
+                let res = rt.block_on(client.query(points));
+                if let Err(e) = res {
+                    eprintln!("Influx write error: {}", e);
+                }
+            }
+            thread::sleep(Duration::from_secs(10));
+        }
+    });
+}
+
+fn should_write(config: &Arc<Mutex<AppConfig>>, name: &str, reg_type: &str) -> bool {
+    let cfg = config.lock().unwrap();
+    match reg_type {
+        "input_register" => {
+            if let Some(regs) = &cfg.modbus_input_register {
+                for map in regs {
+                    if let Some(reg) = map.get(name) {
+                        return reg.influxdb.unwrap_or(false);
+                    }
+                }
+            }
+        }
+        "holding_register" => {
+            if let Some(regs) = &cfg.modbus_holding_register {
+                for map in regs {
+                    if let Some(reg) = map.get(name) {
+                        return reg.influxdb.unwrap_or(false);
+                    }
+                }
+            }
+        }
+        "coils" => {
+            if let Some(coils) = &cfg.modbus_coils {
+                for map in coils {
+                    if let Some(coil) = map.get(name) {
+                        return coil.influxdb.unwrap_or(false);
+                    }
+                }
+            }
+        }
+        _ => {}
+    }
+    false
+}

src/main.rs

@@ -10,6 +10,7 @@ mod config;
 mod modbus;
 mod app_state;
 mod mqtt;
+mod influx;
 pub mod modbus_types;
 use crate::config::{AppConfig, ModbusValueMaps};
 use crate::modbus_types::{ModbusInputRegisterConfig, ModbusHoldingRegisterConfig, ModbusCoilsConfig};
@@ -254,6 +255,12 @@ async fn main() -> std::io::Result<()> {
         mqtt::start_mqtt_thread(Arc::clone(&app_state.config), value_maps);
     }
 
+    // Starte InfluxDB-Thread
+    {
+        let value_maps = Arc::clone(&app_state.value_maps);
+        influx::start_influx_thread(Arc::clone(&app_state.config), value_maps);
+    }
+
     println!("Server läuft auf http://0.0.0.0:8080");
 
     HttpServer::new(move || {

src/modbus_types.rs

@@ -9,7 +9,6 @@ pub struct ModbusCoilsConfig {
     pub comment: Option<String>,
 }
 
-
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct ModbusInputRegisterConfig {
     pub addr: u16,