Implement event / command framework.

The events feature makes it possible to periodically send (proprietary)
events from the Relay gateways.

The commands feature makes it possible to send commands to the Relay
gateways, in which case the output is returned as an event.

Author: brocaar

Cargo.toml

@@ -32,6 +32,8 @@
     "time",
     "fs",
     "sync",
+    "process",
+    "io-util",
   ] }
   hex = "0.4"
   rand = "0.9"

src/cache.rs

@@ -69,6 +69,16 @@ impl From<&packets::MeshPacket> for PayloadCache {
                     .unwrap_or_default()
                     .as_secs() as u32,
             },
+            packets::Payload::Command(v) => PayloadCache {
+                p_type,
+                uplink_id: 0,
+                relay_id: v.relay_id,
+                timestamp: v
+                    .timestamp
+                    .duration_since(UNIX_EPOCH)
+                    .unwrap_or_default()
+                    .as_secs() as u32,
+            },
         }
     }
 }

src/cmd/configfile.rs

@@ -37,12 +37,6 @@ pub fn run() {
   # uplinks and forward these to the proxy API, rather than relaying these.
   border_gateway={{ mesh.border_gateway }}
 
-  # Heartbeat interval (Relay Gateway only).
-  #
-  # This defines the interval in which a Relay Gateway (border_gateway=false)
-  # will emit heartbeat messages.
-  heartbeat_interval="{{ mesh.heartbeat_interval }}"
-
   # Max hop count.
   #
   # This defines the maximum number of hops a relayed payload will pass.
@@ -142,6 +136,63 @@ pub fn run() {
 
     # Command API URL.
     command_url="{{ backend.mesh_concentratord.command_url }}"
+
+
+# Events configuration (Relay only).
+[events]
+
+  # Heartbeat interval (Relay Gateway only).
+  #
+  # This defines the interval in which a Relay Gateway (border_gateway=false)
+  # will emit heartbeat messages.
+  heartbeat_interval="{{ events.heartbeat_interval }}"
+
+  # Commands.
+  #
+  # This configures for each event type the command that must be executed. The
+  # stdout of the command will be used as event payload. Example:
+  #
+  #   128 = ["/path/to/command", "arg1", "arg2"]
+  #
+  [events.commands]
+
+    {{#each events.commands}}
+    {{@key}} = [{{#each this}}"{{this}}", {{/each}}]
+    {{/each}}
+
+  # Event sets (can be repeated).
+  #
+  # This configures sets of events that will be periodically sent by the
+  # relay. Example:
+  #
+  #  [[events.sets]]
+  #    interval = "5min"
+  #    events = [128, 129, 130]
+  #
+  {{#each events.sets}}
+  [[events.sets]]
+    interval = "{{this.interval}}"
+    events = [{{#each this.events}}{{this}}, {{/each}}]
+  {{/each}}
+
+
+# Commands configuration (Relay only).
+[commands]
+
+  # Commands.
+  #
+  # On receiving a command, the Gateway Mesh will execute the command matching
+  # the command type (128 - 255 is for proprietary commands). The payload will
+  # be provided to the command using stdin. The returned stdout will be sent
+  # back as event (using the same type). Example:
+  #
+  #   "129" = ["/path/to/command", "arg1", "arg2"]
+  #
+  [events.commands]
+
+    {{#each events.commands}}
+    {{@key}} = [{{#each this}}"{{this}}", {{/each}}]
+    {{/each}}
 "#;
 
     let conf = config::get();

src/cmd/root.rs

@@ -4,12 +4,13 @@ use signal_hook::consts::signal::*;
 use signal_hook_tokio::Signals;
 
 use crate::config::Configuration;
-use crate::{backend, heartbeat, proxy};
+use crate::{backend, commands, events, proxy};
 
 pub async fn run(conf: &Configuration) -> Result<()> {
     proxy::setup(conf).await?;
     backend::setup(conf).await?;
-    heartbeat::setup(conf).await?;
+    events::setup(conf).await?;
+    commands::setup(conf).await?;
 
     let mut signals = Signals::new([SIGINT, SIGTERM])?;
     let handle = signals.handle();

src/commands.rs

@@ -0,0 +1,119 @@
+use std::collections::HashMap;
+use std::process::Stdio;
+use std::time::SystemTime;
+
+use anyhow::Result;
+use log::error;
+use tokio::io::AsyncWriteExt;
+use tokio::process::Command;
+use tokio::sync::{Mutex, OnceCell};
+
+use crate::{config::Configuration, packets};
+
+static COMMANDS: OnceCell<HashMap<u8, Vec<String>>> = OnceCell::const_new();
+static LAST_TIMESTAMP: OnceCell<Mutex<Option<SystemTime>>> = OnceCell::const_new();
+
+pub async fn setup(conf: &Configuration) -> Result<()> {
+    // Only Relay Gateways process commands.
+    if conf.mesh.border_gateway {
+        return Ok(());
+    }
+
+    // Set commands.
+    COMMANDS
+        .set(
+            conf.commands
+                .commands
+                .iter()
+                .map(|(k, v)| (k.parse().unwrap(), v.clone()))
+                .collect(),
+        )
+        .map_err(|_| anyhow!("OnceCell set error"))?;
+
+    Ok(())
+}
+
+pub async fn execute_commands(pl: &packets::CommandPayload) -> Result<Vec<packets::Event>> {
+    // Validate that the command timestamp did increment, compared to previous
+    // command payload.
+    if let Some(ts) = get_last_timestamp().await {
+        if ts >= pl.timestamp {
+            return Err(anyhow!(
+                "Command timestamp did not increment compared to previous command payload"
+            ));
+        }
+    }
+
+    // Store the command timestamp.
+    set_last_timestamp(pl.timestamp).await;
+
+    // Execute the commands and capture the response events.
+    let mut out = vec![];
+    for cmd in &pl.commands {
+        let resp = match cmd {
+            packets::Command::Proprietary((t, v)) => execute_proprietary(*t, v).await,
+        };
+
+        match resp {
+            Ok(v) => out.push(v),
+            Err(e) => error!("Execute command error: {}", e),
+        }
+    }
+
+    Ok(out)
+}
+
+async fn execute_proprietary(typ: u8, value: &[u8]) -> Result<packets::Event> {
+    let args = COMMANDS
+        .get()
+        .ok_or_else(|| anyhow!("COMMANDS is not set"))?
+        .get(&typ)
+        .ok_or_else(|| anyhow!("Command type {} is not configured", typ))?;
+
+    if args.is_empty() {
+        return Err(anyhow!("Command for command type {} is empty", typ,));
+    }
+
+    let mut cmd = Command::new(&args[0]);
+    cmd.stdin(Stdio::piped());
+    cmd.stdout(Stdio::piped());
+    cmd.stderr(Stdio::piped());
+
+    // Add addition args.
+    if args.len() > 1 {
+        cmd.args(&args[1..]);
+    }
+
+    // Spawn process
+    let mut child = cmd.spawn()?;
+
+    // Write stdin
+    let mut stdin = child.stdin.take().unwrap();
+    tokio::spawn({
+        let b = value.to_vec();
+        async move { stdin.write(&b).await }
+    });
+
+    // Wait for output
+    let out = child.wait_with_output().await?;
+    Ok(packets::Event::Proprietary((typ, out.stdout)))
+}
+
+async fn get_last_timestamp() -> Option<SystemTime> {
+    LAST_TIMESTAMP
+        .get_or_init(|| async { Mutex::new(None) })
+        .await
+        .lock()
+        .await
+        .clone()
+}
+
+async fn set_last_timestamp(ts: SystemTime) {
+    let mut last_ts = LAST_TIMESTAMP
+        .get_or_init(|| async { Mutex::new(None) })
+        .await
+        .lock()
+        .await;
+
+    *last_ts = Some(ts);
+}

src/config.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::sync::{Arc, Mutex, OnceLock};
 use std::time::Duration;
 use std::{env, fs};
@@ -18,6 +19,7 @@ pub struct Configuration {
     pub backend: Backend,
     pub mappings: Mappings,
     pub events: Events,
+    pub commands: Commands,
 }
 
 impl Configuration {
@@ -158,16 +160,34 @@ pub struct DataRate {
 pub struct Events {
     #[serde(with = "humantime_serde")]
     pub heartbeat_interval: Duration,
+    pub commands: HashMap<String, Vec<String>>,
+    pub sets: Vec<EventsSet>,
 }
 
 impl Default for Events {
     fn default() -> Self {
         Events {
             heartbeat_interval: Duration::from_secs(300),
+            commands: Default::default(),
+            sets: Vec::new(),
         }
     }
 }
 
+#[derive(Default, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(default)]
+pub struct EventsSet {
+    #[serde(with = "humantime_serde")]
+    pub interval: Duration,
+    pub events: Vec<u8>,
+}
+
+#[derive(Default, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(default)]
+pub struct Commands {
+    pub commands: HashMap<String, Vec<String>>,
+}
+
 #[derive(Serialize, Deserialize, Clone, Copy, PartialEq, Eq, Default)]
 #[allow(non_camel_case_types)]
 #[allow(clippy::upper_case_acronyms)]