Wait for MQTT connection

kivancsikert · Aug 9, 2024 · 3389ca4 · 3389ca4
1 parent cc72238
commit 3389ca4
Showing 1 changed file with 14 additions and 2 deletions.
diff --git a/src/kernel/mod.rs b/src/kernel/mod.rs
@@ -83,10 +83,12 @@ impl Device {
         #[allow(clippy::arc_with_non_send_sync)]
         let conn = Arc::new(Mutex::new(conn));
 
+        let connected = Arc::new(Signal::<CriticalSectionRawMutex, ()>::new());
         Spawner::for_current_executor()
             .await
-            .spawn(handle_mqtt_events(conn.clone()))
+            .spawn(handle_mqtt_events(conn.clone(), connected.clone()))
             .expect("Couldn't spawn MQTT handler");
+        connected.wait().await;
 
         Ok(Self {
             config,
@@ -182,7 +184,10 @@ async fn init_mqtt(
 }
 
 #[embassy_executor::task]
-async fn handle_mqtt_events(conn: Arc<Mutex<CriticalSectionRawMutex, EspAsyncMqttConnection>>) {
+async fn handle_mqtt_events(
+    conn: Arc<Mutex<CriticalSectionRawMutex, EspAsyncMqttConnection>>,
+    connected: Arc<Signal<CriticalSectionRawMutex, ()>>,
+) {
     let mut conn = conn.lock().await;
     loop {
         let event = conn.next().await.expect("Cannot receive message");
@@ -201,6 +206,13 @@ async fn handle_mqtt_events(conn: Arc<Mutex<CriticalSectionRawMutex, EspAsyncMqt
                     details
                 );
             }
+            EventPayload::Connected(session_present) => {
+                log::info!(
+                    "Connected to MQTT broker (session present: {})",
+                    session_present
+                );
+                connected.signal(());
+            }
             EventPayload::Disconnected => {
                 log::info!("Disconnected from MQTT broker");
                 // TODO Reconnect