allow sending mqtt messages on action

Cargo.toml

@@ -5,23 +5,19 @@ authors = ["Robin Appelman <robin@icewind.nl>"]
 edition = "2018"
 
 [dependencies]
-prometheus-edge-detector = { version = "0.1", git = "https://github.com/icewind1991/prometheus-edge-detector", default-features = false }
+prometheus-edge-detector = { version = "0.1", git = "https://github.com/icewind1991/prometheus-edge-detector", default-features = false, features = ["rustls-tls"] }
 mdns = "1.1"
 tokio = { version = "0.2.4", features = ["macros", "time", "fs"] }
 main_error = "0.1.0"
 futures-util = "0.3.1"
-reqwest = { version = "0.10.0", default-features = false }
+reqwest = { version = "0.10.0", default-features = false, features = ["rustls-tls"] }
 serde = { version = "1.0", features = ["derive"] }
 toml = "0.5"
 log = "0.4"
 env_logger = "0.7"
 err-derive = "0.2.1"
 serde_json = "1.0.45"
+rumqttc = "0.2.0"
 
 [dev-dependencies]
 maplit = "1.0.2"
-
-[features]
-default = ["default-tls"]
-default-tls = ["reqwest/default-tls", "prometheus-edge-detector/default-tls"]
-rustls-tls = ["reqwest/rustls-tls", "prometheus-edge-detector/rustls-tls"]

Dockerfile

@@ -5,7 +5,7 @@ FROM ekidd/rust-musl-builder AS build
 ADD . ./
 RUN sudo chown -R rust:rust .
 
-RUN cargo build --release --no-default-features --features rustls-tls
+RUN cargo build --release
 
 FROM scratch
 

src/config.rs

