eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Replace unwrap() with expect("mutex poisoned") across all mutex locks

Browse Source 
- Replace `.lock().unwrap()` with explicit error messages for poisoned MutexGuard
- Improve robustness against mutex poisoning in renderer, queue and event subsystems
This commit is contained in:
Eric Coissac
2026-04-12 20:27:54 +02:00
parent 82197c8103
commit 39d7c56a99
13 changed files with 219 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pmocontrol/src/events.rs
View File

@@ -19,7 +19,7 @@ impl RendererEventBus {
pub(crate) fn subscribe(&self) -> Receiver<RendererEvent> {
let (tx, rx) = unbounded::<RendererEvent>();
{
let mut subscribers = self.subscribers.lock().unwrap();
let mut subscribers = self.subscribers.lock().expect("event subscribers mutex poisoned");
subscribers.push(tx);
}
rx
@@ -27,7 +27,7 @@ impl RendererEventBus {
#[allow(dead_code)]
pub(crate) fn broadcast(&self, event: RendererEvent) {
let mut subscribers = self.subscribers.lock().unwrap();
let mut subscribers = self.subscribers.lock().expect("event subscribers mutex poisoned");
subscribers.retain(|tx| tx.send(event.clone()).is_ok());
}
}
@@ -47,14 +47,14 @@ impl MediaServerEventBus {
pub fn subscribe(&self) -> Receiver<MediaServerEvent> {
let (tx, rx) = unbounded::<MediaServerEvent>();
{
let mut subscribers = self.subscribers.lock().unwrap();
let mut subscribers = self.subscribers.lock().expect("event subscribers mutex poisoned");
subscribers.push(tx);
}
rx
}
pub(crate) fn broadcast(&self, event: MediaServerEvent) {
let mut subscribers = self.subscribers.lock().unwrap();
let mut subscribers = self.subscribers.lock().expect("event subscribers mutex poisoned");
subscribers.retain(|tx| tx.send(event.clone()).is_ok());
}
}

4
pmocontrol/src/music_renderer/arylic_tcp.rs
View File

@@ -137,7 +137,7 @@ impl RendererFromMediaRendererInfo for ArylicTcpRenderer {
impl ArylicTcpRenderer {
/// Returns true if currently playing a continuous stream (radio without duration)
pub fn is_continuous_stream(&self) -> bool {
*self.continuous_stream.lock().unwrap()
*self.continuous_stream.lock().expect("continuous_stream mutex poisoned")
}
/// Create an ArylicTcpRenderer with a shared queue (for HybridUpnpArylic)
@@ -268,7 +268,7 @@ impl PlaybackPosition for ArylicTcpRenderer {
// Récupérer les métadonnées depuis la queue (avec protection contre diminution de durée)
// Normalement current_index est toujours Some() si la queue n'est pas vide (règle métier)
let mut queue_guard = self.queue.lock().unwrap();
let mut queue_guard = self.queue.lock().expect("queue mutex poisoned");
let queue_item = queue_guard.peek_current().ok().flatten();
if let Some((current_item, _)) = queue_item {

106
pmocontrol/src/music_renderer/capabilities.rs
View File

@@ -4,12 +4,21 @@ use std::sync::{Arc, Mutex};
use crate::queue::{MusicQueue, QueueBackend};
use crate::{errors::ControlPointError, model::PlaybackState, PlaybackItem};
/// Trait for types that have access to a MusicQueue.
/// Marker trait for renderer backends that own a `MusicQueue`.
///
/// Implementing this trait automatically provides the full `QueueBackend`
/// blanket implementation (see `queue/backend.rs`). Backends only need to
/// return a reference to their `Arc<Mutex<MusicQueue>>` field.
pub trait HasQueue {
fn queue(&self) -> &Arc<Mutex<MusicQueue>>;
}
/// Trait for types that track whether they're playing a continuous stream.
/// Marker trait for renderer backends that track stream continuity.
///
/// The flag is `true` while the renderer is playing a continuous stream
/// (e.g. an internet radio station) and `false` for bounded media files.
/// It is used by the watcher to decide whether auto-advance should be
/// suppressed when playback stops.
pub trait HasContinuousStream {
fn continuous_stream(&self) -> &Arc<Mutex<bool>>;
}
@@ -26,7 +35,7 @@ pub trait QueueTransportControl: HasQueue + HasContinuousStream {
/// Play from the queue at the current index (or initialize to 0 if not set).
/// This is the default implementation that handles queue navigation.
fn play_from_queue(&self) -> Result<(), ControlPointError> {
let mut queue = self.queue().lock().unwrap();
let mut queue = self.queue().lock().expect("queue mutex poisoned");
let current_index = match queue.current_index()? {
Some(idx) => idx,
@@ -47,7 +56,7 @@ pub trait QueueTransportControl: HasQueue + HasContinuousStream {
drop(queue);
let is_stream = crate::music_renderer::is_continuous_stream(item.metadata.as_ref(), &item.uri);
*self.continuous_stream().lock().unwrap() = is_stream;
*self.continuous_stream().lock().expect("continuous_stream mutex poisoned") = is_stream;
self.play_item(&item)
}
@@ -55,7 +64,7 @@ pub trait QueueTransportControl: HasQueue + HasContinuousStream {
/// Play the next track from the queue.
fn play_next(&self) -> Result<(), ControlPointError> {
{
let mut queue = self.queue().lock().unwrap();
let mut queue = self.queue().lock().expect("queue mutex poisoned");
if !queue.advance()? {
return Err(ControlPointError::QueueError("No next track".into()));
}
@@ -66,7 +75,7 @@ pub trait QueueTransportControl: HasQueue + HasContinuousStream {
/// Play the previous track from the queue.
fn play_previous(&self) -> Result<(), ControlPointError> {
{
let mut queue = self.queue().lock().unwrap();
let mut queue = self.queue().lock().expect("queue mutex poisoned");
if !queue.rewind()? {
return Err(ControlPointError::QueueError("No previous track".into()));
}
@@ -77,7 +86,7 @@ pub trait QueueTransportControl: HasQueue + HasContinuousStream {
/// Play from a specific index in the queue.
fn play_from_index(&self, index: usize) -> Result<(), ControlPointError> {
{
let mut queue = self.queue().lock().unwrap();
let mut queue = self.queue().lock().expect("queue mutex poisoned");
queue.set_index(Some(index))?;
}
self.play_from_queue()
@@ -98,52 +107,97 @@ pub struct PlaybackPositionInfo {
pub track_metadata: Option<String>, // DIDL-Lite XML from GetPositionInfo
pub track_uri: Option<String>, // Current track URI
}
/// Provides the current playback position and track metadata.
///
/// All time fields use the format `"HH:MM:SS"` (or `None` when unavailable).
/// `track_metadata` carries a raw DIDL-Lite XML fragment returned by the device;
/// callers that only need structured metadata should use `extract_track_metadata`
/// from the watcher module instead.
pub trait PlaybackPosition {
/// Returns the current playback position information.
///
/// Returns `Err` if the renderer is unreachable or the query fails.
fn playback_position(&self) -> Result<PlaybackPositionInfo, ControlPointError>;
}
/// Generic abstraction for playback status (transport state).
/// Generic abstraction for the current transport state.
///
/// For UPnP AV, this is backed by AVTransport::GetTransportInfo.
/// For OpenHome, a future implementation will adapt from OH Info/Time.
/// # Implementations
///
/// - **UPnP AV**: backed by `AVTransport::GetTransportInfo`.
/// - **OpenHome**: adapted from OH `Info` / `Time` services.
/// - **LinkPlay / Arylic**: mapped from the vendor status response.
///
/// # Postconditions
///
/// The returned `PlaybackState` must be one of the canonical values defined by
/// the `PlaybackState` enum. Backend-specific states that have no canonical
/// equivalent should be mapped to the closest approximation (e.g. "BUFFERING"
/// → `PlaybackState::Transitioning`).
pub trait PlaybackStatus {
/// Returns the current transport state of the renderer.
fn playback_state(&self) -> Result<PlaybackState, ControlPointError>;
}
/// Abstraction générique des capacités de transport (lecture / pause / stop / seek)
/// indépendamment du protocole sous-jacent (UPnP AV, OpenHome, ...).
pub trait TransportControl {
/// Set la ressource à lire (URI + métadonnées) et/ou commence la lecture.
/// Generic transport control abstraction (play / pause / stop / seek),
/// independent of the underlying protocol (UPnP AV, OpenHome, …).
///
/// Selon l'implémentation, cette méthode peut soit :
/// - faire un "Set...URI" + "Play" (cas UPnP AV),
/// - ou configurer la file de lecture (cas OpenHome, etc.).
/// # Invariants
///
/// - `play_uri` sets the active resource and begins playback atomically from the
/// caller's perspective. Implementations may split this into two protocol steps
/// (e.g. `SetAVTransportURI` + `Play` for UPnP AV) but the caller should not
/// need to know.
/// - `play` / `pause` / `stop` operate on whatever resource is currently loaded;
/// they do not change the queue pointer.
/// - `seek_rel_time` uses the format `"HH:MM:SS"`. Backends that do not support
/// seeking should return `ControlPointError::NotSupported`.
///
/// # Relation to `QueueTransportControl`
///
/// `TransportControl` knows nothing about the queue. `QueueTransportControl`
/// extends it with queue-aware navigation (`play_next`, `play_previous`, …).
pub trait TransportControl {
/// Load a resource (URI + DIDL-Lite metadata) and begin playback.
///
/// Depending on the backend this may execute as a single atomic operation or as
/// two sequential commands (set resource, then play).
fn play_uri(&self, uri: &str, meta: &str) -> Result<(), ControlPointError>;
/// Démarre ou reprend la lecture.
/// Start or resume playback of the currently loaded resource.
fn play(&self) -> Result<(), ControlPointError>;
/// Met la lecture en pause.
/// Pause the current playback.
fn pause(&self) -> Result<(), ControlPointError>;
/// Arrête la lecture.
/// Stop the current playback and release the loaded resource.
fn stop(&self) -> Result<(), ControlPointError>;
/// Seek à un temps relatif (HH:MM:SS) si supporté.
/// Seek to a relative time position expressed as `"HH:MM:SS"`.
///
/// Returns `ControlPointError::NotSupported` when the backend does not
/// implement seeking.
fn seek_rel_time(&self, hhmmss: &str) -> Result<(), ControlPointError>;
}
/// Abstraction générique des capacités de contrôle de volume / mute.
/// Generic volume and mute control abstraction.
///
/// # Volume scale
///
/// Volume values are expressed on the native scale of each renderer.
/// UPnP AV and OpenHome renderers typically use 0–100. Callers should
/// not assume any particular scale; use the values returned by `volume()`
/// as the baseline for relative adjustments.
pub trait VolumeControl {
/// Retourne le volume logique courant (échelle dépendante du renderer).
/// Returns the current logical volume (renderer-specific scale).
fn volume(&self) -> Result<u16, ControlPointError>;
/// Définit le volume logique (échelle dépendante du renderer).
/// Sets the logical volume (renderer-specific scale).
fn set_volume(&self, v: u16) -> Result<(), ControlPointError>;
/// Indique si le renderer est muet (mute activé).
/// Returns `true` when the renderer is muted.
fn mute(&self) -> Result<bool, ControlPointError>;
/// Active ou désactive le mute.
/// Enables (`true`) or disables (`false`) mute.
fn set_mute(&self, m: bool) -> Result<(), ControlPointError>;
}

4
pmocontrol/src/music_renderer/chromecast_renderer.rs
View File

@@ -172,7 +172,7 @@ impl RendererFromMediaRendererInfo for ChromecastRenderer {
impl ChromecastRenderer {
/// Returns true if currently playing a continuous stream (radio without duration)
pub fn is_continuous_stream(&self) -> bool {
*self.continuous_stream.lock().unwrap()
*self.continuous_stream.lock().expect("continuous_stream mutex poisoned")
}
/// Connect to the device with retry on connection failures.
@@ -205,7 +205,7 @@ impl TransportControl for ChromecastRenderer {
// Détecte si l'URL est un flux continu
let is_stream = crate::music_renderer::is_continuous_stream_url(uri);
*self.continuous_stream.lock().unwrap() = is_stream;
*self.continuous_stream.lock().expect("continuous_stream mutex poisoned") = is_stream;
tracing::debug!(
"ChromecastRenderer play_uri: URI={}, continuous_stream={}",
uri,

6
pmocontrol/src/music_renderer/linkplay_renderer.rs
View File

@@ -91,7 +91,7 @@ impl RendererFromMediaRendererInfo for LinkPlayRenderer {
impl LinkPlayRenderer {
/// Returns true if currently playing a continuous stream (radio without duration)
pub fn is_continuous_stream(&self) -> bool {
*self.continuous_stream.lock().unwrap()
*self.continuous_stream.lock().expect("continuous_stream mutex poisoned")
}
}
@@ -99,7 +99,7 @@ impl TransportControl for LinkPlayRenderer {
fn play_uri(&self, uri: &str, _meta: &str) -> Result<(), ControlPointError> {
// Détecte si l'URL est un flux continu
let is_stream = crate::music_renderer::is_continuous_stream_url(uri);
*self.continuous_stream.lock().unwrap() = is_stream;
*self.continuous_stream.lock().expect("continuous_stream mutex poisoned") = is_stream;
tracing::debug!(
"LinkPlayRenderer play_uri: URI={}, continuous_stream={}",
uri,
@@ -158,7 +158,7 @@ impl PlaybackPosition for LinkPlayRenderer {
let mut position_info = self.fetch_status()?.position_info();
// Use queue metadata instead of direct status metadata to benefit from duration protection
let mut queue_guard = self.queue.lock().unwrap();
let mut queue_guard = self.queue.lock().expect("queue mutex poisoned");
let queue_item = queue_guard.peek_current().ok().flatten();
if let Some((current_item, _)) = queue_item {

96
pmocontrol/src/music_renderer/musicrenderer.rs
View File

@@ -46,9 +46,9 @@ use tracing::warn;
#[derive(Clone, Debug)]
pub struct PlaylistBinding {
/// MediaServer that owns the playlist container.
pub server_id: DeviceId,
pub(crate) server_id: DeviceId,
/// DIDL-Lite object id of the playlist container.
pub container_id: String,
pub(crate) container_id: String,
/// True once at least one ContainerUpdateIDs notification has been seen.
pub(crate) has_seen_update: bool,
/// Flag used internally to signal that the queue should be refreshed
@@ -58,6 +58,18 @@ pub struct PlaylistBinding {
pub(crate) auto_play_on_refresh: bool,
}
impl PlaylistBinding {
/// Returns the ID of the MediaServer that owns this playlist container.
pub fn server_id(&self) -> &DeviceId {
&self.server_id
}
/// Returns the DIDL-Lite object ID of the playlist container.
pub fn container_id(&self) -> &str {
&self.container_id
}
}
/// Backend-agnostic façade exposing transport, volume, and status contracts.
#[derive(Clone, Debug)]
pub enum MusicRendererBackend {
@@ -311,6 +323,14 @@ impl MusicRenderer {
}
/// Main loop for the watcher thread.
///
/// # Error policy
///
/// The watcher thread runs for the lifetime of the renderer and never restarts
/// automatically. Network/device errors during polling are logged at `debug` level
/// and ignored — they are transient and expected when a device is temporarily
/// unreachable. Panics inside `poll_and_emit_changes` are caught and logged at
/// `error` level so they do not kill the watcher thread.
fn watcher_loop(&self, strategy: WatchStrategy, stop_flag: Arc<AtomicBool>) {
let Some(base_interval) = strategy.polling_interval() else {
// Pure push strategy - no polling needed (future implementation)
@@ -341,7 +361,16 @@ impl MusicRenderer {
};
if self.is_online() {
// Wrap in catch_unwind so a panic in poll logic does not terminate the watcher.
let poll_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
self.poll_and_emit_changes(tick);
}));
if let Err(_panic) = poll_result {
error!(
renderer = self.info.friendly_name(),
"poll_and_emit_changes panicked; watcher continues"
);
}
last_activity_time = SystemTime::now();
}
@@ -393,14 +422,19 @@ impl MusicRenderer {
// Step 2: Do all network calls WITHOUT holding any locks
// This prevents blocking other threads that need to read watched_state
let position = self.playback_position().ok();
let raw_state = self.playback_state().ok();
// Errors are logged at trace level — device temporarily unreachable is expected.
let position = self.playback_position()
.inspect_err(|e| tracing::trace!(renderer = self.info.friendly_name(), error = %e, "playback_position failed"))
.ok();
let raw_state = self.playback_state()
.inspect_err(|e| tracing::trace!(renderer = self.info.friendly_name(), error = %e, "playback_state failed"))
.ok();
// Poll volume and mute every other tick (1 second at 500ms interval)
let (volume, mute, is_stream) = if tick % 2 == 0 {
(
self.volume().ok(),
self.mute().ok(),
self.volume().inspect_err(|e| tracing::trace!(renderer = self.info.friendly_name(), error = %e, "volume poll failed")).ok(),
self.mute().inspect_err(|e| tracing::trace!(renderer = self.info.friendly_name(), error = %e, "mute poll failed")).ok(),
Some(self.is_playing_a_stream()),
)
} else {
@@ -490,7 +524,7 @@ impl MusicRenderer {
if let Some(stream_flag) = is_stream {
if stream_flag {
if let Some(ref new_duration) = position.track_duration {
let mut state = self.state.lock().unwrap();
let mut state = self.state.lock().expect("RendererState mutex poisoned");
match &state.current_track_duration {
Some(stored_duration) => {
@@ -647,7 +681,7 @@ impl MusicRenderer {
match state {
PlaybackState::Stopped => {
{
let s = self.state.lock().unwrap();
let s = self.state.lock().expect("RendererState mutex poisoned");
tracing::debug!(
renderer = self.info.friendly_name(),
has_played = s.has_played_since_track_start,
@@ -680,7 +714,7 @@ impl MusicRenderer {
// On autorise l'auto-avance si:
// 1. On a bien vu PLAYING OU
// 2. Le titre a été lancé depuis plus de 20 secondes
let track_start = self.state.lock().unwrap().track_start_time;
let track_start = self.state.lock().expect("RendererState mutex poisoned").track_start_time;
let elapsed = track_start
.and_then(|t| t.elapsed().ok())
.unwrap_or_default();
@@ -737,7 +771,7 @@ impl MusicRenderer {
PlaybackState::NoMedia => {
// Handle end of track (Chromecast returns NoMedia when track ends)
// This is equivalent to Stopped for auto-advance purposes
let s = self.state.lock().unwrap();
let s = self.state.lock().expect("RendererState mutex poisoned");
let playback_source = s.playback_source;
let has_played = s.has_played_since_track_start;
let user_stop = s.user_stop_requested;
@@ -804,7 +838,7 @@ impl MusicRenderer {
self.set_has_played_flag();
}
PlaybackState::Transitioning => {
let s = self.state.lock().unwrap();
let s = self.state.lock().expect("RendererState mutex poisoned");
tracing::trace!(
renderer = self.info.friendly_name(),
has_played = s.has_played_since_track_start,
@@ -1646,13 +1680,13 @@ impl MusicRenderer {
/// Gets the last known track metadata.
pub fn last_metadata(&self) -> Option<TrackMetadata> {
self.state.lock().unwrap().last_metadata.clone()
self.state.lock().expect("RendererState mutex poisoned").last_metadata.clone()
}
/// Sets the last known track metadata.
/// Updates track_start_time and resets current_track_duration only if the metadata actually changes.
pub fn set_last_metadata(&self, metadata: Option<TrackMetadata>) {
let mut state = self.state.lock().unwrap();
let mut state = self.state.lock().expect("RendererState mutex poisoned");
let metadata_changed = state.last_metadata != metadata;
if metadata_changed {
// Pour les flux continus: utiliser dc:date comme track_start_time réel de diffusion.
@@ -1673,23 +1707,23 @@ impl MusicRenderer {
/// Gets the timestamp when the current track started playing.
pub fn track_start_time(&self) -> Option<SystemTime> {
self.state.lock().unwrap().track_start_time
self.state.lock().expect("RendererState mutex poisoned").track_start_time
}
/// Gets the current playback source.
pub fn playback_source(&self) -> PlaybackSource {
self.state.lock().unwrap().playback_source
self.state.lock().expect("RendererState mutex poisoned").playback_source
}
/// Sets the playback source.
pub fn set_playback_source(&self, source: PlaybackSource) {
self.state.lock().unwrap().playback_source = source;
self.state.lock().expect("RendererState mutex poisoned").playback_source = source;
}
/// Checks if currently playing from queue.
pub fn is_playing_from_queue(&self) -> bool {
matches!(
self.state.lock().unwrap().playback_source,
self.state.lock().expect("RendererState mutex poisoned").playback_source,
PlaybackSource::FromQueue
)
}
@@ -1700,7 +1734,7 @@ impl MusicRenderer {
/// Does NOT change None -> External because that would break queue playback
/// (the control_point will set it to FromQueue after play_from_queue succeeds).
pub fn mark_external_if_idle(&self) {
let mut state = self.state.lock().unwrap();
let mut state = self.state.lock().expect("RendererState mutex poisoned");
if matches!(state.playback_source, PlaybackSource::External) {
// Keep External if we were already playing externally
} else {
@@ -1711,12 +1745,12 @@ impl MusicRenderer {
/// Marks that the user requested a stop (to prevent auto-advance).
pub fn mark_user_stop_requested(&self) {
self.state.lock().unwrap().user_stop_requested = true;
self.state.lock().expect("RendererState mutex poisoned").user_stop_requested = true;
}
/// Checks and clears the user stop requested flag.
pub fn check_and_clear_user_stop_requested(&self) -> bool {
let mut state = self.state.lock().unwrap();
let mut state = self.state.lock().expect("RendererState mutex poisoned");
let was_requested = state.user_stop_requested;
state.user_stop_requested = false;
was_requested
@@ -1727,21 +1761,21 @@ impl MusicRenderer {
/// Sets the has_played_since_track_start flag to true.
/// Called when PLAYING state is detected.
fn set_has_played_flag(&self) {
self.state.lock().unwrap().has_played_since_track_start = true;
self.state.lock().expect("RendererState mutex poisoned").has_played_since_track_start = true;
}
/// Clears the has_played_since_track_start flag.
/// Called when stopping playback or starting a new track.
/// This is public so that ControlPoint can reset it when jumping to a new track.
pub fn clear_has_played_flag(&self) {
self.state.lock().unwrap().has_played_since_track_start = false;
self.state.lock().expect("RendererState mutex poisoned").has_played_since_track_start = false;
}
/// Checks and clears the has_played_since_track_start flag.
/// Returns true if PLAYING was seen since last track start, false otherwise.
/// Used to determine if auto-advance should be allowed.
fn check_and_clear_has_played_flag(&self) -> bool {
let mut state = self.state.lock().unwrap();
let mut state = self.state.lock().expect("RendererState mutex poisoned");
let has_played = state.has_played_since_track_start;
state.has_played_since_track_start = false;
has_played
@@ -1757,7 +1791,7 @@ impl MusicRenderer {
/// # Errors
/// Returns an error if the duration is invalid (0 or > 7200 seconds).
pub fn start_sleep_timer(&self, duration_seconds: u32) -> Result<u32, ControlPointError> {
let mut state = self.state.lock().unwrap();
let mut state = self.state.lock().expect("RendererState mutex poisoned");
state
.sleep_timer
.start(duration_seconds)
@@ -1773,7 +1807,7 @@ impl MusicRenderer {
/// # Errors
/// Returns an error if the duration is invalid (0 or > 7200 seconds).
pub fn update_sleep_timer(&self, duration_seconds: u32) -> Result<u32, ControlPointError> {
let mut state = self.state.lock().unwrap();
let mut state = self.state.lock().expect("RendererState mutex poisoned");
state
.sleep_timer
.update(duration_seconds)
@@ -1784,32 +1818,32 @@ impl MusicRenderer {
/// Cancels the sleep timer.
pub fn cancel_sleep_timer(&self) {
self.state.lock().unwrap().sleep_timer.cancel();
self.state.lock().expect("RendererState mutex poisoned").sleep_timer.cancel();
}
/// Returns the remaining seconds of the sleep timer, or None if no timer is active.
pub fn sleep_timer_remaining(&self) -> Option<u32> {
self.state.lock().unwrap().sleep_timer.remaining_seconds()
self.state.lock().expect("RendererState mutex poisoned").sleep_timer.remaining_seconds()
}
/// Returns the configured duration of the sleep timer in seconds.
pub fn sleep_timer_duration(&self) -> u32 {
self.state.lock().unwrap().sleep_timer.duration_seconds()
self.state.lock().expect("RendererState mutex poisoned").sleep_timer.duration_seconds()
}
/// Returns true if the sleep timer is active.
pub fn is_sleep_timer_active(&self) -> bool {
self.state.lock().unwrap().sleep_timer.is_active()
self.state.lock().expect("RendererState mutex poisoned").sleep_timer.is_active()
}
/// Returns true if the sleep timer has expired.
pub fn is_sleep_timer_expired(&self) -> bool {
self.state.lock().unwrap().sleep_timer.is_expired()
self.state.lock().expect("RendererState mutex poisoned").sleep_timer.is_expired()
}
/// Gets the sleep timer state as a tuple (is_active, duration_seconds, remaining_seconds).
pub fn sleep_timer_state(&self) -> (bool, u32, Option<u32>) {
let state = self.state.lock().unwrap();
let state = self.state.lock().expect("RendererState mutex poisoned");
(
state.sleep_timer.is_active(),
state.sleep_timer.duration_seconds(),

20
pmocontrol/src/music_renderer/openhome_renderer.rs
View File

@@ -89,7 +89,7 @@ impl OpenHomeRenderer {
/// Returns true if currently playing a continuous stream (radio without duration)
pub fn is_continuous_stream(&self) -> bool {
*self.continuous_stream.lock().unwrap()
*self.continuous_stream.lock().expect("continuous_stream mutex poisoned")
}
pub fn has_playlist(&self) -> bool {
@@ -176,14 +176,14 @@ impl OpenHomeRenderer {
/// Plus rapide que snapshot_openhome_playlist() pour juste connaître le nombre de pistes.
pub(crate) fn openhome_playlist_len(&self) -> Result<usize, ControlPointError> {
// Use queue.len() which uses cached track_ids() internally
let queue = self.queue.lock().unwrap();
let queue = self.queue.lock().expect("queue mutex poisoned");
queue.len()
}
/// Retourne les IDs des pistes de la playlist OpenHome.
/// Plus rapide que snapshot_openhome_playlist() car ne récupère pas les métadonnées.
pub(crate) fn openhome_playlist_ids(&self) -> Result<Vec<u32>, ControlPointError> {
let queue = self.queue.lock().unwrap();
let queue = self.queue.lock().expect("queue mutex poisoned");
if let Some(oh_queue) = queue.as_openhome() {
oh_queue.track_ids()
} else {
@@ -209,7 +209,7 @@ impl OpenHomeRenderer {
let insert_after = match after_id {
Some(id) => id,
None => {
let queue = self.queue.lock().unwrap();
let queue = self.queue.lock().expect("queue mutex poisoned");
if let Some(oh_queue) = queue.as_openhome() {
oh_queue
.track_ids()?
@@ -365,7 +365,7 @@ impl PlaybackPosition for OpenHomeRenderer {
// Check cache first
{
let mut cache = self.position_cache.lock().unwrap();
let mut cache = self.position_cache.lock().expect("position_cache mutex poisoned");
// Track calls for warning detection
cache.calls_in_last_second.push(now);
@@ -406,7 +406,7 @@ impl PlaybackPosition for OpenHomeRenderer {
let mut track_metadata_xml = None;
// Get track ID from queue (uses cached data)
let queue_guard_for_id = self.queue.lock().unwrap();
let queue_guard_for_id = self.queue.lock().expect("queue mutex poisoned");
if let Some(oh_queue) = queue_guard_for_id.as_openhome() {
match oh_queue.current_track() {
Ok(id_opt) => track_id = id_opt,
@@ -419,7 +419,7 @@ impl PlaybackPosition for OpenHomeRenderer {
drop(queue_guard_for_id);
// Use queue API to get current item with cached metadata
let mut queue_guard = self.queue.lock().unwrap();
let mut queue_guard = self.queue.lock().expect("queue mutex poisoned");
if let Ok(Some((current_item, _))) = queue_guard.peek_current() {
// Use metadata from queue cache (updated via OpenHome events)
track_uri = Some(current_item.uri.clone());
@@ -436,7 +436,7 @@ impl PlaybackPosition for OpenHomeRenderer {
}
// Check if the URI has changed to detect track changes
let mut cached_uri = self.current_track_uri.lock().unwrap();
let mut cached_uri = self.current_track_uri.lock().expect("current_track_uri mutex poisoned");
let uri_changed = cached_uri.as_ref() != Some(&current_item.uri);
if uri_changed {
@@ -448,7 +448,7 @@ impl PlaybackPosition for OpenHomeRenderer {
// Détecte si la nouvelle URL est un flux continu
let is_stream = crate::music_renderer::is_continuous_stream_url(&current_item.uri);
*self.continuous_stream.lock().unwrap() = is_stream;
*self.continuous_stream.lock().expect("continuous_stream mutex poisoned") = is_stream;
tracing::debug!("OpenHome URI changed, continuous_stream={}", is_stream);
*cached_uri = Some(current_item.uri.clone());
@@ -484,7 +484,7 @@ impl PlaybackPosition for OpenHomeRenderer {
// Update cache with fresh data
{
let mut cache = self.position_cache.lock().unwrap();
let mut cache = self.position_cache.lock().expect("position_cache mutex poisoned");
cache.last_position = Some(position_info.clone());
cache.last_update = Some(now);
}

8
pmocontrol/src/music_renderer/stream_detection.rs
View File

@@ -53,7 +53,7 @@ pub fn is_continuous_stream_url(url: &str) -> bool {
// Check cache first
{
let cache = STREAM_CACHE.lock().unwrap();
let cache = STREAM_CACHE.lock().expect("stream cache mutex poisoned");
if let Some(&cached_result) = cache.get(url) {
trace!("Cache hit for {}: is_stream={}", url, cached_result);
return cached_result;
@@ -62,7 +62,7 @@ pub fn is_continuous_stream_url(url: &str) -> bool {
// Check if already being verified
{
let mut pending = PENDING_CHECKS.lock().unwrap();
let mut pending = PENDING_CHECKS.lock().expect("pending checks mutex poisoned");
if pending.contains(url) {
debug!(
"Stream detection already in progress for {}, returning false temporarily",
@@ -93,13 +93,13 @@ pub fn is_continuous_stream_url(url: &str) -> bool {
// Store in cache
{
let mut cache = STREAM_CACHE.lock().unwrap();
let mut cache = STREAM_CACHE.lock().expect("stream cache mutex poisoned");
cache.insert(url_owned.clone(), result);
}
// Remove from pending
{
let mut pending = PENDING_CHECKS.lock().unwrap();
let mut pending = PENDING_CHECKS.lock().expect("pending checks mutex poisoned");
pending.remove(&url_owned);
}

14
pmocontrol/src/music_renderer/upnp_renderer.rs
View File

@@ -117,7 +117,7 @@ impl UpnpRenderer {
/// Returns true if currently playing a continuous stream (radio without duration)
pub fn is_continuous_stream(&self) -> bool {
*self.continuous_stream.lock().unwrap()
*self.continuous_stream.lock().expect("continuous_stream mutex poisoned")
}
}
@@ -192,10 +192,10 @@ impl QueueTransportControl for UpnpRenderer {
let duration = parse_didl_duration(&metadata);
if let Some(ref dur) = duration {
tracing::debug!("Caching duration from queue DIDL: {}", dur);
*self.cached_duration.lock().unwrap() = Some(dur.clone());
*self.cached_duration.lock().expect("cached_duration mutex poisoned") = Some(dur.clone());
} else {
tracing::debug!("No duration to cache from queue DIDL");
*self.cached_duration.lock().unwrap() = None;
*self.cached_duration.lock().expect("cached_duration mutex poisoned") = None;
}
let avt = self.avtransport()?;
@@ -233,7 +233,7 @@ impl TransportControl for UpnpRenderer {
// Détecte si l'URL est un flux continu en interrogeant le serveur HTTP
let is_stream = crate::music_renderer::is_continuous_stream_url(uri);
*self.continuous_stream.lock().unwrap() = is_stream;
*self.continuous_stream.lock().expect("continuous_stream mutex poisoned") = is_stream;
tracing::debug!(
"UpnpRenderer play_uri: URI={}, continuous_stream={}",
uri,
@@ -244,10 +244,10 @@ impl TransportControl for UpnpRenderer {
let duration = parse_didl_duration(meta);
if let Some(ref dur) = duration {
tracing::debug!("Caching duration from DIDL: {}", dur);
*self.cached_duration.lock().unwrap() = Some(dur.clone());
*self.cached_duration.lock().expect("cached_duration mutex poisoned") = Some(dur.clone());
} else {
tracing::debug!("No duration to cache from DIDL");
*self.cached_duration.lock().unwrap() = None;
*self.cached_duration.lock().expect("cached_duration mutex poisoned") = None;
}
let avt = self.avtransport()?;
@@ -341,7 +341,7 @@ impl PlaybackPosition for UpnpRenderer {
let mut track_metadata_xml = None;
let mut track_uri = raw.track_uri.clone();
let mut queue_guard = self.queue.lock().unwrap();
let mut queue_guard = self.queue.lock().expect("queue mutex poisoned");
// Récupérer l'item courant de la queue
// Normalement current_index est toujours Some() si la queue n'est pas vide (règle métier)

22
pmocontrol/src/queue/backend.rs
View File

@@ -37,35 +37,35 @@ use std::sync::{atomic::AtomicBool, Arc, Mutex};
/// All methods simply delegate to the underlying MusicQueue.
impl<T: HasQueue> QueueBackend for T {
fn len(&self) -> Result<usize, ControlPointError> {
self.queue().lock().unwrap().len()
self.queue().lock().expect("queue mutex poisoned").len()
}
fn track_ids(&self) -> Result<Vec<u32>, ControlPointError> {
self.queue().lock().unwrap().track_ids()
self.queue().lock().expect("queue mutex poisoned").track_ids()
}
fn id_to_position(&self, id: u32) -> Result<usize, ControlPointError> {
self.queue().lock().unwrap().id_to_position(id)
self.queue().lock().expect("queue mutex poisoned").id_to_position(id)
}
fn position_to_id(&self, id: usize) -> Result<u32, ControlPointError> {
self.queue().lock().unwrap().position_to_id(id)
self.queue().lock().expect("queue mutex poisoned").position_to_id(id)
}
fn current_track(&self) -> Result<Option<u32>, ControlPointError> {
self.queue().lock().unwrap().current_track()
self.queue().lock().expect("queue mutex poisoned").current_track()
}
fn current_index(&self) -> Result<Option<usize>, ControlPointError> {
self.queue().lock().unwrap().current_index()
self.queue().lock().expect("queue mutex poisoned").current_index()
}
fn queue_snapshot(&self) -> Result<QueueSnapshot, ControlPointError> {
self.queue().lock().unwrap().queue_snapshot()
self.queue().lock().expect("queue mutex poisoned").queue_snapshot()
}
fn set_index(&mut self, index: Option<usize>) -> Result<(), ControlPointError> {
self.queue().lock().unwrap().set_index(index)
self.queue().lock().expect("queue mutex poisoned").set_index(index)
}
fn replace_queue(
@@ -92,11 +92,11 @@ impl<T: HasQueue> QueueBackend for T {
}
fn get_item(&self, index: usize) -> Result<Option<PlaybackItem>, ControlPointError> {
self.queue().lock().unwrap().get_item(index)
self.queue().lock().expect("queue mutex poisoned").get_item(index)
}
fn replace_item(&mut self, index: usize, item: PlaybackItem) -> Result<(), ControlPointError> {
self.queue().lock().unwrap().replace_item(index, item)
self.queue().lock().expect("queue mutex poisoned").replace_item(index, item)
}
fn enqueue_items(
@@ -104,7 +104,7 @@ impl<T: HasQueue> QueueBackend for T {
items: Vec<PlaybackItem>,
mode: EnqueueMode,
) -> Result<(), ControlPointError> {
self.queue().lock().unwrap().enqueue_items(items, mode)
self.queue().lock().expect("queue mutex poisoned").enqueue_items(items, mode)
}
}

10
pmocontrol/src/queue/music_queue.rs
View File

@@ -106,7 +106,7 @@ impl MusicQueue {
on_complete: Box<dyn Fn(usize) + Send + 'static>,
) -> SyncScheduleOutcome {
let (sync_in_progress, sync_pending, sync_cancel_token) = {
let q = queue_arc.lock().unwrap();
let q = queue_arc.lock().expect("MusicQueue mutex poisoned");
(
Arc::clone(&q.sync_in_progress),
Arc::clone(&q.sync_pending),
@@ -144,6 +144,10 @@ impl MusicQueue {
}
}
let _guard = Guard(Arc::clone(&sync_in_progress));
// Error policy: sync errors are logged by sync_worker_loop (warn level) and
// the thread exits normally. No restart — a new sync can be scheduled via
// schedule_sync. The Guard Drop clears sync_in_progress unconditionally,
// even on panic, keeping the AtomicBool protocol consistent.
tracing::debug!(thread = %std::thread::current().name().unwrap_or("?"), "queue-sync thread started");
Self::sync_worker_loop(
queue_arc,
@@ -204,7 +208,7 @@ impl MusicQueue {
);
let result = {
let mut q = queue_arc.lock().unwrap();
let mut q = queue_arc.lock().expect("MusicQueue mutex poisoned");
<MusicQueue as QueueBackend>::sync_queue(
&mut q,
current_items,
@@ -249,7 +253,7 @@ impl MusicQueue {
"queue-sync: completed successfully"
);
if let Some(cb) = on_complete.take() {
let queue_len = queue_arc.lock().unwrap().len().unwrap_or(0);
let queue_len = queue_arc.lock().expect("MusicQueue mutex poisoned").len().unwrap_or(0);
cb(queue_len);
}
}

42
pmocontrol/src/queue/openhome.rs
View File

@@ -233,16 +233,16 @@ impl OpenHomeQueue {
/// Invalide les caches track_ids et read_list (après insert/delete sans impact sur la piste courante).
fn invalidate_track_caches(&self) {
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.track_ids_cache.lock().expect("track_ids_cache mutex poisoned").invalidate();
self.read_list_cache.lock().expect("read_list_cache mutex poisoned").invalidate();
}
/// Invalide tous les caches (après delete_all, seek, stop — opérations qui changent la piste courante).
fn invalidate_all_caches(&self) {
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.current_track_id_cache.lock().unwrap().invalidate();
self.uri_by_id.lock().unwrap().clear();
self.track_ids_cache.lock().expect("track_ids_cache mutex poisoned").invalidate();
self.read_list_cache.lock().expect("read_list_cache mutex poisoned").invalidate();
self.current_track_id_cache.lock().expect("current_track_id_cache mutex poisoned").invalidate();
self.uri_by_id.lock().expect("uri_by_id mutex poisoned").clear();
}
/// Tries to detect a simple append-only or delete-from-end pattern without ReadList.
@@ -390,8 +390,8 @@ impl OpenHomeQueue {
new_metadata: Option<crate::model::TrackMetadata>,
uri: &str,
) {
let mut cache = self.metadata_cache.lock().unwrap();
let mut uri_cache = self.uri_by_id.lock().unwrap();
let mut cache = self.metadata_cache.lock().expect("metadata_cache mutex poisoned");
let mut uri_cache = self.uri_by_id.lock().expect("uri_by_id mutex poisoned");
// Update URI cache
if !uri.is_empty() {
@@ -481,7 +481,7 @@ impl OpenHomeQueue {
// Le cache contient les métadonnées stables mises lors de l'insertion
// Les métadonnées de l'entry (venant de ReadList) changent pour les streams
let metadata = {
let cache = self.metadata_cache.lock().unwrap();
let cache = self.metadata_cache.lock().expect("metadata_cache mutex poisoned");
if let Some(cached_meta) = cache.get(&entry.id) {
// Utiliser les métadonnées stables du cache
tracing::trace!(
@@ -557,7 +557,7 @@ impl OpenHomeQueue {
}
if track_id as usize != playing_id {
self.playlist_client.delete_id_if_exists(track_id)?;
self.metadata_cache.lock().unwrap().remove(&track_id);
self.metadata_cache.lock().expect("metadata_cache mutex poisoned").remove(&track_id);
}
}
@@ -616,7 +616,7 @@ impl OpenHomeQueue {
track_id
);
self.playlist_client.delete_id_if_exists(track_id)?;
self.metadata_cache.lock().unwrap().remove(&track_id);
self.metadata_cache.lock().expect("metadata_cache mutex poisoned").remove(&track_id);
}
}
Ok(())
@@ -858,7 +858,7 @@ impl OpenHomeQueue {
"Using delete_all() for complete replacement (safe - no current track or not in new playlist)"
);
self.playlist_client.delete_all()?;
self.metadata_cache.lock().unwrap().clear();
self.metadata_cache.lock().expect("metadata_cache mutex poisoned").clear();
}
} else {
for idx in (0..current_track_ids.len()).rev() {
@@ -868,7 +868,7 @@ impl OpenHomeQueue {
if !keep_current[idx] {
let track_id = current_track_ids[idx];
self.playlist_client.delete_id_if_exists(track_id)?;
self.metadata_cache.lock().unwrap().remove(&track_id);
self.metadata_cache.lock().expect("metadata_cache mutex poisoned").remove(&track_id);
}
}
}
@@ -1099,7 +1099,7 @@ impl QueueBackend for OpenHomeQueue {
self.ensure_playlist_source_selected()?;
// Lock the cache for the entire operation to prevent race conditions
let mut cache = self.track_ids_cache.lock().unwrap();
let mut cache = self.track_ids_cache.lock().expect("track_ids_cache mutex poisoned");
// Check if cache is valid
if let Some(cached_ids) = cache.get() {
@@ -1147,7 +1147,7 @@ impl QueueBackend for OpenHomeQueue {
fn current_track(&self) -> Result<Option<u32>, ControlPointError> {
// Hold lock during entire operation to prevent race conditions
let mut cache = self.current_track_id_cache.lock().unwrap();
let mut cache = self.current_track_id_cache.lock().expect("current_track_id_cache mutex poisoned");
// Return cached value if valid
if let Some(cached_id) = cache.get() {
@@ -1192,7 +1192,7 @@ impl QueueBackend for OpenHomeQueue {
const MAX_BATCH: usize = 256;
let mut entries = Vec::with_capacity(ids.len());
for chunk in ids.chunks(MAX_BATCH) {
if let Some(cached) = self.read_list_cache.lock().unwrap().get(chunk) {
if let Some(cached) = self.read_list_cache.lock().expect("read_list_cache mutex poisoned").get(chunk) {
trace!(
renderer = self.renderer_id.0.as_str(),
"ReadList cache hit for {} IDs",
@@ -1283,7 +1283,7 @@ impl QueueBackend for OpenHomeQueue {
self.ensure_playlist_source_selected()?;
self.playlist_client.delete_all()?;
self.metadata_cache.lock().unwrap().clear();
self.metadata_cache.lock().expect("metadata_cache mutex poisoned").clear();
// Invalidate caches after delete_all (clears queue and current track)
self.invalidate_all_caches();
@@ -1340,7 +1340,7 @@ impl QueueBackend for OpenHomeQueue {
"sync_queue: Empty playlist - clearing queue with delete_all"
);
self.playlist_client.delete_all()?;
self.metadata_cache.lock().unwrap().clear();
self.metadata_cache.lock().expect("metadata_cache mutex poisoned").clear();
self.invalidate_all_caches();
let post_current_track = self.playlist_client.id().ok();
@@ -1364,7 +1364,7 @@ impl QueueBackend for OpenHomeQueue {
.last()
.copied()
.unwrap_or(OPENHOME_PLAYLIST_HEAD_ID);
let mut uri_cache = self.uri_by_id.lock().unwrap();
let mut uri_cache = self.uri_by_id.lock().expect("uri_by_id mutex poisoned");
for item in &new_items {
if cancel_token.load(SeqCst) {
return Err(ControlPointError::SyncCancelled);
@@ -1583,8 +1583,8 @@ impl QueueBackend for OpenHomeQueue {
.insert(before_id, &item.uri, &metadata)?;
// Mettre à jour le cache avec les nouvelles métadonnées
self.metadata_cache.lock().unwrap().remove(&track_id);
self.uri_by_id.lock().unwrap().remove(&track_id);
self.metadata_cache.lock().expect("metadata_cache mutex poisoned").remove(&track_id);
self.uri_by_id.lock().expect("uri_by_id mutex poisoned").remove(&track_id);
self.cache_metadata(new_id, item.metadata, &item.uri);
if ci == Some(index) {

6
pmocontrol/src/upnp_clients/openhome_client.rs
View File

@@ -817,7 +817,7 @@ impl OhProductClient {
pub fn source_xml(&self) -> Result<Vec<OhProductSource>> {
// Lock the cache for the entire operation to prevent race conditions
let mut cache = self.source_xml_cache.lock().unwrap();
let mut cache = self.source_xml_cache.lock().expect("source_xml_cache mutex poisoned");
// Check if cache is valid
if let Some(cached_sources) = cache.get() {
@@ -841,7 +841,7 @@ impl OhProductClient {
pub fn source_index(&self) -> Result<u32, ControlPointError> {
// Lock the cache for the entire operation to prevent race conditions
let mut cache = self.source_index_cache.lock().unwrap();
let mut cache = self.source_index_cache.lock().expect("source_index_cache mutex poisoned");
// Check if cache is valid
if let Some(cached_index) = cache.get() {
@@ -881,7 +881,7 @@ impl OhProductClient {
// Invalidate cache after write operation
if result.is_ok() {
let mut cache = self.source_index_cache.lock().unwrap();
let mut cache = self.source_index_cache.lock().expect("source_index_cache mutex poisoned");
cache.invalidate();
}