diff --git a/.DS_Store b/.DS_Store index ad016f38..d0612186 100644 Binary files a/.DS_Store and b/.DS_Store differ diff --git a/PMOMusic/src/main.rs b/PMOMusic/src/main.rs index 35c2713b..d0b999dd 100644 --- a/PMOMusic/src/main.rs +++ b/PMOMusic/src/main.rs @@ -1,6 +1,6 @@ use pmoapp::{WebAppExt, Webapp}; use pmomediarenderer::MEDIA_RENDERER; -use pmomediaserver::{MEDIA_SERVER, sources::SourcesExt, ParadiseStreamingExt}; +use pmomediaserver::{MEDIA_SERVER, ParadiseStreamingExt, sources::SourcesExt}; use pmoserver::Server; use pmosource::MusicSourceExt; use pmoupnp::UpnpServerExt; diff --git a/audio_cache/audio_cache.db b/audio_cache/audio_cache.db deleted file mode 100644 index 68743070..00000000 Binary files a/audio_cache/audio_cache.db and /dev/null differ diff --git a/pmoaudio-ext/src/sinks/broadcast_pacing.rs b/pmoaudio-ext/src/sinks/broadcast_pacing.rs index 4d8f9883..0daeb0de 100644 --- a/pmoaudio-ext/src/sinks/broadcast_pacing.rs +++ b/pmoaudio-ext/src/sinks/broadcast_pacing.rs @@ -20,6 +20,8 @@ pub struct BroadcastPacer { max_lead_time: f64, /// Label for logging (e.g., "FLAC" or "OGG") label: String, + /// Pending reset flag - will reset timer on next chunk + pending_reset: bool, } impl BroadcastPacer { @@ -34,15 +36,17 @@ impl BroadcastPacer { start_time: Instant::now(), max_lead_time: max_lead_time.max(0.0), label: label.into(), + pending_reset: false, } } /// Check timing and apply pacing /// /// This function: - /// 1. Detects TopZeroSync (audio_timestamp < 0.1 after >1s) and resets timer - /// 2. Drops frames that are late (audio_ts < elapsed) - /// 3. Sleeps if too far ahead (lead_time > max_lead_time) + /// 1. Detects TopZeroSync (audio_timestamp < 0.1 after >1s) and marks pending reset + /// 2. On next chunk, resets timer with elapsed=0 guarantee + /// 3. Drops frames that are late (audio_ts < elapsed) + /// 4. Sleeps if too far ahead (lead_time > max_lead_time) /// /// # Returns /// @@ -50,28 +54,43 @@ impl BroadcastPacer { /// - `Err(SkipFrame)` if frame is too late and should be dropped pub async fn check_and_pace(&mut self, audio_timestamp: f64) -> Result<(), SkipFrame> { // ╔═══════════════════════════════════════════════════════════════╗ - // ║ 1. DÉTECTION TopZeroSync ║ - // ║ Si le timestamp revient proche de 0, reset l'horloge ║ + // ║ 1. DÉTECTION timestamp proche de 0 → marquer reset ║ + // ║ Quand timestamp < 0.1s, c'est un nouveau morceau ║ // ╚═══════════════════════════════════════════════════════════════╝ - let elapsed_since_start = self.start_time.elapsed().as_secs_f64(); - if audio_timestamp < 0.1 && elapsed_since_start > 1.0 { - self.start_time = Instant::now(); + if audio_timestamp < 0.1 && !self.pending_reset { trace!( - "{} broadcaster: TopZeroSync detected, resetting timer", + "{} broadcaster: Timestamp near zero detected, will reset timer on next chunk", self.label ); + self.pending_reset = true; } // ╔═══════════════════════════════════════════════════════════════╗ - // ║ 2. CALCUL DU LEAD TIME ║ + // ║ 2. RESET TIMER si pending ║ + // ║ Le reset se fait AVANT le calcul d'elapsed pour garantir ║ + // ║ elapsed=0 pour le premier chunk du nouveau morceau ║ + // ╚═══════════════════════════════════════════════════════════════╝ + let elapsed = if self.pending_reset { + self.start_time = Instant::now(); + self.pending_reset = false; + trace!( + "{} broadcaster: Timer reset at audio_ts={:.3}s", + self.label, audio_timestamp + ); + 0.0 // Garantit elapsed=0 pour ce chunk + } else { + self.start_time.elapsed().as_secs_f64() + }; + + // ╔═══════════════════════════════════════════════════════════════╗ + // ║ 3. CALCUL DU LEAD TIME ║ // ║ lead_time > 0 : en avance (OK) ║ // ║ lead_time < 0 : en retard (SKIP) ║ // ╚═══════════════════════════════════════════════════════════════╝ - let elapsed = self.start_time.elapsed().as_secs_f64(); let lead_time = audio_timestamp - elapsed; // ╔═══════════════════════════════════════════════════════════════╗ - // ║ 3. DROP FRAMES EN RETARD (tolérance zéro) ║ + // ║ 4. DROP FRAMES EN RETARD ║ // ╚═══════════════════════════════════════════════════════════════╝ if lead_time < 0.0 { warn!( @@ -82,7 +101,7 @@ impl BroadcastPacer { } // ╔═══════════════════════════════════════════════════════════════╗ - // ║ 4. BACKPRESSURE NATURELLE - Pas de sleep ! ║ + // ║ 5. BACKPRESSURE NATURELLE - Pas de sleep ! ║ // ║ ║ // ║ Le pacing vient de : ║ // ║ - TimerBufferNode en amont (envoi régulier à 50ms/chunk) ║ diff --git a/pmoaudio-ext/src/sinks/timed_broadcast.rs b/pmoaudio-ext/src/sinks/timed_broadcast.rs index 2f06f25a..86d872f6 100644 --- a/pmoaudio-ext/src/sinks/timed_broadcast.rs +++ b/pmoaudio-ext/src/sinks/timed_broadcast.rs @@ -305,6 +305,10 @@ impl Sender { } /// Marque un TopZero : incrémente l'epoch pour les paquets suivants. + /// + /// Reset le timer epoch_start sans effacer le buffer. Les paquets + /// du morceau précédent continueront à être distribués naturellement. + /// Cela évite de perdre les dernières frames FLAC à la transition entre morceaux. pub fn mark_top_zero(&self) { let mut state = self .inner @@ -313,11 +317,8 @@ impl Sender { .expect("timed broadcast mutex poisoned"); state.epoch = state.epoch.wrapping_add(1); state.epoch_start = Instant::now(); - if !state.buffer.is_empty() { - state.head_seq = state.next_seq; - state.buffer.clear(); - self.inner.space_notify.notify_waiters(); - } + // Ne PAS effacer le buffer - laisser les paquets du morceau précédent + // se vider naturellement pour éviter de perdre les dernières frames } /// Nombre actuel de receivers abonnés. diff --git a/pmoaudio-ext/src/sources/playlist_source.rs b/pmoaudio-ext/src/sources/playlist_source.rs index 3c8f863b..27b801d5 100644 --- a/pmoaudio-ext/src/sources/playlist_source.rs +++ b/pmoaudio-ext/src/sources/playlist_source.rs @@ -212,7 +212,13 @@ impl NodeLogic for PlaylistSourceLogic { t }, Ok(None) => { - // Playlist vide, attendre avant retry + // Playlist vide, attendre avant retry et réinitialiser la synchro + if !first_track { + tracing::debug!( + "PlaylistSourceLogic: playlist drained, resetting top-zero sync" + ); + } + first_track = true; tracing::trace!( "PlaylistSourceLogic: playlist empty, waiting {}ms", self.poll_interval_ms @@ -237,14 +243,6 @@ impl NodeLogic for PlaylistSourceLogic { } }; - // Émettre TopZeroSync pour la première piste seulement - if first_track { - tracing::debug!("PlaylistSourceLogic: emitting TopZeroSync"); - let top_zero = AudioSegment::new_top_zero_sync(); - send_to_children!(top_zero); - first_track = false; - } - // Émettre TrackBoundary avec metadata du cache let metadata = match track.track_metadata() { Ok(m) => m, @@ -257,6 +255,28 @@ impl NodeLogic for PlaylistSourceLogic { } }; + let metadata_guard = metadata.read().await; + let artist = metadata_guard + .get_artist() + .await + .ok() + .flatten() + .unwrap_or_else(|| "Unknown artist".to_string()); + let title = metadata_guard + .get_title() + .await + .ok() + .flatten() + .unwrap_or_else(|| "Untitled".to_string()); + drop(metadata_guard); + let remaining = self.playlist_handle.remaining().await.unwrap_or(0); + tracing::info!( + "PlaylistSource: starting track {} - {} ({} remaining)", + artist, + title, + remaining + ); + tracing::debug!("PlaylistSourceLogic: emitting TrackBoundary"); let boundary = AudioSegment::new_track_boundary(0, 0.0, metadata); send_to_children!(boundary); @@ -278,6 +298,10 @@ impl NodeLogic for PlaylistSourceLogic { // Décoder et émettre les chunks PCM // Passer le cache et pk pour gérer le cache progressif let cache_pk = track.cache_pk(); + // Réinitialiser la synchro au début de chaque piste + let emit_top_zero = true; + first_track = false; + match decode_and_emit_track( &file_path, self.chunk_frames, @@ -285,10 +309,12 @@ impl NodeLogic for PlaylistSourceLogic { &stop_token, &self.cache, cache_pk, + emit_top_zero, ) .await { Ok(()) => { + tracing::info!("PlaylistSource: finished track {} - {}", artist, title); // Piste décodée avec succès, transférer vers l'historique si configuré if let Some(ref history) = self.history_playlist { if let Err(e) = history.push(cache_pk.to_string()).await { @@ -306,7 +332,8 @@ impl NodeLogic for PlaylistSourceLogic { } Err(e) => { tracing::error!("PlaylistSourceLogic: error decoding track: {}", e); - let error_marker = AudioSegment::new_error(0, 0.0, format!("Decode error: {}", e)); + let error_marker = + AudioSegment::new_error(0, 0.0, format!("Decode error: {}", e)); send_to_children!(error_marker); // Continue vers la piste suivante } @@ -335,6 +362,7 @@ async fn decode_and_emit_track( stop_token: &CancellationToken, cache: &Arc, cache_pk: &str, + emit_top_zero: bool, ) -> Result<(), AudioError> { // Attendre que le fichier soit suffisamment gros pour le sniffing // Le cache progressif permet de commencer la lecture après le prebuffer (512 KB) @@ -460,6 +488,16 @@ async fn decode_and_emit_track( timestamp_sec, )?; + if emit_top_zero && total_frames == 0 { + tracing::debug!("decode_and_emit_track: emitting TopZeroSync (first chunk)"); + let top_zero = AudioSegment::new_top_zero_sync(); + for tx in output { + tx.send(top_zero.clone()) + .await + .map_err(|_| AudioError::ChildDied)?; + } + } + for tx in output { tx.send(segment.clone()) .await @@ -493,6 +531,13 @@ async fn decode_and_emit_track( .await .map_err(|e| AudioError::ProcessingError(format!("Decode task failed: {}", e)))?; + if !cache.is_download_complete(cache_pk) { + tracing::warn!( + "PlaylistSource: finished reading cache entry {} but download is not complete", + cache_pk + ); + } + Ok(()) } diff --git a/pmoaudiocache/src/lib.rs b/pmoaudiocache/src/lib.rs index 61dfa085..c375b3b7 100755 --- a/pmoaudiocache/src/lib.rs +++ b/pmoaudiocache/src/lib.rs @@ -89,7 +89,10 @@ pub mod openapi; pub mod config_ext; // Re-exports principaux -pub use cache::{add_with_metadata_extraction, get_metadata, new_cache, AudioConfig, Cache}; +pub use cache::{ + add_with_metadata_extraction, get_metadata, new_cache, new_cache_with_consolidation, + AudioConfig, Cache, +}; pub use metadata::AudioMetadata; pub use metadata_ext::{AudioMetadataExt, AudioTrackMetadataExt}; pub use track_metadata::AudioCacheTrackMetadata; diff --git a/pmocache/src/cache.rs b/pmocache/src/cache.rs index 3833ff95..6ddc6b44 100755 --- a/pmocache/src/cache.rs +++ b/pmocache/src/cache.rs @@ -96,8 +96,11 @@ impl Cache { "File with pk {} in cache has no completion marker, will re-download/re-ingest", pk ); - // Supprimer le fichier incomplet + // Supprimer le fichier incomplet ET l'entrée DB let _ = std::fs::remove_file(&file_path); + if let Err(e) = self.db.delete(pk) { + tracing::warn!("Failed to delete DB entry for incomplete file {}: {}", pk, e); + } return Ok(false); } } @@ -141,7 +144,13 @@ impl Cache { /// Finalise l'ajout d'un fichier au cache /// /// Cette fonction helper gère le prébuffering et le nettoyage en background - async fn finalize_download(&self, pk: &str, download: Arc) -> Result { + async fn finalize_download( + &self, + pk: &str, + download: Arc, + collection: Option<&str>, + origin_url: Option<&str>, + ) -> Result { // Attendre le prébuffering (pour le cache progressif) if self.min_prebuffer_size > 0 { download @@ -155,6 +164,16 @@ impl Cache { ); } + // Ajouter à la DB une fois le prébuffer terminé + self.db.add(pk, None, collection)?; + if let Some(url) = origin_url { + self.db.set_origin_url(pk, url)?; + } + + if let Err(e) = self.enforce_limit().await { + tracing::warn!("Error enforcing cache limit: {}", e); + } + // Lancer une tâche de nettoyage et marquage de complétion en background let downloads_clone = self.downloads.clone(); let pk_clone = pk.to_string(); @@ -337,17 +356,9 @@ impl Cache { downloads.insert(pk.clone(), download.clone()); } - // Ajouter immédiatement à la DB - self.db.add(&pk, None, collection)?; - self.db.set_origin_url(&pk, url)?; - - // Appliquer la politique d'éviction LRU si nécessaire - if let Err(e) = self.enforce_limit().await { - tracing::warn!("Error enforcing cache limit: {}", e); - } - // Finaliser avec prébuffering et nettoyage - self.finalize_download(&pk, download).await + self.finalize_download(&pk, download, collection, Some(url)) + .await } /// Ajoute un fichier à partir d'un flux asynchrone. @@ -459,17 +470,9 @@ impl Cache { downloads.insert(pk.clone(), download.clone()); } - self.db.add(&pk, None, collection)?; - if let Some(uri) = source_uri { - self.db.set_origin_url(&pk, uri)?; - } - - if let Err(e) = self.enforce_limit().await { - tracing::warn!("Error enforcing cache limit: {}", e); - } - // Finaliser avec prébuffering et nettoyage - self.finalize_download(&pk, download).await + self.finalize_download(&pk, download, collection, source_uri) + .await } /// Ajoute un fichier local au cache diff --git a/pmocovers/src/cache.rs b/pmocovers/src/cache.rs index 1cf5466d..c0b4c898 100644 --- a/pmocovers/src/cache.rs +++ b/pmocovers/src/cache.rs @@ -77,3 +77,17 @@ pub fn new_cache(dir: &str, limit: usize) -> Result { let transformer_factory = Arc::new(|| create_webp_transformer()); Cache::with_transformer(dir, limit, Some(transformer_factory)) } + +/// Crée un cache de couvertures et lance une consolidation en arrière-plan. +pub async fn new_cache_with_consolidation(dir: &str, limit: usize) -> Result> { + let cache = Arc::new(new_cache(dir, limit)?); + let cache_clone = cache.clone(); + tokio::spawn(async move { + if let Err(e) = cache_clone.consolidate().await { + tracing::warn!("Failed to consolidate cover cache on startup: {}", e); + } else { + tracing::info!("Cover cache consolidated successfully on startup"); + } + }); + Ok(cache) +} diff --git a/pmocovers/src/lib.rs b/pmocovers/src/lib.rs index bc283057..2d4aec1b 100644 --- a/pmocovers/src/lib.rs +++ b/pmocovers/src/lib.rs @@ -60,7 +60,7 @@ pub mod openapi; #[cfg(feature = "pmoconfig")] pub mod config_ext; -pub use cache::{new_cache, Cache, CoversConfig}; +pub use cache::{new_cache, new_cache_with_consolidation, Cache, CoversConfig}; #[cfg(feature = "pmoserver")] pub use openapi::ApiDoc; diff --git a/pmomediaserver/src/paradise_streaming.rs b/pmomediaserver/src/paradise_streaming.rs index 823c40ae..d3ae1349 100644 --- a/pmomediaserver/src/paradise_streaming.rs +++ b/pmomediaserver/src/paradise_streaming.rs @@ -6,16 +6,16 @@ use anyhow::{Context, Result}; use async_trait::async_trait; use axum::{ + Json, Router, body::Body, extract::{Path, State}, http::StatusCode, response::{IntoResponse, Response}, routing::get, - Json, Router, }; -use pmoaudiocache::{get_audio_cache, register_audio_cache, AudioCacheExt, Cache as AudioCache}; -use pmocovers::{get_cover_cache, register_cover_cache, Cache as CoverCache, CoverCacheExt}; -use pmoparadise::{channels::ALL_CHANNELS, ParadiseChannelManager, ParadiseHistoryBuilder}; +use pmoaudiocache::{AudioCacheExt, Cache as AudioCache, get_audio_cache, register_audio_cache}; +use pmocovers::{Cache as CoverCache, CoverCacheExt, get_cover_cache, register_cover_cache}; +use pmoparadise::{ParadiseChannelManager, ParadiseHistoryBuilder, channels::ALL_CHANNELS}; use pmoplaylist::register_audio_cache as register_playlist_audio_cache; use std::sync::Arc; use tokio_util::io::ReaderStream; @@ -99,15 +99,12 @@ impl ParadiseStreamingExt for pmoserver::Server { }; // Créer le builder d'historique - let history_builder = ParadiseHistoryBuilder { - audio_cache: audio_cache.clone(), - cover_cache: cover_cache.clone(), - playlist_prefix: "radioparadise-history".into(), - playlist_title_prefix: Some("Radio Paradise History".into()), - max_history_tracks: Some(500), - collection_prefix: Some("radioparadise".into()), - replay_max_lead_seconds: 1.0, - }; + let mut history_builder = ParadiseHistoryBuilder::default(); + history_builder.playlist_prefix = "radioparadise-history".into(); + history_builder.playlist_title_prefix = Some("Radio Paradise History".into()); + history_builder.max_history_tracks = Some(500); + history_builder.collection_prefix = Some("radioparadise".into()); + history_builder.replay_max_lead_seconds = 1.0; // Créer le manager de canaux info!("📡 Creating ParadiseChannelManager..."); diff --git a/pmomediaserver/src/sources_api.rs b/pmomediaserver/src/sources_api.rs index 6b4d56e1..7034056c 100644 --- a/pmomediaserver/src/sources_api.rs +++ b/pmomediaserver/src/sources_api.rs @@ -141,7 +141,9 @@ async fn register_paradise(Json(params): Json) -> impl IntoRespo use pmosource::api::register_source; // Utiliser l'URL de base depuis les params ou une valeur par défaut - let base_url = params.base_url.unwrap_or_else(|| "http://localhost:8080".to_string()); + let base_url = params + .base_url + .unwrap_or_else(|| "http://localhost:8080".to_string()); // Créer la source Radio Paradise (utilise le singleton PlaylistManager) let source = Arc::new(RadioParadiseSource::new(base_url)); diff --git a/pmoparadise/Cargo.toml b/pmoparadise/Cargo.toml index 98fbe2d9..1a1b93fe 100644 --- a/pmoparadise/Cargo.toml +++ b/pmoparadise/Cargo.toml @@ -111,3 +111,8 @@ path = "examples/now_playing.rs" name = "stream_block" path = "examples/stream_block.rs" required-features = ["full"] + +[[example]] +name = "single_channel_server" +path = "examples/single_channel_server.rs" +required-features = ["full"] diff --git a/pmoparadise/examples/play_and_cache.rs b/pmoparadise/examples/play_and_cache.rs index 862d8446..2a1ca627 100644 --- a/pmoparadise/examples/play_and_cache.rs +++ b/pmoparadise/examples/play_and_cache.rs @@ -28,9 +28,13 @@ use pmoaudio::{AudioPipelineNode, AudioSink, TimerNode}; use pmoaudio_ext::{FlacCacheSink, PlaylistSource}; -use pmoaudiocache::Cache as AudioCache; -use pmocovers::Cache as CoverCache; +use pmoaudiocache::{ + new_cache_with_consolidation as new_audio_cache, + register_audio_cache as register_global_audio_cache, +}; +use pmocovers::{new_cache_with_consolidation as new_cover_cache, register_cover_cache}; use pmoparadise::{RadioParadiseClient, RadioParadiseStreamSource}; +use pmoplaylist::register_audio_cache as register_playlist_audio_cache; use std::env; use std::sync::Arc; use tokio_util::sync::CancellationToken; @@ -98,24 +102,20 @@ async fn main() -> Result<(), Box> { // Créer le cache audio let audio_cache_dir = format!("{}/audio_cache", base_dir); std::fs::create_dir_all(&audio_cache_dir)?; - let audio_cache = Arc::new(AudioCache::new( - &audio_cache_dir, - 1000, // 1000 MB limit - )?); + let audio_cache = new_audio_cache(&audio_cache_dir, 1000).await?; tracing::debug!("Audio cache initialized at: {}", audio_cache_dir); // Créer le cache de covers let cover_cache_dir = format!("{}/cover_cache", base_dir); std::fs::create_dir_all(&cover_cache_dir)?; - let cover_cache = Arc::new(CoverCache::new( - &cover_cache_dir, - 100, // 100 MB limit - )?); + let cover_cache = new_cover_cache(&cover_cache_dir, 100).await?; tracing::debug!("Cover cache initialized at: {}", cover_cache_dir); // Enregistrer le cache audio dans pmoplaylist // (requis par pmoplaylist pour valider les pks) - pmoplaylist::register_audio_cache(audio_cache.clone()); + register_global_audio_cache(audio_cache.clone()); + register_playlist_audio_cache(audio_cache.clone()); + register_cover_cache(cover_cache.clone()); tracing::debug!("Audio cache registered in pmoplaylist"); // Utiliser le gestionnaire de playlist singleton diff --git a/pmoparadise/examples/serve_channels.rs b/pmoparadise/examples/serve_channels.rs index 9081c1c5..b39e18bd 100644 --- a/pmoparadise/examples/serve_channels.rs +++ b/pmoparadise/examples/serve_channels.rs @@ -18,10 +18,13 @@ use axum::{ routing::get, Json, Router, }; -use pmoaudiocache::new_cache as new_audio_cache; -use pmocovers::new_cache as new_cover_cache; +use pmoaudiocache::{ + new_cache_with_consolidation as new_audio_cache, + register_audio_cache as register_global_audio_cache, +}; +use pmocovers::{new_cache_with_consolidation as new_cover_cache, register_cover_cache}; use pmoparadise::{channels::ALL_CHANNELS, ParadiseChannelManager, ParadiseHistoryBuilder}; -use pmoplaylist::register_audio_cache; +use pmoplaylist::register_audio_cache as register_playlist_audio_cache; use pmoserver::{init_logging, ServerBuilder}; use tokio_util::io::ReaderStream; use tracing::{error, info}; @@ -41,9 +44,11 @@ async fn main() -> anyhow::Result<()> { fs::create_dir_all(cover_cache_dir)?; fs::create_dir_all(audio_cache_dir)?; - let cover_cache = Arc::new(new_cover_cache(cover_cache_dir, 500)?); - let audio_cache = Arc::new(new_audio_cache(audio_cache_dir, 1000)?); - register_audio_cache(audio_cache.clone()); + let cover_cache = new_cover_cache(cover_cache_dir, 500).await?; + let audio_cache = new_audio_cache(audio_cache_dir, 1000).await?; + register_global_audio_cache(audio_cache.clone()); + register_playlist_audio_cache(audio_cache.clone()); + register_cover_cache(cover_cache.clone()); let _playlist_manager = pmoplaylist::PlaylistManager(); let history_builder = ParadiseHistoryBuilder { diff --git a/pmoparadise/examples/single_channel_server.rs b/pmoparadise/examples/single_channel_server.rs new file mode 100644 index 00000000..dd975bec --- /dev/null +++ b/pmoparadise/examples/single_channel_server.rs @@ -0,0 +1,122 @@ +//! Simple web server that exposes one Radio Paradise channel over HTTP. +//! +//! Usage: +//! ```bash +//! cargo run --example single_channel_server --features full -- main +//! ``` +//! Valid arguments are either the slug (`main`, `mellow`, `rock`, `eclectic`) or +//! the numeric channel id (`0`..`3`). When no argument is provided, the example +//! defaults to the “main” mix. + +use axum::{ + body::Body, extract::State, http::StatusCode, response::Response, routing::get, Router, +}; +use pmoaudiocache::{ + new_cache_with_consolidation as new_audio_cache, + register_audio_cache as register_global_audio_cache, +}; +use pmocovers::{new_cache_with_consolidation as new_cover_cache, register_cover_cache}; +use pmoparadise::{ + channels::{ChannelDescriptor, ALL_CHANNELS}, + ParadiseHistoryBuilder, ParadiseStreamChannel, ParadiseStreamChannelConfig, +}; +use pmoplaylist::register_audio_cache as register_playlist_audio_cache; +use std::{fs, net::SocketAddr, sync::Arc}; +use tokio::net::TcpListener; +use tokio_util::io::ReaderStream; +use tracing::info; + +#[derive(Clone)] +struct AppState { + channel: Arc, + descriptor: ChannelDescriptor, +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt().with_env_filter("info").init(); + + let descriptor = pick_descriptor(std::env::args().nth(1))?; + info!( + "Selected Radio Paradise channel: {} ({})", + descriptor.display_name, descriptor.slug + ); + + // Prepare caches under ./cache/single-channel + let cache_root = "./cache/single-channel"; + let audio_cache_dir = format!("{}/audio", cache_root); + let cover_cache_dir = format!("{}/covers", cache_root); + fs::create_dir_all(&audio_cache_dir)?; + fs::create_dir_all(&cover_cache_dir)?; + + let audio_cache = new_audio_cache(&audio_cache_dir, 1000).await?; + let cover_cache = new_cover_cache(&cover_cache_dir, 200).await?; + register_global_audio_cache(audio_cache.clone()); + register_playlist_audio_cache(audio_cache.clone()); + register_cover_cache(cover_cache.clone()); + + let mut history_builder = ParadiseHistoryBuilder::new(audio_cache.clone(), cover_cache.clone()); + history_builder.playlist_prefix = format!("single-channel-history-{}", descriptor.slug); + history_builder.collection_prefix = Some(format!("single-channel-{}", descriptor.slug)); + let history_opts = history_builder.build_for_channel(&descriptor).await?; + + let channel = Arc::new( + ParadiseStreamChannel::new( + descriptor, + ParadiseStreamChannelConfig::default(), + Some(cover_cache), + Some(history_opts), + ) + .await?, + ); + + let state = AppState { + channel, + descriptor, + }; + + let app = Router::new() + .route("/stream/flac", get(stream_flac)) + .with_state(state); + + let addr: SocketAddr = ([0, 0, 0, 0], 8080).into(); + info!("HTTP server listening on http://{addr}/stream/flac"); + info!("Connect with a FLAC player (e.g. ffplay http://localhost:8080/stream/flac)"); + + let listener = TcpListener::bind(addr).await?; + axum::serve(listener, app.into_make_service()).await?; + + Ok(()) +} + +async fn stream_flac(State(state): State) -> Result { + let stream = state.channel.subscribe_flac(); + let body = Body::from_stream(ReaderStream::new(stream)); + Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/flac") + .header( + "X-PMO-Channel", + format!( + "{} ({})", + state.descriptor.display_name, state.descriptor.slug + ), + ) + .body(body) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR) +} + +fn pick_descriptor(arg: Option) -> anyhow::Result { + if let Some(token) = arg { + if let Some(desc) = ALL_CHANNELS.iter().find(|c| c.slug == token) { + return Ok(*desc); + } + if let Ok(id) = token.parse::() { + if let Some(desc) = ALL_CHANNELS.iter().find(|c| c.id == id) { + return Ok(*desc); + } + } + anyhow::bail!("Unknown channel identifier: {token}"); + } + Ok(ALL_CHANNELS[0]) +} diff --git a/pmoparadise/src/models.rs b/pmoparadise/src/models.rs index 621ee64b..cdabea3b 100644 --- a/pmoparadise/src/models.rs +++ b/pmoparadise/src/models.rs @@ -247,6 +247,10 @@ pub struct Block { #[serde(default)] pub image_base: Option, + /// Scheduled start time for this block (Unix timestamp in milliseconds, UTC) + #[serde(default)] + pub sched_time_millis: Option, + /// Map of song index (as string) to Song metadata /// Keys are "0", "1", "2", etc. #[serde(default)] @@ -258,6 +262,16 @@ pub struct Block { } impl Block { + /// Scheduled start time in milliseconds if available. + pub fn start_time_millis(&self) -> Option { + if let Some(ts) = self.sched_time_millis { + return Some(ts); + } + self.songs_ordered() + .into_iter() + .find_map(|(_, song)| song.sched_time_millis) + } + /// Get songs in order by index pub fn songs_ordered(&self) -> Vec<(usize, &Song)> { let mut songs: Vec<_> = self diff --git a/pmoparadise/src/playlist_feeder.rs b/pmoparadise/src/playlist_feeder.rs index f02e371e..6b576d0e 100644 --- a/pmoparadise/src/playlist_feeder.rs +++ b/pmoparadise/src/playlist_feeder.rs @@ -3,19 +3,94 @@ //! Architecture simplifiée utilisant les URLs gapless individuelles au lieu du bloc FLAC entier. use crate::{client::RadioParadiseClient, models::EventId}; +use anyhow::Result; use pmoaudiocache::Cache as AudioCache; use pmocovers::Cache as CoversCache; use pmoplaylist::{PlaylistManager, ReadHandle, WriteHandle}; use std::{ - collections::VecDeque, + collections::{HashMap, VecDeque}, sync::Arc, time::{Duration, SystemTime, UNIX_EPOCH}, }; use tokio::sync::Notify; -use anyhow::Result; /// Signal de fin de blocs pub const END_OF_BLOCKS_SIGNAL: EventId = EventId::MAX; +const RECENT_BLOCKS_CACHE_SIZE: usize = 10; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum BlockStatus { + Pending, + InProgress, + Done, +} + +struct RecentBlocks { + states: HashMap, + order: VecDeque, + capacity: usize, +} + +impl RecentBlocks { + fn new(capacity: usize) -> Self { + Self { + states: HashMap::new(), + order: VecDeque::new(), + capacity, + } + } + + fn try_enqueue(&mut self, event_id: EventId) -> bool { + match self.states.get(&event_id) { + Some(_) => false, + None => { + self.order.push_back(event_id); + self.states.insert(event_id, BlockStatus::Pending); + self.evict_old_done(); + true + } + } + } + + fn mark_in_progress(&mut self, event_id: EventId) { + if let Some(state) = self.states.get_mut(&event_id) { + *state = BlockStatus::InProgress; + } else { + self.order.push_back(event_id); + self.states.insert(event_id, BlockStatus::InProgress); + } + self.evict_old_done(); + } + + fn mark_done(&mut self, event_id: EventId) { + if let Some(state) = self.states.get_mut(&event_id) { + *state = BlockStatus::Done; + } else { + self.order.push_back(event_id); + self.states.insert(event_id, BlockStatus::Done); + } + self.evict_old_done(); + } + + fn purge(&mut self, event_id: EventId) { + self.states.remove(&event_id); + } + + fn evict_old_done(&mut self) { + while self.order.len() > self.capacity { + let Some(front) = self.order.front().copied() else { + break; + }; + match self.states.get(&front) { + Some(BlockStatus::Done) | None => { + self.order.pop_front(); + self.states.remove(&front); + } + Some(_) => break, + } + } + } +} /// Feeder qui télécharge les blocs RP et alimente une playlist pub struct RadioParadisePlaylistFeeder { @@ -26,6 +101,7 @@ pub struct RadioParadisePlaylistFeeder { block_queue: Arc>>, notify: Arc, collection: Option, + recent_blocks: tokio::sync::Mutex, } impl RadioParadisePlaylistFeeder { @@ -38,7 +114,9 @@ impl RadioParadisePlaylistFeeder { collection: Option, ) -> Result<(Self, ReadHandle)> { let manager = PlaylistManager::get(); - let write_handle = manager.create_persistent_playlist(playlist_id.clone()).await?; + let write_handle = manager + .create_persistent_playlist(playlist_id.clone()) + .await?; let read_handle = manager.get_read_handle(&playlist_id).await?; Ok(( @@ -50,6 +128,7 @@ impl RadioParadisePlaylistFeeder { block_queue: Arc::new(tokio::sync::Mutex::new(VecDeque::new())), notify: Arc::new(Notify::new()), collection, + recent_blocks: tokio::sync::Mutex::new(RecentBlocks::new(RECENT_BLOCKS_CACHE_SIZE)), }, read_handle, )) @@ -57,6 +136,17 @@ impl RadioParadisePlaylistFeeder { /// Enqueue un bloc pour traitement pub async fn push_block_id(&self, event_id: EventId) { + { + let mut recent = self.recent_blocks.lock().await; + if !recent.try_enqueue(event_id) { + tracing::debug!( + "RadioParadisePlaylistFeeder: Ignoring duplicate enqueue for block {}", + event_id + ); + return; + } + } + { let mut queue = self.block_queue.lock().await; queue.push_back(event_id); @@ -64,6 +154,21 @@ impl RadioParadisePlaylistFeeder { self.notify.notify_one(); } + async fn mark_in_progress(&self, event_id: EventId) { + let mut recent = self.recent_blocks.lock().await; + recent.mark_in_progress(event_id); + } + + async fn mark_done(&self, event_id: EventId) { + let mut recent = self.recent_blocks.lock().await; + recent.mark_done(event_id); + } + + async fn purge_block_state(&self, event_id: EventId) { + let mut recent = self.recent_blocks.lock().await; + recent.purge(event_id); + } + /// Boucle principale de traitement (à exécuter dans une tâche tokio) pub async fn run(self: Arc) -> Result<()> { loop { @@ -73,7 +178,9 @@ impl RadioParadisePlaylistFeeder { let mut queue = self.block_queue.lock().await; if let Some(id) = queue.pop_front() { if id == END_OF_BLOCKS_SIGNAL { - tracing::info!("RadioParadisePlaylistFeeder: END_OF_BLOCKS_SIGNAL received"); + tracing::info!( + "RadioParadisePlaylistFeeder: END_OF_BLOCKS_SIGNAL received" + ); return Ok(()); } break id; @@ -82,9 +189,22 @@ impl RadioParadisePlaylistFeeder { self.notify.notified().await; }; + self.mark_in_progress(event_id).await; + // Traiter le bloc if let Err(e) = self.process_block(event_id).await { - tracing::error!("RadioParadisePlaylistFeeder: Failed to process block {}: {}", event_id, e); + tracing::error!( + "RadioParadisePlaylistFeeder: Failed to process block {}: {}", + event_id, + e + ); + self.purge_block_state(event_id).await; + tracing::debug!( + "RadioParadisePlaylistFeeder: Cleared block {} state after error", + event_id + ); + } else { + self.mark_done(event_id).await; } } } @@ -97,9 +217,7 @@ impl RadioParadisePlaylistFeeder { let block = self.client.get_block(Some(event_id)).await?; // 2. Timestamp actuel - let now_ms = SystemTime::now() - .duration_since(UNIX_EPOCH)? - .as_millis() as u64; + let now_ms = SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() as u64; // 3. Filtrer les chansons encore en lecture ou à venir let songs = block.songs_ordered(); @@ -109,21 +227,28 @@ impl RadioParadisePlaylistFeeder { if !song.is_still_playing(now_ms) { tracing::debug!( "RadioParadisePlaylistFeeder: Skipping finished song {} - {} (ended at {})", - idx, song.title, song.sched_end_time_ms().unwrap_or(0) + idx, + song.title, + song.sched_end_time_ms().unwrap_or(0) ); continue; } // 4. Télécharger la chanson - let gapless_url = song.gapless_url.as_ref() + let gapless_url = song + .gapless_url + .as_ref() .ok_or_else(|| anyhow::anyhow!("Missing gapless_url for song {}", idx))?; tracing::info!( "RadioParadisePlaylistFeeder: Downloading song {} - {} by {}", - idx, song.title, song.artist + idx, + song.title, + song.artist ); - let pk = self.audio_cache + let pk = self + .audio_cache .add_from_url(gapless_url, self.collection.as_deref()) .await?; @@ -131,7 +256,8 @@ impl RadioParadisePlaylistFeeder { self.save_metadata(&pk, song, &block).await?; // 6. Calculer le TTL - let sched_end = song.sched_end_time_ms() + let sched_end = song + .sched_end_time_ms() .ok_or_else(|| anyhow::anyhow!("Cannot calculate TTL without sched_time_millis"))?; let ttl_ms = sched_end.saturating_sub(now_ms); let ttl = Duration::from_millis(ttl_ms); @@ -141,7 +267,9 @@ impl RadioParadisePlaylistFeeder { tracing::info!( "RadioParadisePlaylistFeeder: Added {} to playlist (pk={}, ttl={}s)", - song.title, pk, ttl.as_secs() + song.title, + pk, + ttl.as_secs() ); processed += 1; @@ -149,7 +277,8 @@ impl RadioParadisePlaylistFeeder { tracing::info!( "RadioParadisePlaylistFeeder: Processed block {} - added {} songs to playlist", - event_id, processed + event_id, + processed ); Ok(()) @@ -183,10 +312,17 @@ impl RadioParadisePlaylistFeeder { meta.set_cover_url(Some(cover_url.clone())).await?; // Télécharger la cover - match self.covers_cache.add_from_url(&cover_url, self.collection.as_deref()).await { + match self + .covers_cache + .add_from_url(&cover_url, self.collection.as_deref()) + .await + { Ok(cover_pk) => { meta.set_cover_pk(Some(cover_pk)).await?; - tracing::debug!("RadioParadisePlaylistFeeder: Cached cover for {}", song.title); + tracing::debug!( + "RadioParadisePlaylistFeeder: Cached cover for {}", + song.title + ); } Err(e) => { tracing::warn!("RadioParadisePlaylistFeeder: Failed to cache cover: {}", e); diff --git a/pmoparadise/src/source.rs b/pmoparadise/src/source.rs index 40cfd434..3524a7e8 100644 --- a/pmoparadise/src/source.rs +++ b/pmoparadise/src/source.rs @@ -170,15 +170,9 @@ impl RadioParadiseSource { // Get read handle for the playlist from the singleton let manager = pmoplaylist::PlaylistManager(); - let reader = manager - .get_read_handle(&playlist_id) - .await - .map_err(|e| { - MusicSourceError::BrowseError(format!( - "Failed to get playlist {}: {}", - playlist_id, e - )) - })?; + let reader = manager.get_read_handle(&playlist_id).await.map_err(|e| { + MusicSourceError::BrowseError(format!("Failed to get playlist {}: {}", playlist_id, e)) + })?; // Get entries from playlist let entries = reader.get_entries(offset, count).await.map_err(|e| { @@ -206,18 +200,17 @@ impl RadioParadiseSource { let metadata = &entry.metadata; // Build audio URL from cache - let audio_url = format!( - "{}/cache/audio/{}", - self.base_url, - entry.pk - ); + let audio_url = format!("{}/cache/audio/{}", self.base_url, entry.pk); // Build item Ok(Item { id: format!("radio-paradise:channel:{}:history:track:{}", slug, entry.pk), parent_id: format!("radio-paradise:channel:{}:history", slug), restricted: Some("1".to_string()), - title: metadata.title.clone().unwrap_or_else(|| "Unknown Title".to_string()), + title: metadata + .title + .clone() + .unwrap_or_else(|| "Unknown Title".to_string()), creator: metadata.artist.clone(), class: "object.item.audioItem.musicTrack".to_string(), artist: metadata.artist.clone(), @@ -232,7 +225,9 @@ impl RadioParadiseSource { bits_per_sample: metadata.bits_per_sample.map(|b| b.to_string()), sample_frequency: metadata.sample_rate.map(|s| s.to_string()), nr_audio_channels: Some("2".to_string()), - duration: metadata.duration.map(|d| format!("{}:{:02}:{:02}", d / 3600, (d % 3600) / 60, d % 60)), + duration: metadata + .duration + .map(|d| format!("{}:{:02}:{:02}", d / 3600, (d % 3600) / 60, d % 60)), url: audio_url, }], descriptions: vec![], diff --git a/pmoparadise/src/stream_channel.rs b/pmoparadise/src/stream_channel.rs index b004ed5b..9fe9a490 100644 --- a/pmoparadise/src/stream_channel.rs +++ b/pmoparadise/src/stream_channel.rs @@ -12,23 +12,23 @@ use std::{ Arc, }, task::{Context, Poll}, - time::Duration, + time::{Duration, SystemTime, UNIX_EPOCH}, }; use crate::{ channels::{ChannelDescriptor, ParadiseChannelKind, ALL_CHANNELS}, client::RadioParadiseClient, + models::Block, playlist_feeder::RadioParadisePlaylistFeeder, }; use anyhow::{anyhow, Result}; use pmoaudio::AudioPipelineNode; use pmoaudio_ext::{ FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, OggFlacStreamHandle, - PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, - TrackBoundaryCoverNode, + PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, TrackBoundaryCoverNode, }; -use pmoaudiocache::Cache as AudioCache; -use pmocovers::Cache as CoverCache; +use pmoaudiocache::{get_audio_cache, Cache as AudioCache}; +use pmocovers::{get_cover_cache, Cache as CoverCache}; use pmoflac::EncoderOptions; use pmoplaylist::PlaylistManager; use thiserror::Error; @@ -110,6 +110,16 @@ impl ParadiseHistoryBuilder { } } +impl Default for ParadiseHistoryBuilder { + fn default() -> Self { + let audio_cache = get_audio_cache() + .expect("pmoaudiocache::register_audio_cache must be called before using ParadiseHistoryBuilder::default()"); + let cover_cache = get_cover_cache() + .expect("pmocovers::register_cover_cache must be called before using ParadiseHistoryBuilder::default()"); + Self::new(audio_cache, cover_cache) + } +} + #[cfg(feature = "pmoconfig")] impl ParadiseStreamChannelConfig { pub fn from_config(cfg: &pmoconfig::Config, channel: ParadiseChannelKind) -> Self { @@ -174,6 +184,9 @@ impl ParadiseStreamChannel { cover_cache: Option>, history: Option, ) -> Result { + let cover_cache = cover_cache + .or_else(|| history.as_ref().map(|opts| opts.cover_cache.clone())) + .or_else(|| get_cover_cache()); let manager = PlaylistManager::get(); // 1. Créer la playlist live pour ce canal @@ -189,7 +202,9 @@ impl ParadiseStreamChannel { .await? } else { // Pas d'historique, on a besoin quand même d'un cache audio basique - return Err(anyhow!("History options required for now (audio cache needed)")); + return Err(anyhow!( + "History options required for now (audio cache needed)" + )); }; let feeder = Arc::new(feeder); @@ -317,13 +332,7 @@ impl ParadiseStreamChannel { .channel(descriptor.id) .build() .await?; - Self::with_client( - descriptor, - client, - config, - cover_cache, - history, - ).await + Self::with_client(descriptor, client, config, cover_cache, history).await } /// S'abonne au flux FLAC pur. @@ -458,6 +467,9 @@ impl Drop for ParadiseStreamChannel { } } +const MAX_BLOCK_LEAD: Duration = Duration::from_secs(3600); +const BLOCK_LEAD_CHECK_CHUNK: Duration = Duration::from_secs(300); + struct ChannelState { descriptor: ChannelDescriptor, config: ParadiseStreamChannelConfig, @@ -473,6 +485,24 @@ struct ChannelState { } impl ChannelState { + fn current_unix_millis() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) + } + + fn block_lead_delay(&self, block: &Block) -> Option { + let start = block.start_time_millis()?; + let now = Self::current_unix_millis(); + let max_lead_ms = MAX_BLOCK_LEAD.as_millis() as u64; + if start <= now + max_lead_ms { + None + } else { + Some(Duration::from_millis(start - now - max_lead_ms)) + } + } + fn on_client_added(&self) { if self.active_clients.fetch_add(1, Ordering::SeqCst) == 0 { self.activity_notify.notify_one(); @@ -493,9 +523,38 @@ impl ChannelState { true } + async fn wait_until_block_ready(&self, block: &Block) -> BlockReadiness { + loop { + if self.stop_token.is_cancelled() { + return BlockReadiness::Stopped; + } + if self.active_clients.load(Ordering::SeqCst) == 0 { + return BlockReadiness::NoClients; + } + + if let Some(delay) = self.block_lead_delay(block) { + let sleep_for = delay.min(BLOCK_LEAD_CHECK_CHUNK); + let lead_secs = delay.as_secs_f64(); + info!( + "Block {} scheduled too far in the future ({:.1} min). Sleeping {:?} before retrying.", + block.event, + lead_secs / 60.0, + sleep_for + ); + tokio::select! { + _ = self.stop_token.cancelled() => return BlockReadiness::Stopped, + _ = tokio::time::sleep(sleep_for) => {}, + } + continue; + } + + return BlockReadiness::Ready; + } + } + async fn run_scheduler(self: Arc) { let mut backoff = Duration::from_secs(5); - loop { + 'scheduler: loop { if self.stop_token.is_cancelled() { break; } @@ -506,6 +565,11 @@ impl ChannelState { match self.client.get_block(None).await { Ok(block) => { + match self.wait_until_block_ready(&block).await { + BlockReadiness::Ready => {} + BlockReadiness::NoClients => continue, + BlockReadiness::Stopped => break, + } info!( "Channel {} streaming block {}", self.descriptor.display_name, block.event @@ -524,6 +588,11 @@ impl ChannelState { match self.client.get_block(Some(next_event)).await { Ok(next_block) => { + match self.wait_until_block_ready(&next_block).await { + BlockReadiness::Ready => {} + BlockReadiness::NoClients => break, + BlockReadiness::Stopped => break 'scheduler, + } self.feeder.push_block_id(next_block.event).await; next_event = next_block.end_event; backoff = Duration::from_secs(5); @@ -558,6 +627,12 @@ impl ChannelState { } } +enum BlockReadiness { + Ready, + NoClients, + Stopped, +} + macro_rules! wrap_stream { ($name:ident, $inner:ty) => { pub struct $name {