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

Mise en place de la synchronisation de la playlist du point de contrôle avec celle du serveur de média.

Browse Source
This commit is contained in:
Eric Coissac
2025-12-02 14:41:53 +01:00
parent aeb9332387
commit f99ab2a41e
9 changed files with 19861 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

502
pmocontrol/src/control_point.rs
View File

@@ -7,7 +7,7 @@ use std::time::Duration;
use anyhow::anyhow;
use crossbeam_channel::Receiver;
use pmoupnp::ssdp::SsdpClient;
use tracing::{debug, error, warn};
use tracing::{debug, error, info, warn};
use crate::MusicRenderer;
use crate::capabilities::{
@@ -16,7 +16,9 @@ use crate::capabilities::{
};
use crate::discovery::DiscoveryManager;
use crate::events::{MediaServerEventBus, RendererEventBus};
use crate::media_server::{MediaServerInfo, ServerId};
use crate::media_server::{
MediaBrowser, MediaEntry, MediaResource, MediaServerInfo, MusicServer, ServerId,
};
use crate::media_server_events::spawn_media_server_event_runtime;
use crate::model::{MediaServerEvent, RendererEvent, RendererId, RendererProtocol};
use crate::music_renderer::op_not_supported;
@@ -25,6 +27,25 @@ use crate::provider::HttpXmlDescriptionProvider;
use crate::registry::{DeviceRegistry, DeviceRegistryRead, DeviceUpdate};
use crate::upnp_renderer::UpnpRenderer;
/// Optional attachment between a renderer playback queue and a server-side
/// DIDL-Lite playlist container.
///
/// When a queue is bound to a playlist, the control point will automatically
/// refresh it whenever the server notifies us of changes to that container.
/// User-driven mutations (clear, enqueue, etc.) break the binding automatically.
#[derive(Clone, Debug)]
struct PlaylistBinding {
/// MediaServer that owns the playlist container.
server_id: ServerId,
/// DIDL-Lite object id of the playlist container.
container_id: String,
/// True once at least one ContainerUpdateIDs notification has been seen.
has_seen_update: bool,
/// Flag used internally to signal that the queue should be refreshed
/// from the server container.
pending_refresh: bool,
}
/// Control point minimal :
/// - lance un SsdpClient dans un thread,
/// - passe les SsdpEvent au DiscoveryManager,
@@ -34,6 +55,12 @@ pub struct ControlPoint {
event_bus: RendererEventBus,
media_event_bus: MediaServerEventBus,
runtime: Arc<RuntimeState>,
/// Optional attachment between a renderer playback queue and a
/// server-side DIDL-Lite playlist container.
///
/// Key : RendererId
/// Value : PlaylistBinding
playlist_bindings: Arc<Mutex<HashMap<RendererId, PlaylistBinding>>>,
}
impl ControlPoint {
@@ -45,6 +72,7 @@ impl ControlPoint {
let event_bus = RendererEventBus::new();
let media_event_bus = MediaServerEventBus::new();
let runtime = Arc::new(RuntimeState::new());
let playlist_bindings = Arc::new(Mutex::new(HashMap::new()));
// SsdpClient
let client = SsdpClient::new()?; // pmoupnp::ssdp::SsdpClient
@@ -61,10 +89,11 @@ impl ControlPoint {
// ACTIVE DISCOVERY : envoyer quelques M-SEARCH au démarrage
// pour forcer les devices à répondre rapidement.
let search_targets = [
"ssdp:all",
"urn:schemas-upnp-org:device:MediaRenderer:1",
"urn:av-openhome-org:device:MediaRenderer:1",
"urn:schemas-upnp-org:device:MediaServer:1",
"ssdp:all",
"urn:schemas-upnp-org:device:MediaRenderer:1",
"urn:av-openhome-org:device:MediaRenderer:1",
"urn:schemas-upnp-org:device:MediaServer:1",
"urn:schemas-wiimu-com:service:PlayQueue:1", // <-- AJOUTER
];
for st in &search_targets {
@@ -96,9 +125,11 @@ impl ControlPoint {
event_bus: event_bus.clone(),
media_event_bus: media_event_bus.clone(),
runtime: Arc::clone(&runtime),
playlist_bindings: Arc::clone(&playlist_bindings),
};
thread::spawn(move || {
let mut tick: u32 = 0;
loop {
let infos = {
let reg = runtime_cp.registry.read().unwrap();
@@ -126,6 +157,7 @@ impl ControlPoint {
let mut new_snapshot = prev_snapshot.clone();
let prev_position = prev_snapshot.position.clone();
// Poll position every tick (1s) for smooth UI progress
if let Ok(position) = renderer.playback_position() {
let has_changed = match prev_snapshot.position.as_ref() {
Some(prev) => !playback_position_equal(prev, &position),
@@ -142,6 +174,7 @@ impl ControlPoint {
new_snapshot.position = Some(position);
}
// Poll state every tick to ensure responsive playback control
if let Ok(raw_state) = renderer.playback_state() {
let logical_state = compute_logical_playback_state(
&raw_state,
@@ -154,7 +187,9 @@ impl ControlPoint {
None => true,
};
if has_changed {
// Emit event only for non-transient states to reduce noise
// and avoid overwhelming the renderer during track changes
if has_changed && !matches!(logical_state, PlaybackState::Transitioning) {
runtime_cp.emit_renderer_event(RendererEvent::StateChanged {
id: renderer_id.clone(),
state: logical_state.clone(),
@@ -164,26 +199,30 @@ impl ControlPoint {
new_snapshot.state = Some(logical_state);
}
if let Ok(volume) = renderer.volume() {
if prev_snapshot.last_volume != Some(volume) {
runtime_cp.emit_renderer_event(RendererEvent::VolumeChanged {
id: renderer_id.clone(),
volume,
});
// Poll volume and mute less frequently (every 3 seconds)
// to reduce SOAP overhead without impacting UI responsiveness
if tick % 3 == 0 {
if let Ok(volume) = renderer.volume() {
if prev_snapshot.last_volume != Some(volume) {
runtime_cp.emit_renderer_event(RendererEvent::VolumeChanged {
id: renderer_id.clone(),
volume,
});
}
new_snapshot.last_volume = Some(volume);
}
new_snapshot.last_volume = Some(volume);
}
if let Ok(mute) = renderer.mute() {
if prev_snapshot.last_mute != Some(mute) {
runtime_cp.emit_renderer_event(RendererEvent::MuteChanged {
id: renderer_id.clone(),
mute,
});
}
if let Ok(mute) = renderer.mute() {
if prev_snapshot.last_mute != Some(mute) {
runtime_cp.emit_renderer_event(RendererEvent::MuteChanged {
id: renderer_id.clone(),
mute,
});
new_snapshot.last_mute = Some(mute);
}
new_snapshot.last_mute = Some(mute);
}
runtime_cp
@@ -191,6 +230,8 @@ impl ControlPoint {
.update_snapshot(&renderer_id, new_snapshot);
}
tick = tick.wrapping_add(1);
// Keep 1 second polling for smooth position updates
thread::sleep(Duration::from_secs(1));
}
});
@@ -201,11 +242,138 @@ impl ControlPoint {
timeout_secs,
)?;
// Worker thread to process MediaServerEvent and trigger queue refreshes
// for renderers bound to updated playlist containers
let registry_for_media_worker = Arc::clone(&registry);
let runtime_for_media_worker = Arc::clone(&runtime);
let bindings_for_media_worker = Arc::clone(&playlist_bindings);
let media_rx = media_event_bus.subscribe();
thread::Builder::new()
.name("cp-media-server-event-worker".into())
.spawn(move || {
loop {
let event = match media_rx.recv() {
Ok(e) => e,
Err(_) => {
warn!("MediaServerEvent channel closed, worker exiting");
break;
}
};
match event {
MediaServerEvent::GlobalUpdated {
server_id,
system_update_id,
} => {
info!(
server = server_id.0.as_str(),
system_update_id = system_update_id,
"MediaServer global update"
);
}
MediaServerEvent::ContainersUpdated {
server_id,
container_ids,
} => {
let renderers_to_refresh: Vec<RendererId> = {
let mut bindings = bindings_for_media_worker.lock().unwrap();
let mut to_refresh = Vec::new();
for (renderer_id, binding) in bindings.iter_mut() {
if binding.server_id == server_id
&& container_ids.contains(&binding.container_id)
{
binding.pending_refresh = true;
binding.has_seen_update = true;
to_refresh.push(renderer_id.clone());
}
}
to_refresh
};
for renderer_id in renderers_to_refresh {
debug!(
renderer = renderer_id.0.as_str(),
server = server_id.0.as_str(),
"Triggering queue refresh for bound playlist"
);
if let Err(err) = refresh_attached_queue_for(
&registry_for_media_worker,
&runtime_for_media_worker,
&bindings_for_media_worker,
&renderer_id,
) {
warn!(
renderer = renderer_id.0.as_str(),
server = server_id.0.as_str(),
error = %err,
"Failed to refresh queue from playlist container"
);
}
}
}
}
}
})?;
// Periodic refresh worker for bound playlists
// Every 60 seconds, trigger a refresh for all renderers with active bindings
let registry_for_periodic = Arc::clone(&registry);
let runtime_for_periodic = Arc::clone(&runtime);
let bindings_for_periodic = Arc::clone(&playlist_bindings);
thread::Builder::new()
.name("cp-playlist-periodic-refresh".into())
.spawn(move || {
loop {
// Sleep for 60 seconds between refresh cycles
thread::sleep(Duration::from_secs(60));
// Collect all renderers with active bindings and mark them for refresh
let renderers_to_refresh: Vec<RendererId> = {
let mut bindings = bindings_for_periodic.lock().unwrap();
let mut to_refresh = Vec::new();
for (renderer_id, binding) in bindings.iter_mut() {
binding.pending_refresh = true;
to_refresh.push(renderer_id.clone());
}
to_refresh
};
// Trigger refresh for each bound renderer (outside of lock)
for renderer_id in renderers_to_refresh {
debug!(
renderer = renderer_id.0.as_str(),
"Periodic refresh triggered for bound playlist"
);
if let Err(err) = refresh_attached_queue_for(
&registry_for_periodic,
&runtime_for_periodic,
&bindings_for_periodic,
&renderer_id,
) {
warn!(
renderer = renderer_id.0.as_str(),
error = %err,
"Periodic refresh failed for bound playlist"
);
}
}
}
})?;
Ok(Self {
registry,
event_bus,
media_event_bus,
runtime,
playlist_bindings,
})
}
@@ -308,6 +476,9 @@ impl ControlPoint {
return Err(err);
}
// User-driven mutation: detach any playlist binding
self.detach_binding_on_user_mutation(renderer_id, "clear_queue");
let removed = self
.runtime
.with_queue_mut(renderer_id, |queue| {
@@ -340,6 +511,9 @@ impl ControlPoint {
return Err(err);
}
// User-driven mutation: detach any playlist binding
self.detach_binding_on_user_mutation(renderer_id, "enqueue_items");
let item_count = items.len();
let new_len = self
.runtime
@@ -501,6 +675,92 @@ impl ControlPoint {
self.media_event_bus.subscribe()
}
/// Attach a renderer's playback queue to a server-side playlist container.
///
/// When attached, the queue will be automatically refreshed from the
/// container whenever the server notifies us of changes via ContentDirectory
/// events. The binding is broken if the user explicitly mutates the queue
/// through methods like `clear_queue` or `enqueue_items`.
pub fn attach_queue_to_playlist(
&self,
renderer_id: &RendererId,
server_id: ServerId,
container_id: String,
) {
let mut bindings = self.playlist_bindings.lock().unwrap();
bindings.insert(
renderer_id.clone(),
PlaylistBinding {
server_id: server_id.clone(),
container_id: container_id.clone(),
has_seen_update: false,
pending_refresh: false,
},
);
info!(
renderer = renderer_id.0.as_str(),
server = server_id.0.as_str(),
container = container_id.as_str(),
"Queue attached to playlist container"
);
}
/// Detach a renderer's queue from its associated playlist container.
///
/// After calling this, the queue will no longer be automatically refreshed
/// from the server. If no binding existed, this is a no-op.
pub fn detach_queue_playlist(&self, renderer_id: &RendererId) {
let mut bindings = self.playlist_bindings.lock().unwrap();
if let Some(binding) = bindings.remove(renderer_id) {
info!(
renderer = renderer_id.0.as_str(),
server = binding.server_id.0.as_str(),
container = binding.container_id.as_str(),
"Queue detached from playlist container"
);
} else {
debug!(
renderer = renderer_id.0.as_str(),
"detach_queue_playlist: no binding to remove"
);
}
}
/// Query the current playlist binding for a renderer's queue, if any.
///
/// Returns `(server_id, container_id, has_seen_update)` if the queue is
/// bound to a server playlist container, or `None` otherwise.
pub fn current_queue_playlist_binding(
&self,
renderer_id: &RendererId,
) -> Option<(ServerId, String, bool)> {
let bindings = self.playlist_bindings.lock().unwrap();
bindings.get(renderer_id).map(|binding| {
(
binding.server_id.clone(),
binding.container_id.clone(),
binding.has_seen_update,
)
})
}
/// Internal helper to detach the playlist binding on user-driven mutations.
///
/// This is called by public queue mutation methods (clear, enqueue, etc.)
/// to ensure that explicit user actions break the automatic refresh binding.
fn detach_binding_on_user_mutation(&self, renderer_id: &RendererId, reason: &str) {
let mut bindings = self.playlist_bindings.lock().unwrap();
if let Some(binding) = bindings.remove(renderer_id) {
info!(
renderer = renderer_id.0.as_str(),
server = binding.server_id.0.as_str(),
container = binding.container_id.as_str(),
reason = reason,
"Playlist binding auto-detached due to user mutation"
);
}
}
pub(crate) fn emit_renderer_event(&self, event: RendererEvent) {
self.handle_renderer_event(&event);
self.event_bus.broadcast(event);
@@ -657,6 +917,202 @@ impl RuntimeState {
}
}
/// Internal helper to refresh a renderer's playback queue from its bound
/// playlist container.
///
/// This function is called automatically when a ContentDirectory event indicates
/// that the bound container has been updated. It attempts to preserve the
/// currently playing item when possible.
fn refresh_attached_queue_for(
registry: &Arc<RwLock<DeviceRegistry>>,
runtime: &Arc<RuntimeState>,
bindings: &Arc<Mutex<HashMap<RendererId, PlaylistBinding>>>,
renderer_id: &RendererId,
) -> anyhow::Result<()> {
// Step 1: Check binding and mark refresh as in-progress
let (server_id, container_id) = {
let mut bindings_lock = bindings.lock().unwrap();
let binding = match bindings_lock.get_mut(renderer_id) {
Some(b) => b,
None => {
debug!(
renderer = renderer_id.0.as_str(),
"refresh_attached_queue_for: no binding present"
);
return Ok(());
}
};
if !binding.pending_refresh {
debug!(
renderer = renderer_id.0.as_str(),
"refresh_attached_queue_for: pending_refresh is false, nothing to do"
);
return Ok(());
}
// Mark as processed
binding.pending_refresh = false;
(binding.server_id.clone(), binding.container_id.clone())
};
// Step 2: Fetch MediaServerInfo from registry
let server_info = {
let reg = registry.read().unwrap();
reg.get_server(&server_id)
};
let server_info = match server_info {
Some(info) => info,
None => {
warn!(
renderer = renderer_id.0.as_str(),
server = server_id.0.as_str(),
"refresh_attached_queue_for: server not found in registry"
);
return Ok(());
}
};
if !server_info.online {
debug!(
renderer = renderer_id.0.as_str(),
server = server_id.0.as_str(),
"refresh_attached_queue_for: server offline, skipping refresh"
);
return Ok(());
}
if !server_info.has_content_directory {
debug!(
renderer = renderer_id.0.as_str(),
server = server_id.0.as_str(),
"refresh_attached_queue_for: server has no ContentDirectory"
);
return Ok(());
}
// Step 3: Create MusicServer and browse container
let music_server = MusicServer::from_info(&server_info, Duration::from_secs(5))?;
let entries = match music_server.browse_children(&container_id, 0, 64) {
Ok(e) => e,
Err(err) => {
warn!(
renderer = renderer_id.0.as_str(),
server = server_id.0.as_str(),
container = container_id.as_str(),
error = %err,
"Failed to browse playlist container for refresh"
);
return Err(err);
}
};
// Step 4: Convert MediaEntry to PlaybackItem
let new_items: Vec<PlaybackItem> = entries
.iter()
.filter_map(|entry| playback_item_from_entry(&music_server, entry))
.collect();
if new_items.is_empty() {
debug!(
renderer = renderer_id.0.as_str(),
server = server_id.0.as_str(),
container = container_id.as_str(),
"Refreshed playlist is empty, clearing queue"
);
runtime.with_queue_mut(renderer_id, |queue| queue.clear());
return Ok(());
}
// Step 5: Intelligent refresh: try to keep current item if it's still in the new list
let current_item = runtime.queue_snapshot(renderer_id).and_then(|q| q.first().cloned());
let item_found_at = current_item.as_ref().and_then(|current| {
new_items.iter().position(|new_item| {
// Match by object_id if both have it
if let (Some(current_obj), Some(new_obj)) = (&current.object_id, &new_item.object_id)
{
return current_obj == new_obj;
}
// Fallback: match by URI
current.uri == new_item.uri
})
});
runtime.with_queue_mut(renderer_id, |queue| {
queue.clear();
if let Some(idx) = item_found_at {
// Current item found: reconstruct queue starting from that item
for i in idx..new_items.len() {
queue.enqueue(new_items[i].clone());
}
info!(
renderer = renderer_id.0.as_str(),
server = server_id.0.as_str(),
container = container_id.as_str(),
queue_len = new_items.len() - idx,
current_preserved = true,
"Refreshed queue from playlist container"
);
} else {
// Current item not found: replace with full new list
for item in new_items.iter() {
queue.enqueue(item.clone());
}
info!(
renderer = renderer_id.0.as_str(),
server = server_id.0.as_str(),
container = container_id.as_str(),
queue_len = new_items.len(),
current_preserved = false,
"Refreshed queue from playlist container (current item not found)"
);
}
});
Ok(())
}
/// Helper to detect if a MediaResource is audio content.
fn is_audio_resource(res: &MediaResource) -> bool {
let lower = res.protocol_info.to_ascii_lowercase();
if lower.contains("audio/") {
return true;
}
// Check MIME type in protocolInfo (format: protocol:network:contentFormat:additionalInfo)
lower
.split(':')
.nth(2)
.map(|mime| mime.starts_with("audio/"))
.unwrap_or(false)
}
/// Helper to convert a MediaEntry to a PlaybackItem.
fn playback_item_from_entry(server: &MusicServer, entry: &MediaEntry) -> Option<PlaybackItem> {
// Ignore containers
if entry.is_container {
return None;
}
// Skip "live stream" entries (heuristic from example)
if entry.title.to_ascii_lowercase().contains("live stream") {
return None;
}
// Find an audio resource
let resource = entry.resources.iter().find(|res| is_audio_resource(res))?;
let mut item = PlaybackItem::new(resource.uri.clone());
item.title = Some(entry.title.clone());
item.server_id = Some(server.id().clone());
item.object_id = Some(entry.id.clone());
Some(item)
}
/// Parse "HH:MM:SS" style time strings to seconds.
///
/// Returns None for empty or sentinel values such as "NOT_IMPLEMENTED" or "-:--:--".