fix(mqtt): decouple eventloop from status watcher and publish playlist to state and subtopic

jellyfin-kiosk-rs/src/mqtt.rs

@@ -109,63 +109,79 @@ async fn run_mqtt_loop(config: AppConfig, player: Arc<PlayerController>) {
         device_name
     );
 
-    let mut status_rx = player.status_watch_rx();
+    let is_connected = Arc::new(std::sync::atomic::AtomicBool::new(false));
-    let mut last_published: Option<MqttPlaybackState> = None;
 
     loop {
         let (client, mut eventloop) = AsyncClient::new(mqtt_opts.clone(), 100);
-        let mut is_connected = false;
+        is_connected.store(false, std::sync::atomic::Ordering::SeqCst);
 
-        loop {
+        let (cancel_tx, mut cancel_rx) = tokio::sync::oneshot::channel::<()>();
-            tokio::select! {
+        let client_clone = client.clone();
-                event_res = eventloop.poll() => {
+        let mqtt_cfg_clone = mqtt_cfg.clone();
-                    match event_res {
+        let device_id_string = device_id.to_string();
-                        Ok(rumqttc::Event::Incoming(Packet::ConnAck(_))) => {
+        let is_connected_clone = is_connected.clone();
-                            is_connected = true;
+        let mut status_rx_clone = player.status_watch_rx();
-                            tracing::info!("MQTT broker connected ({}:{})", mqtt_cfg.host, mqtt_cfg.port);
 
-                            // 1. Announce online availability
+        let watcher_handle = tokio::spawn(async move {
-                            let _ = client.publish(&avail_topic, QoS::AtLeastOnce, true, "online").await;
+            let mut last_published: Option<MqttPlaybackState> = None;
-
+            loop {
-                            // 2. Subscribe to control commands
+                tokio::select! {
-                            let _ = client.subscribe(&command_topic, QoS::AtLeastOnce).await;
+                    _ = &mut cancel_rx => break,
-
+                    res = status_rx_clone.changed() => {
-                            // 3. Publish Home Assistant Discovery configs
+                        if res.is_err() {
-                            publish_homeassistant_discovery(&client, mqtt_cfg, device_id, device_name).await;
+                            break;
-
-                            // 4. Publish current playback state
-                            let current_st = {
-                                let guard = player.status.lock().await;
-                                MqttPlaybackState::from_status(&guard)
-                            };
-                            publish_playback_state(&client, mqtt_cfg, device_id, &current_st).await;
-                            last_published = Some(current_st);
                         }
-                        Ok(rumqttc::Event::Incoming(Packet::Publish(p))) => {
+                        if is_connected_clone.load(std::sync::atomic::Ordering::SeqCst) {
-                            if p.topic == command_topic {
+                            let st = status_rx_clone.borrow().clone();
-                                if let Ok(cmd_str) = std::str::from_utf8(&p.payload) {
+                            let new_state = MqttPlaybackState::from_status(&st);
-                                    handle_remote_command(&player, cmd_str.trim()).await;
+                            if last_published.as_ref() != Some(&new_state) {
-                                }
+                                publish_playback_state(&client_clone, &mqtt_cfg_clone, &device_id_string, &new_state).await;
+                                last_published = Some(new_state);
                             }
                         }
-                        Ok(_) => {}
-                        Err(e) => {
-                            tracing::warn!("MQTT connection lost or error: {}. Reconnecting...", e);
-                            tokio::time::sleep(Duration::from_secs(3)).await;
-                            break; // break inner loop to re-create client/eventloop
-                        }
                     }
                 }
-                changed_res = status_rx.changed() => {
+            }
-                    if changed_res.is_ok() && is_connected {
+        });
-                        let st = status_rx.borrow().clone();
+
-                        let new_state = MqttPlaybackState::from_status(&st);
+        loop {
-                        if last_published.as_ref() != Some(&new_state) {
+            match eventloop.poll().await {
-                            publish_playback_state(&client, mqtt_cfg, device_id, &new_state).await;
+                Ok(rumqttc::Event::Incoming(Packet::ConnAck(_))) => {
-                            last_published = Some(new_state);
+                    is_connected.store(true, std::sync::atomic::Ordering::SeqCst);
+                    tracing::info!("MQTT broker connected ({}:{})", mqtt_cfg.host, mqtt_cfg.port);
+
+                    // 1. Announce online availability
+                    let _ = client.publish(&avail_topic, QoS::AtLeastOnce, true, "online").await;
+
+                    // 2. Subscribe to control commands
+                    let _ = client.subscribe(&command_topic, QoS::AtLeastOnce).await;
+
+                    // 3. Publish Home Assistant Discovery configs
+                    publish_homeassistant_discovery(&client, mqtt_cfg, device_id, device_name).await;
+
+                    // 4. Publish current playback state
+                    let current_st = {
+                        let guard = player.status.lock().await;
+                        MqttPlaybackState::from_status(&guard)
+                    };
+                    publish_playback_state(&client, mqtt_cfg, device_id, &current_st).await;
+                }
+                Ok(rumqttc::Event::Incoming(Packet::Publish(p))) => {
+                    if p.topic == command_topic {
+                        if let Ok(cmd_str) = std::str::from_utf8(&p.payload) {
+                            handle_remote_command(&player, cmd_str.trim()).await;
                         }
                     }
                 }
+                Ok(_) => {}
+                Err(e) => {
+                    is_connected.store(false, std::sync::atomic::Ordering::SeqCst);
+                    tracing::warn!("MQTT connection lost or error: {}. Reconnecting...", e);
+                    let _ = cancel_tx.send(());
+                    watcher_handle.abort();
+                    tokio::time::sleep(Duration::from_secs(3)).await;
+                    break; // break inner loop to re-create client/eventloop
+                }
             }
         }
     }
@@ -334,6 +350,7 @@ pub async fn publish_playback_state(
         "media_type": state.media_type,
         "title": state.title,
         "artist": state.artist,
+        "playlist": state.playlist,
         "genre": state.genre,
         "stream_title": state.stream_title,
     });
@@ -345,15 +362,17 @@ pub async fn publish_playback_state(
     let _ = client.publish(format!("{}/{}/media_type", mqtt_cfg.topic_prefix, device_id), QoS::AtLeastOnce, true, state.media_type.as_str()).await;
     let _ = client.publish(format!("{}/{}/title", mqtt_cfg.topic_prefix, device_id), QoS::AtLeastOnce, true, state.title.as_str()).await;
     let _ = client.publish(format!("{}/{}/artist", mqtt_cfg.topic_prefix, device_id), QoS::AtLeastOnce, true, state.artist.as_str()).await;
+    let _ = client.publish(format!("{}/{}/playlist", mqtt_cfg.topic_prefix, device_id), QoS::AtLeastOnce, true, state.playlist.as_str()).await;
     let _ = client.publish(format!("{}/{}/genre", mqtt_cfg.topic_prefix, device_id), QoS::AtLeastOnce, true, state.genre.as_str()).await;
     let _ = client.publish(format!("{}/{}/stream_title", mqtt_cfg.topic_prefix, device_id), QoS::AtLeastOnce, true, state.stream_title.as_str()).await;
     let _ = client.publish(format!("{}/{}/playback_state", mqtt_cfg.topic_prefix, device_id), QoS::AtLeastOnce, true, state.state.as_str()).await;
 
     tracing::info!(
-        "MQTT state published: media_type='{}', title='{}', artist='{}', genre='{}', stream_title='{}', state='{}'",
+        "MQTT state published: media_type='{}', title='{}', artist='{}', playlist='{}', genre='{}', stream_title='{}', state='{}'",
         state.media_type,
         state.title,
         state.artist,
+        state.playlist,
         state.genre,
         state.stream_title,
         state.state