Configurable QoS

1 files changed, 4 insertions, 5 deletions

diff --git a/src/main.rs b/src/main.rs
index 0ba13f4..84ee406 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -73,9 +73,8 @@ async fn main() -> anyhow::Result<()> {
     let conf = config::load(&conf_path)
         .with_context(|| format!("Failed to load config: {:?}", &conf_path))?;
     let (client, mut event_loop) = conf.mqtt_client();
-    for topic in conf.routes.keys() {
-        // TODO: Configurable subscription QoS
-        if let Err(e) = client.subscribe(topic, QoS::ExactlyOnce).await {
+    for (topic, route) in conf.routes.iter() {
+        if let Err(e) = client.subscribe(topic, route.qos.unwrap_or(conf.qos)).await {
             eprintln!("warning: Failed to subscribe to '{topic}': {e:?}");
         }
     }
@@ -83,11 +82,11 @@ async fn main() -> anyhow::Result<()> {
         let notification = event_loop.poll().await;
         match notification? {
             Incoming(Packet::Publish(p)) => {
-                for (topic, programs) in conf.routes.iter() {
+                for (topic, route) in conf.routes.iter() {
                     if !topic_match(&topic, &p.topic) {
                         continue;
                     }
-                    for program in programs {
+                    for program in &route.programs {
                         // TODO: Switch to moro_local to avoid this ewwyness
                         let program = program.clone();
                         let p = p.clone();