From a559375176bdfe4d1199566448729da0fb261b94 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sat, 29 Nov 2025 01:08:07 +0100 Subject: [PATCH] Ajout la playlist live --- .../variables/containerupdateids.rs | 9 + .../src/contentdirectory/variables/mod.rs | 2 + pmomediaserver/src/sources.rs | 7 +- pmoparadise/src/source.rs | 284 +++++++++++++++++- pmoplaylist/src/handle/write.rs | 18 ++ pmoplaylist/src/manager.rs | 34 ++- 6 files changed, 350 insertions(+), 4 deletions(-) create mode 100644 pmomediaserver/src/contentdirectory/variables/containerupdateids.rs diff --git a/pmomediaserver/src/contentdirectory/variables/containerupdateids.rs b/pmomediaserver/src/contentdirectory/variables/containerupdateids.rs new file mode 100644 index 00000000..8a0ace80 --- /dev/null +++ b/pmomediaserver/src/contentdirectory/variables/containerupdateids.rs @@ -0,0 +1,9 @@ +use pmoupnp::define_variable; + +// Liste des conteneurs modifiés (format "id,updateId,id,updateId,...") +define_variable! { + pub static CONTAINERUPDATEIDS: String = "ContainerUpdateIDs" { + evented: true, + // valeur initiale vide + } +} diff --git a/pmomediaserver/src/contentdirectory/variables/mod.rs b/pmomediaserver/src/contentdirectory/variables/mod.rs index 6660c23c..7277e589 100644 --- a/pmomediaserver/src/contentdirectory/variables/mod.rs +++ b/pmomediaserver/src/contentdirectory/variables/mod.rs @@ -7,6 +7,7 @@ mod a_arg_type_result; mod a_arg_type_searchcriteria; mod a_arg_type_sortcriteria; mod a_arg_type_updateid; +mod containerupdateids; mod searchcapabilities; mod sortcapabilities; mod systemupdateid; @@ -20,6 +21,7 @@ pub use a_arg_type_result::A_ARG_TYPE_RESULT; pub use a_arg_type_searchcriteria::A_ARG_TYPE_SEARCHCRITERIA; pub use a_arg_type_sortcriteria::A_ARG_TYPE_SORTCRITERIA; pub use a_arg_type_updateid::A_ARG_TYPE_UPDATEID; +pub use containerupdateids::CONTAINERUPDATEIDS; pub use searchcapabilities::SEARCHCAPABILITIES; pub use sortcapabilities::SORTCAPABILITIES; pub use systemupdateid::SYSTEMUPDATEID; diff --git a/pmomediaserver/src/sources.rs b/pmomediaserver/src/sources.rs index 370809ba..73d9f8d8 100644 --- a/pmomediaserver/src/sources.rs +++ b/pmomediaserver/src/sources.rs @@ -181,10 +181,13 @@ impl SourcesExt for Server { let base_url = self.base_url(); // Créer la source Radio Paradise (utilise le singleton PlaylistManager) - let source = RadioParadiseSource::new(base_url.to_string()); + let source = Arc::new(RadioParadiseSource::new(base_url.to_string())); + + // Brancher les callbacks de playlists (live/history) pour signaler les updates + source.attach_playlist_callbacks(); // Enregistrer la source - self.register_music_source(Arc::new(source)).await; + self.register_music_source(source.clone()).await; tracing::info!("✅ Radio Paradise source registered successfully"); diff --git a/pmoparadise/src/source.rs b/pmoparadise/src/source.rs index 7e149328..7b74d91e 100644 --- a/pmoparadise/src/source.rs +++ b/pmoparadise/src/source.rs @@ -28,6 +28,8 @@ const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp"); /// - Root: `radio-paradise` /// - Channel container: `radio-paradise:channel:{slug}` /// - Live stream item: `radio-paradise:channel:{slug}:live` +/// - Live playlist container: `radio-paradise:channel:{slug}:liveplaylist` +/// - Live playlist track: `radio-paradise:channel:{slug}:liveplaylist:track:{pk}` /// - History container: `radio-paradise:channel:{slug}:history` /// - History track: `radio-paradise:channel:{slug}:history:track:{pk}` #[derive(Debug, Clone)] @@ -38,6 +40,8 @@ pub struct RadioParadiseSource { update_counter: Arc>, /// Last change timestamp last_change: Arc>, + /// Tokens des callbacks enregistrés auprès du PlaylistManager + callback_tokens: Arc>>, } impl RadioParadiseSource { @@ -56,6 +60,7 @@ impl RadioParadiseSource { base_url: base_url.into(), update_counter: Arc::new(RwLock::new(0)), last_change: Arc::new(RwLock::new(SystemTime::now())), + callback_tokens: Arc::new(std::sync::Mutex::new(Vec::new())), } } @@ -69,6 +74,49 @@ impl RadioParadiseSource { format!("{}/radioparadise/stream/{}/ogg", self.base_url, slug) } + /// Incrémente l'update_counter et met à jour last_change + async fn bump_update_counter(&self) { + { + let mut c = self.update_counter.write().await; + *c = c.wrapping_add(1).max(1); + } + let mut lc = self.last_change.write().await; + *lc = SystemTime::now(); + } + + /// Enregistre des callbacks sur les playlists live/historique pour notifier les changements + pub fn attach_playlist_callbacks(self: &Arc) { + use pmoplaylist::PlaylistManager; + + // Préparer les IDs de playlists à surveiller (live + history pour chaque canal) + let ids: Vec = ALL_CHANNELS + .iter() + .flat_map(|ch| { + vec![ + Self::live_playlist_id(ch.slug), + Self::history_playlist_id(ch.slug), + ] + }) + .collect(); + + let mgr = PlaylistManager(); + let mut tokens = self.callback_tokens.lock().unwrap(); + + for pid in ids { + let weak = Arc::downgrade(self); + let token = mgr.register_callback(move |changed_id| { + if changed_id == pid { + if let Some(strong) = weak.upgrade() { + tokio::spawn(async move { + strong.bump_update_counter().await; + }); + } + } + }); + tokens.push(token); + } + } + /// URL de fallback pour l'image par défaut de la source fn default_cover_url(&self) -> String { format!("{}/api/sources/{}/image", self.base_url, self.id()) @@ -152,6 +200,11 @@ impl RadioParadiseSource { format!("radio-paradise-history-{}", slug) } + /// Live playlist id for a channel + fn live_playlist_id(slug: &str) -> String { + format!("radio-paradise-live-{}", slug) + } + /// Get channel descriptor by slug fn get_channel_by_slug(slug: &str) -> Option<&'static ChannelDescriptor> { ALL_CHANNELS.iter().find(|ch| ch.slug == slug) @@ -168,6 +221,15 @@ impl RadioParadiseSource { ["radio-paradise", "channel", slug, "live"] => ObjectIdType::LiveStream { slug: (*slug).to_string(), }, + ["radio-paradise", "channel", slug, "liveplaylist"] => ObjectIdType::LivePlaylist { + slug: (*slug).to_string(), + }, + ["radio-paradise", "channel", slug, "liveplaylist", "track", pk] => { + ObjectIdType::LivePlaylistTrack { + slug: (*slug).to_string(), + pk: (*pk).to_string(), + } + } ["radio-paradise", "channel", slug, "history"] => ObjectIdType::History { slug: (*slug).to_string(), }, @@ -196,6 +258,21 @@ impl RadioParadiseSource { } } + /// Build the live playlist container for a channel + fn build_live_playlist_container(&self, descriptor: &ChannelDescriptor) -> Container { + Container { + id: format!("radio-paradise:channel:{}:liveplaylist", descriptor.slug), + parent_id: format!("radio-paradise:channel:{}", descriptor.slug), + restricted: Some("1".to_string()), + child_count: None, + searchable: Some("0".to_string()), + title: format!("{} - Live Playlist", descriptor.display_name), + class: "object.container.playlistContainer".to_string(), + containers: vec![], + items: vec![], + } + } + /// Build a live stream item for a channel fn build_live_stream_item(&self, descriptor: &ChannelDescriptor) -> Item { let stream_url = self.build_live_url(descriptor.slug); @@ -332,6 +409,82 @@ impl RadioParadiseSource { Ok(items) } + /// Get items from live playlist (current stream queue) + #[cfg(feature = "playlist")] + async fn get_live_playlist_items( + &self, + slug: &str, + _offset: usize, + count: usize, + ) -> Result> { + let playlist_id = Self::live_playlist_id(slug); + + let manager = pmoplaylist::PlaylistManager(); + let reader = manager.get_read_handle(&playlist_id).await.map_err(|e| { + MusicSourceError::BrowseError(format!("Failed to get live playlist {}: {}", playlist_id, e)) + })?; + + let mut items = reader.to_items(count).await.map_err(|e| { + MusicSourceError::BrowseError(format!( + "Failed to read live playlist entries: {}", + e + )) + })?; + + for item in items.iter_mut() { + // Ajuster id/parent/url pour coller au schéma Radio Paradise + if let Some(resource) = item.resources.first_mut() { + if let Some(pk) = resource.url.split('/').last() { + item.id = format!( + "radio-paradise:channel:{}:liveplaylist:track:{}", + slug, pk + ); + item.parent_id = format!( + "radio-paradise:channel:{}:liveplaylist", + slug + ); + + if resource.url.starts_with('/') { + resource.url = format!("{}{}", self.base_url, resource.url); + } + } + } + + if item.genre.is_none() { + item.genre = Some("Radio Paradise".to_string()); + } + + if let Some(art) = item.album_art.as_mut() { + if art.starts_with('/') { + *art = format!("{}{}", self.base_url, art); + } + } else { + item.album_art = Some(self.default_cover_url()); + } + } + + Ok(items) + } + + /// Get a single item from the live playlist by pk + #[cfg(feature = "playlist")] + async fn get_live_playlist_item(&self, slug: &str, pk: &str) -> Result { + let items = self.get_live_playlist_items(slug, 0, 1000).await?; + let expected_id = format!( + "radio-paradise:channel:{}:liveplaylist:track:{}", + slug, pk + ); + for item in items { + if item.id == expected_id { + return Ok(item); + } + } + Err(MusicSourceError::ObjectNotFound(format!( + "Track with pk {} not found in live playlist", + pk + ))) + } + } /// Types of object IDs in the Radio Paradise source @@ -340,6 +493,8 @@ enum ObjectIdType { Root, Channel { slug: String }, LiveStream { slug: String }, + LivePlaylist { slug: String }, + LivePlaylistTrack { slug: String, pk: String }, History { slug: String }, HistoryTrack { slug: String, pk: String }, Unknown, @@ -393,6 +548,7 @@ impl MusicSource for RadioParadiseSource { })?; let live_item = self.build_live_stream_item(descriptor); + let live_playlist_container = self.build_live_playlist_container(descriptor); #[cfg(feature = "playlist")] let history_container = self.build_history_container_with_count(descriptor).await; @@ -400,7 +556,7 @@ impl MusicSource for RadioParadiseSource { let history_container = self.build_history_container(descriptor); Ok(BrowseResult::Mixed { - containers: vec![history_container], + containers: vec![live_playlist_container, history_container], items: vec![live_item], }) } @@ -439,12 +595,52 @@ impl MusicSource for RadioParadiseSource { Ok(BrowseResult::Items(vec![item])) } + ObjectIdType::LivePlaylist { slug } => { + // Playlist du live : container + items + let descriptor = Self::get_channel_by_slug(&slug).ok_or_else(|| { + MusicSourceError::ObjectNotFound(format!("Unknown channel: {}", slug)) + })?; + + #[cfg(feature = "playlist")] + { + let container = self.build_live_playlist_container(descriptor); + let items = self.get_live_playlist_items(&slug, 0, 100).await?; + Ok(BrowseResult::Mixed { + containers: vec![container], + items, + }) + } + + #[cfg(not(feature = "playlist"))] + { + let container = self.build_live_playlist_container(descriptor); + Ok(BrowseResult::Containers(vec![container])) + } + } + ObjectIdType::HistoryTrack { slug, pk } => { // Return metadata for the history track item let item = self.get_item(object_id).await?; Ok(BrowseResult::Items(vec![item])) } + ObjectIdType::LivePlaylistTrack { slug, pk } => { + // Détails d'un titre du live (playlist live) + #[cfg(feature = "playlist")] + { + let item = self.get_live_playlist_item(&slug, &pk).await?; + Ok(BrowseResult::Items(vec![item])) + } + + #[cfg(not(feature = "playlist"))] + { + let _ = (slug, pk); + Err(MusicSourceError::NotSupported( + "Playlist feature not enabled".to_string(), + )) + } + } + ObjectIdType::Unknown => Err(MusicSourceError::ObjectNotFound(format!( "Unknown object ID: {}", object_id @@ -464,6 +660,11 @@ impl MusicSource for RadioParadiseSource { Ok(format!("{}/cache/audio/{}", self.base_url, pk)) } + ObjectIdType::LivePlaylistTrack { pk, .. } => { + // Return cached audio URL + Ok(format!("{}/cache/audio/{}", self.base_url, pk)) + } + _ => Err(MusicSourceError::ObjectNotFound(format!( "Cannot resolve URI for object: {}", object_id @@ -514,6 +715,14 @@ impl MusicSource for RadioParadiseSource { bitrate: None, channels: Some(2), }]), + ObjectIdType::LivePlaylistTrack { .. } => Ok(vec![AudioFormat { + format_id: "flac".to_string(), + mime_type: "audio/flac".to_string(), + sample_rate: Some(44100), + bit_depth: Some(16), + bitrate: None, + channels: Some(2), + }]), _ => Err(MusicSourceError::ObjectNotFound(format!( "Cannot list formats for object: {}", object_id @@ -603,6 +812,79 @@ impl MusicSource for RadioParadiseSource { } } + ObjectIdType::LivePlaylistTrack { slug, pk } => { + #[cfg(feature = "playlist")] + { + let playlist_id = Self::live_playlist_id(&slug); + let manager = pmoplaylist::PlaylistManager(); + let reader = manager.get_read_handle(&playlist_id).await.map_err(|e| { + MusicSourceError::BrowseError(format!( + "Failed to get live playlist {}: {}", + playlist_id, e + )) + })?; + + let items = reader.to_items(1000).await.map_err(|e| { + MusicSourceError::BrowseError(format!( + "Failed to read live playlist entries: {}", + e + )) + })?; + + for mut item in items { + if let Some(resource) = item.resources.first_mut() { + if let Some(pk2) = resource.url.split('/').last() { + item.id = format!( + "radio-paradise:channel:{}:liveplaylist:track:{}", + slug, pk2 + ); + item.parent_id = format!( + "radio-paradise:channel:{}:liveplaylist", + slug + ); + + if resource.url.starts_with('/') { + resource.url = format!("{}{}", self.base_url, resource.url); + } + } + } + + if item.genre.is_none() { + item.genre = Some("Radio Paradise".to_string()); + } + + if let Some(art) = item.album_art.as_mut() { + if art.starts_with('/') { + *art = format!("{}{}", self.base_url, art); + } + } else { + item.album_art = Some(self.default_cover_url()); + } + + let expected_id = format!( + "radio-paradise:channel:{}:liveplaylist:track:{}", + slug, pk + ); + if item.id == expected_id { + return Ok(item); + } + } + + Err(MusicSourceError::ObjectNotFound(format!( + "Track with pk {} not found in live playlist", + pk + ))) + } + + #[cfg(not(feature = "playlist"))] + { + let _ = (slug, pk); + Err(MusicSourceError::NotSupported( + "Playlist feature not enabled".to_string(), + )) + } + } + _ => Err(MusicSourceError::ObjectNotFound(format!( "Cannot get item for object: {}", object_id diff --git a/pmoplaylist/src/handle/write.rs b/pmoplaylist/src/handle/write.rs index 366016d1..c93b390e 100644 --- a/pmoplaylist/src/handle/write.rs +++ b/pmoplaylist/src/handle/write.rs @@ -47,6 +47,9 @@ impl WriteHandle { self.save_to_db().await?; } + // Notifier le manager + crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + Ok(()) } @@ -75,6 +78,8 @@ impl WriteHandle { self.save_to_db().await?; } + crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + Ok(()) } @@ -107,6 +112,8 @@ impl WriteHandle { self.save_to_db().await?; } + crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + Ok(()) } @@ -126,6 +133,8 @@ impl WriteHandle { self.save_to_db().await?; } + crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + Ok(()) } @@ -141,6 +150,9 @@ impl WriteHandle { // Supprimer du manager crate::manager::delete_playlist_internal(&self.playlist.id).await?; + // Notifier la suppression + crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + Ok(()) } @@ -156,6 +168,8 @@ impl WriteHandle { self.save_to_db().await?; } + crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + Ok(()) } @@ -175,6 +189,8 @@ impl WriteHandle { self.save_to_db().await?; } + crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + Ok(()) } @@ -194,6 +210,8 @@ impl WriteHandle { self.save_to_db().await?; } + crate::manager::PlaylistManager().notify_playlist_changed(&self.playlist.id); + Ok(()) } diff --git a/pmoplaylist/src/manager.rs b/pmoplaylist/src/manager.rs index d28678c6..109c8a32 100644 --- a/pmoplaylist/src/manager.rs +++ b/pmoplaylist/src/manager.rs @@ -8,9 +8,10 @@ use crate::Result; use once_cell::sync::OnceCell; use std::collections::HashMap; use std::path::PathBuf; -use std::sync::Arc; +use std::sync::{Arc, atomic::{AtomicU64, Ordering}}; use std::time::Duration; use tokio::sync::RwLock; +use std::sync::RwLock as StdRwLock; /// Singleton PlaylistManager static PLAYLIST_MANAGER: OnceCell = OnceCell::new(); @@ -22,6 +23,8 @@ static AUDIO_CACHE: OnceCell> = OnceCell::new(); struct ManagerInner { playlists: RwLock>>, persistence: Option>, + callbacks: StdRwLock>>, + cb_counter: AtomicU64, } /// Gestionnaire central de playlists @@ -47,6 +50,8 @@ impl PlaylistManager { inner: Arc::new(ManagerInner { playlists: RwLock::new(HashMap::new()), persistence: Some(persistence.clone()), + callbacks: StdRwLock::new(HashMap::new()), + cb_counter: AtomicU64::new(1), }), }; @@ -123,6 +128,33 @@ impl PlaylistManager { Ok(WriteHandle::new(playlist, write_token)) } + /// Enregistre un callback de modification de playlist. + /// + /// Retourne un jeton (u64) pour désenregistrer plus tard. + pub fn register_callback(&self, cb: F) -> u64 + where + F: Fn(&str) + Send + Sync + 'static, + { + let token = self.inner.cb_counter.fetch_add(1, Ordering::Relaxed); + let mut guard = self.inner.callbacks.write().unwrap(); + guard.insert(token, Arc::new(cb)); + token + } + + /// Désenregistre un callback via son jeton. + pub fn unregister_callback(&self, token: u64) { + let mut guard = self.inner.callbacks.write().unwrap(); + guard.remove(&token); + } + + /// Notifie tous les callbacks qu'une playlist a changé. + pub(crate) fn notify_playlist_changed(&self, id: &str) { + let guard = self.inner.callbacks.read().unwrap(); + for cb in guard.values() { + cb(id); + } + } + /// R�cup�re un write handle (cr�e �ph�m�re si n'existe pas) pub async fn get_write_handle(&self, id: String) -> Result { let mut playlists = self.inner.playlists.write().await;