Process timeout after 60 seconds

3 files changed, 8 insertions, 6 deletions

diff --git a/Cargo.toml b/Cargo.toml
index 001d835..654a513 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -10,5 +10,5 @@ edition = "2021"
 anyhow = "1.0.98"
 rumqttc = "0.24.0"
 serde = { version = "1.0.219", features = ["derive"] }
-tokio = { version = "1.45.1", features = ["rt", "macros", "process"] }
+tokio = { version = "1.45.1", features = ["rt", "macros", "process", "time"] }
 toml = { version = "0.8.22", default-features = false, features = ["parse"] }
diff --git a/README.md b/README.md
index a74bc7c..610ed47 100644
--- a/README.md
+++ b/README.md
@@ -73,7 +73,7 @@ it being ran every time a new MQTT message is published to this topic.
 
 ## Missing Features
 
-* Timeouts (existent, configurable, per process configurable)
+* Configurable timeouts (eventually configurable per process)
 * Permission checks on `mqttr.toml` if it contains a password (to ensure the
   password isn't being exposed)
 * Configurable QoS for each subscription (default is 0 (at most once))
diff --git a/src/main.rs b/src/main.rs
index bc9a7b5..cf451dd 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -3,11 +3,11 @@
 
 // TODO: Log levels
 
-use std::{path::PathBuf, process::Stdio};
+use std::{path::PathBuf, process::Stdio, time::Duration};
 
 use anyhow::Context;
 use rumqttc::{Event::Incoming, Packet, Publish, QoS};
-use tokio::{io::AsyncWriteExt, process::Command};
+use tokio::{io::AsyncWriteExt, process::Command, time::timeout};
 
 mod config;
 
@@ -87,8 +87,10 @@ async fn main() -> anyhow::Result<()> {
                         let program = program.clone();
                         let p = p.clone();
                         tokio::spawn(async move {
-                            if let Err(e) = run(&program, &p).await {
-                                eprintln!("error: Failed to run {program:?}: {e:?}");
+                            match timeout(Duration::from_secs(60), run(&program, &p)).await {
+                                Err(_) => eprintln!("error: Execution of {program:?} for message {p:?} timed out"),
+                                Ok(Err(e)) => eprintln!("error: Failed to run {program:?}: {e:?}"),
+                                _ => (),
                             }
                         });
                     }