@@ -64,6 +64,7 @@ impl Parameter {
 
 #[derive(Debug, Clone, Deserialize)]
 pub struct Condition {
+    #[serde(default)]
     pub params: HashMap<String, Parameter>,
     pub query: String,
     pub from: u64,
@@ -73,13 +74,17 @@ pub struct Condition {
 #[derive(Debug, Clone, Deserialize)]
 pub struct Action {
     pub method: Method,
+    #[serde(default)]
     pub params: HashMap<String, Parameter>,
-    pub url: String,
+    pub url: Option<String>,
+    pub topic: Option<String>,
+    pub payload: Option<String>,
 }
 
 #[derive(Debug, Clone, Deserialize)]
 pub struct Config {
     pub prometheus: PrometheusConfig,
+    pub mqtt: Option<MqttConfig>,
     #[serde(rename = "trigger")]
     pub triggers: Vec<Trigger>,
 }
@@ -89,6 +94,12 @@ pub struct PrometheusConfig {
     pub url: String,
 }
 
+#[derive(Debug, Clone, Deserialize)]
+pub struct MqttConfig {
+    pub host: String,
+    pub port: Option<u16>
+}
+
 #[derive(Debug, Clone, Deserialize)]
 pub struct Trigger {
     pub name: String,
@@ -97,12 +108,13 @@ pub struct Trigger {
     pub action: Action,
 }
 
-#[derive(Debug, Clone, Deserialize)]
+#[derive(Debug, Clone, Deserialize, Copy)]
 #[serde(rename_all = "UPPERCASE")]
 pub enum Method {
     Get,
     Put,
     Post,
+    Mqtt
 }
 
 #[derive(Debug, Clone, Deserialize)]

src/trigger.rs

@@ -1,4 +1,4 @@
-use crate::config::{Action, Condition, Config, Method, Parameter, ParameterError, Trigger};
+use crate::config::{Action, Condition, Config, Method, Parameter, ParameterError, Trigger, MqttConfig};
 use err_derive::Error;
 use futures_util::future::try_join_all;
 use log::{error, info};
@@ -8,9 +8,11 @@ use reqwest::Client;
 use std::collections::HashMap;
 use std::time::{Duration, SystemTime};
 use tokio::time::delay_for;
+use rumqttc::{MqttOptions, AsyncClient, QoS};
 
 pub struct TriggerManager {
     http_client: Client,
+    mqtt_config: Option<MqttConfig>,
     edge_detector: EdgeDetector,
     triggers: Vec<Trigger>,
 }
@@ -34,6 +36,8 @@ pub enum TriggerError {
     Edge(#[error(source)] prometheus_edge_detector::Error),
     #[error(display = "{}", _0)]
     Network(#[error(source)] reqwest::Error),
+    #[error(display = "{}", _0)]
+    Mqtt(#[error(source)] rumqttc::ClientError),
 }
 
 impl TriggerManager {
@@ -42,6 +46,7 @@ impl TriggerManager {
 
         TriggerManager {
             http_client: Client::new(),
+            mqtt_config: config.mqtt,
             edge_detector,
             triggers: config.triggers,
         }
@@ -53,7 +58,7 @@ impl TriggerManager {
                 .iter()
                 .map(|trigger| self.run_trigger(trigger)),
         )
-        .await?;
+            .await?;
 
         Ok(())
     }
@@ -78,7 +83,7 @@ impl TriggerManager {
                     match self.get_edge(&trigger.condition, delay).await {
                         Ok(Some(new_edge)) if new_edge == edge => {
                             info!("[{}] Edge still valid, triggering", trigger.name);
-                            if let Err(e) = run_action(&trigger.action, &self.http_client).await {
+                            if let Err(e) = run_action(&trigger.action, &self.http_client, self.mqtt_config.as_ref()).await {
                                 error!("[{}]: {}", trigger.name, e);
                             }
                             delay_for(delay_duration).await;
@@ -125,6 +130,16 @@ impl TriggerManager {
     }
 }
 
+async fn interpolate_option_params(
+    input: &Option<String>,
+    params: &HashMap<String, Parameter>,
+) -> Result<Option<String>, ParameterError> {
+    match input.as_ref() {
+        Some(input) => Ok(Some(interpolate_params(input, params).await?)),
+        None => Ok(None)
+    }
+}
+
 async fn interpolate_params(
     input: &str,
     params: &HashMap<String, Parameter>,
@@ -153,19 +168,41 @@ async fn test_interpolate() {
             "param".to_string() => Parameter::Value{value: "bar".to_string()}
         },
     )
-    .await;
+        .await;
     assert_eq!("foo_bar".to_string(), result.unwrap());
 }
 
-async fn run_action(action: &Action, client: &Client) -> Result<(), TriggerError> {
-    let url = interpolate_params(&action.url, &action.params).await?;
+async fn run_action(action: &Action, client: &Client, mqtt_config: Option<&MqttConfig>) -> Result<(), TriggerError> {
+    let url = interpolate_option_params(&action.url, &action.params).await?;
+    let topic = interpolate_option_params(&action.topic, &action.params).await?;
+    let payload = interpolate_option_params(&action.payload, &action.params).await?;
 
-    let req = match action.method {
-        Method::Put => client.put(&url),
-        Method::Post => client.post(&url),
-        Method::Get => client.get(&url),
+    match (action.method, url, topic, payload) {
+        (Method::Put, Some(url), _, _) => {
+            client.put(&url).send().await?;
+        },
+        (Method::Post, Some(url), _, _) => {
+            client.post(&url).send().await?;
+        }
+        (Method::Get, Some(url), _, _) => {
+            client.get(&url).send().await?;
+        }
+        (Method::Mqtt, _, Some(topic), Some(payload)) => {
+            if let Some(mqtt_config) = mqtt_config {
+                let mqtt_options = MqttOptions::new("rumqtt-async", mqtt_config.host.as_str(), mqtt_config.port.unwrap_or(1883));
+
+                let (mqtt_client, mut event_loop) = AsyncClient::new(mqtt_options, 10);
+                mqtt_client.publish(topic, QoS::AtMostOnce, false, payload).await?;
+
+                let _ = tokio::time::timeout(Duration::from_secs(1), async move {
+                    loop {
+                        let _ = event_loop.poll().await;
+                    }
+                }).await;
+            }
+        },
+        _ => {}
     };
-    req.send().await?;
 
     Ok(())
 }