From a0c1ab4228d2ea2ce442bcf1b62a266aa971160d Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Fri, 23 Jan 2026 22:41:54 +0100 Subject: [PATCH] Bump version to 0.3.12 and enhance Radio France source integration Bump version from 0.3.11 to 0.3.12 - Update Cargo.toml and version.txt to 0.3.12 - Refactor Radio France source registration to share Arc with API - Enhance API routes to start/stop metadata refresh on stream access - Add cleanup logic for stream disconnection - Improve logging for metadata refresh operations - Update Radio France source methods to be public for external access --- Cargo.lock | 2 +- PMOMusic/Cargo.toml | 2 +- pmomediaserver/src/sources.rs | 15 +++-- pmoradiofrance/src/api_rest.rs | 60 ++++++++++++++++- pmoradiofrance/src/pmoserver_ext.rs | 17 +++++ pmoradiofrance/src/pmoserver_impl.rs | 25 +++++++ pmoradiofrance/src/source.rs | 99 +++++++++++++++++++++++++++- version.txt | 2 +- 8 files changed, 209 insertions(+), 13 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 197d6b5b..242c47ae 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,7 +4,7 @@ version = 4 [[package]] name = "PMOMusic" -version = "0.3.11" +version = "0.3.12" dependencies = [ "axum 0.8.7", "console-subscriber", diff --git a/PMOMusic/Cargo.toml b/PMOMusic/Cargo.toml index e075335a..081aece7 100644 --- a/PMOMusic/Cargo.toml +++ b/PMOMusic/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "PMOMusic" -version = "0.3.11" +version = "0.3.12" 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..3ef56a96 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 ============ @@ -95,6 +96,25 @@ 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); + + if let Some(ref source) = state.source { + // Spawn refresh task (non-blocking) + let source_clone = Arc::clone(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); + } + }); + } else { + #[cfg(feature = "logging")] + tracing::warn!("No source available to start metadata refresh"); + } + // Get the stream URL let stream_url = state .client @@ -116,12 +136,48 @@ 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 = state.source.clone(); + 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); + if let Some(src) = source { + src.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); + if let Some(src) = source { + src.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..bd46af26 100644 --- a/pmoradiofrance/src/pmoserver_ext.rs +++ b/pmoradiofrance/src/pmoserver_ext.rs @@ -12,14 +12,21 @@ use crate::stateful_client::RadioFranceStatefulClient; #[derive(Clone)] pub struct RadioFranceState { pub client: Arc, + pub source: Option>, } impl RadioFranceState { pub fn new(client: RadioFranceStatefulClient) -> Self { Self { client: Arc::new(client), + source: None, } } + + pub fn with_source(mut self, source: Arc) -> Self { + self.source = Some(source); + self + } } /// Trait pour étendre pmoserver avec les fonctionnalités Radio France @@ -80,6 +87,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..ae9bd586 100644 --- a/pmoradiofrance/src/pmoserver_impl.rs +++ b/pmoradiofrance/src/pmoserver_impl.rs @@ -56,4 +56,29 @@ 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 le client stateful + let config = pmoconfig::get_config(); + let client = RadioFranceStatefulClient::new(config) + .await + .map_err(|e| anyhow::anyhow!("Failed to create Radio France client: {}", e))?; + + // Créer l'état partagé avec la source + let state = RadioFranceState::new(client).with_source(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..8e7f90f1 100644 --- a/pmoradiofrance/src/source.rs +++ b/pmoradiofrance/src/source.rs @@ -151,7 +151,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 @@ -177,11 +177,51 @@ impl RadioFranceSource { Ok(metadata) => { let delay = std::time::Duration::from_millis(metadata.delay_to_refresh); + #[cfg(feature = "logging")] + { + 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 + ); + } + // Update the playlist metadata #[cfg(feature = "cache")] { let mut pls = playlists.write().await; + + #[cfg(feature = "logging")] + tracing::debug!( + "Looking for playlist '{}' in cache, found: {}", + slug, + pls.contains_key(&slug) + ); + if let Some(playlist) = pls.get_mut(&slug) { + let old_title = playlist.stream_item.title.clone(); + + #[cfg(feature = "logging")] + tracing::debug!( + "Updating playlist for {}: current title = '{}'", + slug, + old_title + ); + let _: Result<()> = playlist .update_metadata( &metadata, @@ -190,6 +230,18 @@ impl RadioFranceSource { ) .await; + let new_title = playlist.stream_item.title.clone(); + + #[cfg(feature = "logging")] + if old_title != new_title { + tracing::info!( + "Metadata updated for {}: {} -> {}", + slug, + old_title, + new_title + ); + } + // Update change tracking *update_id.write().await = update_id.read().await.wrapping_add(1); *last_change.write().await = Some(SystemTime::now()); @@ -198,6 +250,13 @@ impl RadioFranceSource { if let Some(ref notifier) = container_notifier { // Notify the station's stream item container let container_id = format!("radiofrance:{}", slug); + + #[cfg(feature = "logging")] + tracing::info!( + "Notifying UPnP container update: {}", + container_id + ); + notifier(&[container_id]); } } @@ -206,12 +265,41 @@ impl RadioFranceSource { #[cfg(not(feature = "cache"))] { let mut pls = playlists.write().await; + + #[cfg(feature = "logging")] + tracing::debug!( + "Looking for playlist '{}' in cache, found: {}", + slug, + pls.contains_key(&slug) + ); + if let Some(playlist) = pls.get_mut(&slug) { + let old_title = playlist.stream_item.title.clone(); + + #[cfg(feature = "logging")] + tracing::debug!( + "Updating playlist for {}: current title = '{}'", + slug, + old_title + ); + let _: Result<()> = playlist.update_metadata_no_cache( &metadata, server_base_url.as_deref(), ); + let new_title = playlist.stream_item.title.clone(); + + #[cfg(feature = "logging")] + if old_title != new_title { + tracing::info!( + "Metadata updated for {}: {} -> {}", + slug, + old_title, + new_title + ); + } + // Update change tracking *update_id.write().await = update_id.read().await.wrapping_add(1); *last_change.write().await = Some(SystemTime::now()); @@ -220,6 +308,13 @@ impl RadioFranceSource { if let Some(ref notifier) = container_notifier { // Notify the station's stream item container let container_id = format!("radiofrance:{}", slug); + + #[cfg(feature = "logging")] + tracing::info!( + "Notifying UPnP container update: {}", + container_id + ); + notifier(&[container_id]); } } @@ -246,7 +341,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(); diff --git a/version.txt b/version.txt index 20805912..0b9c0199 100644 --- a/version.txt +++ b/version.txt @@ -1 +1 @@ -0.3.11 +0.3.12