diff --git a/pmocontrol/src/music_renderer/capabilities.rs b/pmocontrol/src/music_renderer/capabilities.rs index b7e1559e..dcc376a5 100644 --- a/pmocontrol/src/music_renderer/capabilities.rs +++ b/pmocontrol/src/music_renderer/capabilities.rs @@ -46,7 +46,7 @@ pub trait QueueTransportControl: HasQueue + HasContinuousStream { drop(queue); - let is_stream = crate::music_renderer::is_continuous_stream_url(&item.uri); + let is_stream = crate::music_renderer::is_continuous_stream(item.metadata.as_ref(), &item.uri); *self.continuous_stream().lock().unwrap() = is_stream; self.play_item(&item) diff --git a/pmocontrol/src/music_renderer/mod.rs b/pmocontrol/src/music_renderer/mod.rs index 228c2a8b..9d95d684 100644 --- a/pmocontrol/src/music_renderer/mod.rs +++ b/pmocontrol/src/music_renderer/mod.rs @@ -23,7 +23,7 @@ pub use crate::music_renderer::capabilities::{ }; pub use crate::music_renderer::musicrenderer::{MusicRenderer, PlaylistBinding}; pub use crate::music_renderer::sleep_timer::SleepTimer; -pub use crate::music_renderer::stream_detection::is_continuous_stream_url; +pub use crate::music_renderer::stream_detection::{is_continuous_stream, is_continuous_stream_url}; use crate::{ errors::ControlPointError, music_renderer::musicrenderer::MusicRendererBackend, RendererInfo, }; diff --git a/pmocontrol/src/music_renderer/musicrenderer.rs b/pmocontrol/src/music_renderer/musicrenderer.rs index 50656c77..ae456117 100644 --- a/pmocontrol/src/music_renderer/musicrenderer.rs +++ b/pmocontrol/src/music_renderer/musicrenderer.rs @@ -492,39 +492,19 @@ impl MusicRenderer { if let Some(ref new_duration) = position.track_duration { let mut state = self.state.lock().unwrap(); - // Parse durations to compare (HH:MM:SS format) - let parse_duration = |dur_str: &str| -> Option { - let parts: Vec<&str> = dur_str.split(':').collect(); - if parts.len() == 3 { - let h: u32 = parts[0].parse().ok()?; - let m: u32 = parts[1].parse().ok()?; - let s: u32 = parts[2].parse().ok()?; - Some(h * 3600 + m * 60 + s) - } else { - None - } - }; - match &state.current_track_duration { Some(stored_duration) => { - // Compare new duration with stored one - if let (Some(stored_secs), Some(new_secs)) = ( - parse_duration(stored_duration), - parse_duration(new_duration), - ) { - if new_secs > stored_secs { - // Duration increased: update stored value and use new one - tracing::debug!( - "MusicRenderer [{}]: Stream duration increased: {} -> {}", - self.info.friendly_name(), - stored_duration, - new_duration - ); - state.current_track_duration = Some(new_duration.clone()); - } else { - // Duration decreased or equal: keep stored value - position.track_duration = Some(stored_duration.clone()); - } + if crate::queue::stream_duration_increased(stored_duration, new_duration) { + tracing::debug!( + "MusicRenderer [{}]: Stream duration increased: {} -> {}", + self.info.friendly_name(), + stored_duration, + new_duration + ); + state.current_track_duration = Some(new_duration.clone()); + } else if crate::queue::stream_duration_decreased(stored_duration, new_duration) { + // Duration decreased: keep stored value + position.track_duration = Some(stored_duration.clone()); } } None => { diff --git a/pmocontrol/src/music_renderer/stream_detection.rs b/pmocontrol/src/music_renderer/stream_detection.rs index 9cda7ff1..f4c4176c 100644 --- a/pmocontrol/src/music_renderer/stream_detection.rs +++ b/pmocontrol/src/music_renderer/stream_detection.rs @@ -195,6 +195,17 @@ fn check_stream_headers(url: &str) -> Result { Ok(is_stream) } +/// Canonical check: returns `true` if this item should be treated as a continuous stream. +/// +/// Checks `metadata.is_continuous_stream` first (already computed at ingest time), +/// then falls back to the URL-based HTTP detection. +/// +/// Use this function everywhere transport-layer code needs to decide whether playback is +/// a continuous stream (radio) vs bounded media (file/album track). +pub fn is_continuous_stream(metadata: Option<&crate::model::TrackMetadata>, uri: &str) -> bool { + metadata.map(|m| m.is_continuous_stream).unwrap_or(false) || is_continuous_stream_url(uri) +} + #[cfg(test)] mod tests { use super::*; diff --git a/pmocontrol/src/queue/mod.rs b/pmocontrol/src/queue/mod.rs index 2dfd125c..a1843783 100644 --- a/pmocontrol/src/queue/mod.rs +++ b/pmocontrol/src/queue/mod.rs @@ -19,7 +19,7 @@ use crate::{errors::ControlPointError, RendererInfo}; /// Returns true if `new_dur` < `old_dur` (both parseable as HH:MM:SS/MM:SS/SS). /// Used to protect stream durations from decreasing for the same track. -pub(super) fn stream_duration_decreased(old_dur: &str, new_dur: &str) -> bool { +pub(crate) fn stream_duration_decreased(old_dur: &str, new_dur: &str) -> bool { match ( parse_time_flexible(old_dur).ok(), parse_time_flexible(new_dur).ok(), @@ -30,7 +30,7 @@ pub(super) fn stream_duration_decreased(old_dur: &str, new_dur: &str) -> bool { } /// Returns true if `new_dur` > `old_dur` (both parseable as HH:MM:SS/MM:SS/SS). -pub(super) fn stream_duration_increased(old_dur: &str, new_dur: &str) -> bool { +pub(crate) fn stream_duration_increased(old_dur: &str, new_dur: &str) -> bool { match ( parse_time_flexible(old_dur).ok(), parse_time_flexible(new_dur).ok(), diff --git a/pmocontrol/src/queue/music_queue.rs b/pmocontrol/src/queue/music_queue.rs index 4b9318ed..642f8c7b 100644 --- a/pmocontrol/src/queue/music_queue.rs +++ b/pmocontrol/src/queue/music_queue.rs @@ -129,6 +129,14 @@ impl MusicQueue { thread::Builder::new() .name(thread_name) .spawn(move || { + // Protocol for the three AtomicBools: + // sync_in_progress : set to true before spawn, cleared on Drop via Guard. + // sync_pending : set to true by a concurrent caller that arrives while + // a sync is already running. The worker re-fetches items + // and loops when it detects this flag on exit. + // sync_cancel_token: set to true when a new sync request interrupts an + // in-progress one. Passed into QueueBackend::sync_queue + // so it can abort early. struct Guard(Arc); impl Drop for Guard { fn drop(&mut self) { @@ -136,113 +144,133 @@ impl MusicQueue { } } let _guard = Guard(Arc::clone(&sync_in_progress)); - - let mut current_items = items; - let mut current_on_ready = Some(on_ready); - let mut on_complete = Some(on_complete); - tracing::debug!(thread = %std::thread::current().name().unwrap_or("?"), "queue-sync thread started"); - - loop { - sync_pending.store(false, SeqCst); - sync_cancel_token.store(false, SeqCst); - - // Extract the real on_ready BEFORE locking the queue. - // on_ready may call play_from_queue() which re-locks the queue, - // so we must NOT call it while holding queue_arc. - let real_on_ready = current_on_ready.take().flatten(); - let on_ready_triggered = Arc::new(AtomicBool::new(false)); - let proxy_on_ready: Option> = - real_on_ready.as_ref().map(|_| { - let flag = Arc::clone(&on_ready_triggered); - Box::new(move || { - flag.store(true, SeqCst); - }) as Box - }); - - tracing::debug!( - thread = %std::thread::current().name().unwrap_or("?"), - items = current_items.len(), - has_on_ready = real_on_ready.is_some(), - "queue-sync: calling sync_queue" - ); - - let result = { - let mut q = queue_arc.lock().unwrap(); - ::sync_queue( - &mut q, - current_items, - &sync_cancel_token, - proxy_on_ready, - ) - }; - // Queue lock is released here. - // Now safe to call on_ready (which may re-lock the queue). - // If on_ready was triggered by the proxy, consume and call it. - // If not (cancelled before first insert), keep it to pass to retry. - let carry_on_ready = if on_ready_triggered.load(SeqCst) { - tracing::debug!( - thread = %std::thread::current().name().unwrap_or("?"), - "queue-sync: on_ready triggered, calling callback" - ); - if let Some(f) = real_on_ready { - f(); - } - None - } else { - if real_on_ready.is_some() { - tracing::debug!( - thread = %std::thread::current().name().unwrap_or("?"), - "queue-sync: on_ready not triggered, carrying to next attempt" - ); - } - real_on_ready - }; - - match result { - Err(ControlPointError::SyncCancelled) => { - tracing::debug!( - thread = %std::thread::current().name().unwrap_or("?"), - "queue-sync: cancelled" - ); - } - Err(e) => { - tracing::warn!("queue-sync error: {}", e); - } - Ok(()) => { - tracing::debug!( - thread = %std::thread::current().name().unwrap_or("?"), - "queue-sync: completed successfully" - ); - if let Some(cb) = on_complete.take() { - let queue_len = queue_arc.lock().unwrap().len().unwrap_or(0); - cb(queue_len); - } - } - } - - if !sync_pending.load(SeqCst) { - break; - } - - match pending_items_fn() { - Ok(new_items) => { - current_items = new_items; - current_on_ready = Some(carry_on_ready); - } - Err(e) => { - tracing::warn!("queue-sync pending re-fetch error: {}", e); - break; - } - } - } - + Self::sync_worker_loop( + queue_arc, + items, + pending_items_fn, + on_ready, + on_complete, + sync_pending, + sync_cancel_token, + ); tracing::debug!(thread = %std::thread::current().name().unwrap_or("?"), "queue-sync thread done"); }) .expect("Failed to spawn queue-sync thread"); SyncScheduleOutcome::Scheduled } + + /// Inner loop executed by the sync worker thread. + /// + /// Runs at least once with `initial_items`. If a new sync request arrives while the + /// loop is running (`sync_pending` becomes true), it re-fetches items via + /// `pending_items_fn` and iterates again, allowing the latest playlist state to win. + fn sync_worker_loop( + queue_arc: Arc>, + initial_items: Vec, + pending_items_fn: Box Result, ControlPointError> + Send>, + initial_on_ready: Option>, + on_complete: Box, + sync_pending: Arc, + sync_cancel_token: Arc, + ) { + let mut current_items = initial_items; + let mut current_on_ready = Some(initial_on_ready); + let mut on_complete = Some(on_complete); + + loop { + sync_pending.store(false, SeqCst); + sync_cancel_token.store(false, SeqCst); + + // Extract the real on_ready BEFORE locking the queue. + // on_ready may call play_from_queue() which re-locks the queue, + // so we must NOT call it while holding queue_arc. + let real_on_ready = current_on_ready.take().flatten(); + let on_ready_triggered = Arc::new(AtomicBool::new(false)); + let proxy_on_ready: Option> = + real_on_ready.as_ref().map(|_| { + let flag = Arc::clone(&on_ready_triggered); + Box::new(move || { + flag.store(true, SeqCst); + }) as Box + }); + + tracing::debug!( + thread = %std::thread::current().name().unwrap_or("?"), + items = current_items.len(), + has_on_ready = real_on_ready.is_some(), + "queue-sync: calling sync_queue" + ); + + let result = { + let mut q = queue_arc.lock().unwrap(); + ::sync_queue( + &mut q, + current_items, + &sync_cancel_token, + proxy_on_ready, + ) + }; + // Queue lock is released here. + // Now safe to call on_ready (which may re-lock the queue). + let carry_on_ready = if on_ready_triggered.load(SeqCst) { + tracing::debug!( + thread = %std::thread::current().name().unwrap_or("?"), + "queue-sync: on_ready triggered, calling callback" + ); + if let Some(f) = real_on_ready { + f(); + } + None + } else { + if real_on_ready.is_some() { + tracing::debug!( + thread = %std::thread::current().name().unwrap_or("?"), + "queue-sync: on_ready not triggered, carrying to next attempt" + ); + } + real_on_ready + }; + + match result { + Err(ControlPointError::SyncCancelled) => { + tracing::debug!( + thread = %std::thread::current().name().unwrap_or("?"), + "queue-sync: cancelled" + ); + } + Err(e) => { + tracing::warn!("queue-sync error: {}", e); + } + Ok(()) => { + tracing::debug!( + thread = %std::thread::current().name().unwrap_or("?"), + "queue-sync: completed successfully" + ); + if let Some(cb) = on_complete.take() { + let queue_len = queue_arc.lock().unwrap().len().unwrap_or(0); + cb(queue_len); + } + } + } + + if !sync_pending.load(SeqCst) { + break; + } + + match pending_items_fn() { + Ok(new_items) => { + current_items = new_items; + current_on_ready = Some(carry_on_ready); + } + Err(e) => { + tracing::warn!("queue-sync pending re-fetch error: {}", e); + break; + } + } + } + } } impl QueueBackend for MusicQueue {