diff --git a/src/runtime/nodes/mqtt_shared.rs b/src/runtime/nodes/mqtt_shared.rs index a5b0e3e..ca7d5e0 100644 --- a/src/runtime/nodes/mqtt_shared.rs +++ b/src/runtime/nodes/mqtt_shared.rs @@ -164,16 +164,26 @@ async fn drive( loop { match poller.poll().await { Ok(PollEvent::Connected) => { + // Re-subscribe and publish birth, but NOT inline: each client + // request enqueues into a bounded channel that only this driver + // drains by polling. Awaiting many requests here (e.g. dozens of + // mqtt-in filters) fills the queue while the loop is not polling, + // which deadlocks the connection. Hand the work to a detached + // task so the event loop keeps being polled and the queue drains. let filters: Vec<(String, u8)> = subs .lock() .unwrap() .iter() .map(|s| (s.filter.clone(), s.qos)) .collect(); - for (filter, qos) in filters { - let _ = client.subscribe(&filter, qos).await; - } - birth.publish(client.as_ref()).await; + let client = client.clone(); + let birth = birth.clone(); + tokio::spawn(async move { + for (filter, qos) in filters { + let _ = client.subscribe(&filter, qos).await; + } + birth.publish(client.as_ref()).await; + }); } Ok(PollEvent::Message(m)) => { let subs = subs.lock().unwrap();