diff --git a/jellyfin-kiosk-rs/src/mqtt.rs b/jellyfin-kiosk-rs/src/mqtt.rs index 2e96486..79490f9 100644 --- a/jellyfin-kiosk-rs/src/mqtt.rs +++ b/jellyfin-kiosk-rs/src/mqtt.rs @@ -109,63 +109,79 @@ async fn run_mqtt_loop(config: AppConfig, player: Arc) { device_name ); - let mut status_rx = player.status_watch_rx(); - let mut last_published: Option = None; + let is_connected = Arc::new(std::sync::atomic::AtomicBool::new(false)); 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 { - tokio::select! { - event_res = eventloop.poll() => { - match event_res { - Ok(rumqttc::Event::Incoming(Packet::ConnAck(_))) => { - is_connected = true; - tracing::info!("MQTT broker connected ({}:{})", mqtt_cfg.host, mqtt_cfg.port); + let (cancel_tx, mut cancel_rx) = tokio::sync::oneshot::channel::<()>(); + let client_clone = client.clone(); + let mqtt_cfg_clone = mqtt_cfg.clone(); + let device_id_string = device_id.to_string(); + let is_connected_clone = is_connected.clone(); + let mut status_rx_clone = player.status_watch_rx(); - // 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, ¤t_st).await; - last_published = Some(current_st); + let watcher_handle = tokio::spawn(async move { + let mut last_published: Option = None; + loop { + tokio::select! { + _ = &mut cancel_rx => break, + res = status_rx_clone.changed() => { + if res.is_err() { + break; } - 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; - } + if is_connected_clone.load(std::sync::atomic::Ordering::SeqCst) { + let st = status_rx_clone.borrow().clone(); + let new_state = MqttPlaybackState::from_status(&st); + 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); - if last_published.as_ref() != Some(&new_state) { - publish_playback_state(&client, mqtt_cfg, device_id, &new_state).await; - last_published = Some(new_state); + } + }); + + loop { + match eventloop.poll().await { + Ok(rumqttc::Event::Incoming(Packet::ConnAck(_))) => { + 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, ¤t_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