diff --git a/Blackboard/Todo/metadata_RF_cache.md b/Blackboard/Todo/metadata_RF_cache.md new file mode 100644 index 00000000..bc39b4f4 --- /dev/null +++ b/Blackboard/Todo/metadata_RF_cache.md @@ -0,0 +1,22 @@ +**Tu réaliseras ce travail en appliquant scrupuleusement les règles définies dans [@Rules.md](file:///Users/coissac/Sync/maison/Petite_maisons/src/pmomusic/Blackboard/Rules.md)** + +**On ne travaille que dans la Crate PMORadioFrance** + +À partir de maintenant, tu ne prends plus en compte ce que tu pensais avant et tu écoutes bien. Et tu construis un plan d'implémentation que je dois valider. Tu arrêtes de prendre des initiatives et de faire des bêtises. + +- Tu as des fonctions d'interrogation de l'API Radio France qu'il faut utiliser au minimum. Mais Radio France nous donne des dates d'invalidation des métadonnées. Globalement, on doit gérer une grosse map où les valeurs ont des TTL. +- Quand un client demande une donnée du cache, si le TTL est atteint, il commence par utiliser l'API Radio France, modifie le cache, puis la retourne. Dans le cas contraire, il retourne directement la donnée. + +A chaque fois qu'il fait un appel de l'API Radio France pour modifier ses valeurs, Le cache émet un événement ou avertit ses abonnés, comme quoi les données d'un slug particulier ont été modifiées. Comme ça tout le monde peut se synchroniser. + +Les clients, par exemple la fonction Browse, n'interrogent que le cache qui a forcément des données à jour. Les métadonnées ne sont jamais stockées hors du cache, on se réfère toujours à elles. + +Nous devons maintenant considérer le fonctionnement du control point. Celui-ci est capable de s'abonner à une playlist pour suivre ses modifications. Il ne peut pas s'abonner à un item. + +Dans le cas d'une radio, on peut considérer que chaque canal, chaque slug, est en réalité une playlist à un item qu'il faut suivre. Ainsi, le Control Point peut décider de jouer cette playlist en s'abonnant à elle et être tenu au courant des modifications par des événements GENA. + +La source Radio France doit donc s'abonner aux événements du Cache. A chaque fois qu'un slug est modifié, elle avertit par un événement Jenna que la playlist à un item qui correspond à ce Slug est modifiée. + +Maintenant, il y a le cache des stations. Le cache des stations finalement il ne stock qu'un emboîtement de listes de slug. Ça, normalement, ça ne bouge quasiment pas. On peut dire que une fois par jour, on met à jour ce cache. Les listes de slug ont donc un TTL mais très long. + +A chaque browse, on reconstruit un document didl à partir des métadonnées à jour provenant du cache. diff --git a/Cargo.lock b/Cargo.lock index 197d6b5b..de470c04 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,7 +4,7 @@ version = 4 [[package]] name = "PMOMusic" -version = "0.3.11" +version = "0.3.13" dependencies = [ "axum 0.8.7", "console-subscriber", diff --git a/PMOMusic/Cargo.toml b/PMOMusic/Cargo.toml index e075335a..bf06997e 100644 --- a/PMOMusic/Cargo.toml +++ b/PMOMusic/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "PMOMusic" -version = "0.3.11" +version = "0.3.13" edition = "2024" [dependencies] diff --git a/pmomediaserver/src/sources.rs b/pmomediaserver/src/sources.rs index 30ef3f13..bd17eb99 100644 --- a/pmomediaserver/src/sources.rs +++ b/pmomediaserver/src/sources.rs @@ -270,13 +270,16 @@ impl SourcesExt for Server { }); let source = source.with_container_notifier(notifier); - // Enregistrer la source - self.register_music_source(Arc::new(source)).await; + // Enregistrer la source (Arc pour partage avec l'API) + let source_arc = Arc::new(source); + self.register_music_source(source_arc.clone()).await; - // Initialiser les routes API Radio France - self.init_radiofrance().await.map_err(|e| { - SourceInitError::RadioFranceError(format!("Failed to init API routes: {}", e)) - })?; + // Initialiser les routes API Radio France avec la source + self.init_radiofrance_with_source(source_arc) + .await + .map_err(|e| { + SourceInitError::RadioFranceError(format!("Failed to init API routes: {}", e)) + })?; tracing::info!("✅ Radio France source registered successfully"); diff --git a/pmoradiofrance/src/api_rest.rs b/pmoradiofrance/src/api_rest.rs index 95f6eefa..a37633e4 100644 --- a/pmoradiofrance/src/api_rest.rs +++ b/pmoradiofrance/src/api_rest.rs @@ -16,6 +16,7 @@ use axum::{ }; use futures::StreamExt; use serde_json; +use std::sync::Arc; // ============ Gestion des erreurs ============ @@ -65,6 +66,7 @@ async fn get_stations( State(state): State, ) -> Result, AppError> { let stations = state + .source .client .get_stations() .await @@ -81,6 +83,7 @@ async fn get_metadata( Path(slug): Path, ) -> Result, AppError> { let metadata = state + .source .client .get_live_metadata(&slug) .await @@ -95,8 +98,23 @@ async fn proxy_stream( State(state): State, Path(slug): Path, ) -> Result { + // Start metadata refresh when stream is accessed + #[cfg(feature = "logging")] + tracing::info!("Stream proxy accessed for station: {}", slug); + + // Spawn refresh task (non-blocking) + let source_clone = Arc::clone(&state.source); + let slug_clone = slug.clone(); + tokio::spawn(async move { + if let Err(e) = source_clone.start_metadata_refresh(&slug_clone).await { + #[cfg(feature = "logging")] + tracing::error!("Failed to start metadata refresh for {}: {}", slug_clone, e); + } + }); + // Get the stream URL let stream_url = state + .source .client .get_stream_url(&slug) .await @@ -116,12 +134,44 @@ async fn proxy_stream( headers.insert("content-type", "audio/aac".parse().unwrap()); headers.insert("cache-control", "no-cache".parse().unwrap()); - // Create streaming body + // Create streaming body with cleanup on disconnect let stream = response .bytes_stream() .map(|chunk| chunk.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))); - let body = Body::from_stream(stream); + // Wrap the stream to detect when client disconnects + let source_for_cleanup = Arc::clone(&state.source); + let slug_for_cleanup = slug.clone(); + let monitored_stream = + futures::stream::unfold((stream, false), move |(mut stream, mut done)| { + let source = source_for_cleanup.clone(); + let slug = slug_for_cleanup.clone(); + async move { + if done { + return None; + } + + match stream.next().await { + Some(Ok(chunk)) => Some((Ok(chunk), (stream, false))), + Some(Err(e)) => { + // Error occurred, stop refresh + #[cfg(feature = "logging")] + tracing::info!("Stream error for {}, stopping refresh", slug); + source.stop_metadata_refresh(&slug).await; + Some((Err(e), (stream, true))) + } + None => { + // Stream ended normally, stop refresh + #[cfg(feature = "logging")] + tracing::info!("Stream ended for {}, stopping refresh", slug); + source.stop_metadata_refresh(&slug).await; + None + } + } + } + }); + + let body = Body::from_stream(monitored_stream); Ok((headers, body).into_response()) } diff --git a/pmoradiofrance/src/pmoserver_ext.rs b/pmoradiofrance/src/pmoserver_ext.rs index aa1f7bd1..1eb7e71f 100644 --- a/pmoradiofrance/src/pmoserver_ext.rs +++ b/pmoradiofrance/src/pmoserver_ext.rs @@ -11,14 +11,12 @@ use crate::stateful_client::RadioFranceStatefulClient; /// État partagé pour les handlers Radio France #[derive(Clone)] pub struct RadioFranceState { - pub client: Arc, + pub source: Arc, } impl RadioFranceState { - pub fn new(client: RadioFranceStatefulClient) -> Self { - Self { - client: Arc::new(client), - } + pub fn new(source: Arc) -> Self { + Self { source } } } @@ -80,6 +78,16 @@ pub trait RadioFranceExt { /// server.init_radiofrance().await?; /// ``` async fn init_radiofrance(&mut self) -> Result>; + + /// Initialise l'extension Radio France avec une source existante + /// + /// Cette méthode est similaire à `init_radiofrance()` mais utilise une source + /// déjà créée et enregistrée, permettant de partager la même instance entre + /// le MediaServer UPnP et les routes API REST. + async fn init_radiofrance_with_source( + &mut self, + source: Arc, + ) -> Result>; } // L'implémentation du trait sera dans un module séparé (pmoserver_impl.rs) diff --git a/pmoradiofrance/src/pmoserver_impl.rs b/pmoradiofrance/src/pmoserver_impl.rs index 68017838..d4b1a174 100644 --- a/pmoradiofrance/src/pmoserver_impl.rs +++ b/pmoradiofrance/src/pmoserver_impl.rs @@ -38,14 +38,14 @@ impl RadioFranceExt for Server { async fn init_radiofrance(&mut self) -> Result> { info!("Initializing Radio France API..."); - // Créer le client stateful + // Créer une source dédiée pour l'API (sans enregistrement UPnP) let config = pmoconfig::get_config(); - let client = RadioFranceStatefulClient::new(config) + let source = crate::source::RadioFranceSource::new(config) .await - .map_err(|e| anyhow::anyhow!("Failed to create Radio France client: {}", e))?; + .map_err(|e| anyhow::anyhow!("Failed to create Radio France source: {}", e))?; - // Créer l'état partagé (RadioFranceState est Clone et contient déjà un Arc) - let state = RadioFranceState::new(client); + // Créer l'état partagé + let state = RadioFranceState::new(Arc::new(source)); // Créer et enregistrer le router let router = create_router(state.clone()); @@ -56,4 +56,23 @@ impl RadioFranceExt for Server { Ok(Arc::new(state)) } + + async fn init_radiofrance_with_source( + &mut self, + source: Arc, + ) -> Result> { + info!("Initializing Radio France API with existing source..."); + + // Créer l'état partagé (simplement une référence à la source) + let state = RadioFranceState::new(source); + + // Créer et enregistrer le router + let router = create_router(state.clone()); + self.add_router("/api/radiofrance", router).await; + + info!("Radio France API initialized with source"); + info!("API endpoints available at /api/radiofrance/*"); + + Ok(Arc::new(state)) + } } diff --git a/pmoradiofrance/src/source.rs b/pmoradiofrance/src/source.rs index b8426f7e..8a758a78 100644 --- a/pmoradiofrance/src/source.rs +++ b/pmoradiofrance/src/source.rs @@ -33,10 +33,7 @@ pub const RADIOFRANCE_DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/radiofran /// - Hierarchical organization (standalone, groups, local radios) pub struct RadioFranceSource { /// Stateful client with automatic caching - client: RadioFranceStatefulClient, - - /// Cache of playlists by station slug (volatile metadata) - playlists: Arc>>, + pub(crate) client: RadioFranceStatefulClient, /// Background tasks for metadata refresh refresh_handles: Arc>>>, @@ -83,7 +80,6 @@ impl RadioFranceSource { Ok(Self { client, - playlists: Arc::new(RwLock::new(HashMap::new())), refresh_handles: Arc::new(RwLock::new(HashMap::new())), #[cfg(feature = "cache")] cover_cache: None, @@ -151,7 +147,7 @@ impl RadioFranceSource { } /// Start metadata refresh task for a station - async fn start_metadata_refresh(&self, station_slug: &str) -> Result<()> { + pub async fn start_metadata_refresh(&self, station_slug: &str) -> Result<()> { let mut handles = self.refresh_handles.write().await; // If already running, do nothing @@ -160,69 +156,54 @@ impl RadioFranceSource { } let client = self.client.clone(); - let playlists = self.playlists.clone(); let slug = station_slug.to_string(); let update_id = self.update_id.clone(); let last_change = self.last_change.clone(); let container_notifier = self.container_notifier.clone(); - #[cfg(feature = "cache")] - let cover_cache = self.cover_cache.clone(); - - let server_base_url = self.server_base_url.clone(); - let handle = tokio::spawn(async move { loop { - match client.get_live_metadata(&slug).await { + // Force refresh metadata (bypass cache to get fresh data) + match client.refresh_live_metadata(&slug).await { Ok(metadata) => { let delay = std::time::Duration::from_millis(metadata.delay_to_refresh); - // Update the playlist metadata - #[cfg(feature = "cache")] + #[cfg(feature = "logging")] { - let mut pls = playlists.write().await; - if let Some(playlist) = pls.get_mut(&slug) { - let _: Result<()> = playlist - .update_metadata( - &metadata, - cover_cache.as_ref(), - server_base_url.as_deref(), - ) - .await; - - // Update change tracking - *update_id.write().await = update_id.read().await.wrapping_add(1); - *last_change.write().await = Some(SystemTime::now()); - - // Notify UPnP ContentDirectory of the change - if let Some(ref notifier) = container_notifier { - // Notify the station's stream item container - let container_id = format!("radiofrance:{}", slug); - notifier(&[container_id]); - } - } + let artist = metadata + .now + .song + .as_ref() + .and_then(|s| { + if s.interpreters.is_empty() { + None + } else { + Some(s.artists_display()) + } + }) + .unwrap_or_else(|| "".to_string()); + tracing::debug!( + "Refreshed metadata for {}: title='{}' artist='{}' delay={}ms", + slug, + metadata.now.first_line.title.as_deref().unwrap_or(""), + artist, + metadata.delay_to_refresh + ); } - #[cfg(not(feature = "cache"))] - { - let mut pls = playlists.write().await; - if let Some(playlist) = pls.get_mut(&slug) { - let _: Result<()> = playlist.update_metadata_no_cache( - &metadata, - server_base_url.as_deref(), - ); + // Update change tracking + *update_id.write().await = update_id.read().await.wrapping_add(1); + *last_change.write().await = Some(SystemTime::now()); - // Update change tracking - *update_id.write().await = update_id.read().await.wrapping_add(1); - *last_change.write().await = Some(SystemTime::now()); + // Notify UPnP ContentDirectory of the change + if let Some(ref notifier) = container_notifier { + // Notify the station's stream item container + let container_id = format!("radiofrance:{}", slug); - // Notify UPnP ContentDirectory of the change - if let Some(ref notifier) = container_notifier { - // Notify the station's stream item container - let container_id = format!("radiofrance:{}", slug); - notifier(&[container_id]); - } - } + #[cfg(feature = "logging")] + tracing::debug!("Notifying UPnP container update: {}", container_id); + + notifier(&[container_id]); } tokio::time::sleep(delay).await; @@ -246,7 +227,7 @@ impl RadioFranceSource { } /// Stop metadata refresh task for a station - async fn stop_metadata_refresh(&self, station_slug: &str) { + pub async fn stop_metadata_refresh(&self, station_slug: &str) { let mut handles = self.refresh_handles.write().await; if let Some(handle) = handles.remove(station_slug) { handle.abort(); @@ -394,25 +375,13 @@ impl RadioFranceSource { station.slug ); - let playlists = self.playlists.read().await; - - // If we already have this station in cache, use it - if let Some(existing) = playlists.get(&station.slug) { - #[cfg(feature = "logging")] - tracing::debug!("Using cached item for: {}", station.slug); - return Ok(existing.stream_item.clone()); - } - - // Release read lock before fetching metadata - drop(playlists); - - // Fetch metadata from API + // Fetch metadata from API (cached by RadioFranceStatefulClient) let metadata = self.client.get_live_metadata(&station.slug).await?; - // Create playlist with metadata + // Build item from live metadata (no caching here, rely on client cache) #[cfg(feature = "cache")] - let playlist = StationPlaylist::from_live_metadata( - station.clone(), + let item = StationPlaylist::build_item_from_metadata( + station, &metadata, self.cover_cache.as_ref(), self.server_base_url.as_deref(), @@ -420,17 +389,12 @@ impl RadioFranceSource { .await?; #[cfg(not(feature = "cache"))] - let playlist = StationPlaylist::from_live_metadata_no_cache( - station.clone(), + let item = StationPlaylist::build_item_from_metadata_sync( + station, &metadata, self.server_base_url.as_deref(), )?; - // Cache it - let mut playlists_write = self.playlists.write().await; - playlists_write.insert(station.slug.clone(), playlist.clone()); - drop(playlists_write); - // Note: We don't start metadata refresh here to avoid blocking during browse. // Refresh will be started in resolve_uri() when the stream is actually played. @@ -438,11 +402,11 @@ impl RadioFranceSource { tracing::debug!( "Built item for {}: {} resources, album_art: {:?}", station.slug, - playlist.stream_item.resources.len(), - playlist.stream_item.album_art.is_some() + item.resources.len(), + item.album_art.is_some() ); - Ok(playlist.stream_item) + Ok(item) } } @@ -535,22 +499,18 @@ impl MusicSource for RadioFranceSource { // Build items for this group only (main + webradios) let group_id = format!("radiofrance:group:{}", slug); - let mut main_item = self - .build_station_item(&group.main) - .await - .map_err(|e| MusicSourceError::BrowseError(e.to_string()))?; - - // Fix parent_id to point to the group container - main_item.parent_id = group_id.clone(); - let mut items = vec![main_item]; - + // Paralléliser les fetches pour éviter les timeouts + let mut futures = vec![self.build_station_item(&group.main)]; for webradio in &group.webradios { - let mut item = self - .build_station_item(webradio) - .await - .map_err(|e| MusicSourceError::BrowseError(e.to_string()))?; + futures.push(self.build_station_item(webradio)); + } - // Fix parent_id to point to the group container + let results = futures::future::join_all(futures).await; + + let mut items = Vec::new(); + for result in results { + let mut item = + result.map_err(|e| MusicSourceError::BrowseError(e.to_string()))?; item.parent_id = group_id.clone(); items.push(item); } diff --git a/version.txt b/version.txt index 20805912..e4737652 100644 --- a/version.txt +++ b/version.txt @@ -1 +1 @@ -0.3.11 +0.3.13