diff --git a/PMOMusic/src/main.rs b/PMOMusic/src/main.rs index 173adc2b..9919b170 100644 --- a/PMOMusic/src/main.rs +++ b/PMOMusic/src/main.rs @@ -81,6 +81,11 @@ async fn main() -> Result<(), Box> { .await .expect("Failed to register MediaServer"); + // Enregistrer l'instance ContentDirectory pour les notifications GENA + if let Some(cd_service) = server_instance.get_service("ContentDirectory") { + pmomediaserver::contentdirectory::state::register_instance(&cd_service); + } + // Initialiser les ProtocolInfo du MediaServer server_instance.init_protocol_info(); diff --git a/pmomediaserver/src/contentdirectory/mod.rs b/pmomediaserver/src/contentdirectory/mod.rs index 4f1054c3..ad8340de 100644 --- a/pmomediaserver/src/contentdirectory/mod.rs +++ b/pmomediaserver/src/contentdirectory/mod.rs @@ -77,12 +77,13 @@ use pmoupnp::define_service; pub mod actions; pub mod handlers; pub mod variables; +pub mod state; use actions::{BROWSE, GETSEARCHCAPABILITIES, GETSORTCAPABILITIES, GETSYSTEMUPDATEID, SEARCH}; use variables::{ A_ARG_TYPE_BROWSEFLAG, A_ARG_TYPE_COUNT, A_ARG_TYPE_FILTER, A_ARG_TYPE_INDEX, A_ARG_TYPE_OBJECTID, A_ARG_TYPE_RESULT, A_ARG_TYPE_SEARCHCRITERIA, A_ARG_TYPE_SORTCRITERIA, - A_ARG_TYPE_UPDATEID, SEARCHCAPABILITIES, SORTCAPABILITIES, SYSTEMUPDATEID, + A_ARG_TYPE_UPDATEID, SEARCHCAPABILITIES, SORTCAPABILITIES, SYSTEMUPDATEID, CONTAINERUPDATEIDS, }; // Service ContentDirectory:1 conforme à la spécification UPnP AV pour MediaServer @@ -102,6 +103,7 @@ define_service! { SEARCHCAPABILITIES, SORTCAPABILITIES, SYSTEMUPDATEID, + CONTAINERUPDATEIDS, ], actions: [ BROWSE, diff --git a/pmomediaserver/src/contentdirectory/state.rs b/pmomediaserver/src/contentdirectory/state.rs new file mode 100644 index 00000000..3f91f20c --- /dev/null +++ b/pmomediaserver/src/contentdirectory/state.rs @@ -0,0 +1,58 @@ +use once_cell::sync::OnceCell; +use pmoupnp::{services::ServiceInstance, variable_types::StateValue}; +use std::sync::{atomic::{AtomicU32, Ordering}, Arc, Weak, Mutex}; + +static CONTENTDIR_INSTANCE: OnceCell> = OnceCell::new(); +static SYSTEM_UPDATE_ID: AtomicU32 = AtomicU32::new(1); +static CONTAINER_UPDATE_IDS: Mutex = Mutex::new(String::new()); + +/// Enregistre l'instance ContentDirectory pour pouvoir pousser des notifications GENA. +pub fn register_instance(instance: &Arc) { + let _ = CONTENTDIR_INSTANCE.set(Arc::downgrade(instance)); + // Initialiser les valeurs + set_system_update_id(1); + set_container_update_ids(""); +} + +/// Notifie une mise à jour en incrémentant SystemUpdateID et ContainerUpdateIDs. +/// `container_ids` doit contenir les IDs des conteneurs impactés. +pub fn notify_containers_updated(container_ids: &[&str]) { + let new_id = SYSTEM_UPDATE_ID.fetch_add(1, Ordering::Relaxed).saturating_add(1); + set_system_update_id(new_id); + + if !container_ids.is_empty() { + let mut buf = String::new(); + for (idx, cid) in container_ids.iter().enumerate() { + if idx > 0 { + buf.push(','); + } + buf.push_str(cid); + buf.push(','); + buf.push_str(&new_id.to_string()); + } + set_container_update_ids(&buf); + } +} + +fn set_system_update_id(id: u32) { + tracing::info!("ContentDirectory: SystemUpdateID -> {}", id); + if let Some(service) = CONTENTDIR_INSTANCE.get().and_then(|w| w.upgrade()) { + if let Some(var) = service.get_variable("SystemUpdateID") { + let _ = var.set_value(StateValue::UI4(id)); + } + } +} + +fn set_container_update_ids(value: &str) { + tracing::info!("ContentDirectory: ContainerUpdateIDs -> {}", value); + { + let mut guard = CONTAINER_UPDATE_IDS.lock().unwrap(); + *guard = value.to_string(); + } + + if let Some(service) = CONTENTDIR_INSTANCE.get().and_then(|w| w.upgrade()) { + if let Some(var) = service.get_variable("ContainerUpdateIDs") { + let _ = var.set_value(StateValue::String(value.to_string())); + } + } +} diff --git a/pmomediaserver/src/sources.rs b/pmomediaserver/src/sources.rs index 73d9f8d8..89a1bb93 100644 --- a/pmomediaserver/src/sources.rs +++ b/pmomediaserver/src/sources.rs @@ -173,7 +173,8 @@ impl SourcesExt for Server { #[cfg(feature = "paradise")] async fn register_paradise(&mut self) -> Result<()> { - use pmoparadise::{RadioParadiseClient, RadioParadiseExt, RadioParadiseSource}; + use pmoparadise::{RadioParadiseExt, RadioParadiseSource}; + use crate::contentdirectory::state; tracing::info!("Initializing Radio Paradise source..."); @@ -181,7 +182,13 @@ impl SourcesExt for Server { let base_url = self.base_url(); // Créer la source Radio Paradise (utilise le singleton PlaylistManager) - let source = Arc::new(RadioParadiseSource::new(base_url.to_string())); + let notifier = Arc::new(|containers: &[String]| { + let refs: Vec<&str> = containers.iter().map(|s| s.as_str()).collect(); + state::notify_containers_updated(&refs); + }); + let source = Arc::new( + RadioParadiseSource::new(base_url.to_string()).with_container_notifier(notifier), + ); // Brancher les callbacks de playlists (live/history) pour signaler les updates source.attach_playlist_callbacks(); diff --git a/pmoparadise/src/source.rs b/pmoparadise/src/source.rs index 7b74d91e..29c4c19f 100644 --- a/pmoparadise/src/source.rs +++ b/pmoparadise/src/source.rs @@ -4,6 +4,7 @@ //! exposing live streams and historical playlists for all 4 channels. use crate::channels::{ChannelDescriptor, ALL_CHANNELS}; +use std::fmt; use pmosource::pmodidl::{Container, Item, Resource}; use pmosource::{ async_trait, AudioFormat, BrowseResult, MusicSource, MusicSourceError, Result, @@ -32,7 +33,7 @@ const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp"); /// - 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)] +#[derive(Clone)] pub struct RadioParadiseSource { /// Base URL for streaming server (e.g., "http://localhost:8080") base_url: String, @@ -42,6 +43,16 @@ pub struct RadioParadiseSource { last_change: Arc>, /// Tokens des callbacks enregistrés auprès du PlaylistManager callback_tokens: Arc>>, + /// Notifier optionnel pour signaler les mises à jour de conteneurs au ContentDirectory + container_notifier: Option>, +} + +impl fmt::Debug for RadioParadiseSource { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("RadioParadiseSource") + .field("base_url", &self.base_url) + .finish_non_exhaustive() + } } impl RadioParadiseSource { @@ -61,9 +72,19 @@ impl RadioParadiseSource { 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())), + container_notifier: None, } } + /// Injecte un notifier pour propager les changements de playlists vers le ContentDirectory + pub fn with_container_notifier( + mut self, + notifier: Arc, + ) -> Self { + self.container_notifier = Some(notifier); + self + } + /// Build a live stream URL for a channel fn build_live_url(&self, slug: &str) -> String { format!("{}/radioparadise/stream/{}/flac", self.base_url, slug) @@ -104,11 +125,35 @@ impl RadioParadiseSource { for pid in ids { let weak = Arc::downgrade(self); + let pid_clone = pid.clone(); let token = mgr.register_callback(move |changed_id| { + let pid = pid_clone.clone(); if changed_id == pid { if let Some(strong) = weak.upgrade() { tokio::spawn(async move { strong.bump_update_counter().await; + // Notifier ContentDirectory des conteneurs concernés + let containers: Vec = if pid.contains("history") { + // history playlist -> container history + ALL_CHANNELS + .iter() + .find(|ch| pid.ends_with(ch.slug)) + .map(|ch| vec![format!("radio-paradise:channel:{}:history", ch.slug)]) + .unwrap_or_default() + } else { + // live playlist -> container liveplaylist + ALL_CHANNELS + .iter() + .find(|ch| pid.ends_with(ch.slug)) + .map(|ch| vec![format!("radio-paradise:channel:{}:liveplaylist", ch.slug)]) + .unwrap_or_default() + }; + + if !containers.is_empty() { + if let Some(notifier) = strong.container_notifier.as_ref() { + notifier(&containers); + } + } }); } } diff --git a/pmoserver/src/server.rs b/pmoserver/src/server.rs index 73c3b033..3c72be15 100644 --- a/pmoserver/src/server.rs +++ b/pmoserver/src/server.rs @@ -17,7 +17,7 @@ use crate::logs::{LogState, init_logging, log_dump, log_sse}; use axum::extract::State; use axum::handler::Handler; use axum::response::Redirect; -use axum::routing::{get, post}; +use axum::routing::{any, get, post}; use axum::{Json, Router}; use axum_embed::ServeEmbed; use pmoconfig::get_config; @@ -240,6 +240,29 @@ impl Server { }; } + /// Ajoute un handler qui accepte tous les verbes HTTP (ANY) avec état + pub async fn add_any_handler_with_state( + &mut self, + path: &str, + handler: H, + state: S, + ) where + H: Handler + Clone + 'static, + T: 'static, + S: Clone + Send + Sync + 'static, + { + let route = Router::new() + .route("/", any(handler.clone())) + .with_state(state.clone()); + + let mut r = self.router.write().await; + *r = if path == "/" { + std::mem::take(&mut *r).merge(route) + } else { + std::mem::take(&mut *r).nest(path, route) + }; + } + /// Ajoute un répertoire statique pub async fn add_dir(&mut self, path: &str) where diff --git a/pmoupnp/src/services/service_instance.rs b/pmoupnp/src/services/service_instance.rs index 032c57fe..efc8db11 100644 --- a/pmoupnp/src/services/service_instance.rs +++ b/pmoupnp/src/services/service_instance.rs @@ -579,10 +579,10 @@ impl ServiceInstance { .add_post_handler_with_state(&self.control_route(), control_handler, instance_control) .await; - // Handler événements + // Handler événements (SUBSCRIBE/UNSUBSCRIBE sont des verbes spécifiques, pas GET) let instance_event = self.clone(); server - .add_handler_with_state(&self.event_route(), event_sub_handler, instance_event) + .add_any_handler_with_state(&self.event_route(), event_sub_handler, instance_event) .await; Ok(())