1 files changed, 6 insertions, 5 deletions
diff --git a/src/main.rs b/src/main.rs
index 14ccb7a..07288e5 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -64,19 +64,21 @@ fn topic_match(filter: &str, topic: &str) -> bool {
     }
 }
 
-fn main() -> anyhow::Result<()> {
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> anyhow::Result<()> {
     let mut conf_path: PathBuf = option_env!("SYSCONFDIR").unwrap_or("/usr/local/etc").into();
     conf_path.push(format!("{PROGRAM}.toml"));
     let conf = config::load(&conf_path)
         .with_context(|| format!("Failed to load config: {:?}", &conf_path))?;
-    let (client, mut connection) = conf.mqtt_client();
+    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::AtMostOnce) {
+        if let Err(e) = client.subscribe(topic, QoS::AtMostOnce).await {
             eprintln!("warning: Failed to subscribe to '{topic}': {e:?}");
         }
     }
-    for notification in connection.iter() {
+    loop {
+        let notification = event_loop.poll().await;
         match notification? {
             Incoming(Packet::Publish(p)) => {
                 for (topic, programs) in conf.routes.iter() {
@@ -92,7 +94,6 @@ fn main() -> anyhow::Result<()> {
             _ => (),
         }
     }
-    Ok(())
 }
 
 #[cfg(test)]