Pass packet metadata as arguments

DUP, QoS, RETAIN and the packet identifier

3 files changed, 21 insertions, 8 deletions

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 176a5a0..73e40cd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,8 @@
 ### Added
 
 * Error if the config contains credentials but is group or world readable.
+* The message DUP, QoS, and RETAIN fields as well as the Packet Identifier (when
+  QoS is 1 or 2) are now appended (in that order) to the process arguments.
 
 ### Fixed
 
diff --git a/README.md b/README.md
index 8bf7818..24c635f 100644
--- a/README.md
+++ b/README.md
@@ -55,14 +55,19 @@ The routes follow the following format:
 On startup, `mqttr` will read the config file and subscribe to all the topics.
 When a message is received, `mqttr` will match the message topic against the
 routes and execute every program which is part of a matching route. Each program
-will have the message topic appended as an additional argument and will receive
-the message on stdin.
+will have the message topic as well as the DUP, QoS, RETAIN and Packet
+Identifier (when applicable) appended as an additional arguments (in that order)
+and will receive the message on stdin.
 
 An example program is this script:
 
 ```bash
 #!/usr/bin/env bash
 topic=$1
+dup=$2
+qos=$3
+retain=$4
+((qos == 1 || qos == 2)) && pkid=$5
 action=$(jq --raw-output .action)
 [[ $action == "toggle" ]] || exit 0
 mosquitto_pub --topic "zigbee2mqtt/light/set" --message '{"state":"TOGGLE"}'
@@ -75,7 +80,6 @@ it being ran every time a new MQTT message is published to this topic.
 
 * Configurable timeouts (eventually configurable per process)
 * Configurable QoS for each subscription (default is 0 (at most once))
-* Pass message metadata via the environment (QoS, retain, dup)
 * Configurable logging
 * Ability to configure programs with non-UTF-8 in paths
 * Maybe config reloading on SIGHUP
diff --git a/src/main.rs b/src/main.rs
index cf451dd..bc59b0b 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -14,12 +14,17 @@ mod config;
 const PROGRAM: &str = "mqttr";
 
 async fn run(program: &[String], message: &Publish) -> anyhow::Result<()> {
-    // TODO: Set environment variables
-    let mut proc = Command::new(&program[0])
+    let mut command = Command::new(&program[0]);
+    command
         .args(&program[1..])
         .arg(&message.topic)
-        .stdin(Stdio::piped())
-        .spawn()?;
+        .arg(format!("{}", message.dup as u8))
+        .arg(format!("{}", message.qos as u8))
+        .arg(format!("{}", message.retain as u8));
+    if message.qos == QoS::AtLeastOnce || message.qos == QoS::ExactlyOnce {
+        command.arg(format!("{}", message.pkid));
+    }
+    let mut proc = command.stdin(Stdio::piped()).spawn()?;
     let mut stdin = proc.stdin.take().context("No stdin")?;
     stdin.write(&message.payload).await?;
     drop(stdin);
@@ -88,7 +93,9 @@ async fn main() -> anyhow::Result<()> {
                         let p = p.clone();
                         tokio::spawn(async move {
                             match timeout(Duration::from_secs(60), run(&program, &p)).await {
-                                Err(_) => eprintln!("error: Execution of {program:?} for message {p:?} timed out"),
+                                Err(_) => eprintln!(
+                                    "error: Execution of {program:?} for message {p:?} timed out"
+                                ),
                                 Ok(Err(e)) => eprintln!("error: Failed to run {program:?}: {e:?}"),
                                 _ => (),
                             }