From 0a03f7246718faca83fd29eb04ead549b4cd7a41 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sat, 29 Nov 2025 14:19:16 +0100 Subject: [PATCH] =?UTF-8?q?Changement=20du=20m=C3=A9canisme=20d'attention?= =?UTF-8?q?=20sur=20les=20channels=20Radio=20Paradise.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 1 + PMOMusic/src/main.rs | 4 +- pmoaudio-ext/src/sinks/byte_stream_reader.rs | 12 +- pmoaudio-ext/src/sinks/chunk_to_pcm.rs | 5 +- pmoaudio-ext/src/sinks/flac_cache_sink.rs | 44 ++++- pmoaudio-ext/src/sinks/flac_frame_utils.rs | 8 +- pmoaudio-ext/src/sinks/streaming_flac_sink.rs | 3 +- .../src/sinks/streaming_icyflac_sink.rs | 21 ++- .../src/sinks/streaming_ogg_flac_sink.rs | 4 +- .../src/sinks/streaming_sink_common.rs | 4 +- pmoaudio-ext/src/sinks/timed_broadcast.rs | 19 ++- pmoaudio-ext/src/sources/playlist_source.rs | 14 +- pmoaudiocache/src/api.rs | 5 +- pmoaudiocache/src/cache.rs | 7 +- pmoaudiocache/src/lib.rs | 2 +- pmoaudiocache/src/metadata_ext.rs | 21 ++- pmocache/src/cache.rs | 15 +- pmocache/tests/test_db.rs | 4 +- pmocovers/src/lib.rs | 18 +-- pmodidl/examples/test_serialization.rs | 9 +- pmodidl/src/lib.rs | 25 +-- pmoflac/src/encoder.rs | 4 +- pmomediaserver/src/content_handler.rs | 20 +-- pmomediaserver/src/contentdirectory/mod.rs | 4 +- pmomediaserver/src/contentdirectory/state.rs | 9 +- pmomediaserver/src/device_ext.rs | 1 - pmomediaserver/src/paradise_streaming.rs | 73 ++++++++- pmomediaserver/src/sources.rs | 2 +- pmometadata/src/lib.rs | 31 ++-- pmoparadise/Cargo.toml | 1 + pmoparadise/examples/single_channel_server.rs | 32 ++-- pmoparadise/src/pmoserver_ext.rs | 6 +- pmoparadise/src/source.rs | 153 +++++++++++++----- pmoparadise/src/stream_channel.rs | 114 +++++++++++-- pmoplaylist/src/handle/read.rs | 7 +- pmoplaylist/src/handle/write.rs | 10 ++ pmoplaylist/src/lib.rs | 14 +- pmoplaylist/src/manager.rs | 25 +-- pmoplaylist/src/sse.rs | 4 +- pmoserver/src/logs/mod.rs | 15 +- pmoserver/src/server.rs | 8 +- pmosource/src/api.rs | 4 +- pmoupnp/src/actions/argument_methods.rs | 5 +- pmoupnp/src/upnp_server.rs | 8 +- 44 files changed, 564 insertions(+), 231 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 15180057..aad9c1b1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3280,6 +3280,7 @@ dependencies = [ "futures", "futures-util", "hex", + "once_cell", "pmoaudio", "pmoaudio-ext", "pmoaudiocache", diff --git a/PMOMusic/src/main.rs b/PMOMusic/src/main.rs index 9919b170..e4892c63 100644 --- a/PMOMusic/src/main.rs +++ b/PMOMusic/src/main.rs @@ -1,6 +1,8 @@ use pmoapp::{WebAppExt, Webapp}; use pmomediarenderer::MEDIA_RENDERER; -use pmomediaserver::{MEDIA_SERVER, MediaServerDeviceExt, ParadiseStreamingExt, sources::SourcesExt}; +use pmomediaserver::{ + MEDIA_SERVER, MediaServerDeviceExt, ParadiseStreamingExt, sources::SourcesExt, +}; use pmoserver::Server; use pmosource::MusicSourceExt; use pmoupnp::UpnpServerExt; diff --git a/pmoaudio-ext/src/sinks/byte_stream_reader.rs b/pmoaudio-ext/src/sinks/byte_stream_reader.rs index 9de6f3cd..acee39fe 100644 --- a/pmoaudio-ext/src/sinks/byte_stream_reader.rs +++ b/pmoaudio-ext/src/sinks/byte_stream_reader.rs @@ -1,7 +1,15 @@ use std::io; -use std::{collections::VecDeque, pin::Pin, sync::Arc, task::{Context, Poll}}; +use std::{ + collections::VecDeque, + pin::Pin, + sync::Arc, + task::{Context, Poll}, +}; -use tokio::{io::{AsyncRead, ReadBuf}, sync::{RwLock, mpsc}}; +use tokio::{ + io::{AsyncRead, ReadBuf}, + sync::{mpsc, RwLock}, +}; /// PCM chunk with audio data and timestamp for precise pacing. #[derive(Debug)] diff --git a/pmoaudio-ext/src/sinks/chunk_to_pcm.rs b/pmoaudio-ext/src/sinks/chunk_to_pcm.rs index d9652ba8..829110f9 100644 --- a/pmoaudio-ext/src/sinks/chunk_to_pcm.rs +++ b/pmoaudio-ext/src/sinks/chunk_to_pcm.rs @@ -1,7 +1,10 @@ use pmoaudio::{AudioChunk, AudioError}; /// Convert an AudioChunk to PCM bytes with specified bit depth. -pub(crate) fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result, AudioError> { +pub(crate) fn chunk_to_pcm_bytes( + chunk: &AudioChunk, + bits_per_sample: u8, +) -> Result, AudioError> { match chunk { AudioChunk::F32(_) | AudioChunk::F64(_) => { return Err(AudioError::ProcessingError( diff --git a/pmoaudio-ext/src/sinks/flac_cache_sink.rs b/pmoaudio-ext/src/sinks/flac_cache_sink.rs index 1db1471b..32a99b42 100755 --- a/pmoaudio-ext/src/sinks/flac_cache_sink.rs +++ b/pmoaudio-ext/src/sinks/flac_cache_sink.rs @@ -298,28 +298,44 @@ impl NodeLogic for FlacCacheSinkLogic { if let Some(sr) = transform.sample_rate { if let Err(e) = meta.set_sample_rate(Some(sr)).await { - tracing::error!("FlacCacheSink: Failed to set sample_rate for pk {}: {:?}", pk, e); + tracing::error!( + "FlacCacheSink: Failed to set sample_rate for pk {}: {:?}", + pk, + e + ); } else { tracing::debug!("FlacCacheSink: Set sample_rate={} for pk {}", sr, pk); } } if let Some(bps) = transform.bits_per_sample { if let Err(e) = meta.set_bits_per_sample(Some(bps)).await { - tracing::error!("FlacCacheSink: Failed to set bits_per_sample for pk {}: {:?}", pk, e); + tracing::error!( + "FlacCacheSink: Failed to set bits_per_sample for pk {}: {:?}", + pk, + e + ); } else { tracing::debug!("FlacCacheSink: Set bits_per_sample={} for pk {}", bps, pk); } } if let Some(ch) = transform.channels { if let Err(e) = meta.set_channels(Some(ch)).await { - tracing::error!("FlacCacheSink: Failed to set channels for pk {}: {:?}", pk, e); + tracing::error!( + "FlacCacheSink: Failed to set channels for pk {}: {:?}", + pk, + e + ); } else { tracing::debug!("FlacCacheSink: Set channels={} for pk {}", ch, pk); } } if let Some(ts) = transform.total_samples { if let Err(e) = meta.set_total_samples(Some(ts)).await { - tracing::error!("FlacCacheSink: Failed to set total_samples for pk {}: {:?}", pk, e); + tracing::error!( + "FlacCacheSink: Failed to set total_samples for pk {}: {:?}", + pk, + e + ); } else { tracing::debug!("FlacCacheSink: Set total_samples={} for pk {}", ts, pk); } @@ -329,10 +345,19 @@ impl NodeLogic for FlacCacheSinkLogic { if sr > 0 { use std::time::Duration; let secs = (ts as f64 / sr as f64).round() as u64; - if let Err(e) = meta.set_duration(Some(Duration::from_secs(secs))).await { - tracing::error!("FlacCacheSink: Failed to set duration for pk {}: {:?}", pk, e); + if let Err(e) = meta.set_duration(Some(Duration::from_secs(secs))).await + { + tracing::error!( + "FlacCacheSink: Failed to set duration for pk {}: {:?}", + pk, + e + ); } else { - tracing::debug!("FlacCacheSink: Set duration={} secs for pk {}", secs, pk); + tracing::debug!( + "FlacCacheSink: Set duration={} secs for pk {}", + secs, + pk + ); } } } @@ -340,7 +365,10 @@ impl NodeLogic for FlacCacheSinkLogic { drop(meta); // Libérer le lock explicitement } else { - tracing::warn!("FlacCacheSink: No transform metadata available for pk {}", pk); + tracing::warn!( + "FlacCacheSink: No transform metadata available for pk {}", + pk + ); } // Phase 2: Prebuffer terminé! Copier les métadonnées et pusher à la playlist diff --git a/pmoaudio-ext/src/sinks/flac_frame_utils.rs b/pmoaudio-ext/src/sinks/flac_frame_utils.rs index 90668afe..6c7cf0b6 100644 --- a/pmoaudio-ext/src/sinks/flac_frame_utils.rs +++ b/pmoaudio-ext/src/sinks/flac_frame_utils.rs @@ -17,7 +17,6 @@ pub(crate) enum FlacStreamState { Streaming, } - /// Validate and parse FLAC block size from frame header /// /// Returns the number of samples in the frame if the header is valid, or None if: @@ -376,7 +375,6 @@ pub(crate) fn find_complete_frames_with_samples(data: &[u8]) -> (usize, u64) { } } - /// Extract sample rate from STREAMINFO block in FLAC header pub(crate) fn extract_sample_rate_from_streaminfo(flac_header: &[u8]) -> Result { // Verify we have at least "fLaC" magic + STREAMINFO block header @@ -424,7 +422,9 @@ pub(crate) fn extract_sample_rate_from_streaminfo(flac_header: &[u8]) -> Result< } /// Read FLAC header (fLaC + all metadata blocks until first frame) -pub(crate) async fn read_flac_header(stream: &mut FlacEncodedStream) -> Result, AudioError> { +pub(crate) async fn read_flac_header( + stream: &mut FlacEncodedStream, +) -> Result, AudioError> { let mut header = Vec::new(); let mut buffer = [0u8; 4]; @@ -472,8 +472,6 @@ pub(crate) async fn read_flac_header(stream: &mut FlacEncodedStream) -> Result State { let entry = oentry.unwrap(); trace!( "TimedBroadcast[{}]: pruning played packet (@{} epoch={})", - self.name, entry.seq, entry.epoch + self.name, + entry.seq, + entry.epoch ); self.head_seq += 1; @@ -311,15 +313,13 @@ impl Sender { } // 2. Vérifier si un slot est disponible et insérer - let is_top_zero = - audio_timestamp.abs() < TOP_ZERO_EPSILON + let is_top_zero = audio_timestamp.abs() < TOP_ZERO_EPSILON && segment_duration >= TOP_ZERO_EPSILON; let is_zero_header = - audio_timestamp.abs() < TOP_ZERO_EPSILON - && segment_duration < TOP_ZERO_EPSILON; + audio_timestamp.abs() < TOP_ZERO_EPSILON && segment_duration < TOP_ZERO_EPSILON; if state.buffer.len() < self.inner.capacity { - if !state.initialized { - if !is_top_zero && segment_duration >= TOP_ZERO_EPSILON { + if !state.initialized { + if !is_top_zero && segment_duration >= TOP_ZERO_EPSILON { warn!( "TimedBroadcast[{}]: First packet has non-zero timestamp {:.1}ms - Duration={:.1}ms, treating as epoch start anyway", state.name, @@ -352,13 +352,12 @@ impl Sender { state.name, state.epoch, state.last_segment_end.is_some(), - segment_duration*1000.0 + segment_duration * 1000.0 ); } let expires_at = state.epoch_start - + Duration::from_secs_f64(audio_timestamp - + segment_duration); + + Duration::from_secs_f64(audio_timestamp + segment_duration); let is_first_packet = state.next_seq == 0; if !is_first_packet && !is_top_zero && !is_zero_header && expires_at <= now { diff --git a/pmoaudio-ext/src/sources/playlist_source.rs b/pmoaudio-ext/src/sources/playlist_source.rs index 54708117..d120fcd7 100644 --- a/pmoaudio-ext/src/sources/playlist_source.rs +++ b/pmoaudio-ext/src/sources/playlist_source.rs @@ -322,9 +322,19 @@ impl NodeLogic for PlaylistSourceLogic { Ok(()) => { tracing::info!("PlaylistSource: finished track {} - {}", artist, title); // Piste décodée avec succès, transférer vers l'historique si configuré - tracing::warn!("🔍 HISTORY DEBUG: history_playlist is {:?}", if self.history_playlist.is_some() { "Some" } else { "None" }); + tracing::warn!( + "🔍 HISTORY DEBUG: history_playlist is {:?}", + if self.history_playlist.is_some() { + "Some" + } else { + "None" + } + ); if let Some(ref history) = self.history_playlist { - tracing::warn!("🔍 HISTORY DEBUG: Attempting to push cache_pk={} to history", cache_pk); + tracing::warn!( + "🔍 HISTORY DEBUG: Attempting to push cache_pk={} to history", + cache_pk + ); if let Err(e) = history.push(cache_pk.to_string()).await { tracing::warn!( "PlaylistSourceLogic: failed to add track to history: {}", diff --git a/pmoaudiocache/src/api.rs b/pmoaudiocache/src/api.rs index f548eb80..7962b8f2 100644 --- a/pmoaudiocache/src/api.rs +++ b/pmoaudiocache/src/api.rs @@ -82,10 +82,7 @@ pub async fn get_cover_url( Ok(Some(cover_pk)) if !cover_pk.is_empty() => (cover_pk, "cover_pk".to_string()), _ => match metadata_guard.get_cover_url().await { Ok(Some(url)) if !url.is_empty() => (url, "cover_url".to_string()), - _ => ( - pmometadata::get_default_cover_url(), - "default".to_string(), - ), + _ => (pmometadata::get_default_cover_url(), "default".to_string()), }, }; diff --git a/pmoaudiocache/src/cache.rs b/pmoaudiocache/src/cache.rs index 969b2871..3b0d7ac4 100644 --- a/pmoaudiocache/src/cache.rs +++ b/pmoaudiocache/src/cache.rs @@ -4,10 +4,10 @@ //! spécifiques aux fichiers audio : conversion FLAC automatique et stockage //! des métadonnées en JSON dans la base de données. -use anyhow::Result; use crate::metadata_ext::AudioTrackMetadataExt; -use pmocache::CacheConfig; +use anyhow::Result; use pmocache::download::TransformMetadata; +use pmocache::CacheConfig; use serde_json::Value; use std::sync::Arc; @@ -277,8 +277,7 @@ fn parse_flac_streaminfo(data: &[u8]) -> Option<(u32, u8, u64)> { let s = &data[8..8 + 34]; // sample_rate: 20 bits: bytes 10..12 - let sample_rate = - ((s[10] as u32) << 12) | ((s[11] as u32) << 4) | ((s[12] as u32 & 0xF0) >> 4); + let sample_rate = ((s[10] as u32) << 12) | ((s[11] as u32) << 4) | ((s[12] as u32 & 0xF0) >> 4); // bits_per_sample: 5 bits spanning byte 12 (lsb) and byte 13 (msb) let bps_raw = (((s[12] & 0x01) as u8) << 4) | ((s[13] & 0xF0) >> 4); diff --git a/pmoaudiocache/src/lib.rs b/pmoaudiocache/src/lib.rs index 1bef4129..bb813f3b 100755 --- a/pmoaudiocache/src/lib.rs +++ b/pmoaudiocache/src/lib.rs @@ -231,7 +231,7 @@ impl AudioCacheExt for pmoserver::Server { "/{pk}/cover-url", axum::routing::get(crate::api::get_cover_url), ) - .with_state(cache.clone()) + .with_state(cache.clone()), ); let openapi = crate::ApiDoc::openapi(); diff --git a/pmoaudiocache/src/metadata_ext.rs b/pmoaudiocache/src/metadata_ext.rs index 73e24cb2..0eb59580 100644 --- a/pmoaudiocache/src/metadata_ext.rs +++ b/pmoaudiocache/src/metadata_ext.rs @@ -124,9 +124,24 @@ impl TrackMetadataDidlExt for dyn TrackMetadata { pmodidl::Resource { // Aligne sur Sink du renderer (audio/flac) avec PN explicite. protocol_info: "http-get:*:audio/flac:DLNA.ORG_PN=FLAC".to_string(), - bits_per_sample: self.get_bits_per_sample().await.ok().flatten().map(|b| b.to_string()), - sample_frequency: self.get_sample_rate().await.ok().flatten().map(|sr| sr.to_string()), - nr_audio_channels: self.get_channels().await.ok().flatten().map(|ch| ch.to_string()), + bits_per_sample: self + .get_bits_per_sample() + .await + .ok() + .flatten() + .map(|b| b.to_string()), + sample_frequency: self + .get_sample_rate() + .await + .ok() + .flatten() + .map(|sr| sr.to_string()), + nr_audio_channels: self + .get_channels() + .await + .ok() + .flatten() + .map(|ch| ch.to_string()), duration, url, } diff --git a/pmocache/src/cache.rs b/pmocache/src/cache.rs index 61a3b140..76d56aae 100755 --- a/pmocache/src/cache.rs +++ b/pmocache/src/cache.rs @@ -237,22 +237,27 @@ impl Cache { ); if let Some(sr) = transform.sample_rate { - self.db.set_a_metadata(pk, "sample_rate", serde_json::json!(sr))?; + self.db + .set_a_metadata(pk, "sample_rate", serde_json::json!(sr))?; } if let Some(bps) = transform.bits_per_sample { - self.db.set_a_metadata(pk, "bits_per_sample", serde_json::json!(bps))?; + self.db + .set_a_metadata(pk, "bits_per_sample", serde_json::json!(bps))?; } if let Some(ch) = transform.channels { - self.db.set_a_metadata(pk, "channels", serde_json::json!(ch))?; + self.db + .set_a_metadata(pk, "channels", serde_json::json!(ch))?; } if let Some(ts) = transform.total_samples { - self.db.set_a_metadata(pk, "total_samples", serde_json::json!(ts))?; + self.db + .set_a_metadata(pk, "total_samples", serde_json::json!(ts))?; // Calculer la durée à partir de total_samples et sample_rate if let Some(sr) = transform.sample_rate { if sr > 0 { let secs = (ts as f64 / sr as f64).round() as u64; - self.db.set_a_metadata(pk, "duration_secs", serde_json::json!(secs))?; + self.db + .set_a_metadata(pk, "duration_secs", serde_json::json!(secs))?; } } } diff --git a/pmocache/tests/test_db.rs b/pmocache/tests/test_db.rs index c60f7249..3c814320 100644 --- a/pmocache/tests/test_db.rs +++ b/pmocache/tests/test_db.rs @@ -269,7 +269,9 @@ fn test_get_pk_by_origin_url() { assert_eq!(found_pk, Some(pk.to_string())); // Rechercher une URL qui n'existe pas - let not_found = db.get_pk_by_origin_url("https://example.com/notfound.jpg").unwrap(); + let not_found = db + .get_pk_by_origin_url("https://example.com/notfound.jpg") + .unwrap(); assert_eq!(not_found, None); } diff --git a/pmocovers/src/lib.rs b/pmocovers/src/lib.rs index 7597f881..b090caca 100644 --- a/pmocovers/src/lib.rs +++ b/pmocovers/src/lib.rs @@ -54,7 +54,6 @@ pub mod cache; pub mod webp; - #[cfg(feature = "pmoserver")] pub mod openapi; @@ -201,7 +200,10 @@ async fn serve_jpeg_internal( use image::ImageFormat; use std::io::Cursor; - let path = cache.get_file_path_with_qualifier(&pk, ::default_param()); + let path = cache.get_file_path_with_qualifier( + &pk, + ::default_param(), + ); if !path.exists() { return (StatusCode::NOT_FOUND, "File not found").into_response(); } @@ -218,12 +220,7 @@ async fn serve_jpeg_internal( .await; match res { - Ok(Ok(data)) => ( - StatusCode::OK, - [("content-type", "image/jpeg")], - data, - ) - .into_response(), + Ok(Ok(data)) => (StatusCode::OK, [("content-type", "image/jpeg")], data).into_response(), Ok(Err(e)) => { tracing::warn!("JPEG transcode error for {}: {}", pk, e); (StatusCode::INTERNAL_SERVER_ERROR, "Transcode error").into_response() @@ -294,10 +291,7 @@ impl CoverCacheExt for pmoserver::Server { // Router JPEG (transcodage à la volée depuis le WebP stocké) // Routes: GET /covers/jpeg/{pk} et GET /covers/jpeg/{pk}/{size} let jpeg_router = axum::Router::new() - .route( - "/covers/jpeg/{pk}", - axum::routing::get(serve_cover_jpeg), - ) + .route("/covers/jpeg/{pk}", axum::routing::get(serve_cover_jpeg)) .route( "/covers/jpeg/{pk}/{size}", axum::routing::get(serve_cover_jpeg_with_size), diff --git a/pmodidl/examples/test_serialization.rs b/pmodidl/examples/test_serialization.rs index 7289b9cc..53795255 100644 --- a/pmodidl/examples/test_serialization.rs +++ b/pmodidl/examples/test_serialization.rs @@ -9,9 +9,9 @@ fn main() { creator: Some("Test Artist".to_string()), class: "object.item.audioItem.musicTrack".to_string(), artist: Some("Test Artist".to_string()), - album: None, // Pas d'album + album: None, // Pas d'album genre: None, - album_art: None, // Pas d'albumArtURI + album_art: None, // Pas d'albumArtURI album_art_pk: None, date: None, original_track_number: None, @@ -42,7 +42,10 @@ fn main() { println!("=== Output from quick_xml::se::to_string() ==="); println!("{}", xml); println!("\n=== Length: {} bytes ===", xml.len()); - println!("\n=== Starts with '{}", xml); println!("{}", with_decl); diff --git a/pmodidl/src/lib.rs b/pmodidl/src/lib.rs index c39bd6eb..1e4f4028 100644 --- a/pmodidl/src/lib.rs +++ b/pmodidl/src/lib.rs @@ -554,7 +554,8 @@ impl ToXmlElement for Container { fn to_xml_element(&self) -> Element { let mut elem = Element::new("container"); elem.attributes.insert("id".into(), self.id.clone()); - elem.attributes.insert("parentID".into(), self.parent_id.clone()); + elem.attributes + .insert("parentID".into(), self.parent_id.clone()); if let Some(ref r) = self.restricted { elem.attributes.insert("restricted".into(), r.clone()); } @@ -562,11 +563,14 @@ impl ToXmlElement for Container { elem.attributes.insert("childCount".into(), cc.clone()); } if let Some(ref searchable) = self.searchable { - elem.attributes.insert("searchable".into(), searchable.clone()); + elem.attributes + .insert("searchable".into(), searchable.clone()); } - elem.children.push(XMLNode::Element(text_element("dc:title", &self.title))); - elem.children.push(XMLNode::Element(text_element("upnp:class", &self.class))); + elem.children + .push(XMLNode::Element(text_element("dc:title", &self.title))); + elem.children + .push(XMLNode::Element(text_element("upnp:class", &self.class))); for c in &self.containers { elem.children.push(XMLNode::Element(c.to_xml_element())); @@ -583,15 +587,18 @@ impl ToXmlElement for Item { fn to_xml_element(&self) -> Element { let mut elem = Element::new("item"); elem.attributes.insert("id".into(), self.id.clone()); - elem.attributes.insert("parentID".into(), self.parent_id.clone()); + elem.attributes + .insert("parentID".into(), self.parent_id.clone()); if let Some(ref r) = self.restricted { elem.attributes.insert("restricted".into(), r.clone()); } - elem.children.push(XMLNode::Element(text_element("dc:title", &self.title))); + elem.children + .push(XMLNode::Element(text_element("dc:title", &self.title))); if let Some(ref c) = self.creator { - elem.children.push(XMLNode::Element(text_element("dc:creator", c))); + elem.children + .push(XMLNode::Element(text_element("dc:creator", c))); } elem.children @@ -644,7 +651,8 @@ impl ToXmlElement for Resource { elem.attributes.insert("bitsPerSample".into(), bps.clone()); } if let Some(ref freq) = self.sample_frequency { - elem.attributes.insert("sampleFrequency".into(), freq.clone()); + elem.attributes + .insert("sampleFrequency".into(), freq.clone()); } if let Some(ref ch) = self.nr_audio_channels { elem.attributes.insert("nrAudioChannels".into(), ch.clone()); @@ -678,7 +686,6 @@ impl ToXmlElement for Description { } } - // ============= Itérateurs personnalisés ============= struct AllContainersIter<'a> { diff --git a/pmoflac/src/encoder.rs b/pmoflac/src/encoder.rs index 5292932b..a32decef 100755 --- a/pmoflac/src/encoder.rs +++ b/pmoflac/src/encoder.rs @@ -554,7 +554,9 @@ fn run_encoder( "set_total_samples_estimate failed", )?; } else { - tracing::warn!("FLAC encoder: total_samples is None, STREAMINFO will have total_samples=0"); + tracing::warn!( + "FLAC encoder: total_samples is None, STREAMINFO will have total_samples=0" + ); } if let Some(block_size) = options.block_size { ensure( diff --git a/pmomediaserver/src/content_handler.rs b/pmomediaserver/src/content_handler.rs index 24e4bdbe..327f56e4 100644 --- a/pmomediaserver/src/content_handler.rs +++ b/pmomediaserver/src/content_handler.rs @@ -10,10 +10,10 @@ //! - **Search** : Recherche dans les sources qui le supportent //! - **Update ID** : Suivi des changements pour les notifications UPnP -use pmoutils::ToXmlElement; use pmodidl::{Container, DIDLLite}; use pmosource::api::{get_source as get_source_from_registry, list_all_sources}; use pmosource::{BrowseResult, MusicSource, MusicSourceError}; +use pmoutils::ToXmlElement; use std::collections::HashSet; use std::sync::Arc; @@ -123,20 +123,14 @@ impl ContentHandler { container.child_count = None; // compatibilité CP let didl = to_didl_lite(&[container], &[])?; let update_id = source.update_id().await.max(1); - tracing::debug!( - "BrowseMetadata root (flatten) didl_len={}B", - didl.len() - ); + tracing::debug!("BrowseMetadata root (flatten) didl_len={}B", didl.len()); return Ok((didl, 1, 1, update_id)); } // Sinon retourner le container racine agrégé let root = self.build_root_container().await; let didl = to_didl_lite(&[root], &[])?; - tracing::debug!( - "BrowseMetadata root (aggregate) didl_len={}B", - didl.len() - ); + tracing::debug!("BrowseMetadata root (aggregate) didl_len={}B", didl.len()); Ok((didl, 1, 1, 1)) } else { // Essayer de trouver l'objet dans les sources @@ -300,7 +294,13 @@ impl ContentHandler { match source.browse(object_id).await { Ok(result) => { return self - .browse_result_to_didl(object_id, result, source, starting_index, requested_count) + .browse_result_to_didl( + object_id, + result, + source, + starting_index, + requested_count, + ) .await; } Err(MusicSourceError::ObjectNotFound(_)) => continue, diff --git a/pmomediaserver/src/contentdirectory/mod.rs b/pmomediaserver/src/contentdirectory/mod.rs index ad8340de..fcec5894 100644 --- a/pmomediaserver/src/contentdirectory/mod.rs +++ b/pmomediaserver/src/contentdirectory/mod.rs @@ -76,14 +76,14 @@ use pmoupnp::define_service; pub mod actions; pub mod handlers; -pub mod variables; pub mod state; +pub mod variables; 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, CONTAINERUPDATEIDS, + A_ARG_TYPE_UPDATEID, CONTAINERUPDATEIDS, SEARCHCAPABILITIES, SORTCAPABILITIES, SYSTEMUPDATEID, }; // Service ContentDirectory:1 conforme à la spécification UPnP AV pour MediaServer diff --git a/pmomediaserver/src/contentdirectory/state.rs b/pmomediaserver/src/contentdirectory/state.rs index 3f91f20c..6627f7a0 100644 --- a/pmomediaserver/src/contentdirectory/state.rs +++ b/pmomediaserver/src/contentdirectory/state.rs @@ -1,6 +1,9 @@ use once_cell::sync::OnceCell; use pmoupnp::{services::ServiceInstance, variable_types::StateValue}; -use std::sync::{atomic::{AtomicU32, Ordering}, Arc, Weak, Mutex}; +use std::sync::{ + Arc, Mutex, Weak, + atomic::{AtomicU32, Ordering}, +}; static CONTENTDIR_INSTANCE: OnceCell> = OnceCell::new(); static SYSTEM_UPDATE_ID: AtomicU32 = AtomicU32::new(1); @@ -17,7 +20,9 @@ pub fn register_instance(instance: &Arc) { /// 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); + 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() { diff --git a/pmomediaserver/src/device_ext.rs b/pmomediaserver/src/device_ext.rs index 9189ce92..1e70c09f 100644 --- a/pmomediaserver/src/device_ext.rs +++ b/pmomediaserver/src/device_ext.rs @@ -1,5 +1,4 @@ ///! Extension trait pour initialiser le PMO Music MediaServer UPnP - use pmoupnp::devices::DeviceInstance; use pmoupnp::variable_types::StateValue; use std::sync::Arc; diff --git a/pmomediaserver/src/paradise_streaming.rs b/pmomediaserver/src/paradise_streaming.rs index c18982c2..b845f74c 100644 --- a/pmomediaserver/src/paradise_streaming.rs +++ b/pmomediaserver/src/paradise_streaming.rs @@ -6,7 +6,6 @@ use anyhow::{Context, Result}; use async_trait::async_trait; use axum::{ - Json, Router, body::Body, extract::{Path, State}, http::{ @@ -15,11 +14,17 @@ use axum::{ }, 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::{ChannelDescriptor, ALL_CHANNELS}, + stream_channel::register_global_channel_manager, + 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 pmoplaylist::{self, PlaylistEventKind}; use std::sync::Arc; use tokio_util::io::ReaderStream; use tracing::{error, info}; @@ -64,7 +69,9 @@ impl ParadiseStreamingExt for pmoserver::Server { async fn init_paradise_streaming(&mut self) -> Result> { info!("🎵 Initializing Radio Paradise streaming channels..."); // Sentinel log pour vérifier qu'on exécute bien cette version du binaire - tracing::warn!("🔍 Rien de neuf: entering init_paradise_streaming with caches+history setup"); + tracing::warn!( + "🔍 Rien de neuf: entering init_paradise_streaming with caches+history setup" + ); // Récupérer ou initialiser les caches singletons info!("📦 Getting cache singletons..."); @@ -143,6 +150,9 @@ impl ParadiseStreamingExt for pmoserver::Server { } }; + register_global_channel_manager(manager.clone()); + spawn_playlist_event_handler(manager.clone()); + let state = Arc::new(ParadiseStreamingState { manager: manager.clone(), }); @@ -317,3 +327,56 @@ async fn stream_history_ogg( .body(Body::from_stream(ReaderStream::new(stream))) .unwrap()) } + +fn spawn_playlist_event_handler(manager: Arc) { + tokio::spawn(async move { + let mut rx = pmoplaylist::subscribe_events(); + while let Ok(envelope) = rx.recv().await { + if let PlaylistEventKind::TrackPlayed { cache_pk, .. } = envelope.event.kind { + if let Some(descriptor) = channel_from_live_playlist(&envelope.event.playlist_id) { + if let Err(e) = manager.prefetch_until_horizon(descriptor.id).await { + tracing::warn!( + "Failed to prefetch for channel {}: {}", + descriptor.display_name, + e + ); + } + if let Err(e) = append_track_to_history(descriptor, &cache_pk).await { + tracing::warn!( + "Failed to update history for channel {}: {}", + descriptor.display_name, + e + ); + } + } + } + } + }); +} + +fn channel_from_live_playlist(playlist_id: &str) -> Option<&'static ChannelDescriptor> { + const PREFIX: &str = "radio-paradise-live-"; + let slug = playlist_id.strip_prefix(PREFIX)?; + ALL_CHANNELS + .iter() + .find(|descriptor| descriptor.slug == slug) +} + +async fn append_track_to_history(descriptor: &ChannelDescriptor, cache_pk: &str) -> Result<()> { + let playlist_id = format!("radio-paradise-history-{}", descriptor.slug); + let manager = pmoplaylist::PlaylistManager(); + let handle = manager + .get_persistent_write_handle(playlist_id.clone()) + .await + .with_context(|| format!("Failed to get history playlist {}", playlist_id))?; + + if handle.contains_pk(cache_pk).await? { + return Ok(()); + } + + handle + .push(cache_pk.to_string()) + .await + .with_context(|| format!("Failed to append {} to {}", cache_pk, playlist_id))?; + Ok(()) +} diff --git a/pmomediaserver/src/sources.rs b/pmomediaserver/src/sources.rs index 89a1bb93..6c6b2ff0 100644 --- a/pmomediaserver/src/sources.rs +++ b/pmomediaserver/src/sources.rs @@ -173,8 +173,8 @@ impl SourcesExt for Server { #[cfg(feature = "paradise")] async fn register_paradise(&mut self) -> Result<()> { - use pmoparadise::{RadioParadiseExt, RadioParadiseSource}; use crate::contentdirectory::state; + use pmoparadise::{RadioParadiseExt, RadioParadiseSource}; tracing::info!("Initializing Radio Paradise source..."); diff --git a/pmometadata/src/lib.rs b/pmometadata/src/lib.rs index 84e66d02..7243a45b 100755 --- a/pmometadata/src/lib.rs +++ b/pmometadata/src/lib.rs @@ -552,8 +552,23 @@ where let src_guard = src.read().await; copy_metadata!( - src_guard, dest, title, artist, album, year, duration, sample_rate, total_samples, - bits_per_sample, track_id, channel_id, event, rating, cover_url, cover_pk, extra + src_guard, + dest, + title, + artist, + album, + year, + duration, + sample_rate, + total_samples, + bits_per_sample, + track_id, + channel_id, + event, + rating, + cover_url, + cover_pk, + extra ); // Try to update the timestamp, but ignore transient errors @@ -828,8 +843,6 @@ impl TrackMetadata for MemoryTrackMetadata { self.updated_at = Some(SystemTime::now()); Ok(Some(())) } - - } #[cfg(test)] @@ -1248,10 +1261,7 @@ mod tests { #[tokio::test] async fn test_get_cover_url_with_fallback_none() { let metadata = MemoryTrackMetadata::new(); - assert_eq!( - metadata.get_cover_url_with_fallback().await.unwrap(), - None - ); + assert_eq!(metadata.get_cover_url_with_fallback().await.unwrap(), None); } #[tokio::test] @@ -1278,10 +1288,7 @@ mod tests { .await .unwrap(); - assert_eq!( - metadata.get_cover_url_or_default().await.unwrap(), - "abc123" - ); + assert_eq!(metadata.get_cover_url_or_default().await.unwrap(), "abc123"); } #[tokio::test] diff --git a/pmoparadise/Cargo.toml b/pmoparadise/Cargo.toml index e3fe0cda..26a23d78 100644 --- a/pmoparadise/Cargo.toml +++ b/pmoparadise/Cargo.toml @@ -29,6 +29,7 @@ hex = "0.4" tokio-util = { version = "0.7", features = ["io"] } async-stream = "0.3" rusqlite = { version = "0.37", features = ["bundled"] } +once_cell = "1.20" # Gestion des erreurs thiserror = "2.0.17" diff --git a/pmoparadise/examples/single_channel_server.rs b/pmoparadise/examples/single_channel_server.rs index ceb84ef8..c83b3475 100644 --- a/pmoparadise/examples/single_channel_server.rs +++ b/pmoparadise/examples/single_channel_server.rs @@ -16,16 +16,18 @@ use axum::{ routing::get, Json, Router, }; +use pmoaudio_ext::StreamingSinkOptions; 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, Cache as CoverCache}; +use pmocovers::{ + new_cache_with_consolidation as new_cover_cache, register_cover_cache, Cache as CoverCache, +}; use pmoparadise::{ channels::{ChannelDescriptor, ALL_CHANNELS}, ParadiseHistoryBuilder, ParadiseStreamChannel, ParadiseStreamChannelConfig, }; -use pmoaudio_ext::StreamingSinkOptions; use pmoplaylist::register_audio_cache as register_playlist_audio_cache; use std::{fs, net::SocketAddr, sync::Arc}; use tokio::net::TcpListener; @@ -44,9 +46,7 @@ async fn main() -> anyhow::Result<()> { let env_filter = tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")); - tracing_subscriber::fmt() - .with_env_filter(env_filter) - .init(); + tracing_subscriber::fmt().with_env_filter(env_filter).init(); let descriptor = pick_descriptor(std::env::args().nth(1))?; info!( @@ -206,22 +206,16 @@ async fn get_cover( ) -> Result { // Récupérer le chemin de la cover depuis le cache // Le cache retourne un PathBuf pointant vers le fichier .webp - let cover_path = state - .cover_cache - .get(&pk) - .await - .map_err(|e| { - tracing::error!("Failed to get cover path for {}: {}", pk, e); - StatusCode::NOT_FOUND - })?; + let cover_path = state.cover_cache.get(&pk).await.map_err(|e| { + tracing::error!("Failed to get cover path for {}: {}", pk, e); + StatusCode::NOT_FOUND + })?; // Lire le fichier - let cover_data = tokio::fs::read(&cover_path) - .await - .map_err(|e| { - tracing::error!("Failed to read cover file {:?}: {}", cover_path, e); - StatusCode::NOT_FOUND - })?; + let cover_data = tokio::fs::read(&cover_path).await.map_err(|e| { + tracing::error!("Failed to read cover file {:?}: {}", cover_path, e); + StatusCode::NOT_FOUND + })?; Response::builder() .status(StatusCode::OK) diff --git a/pmoparadise/src/pmoserver_ext.rs b/pmoparadise/src/pmoserver_ext.rs index da22964f..4b95acb5 100644 --- a/pmoparadise/src/pmoserver_ext.rs +++ b/pmoparadise/src/pmoserver_ext.rs @@ -429,11 +429,7 @@ async fn get_cover_url( })?; let song = block.get_song(song_index).ok_or_else(|| { - tracing::warn!( - "Song index {} not found in block {}", - song_index, - event_id - ); + tracing::warn!("Song index {} not found in block {}", song_index, event_id); StatusCode::NOT_FOUND })?; diff --git a/pmoparadise/src/source.rs b/pmoparadise/src/source.rs index 28f8eac3..459998dd 100644 --- a/pmoparadise/src/source.rs +++ b/pmoparadise/src/source.rs @@ -4,20 +4,26 @@ //! 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, SourceCapabilities, }; +use std::fmt; use std::sync::Arc; -use std::time::SystemTime; +use std::time::{Duration, Instant, SystemTime}; use tokio::sync::RwLock; - /// Default Radio Paradise image (embedded in binary) const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp"); +#[cfg(feature = "playlist")] +const LIVE_PLAYLIST_MIN_READY_ITEMS: usize = 5; +#[cfg(feature = "playlist")] +const LIVE_PLAYLIST_READY_TIMEOUT: Duration = Duration::from_secs(10); +#[cfg(feature = "playlist")] +const LIVE_PLAYLIST_READY_POLL: Duration = Duration::from_millis(200); + /// RadioParadiseSource - UPnP ContentDirectory source for Radio Paradise /// /// Provides access to: @@ -142,14 +148,21 @@ impl RadioParadiseSource { ALL_CHANNELS .iter() .find(|ch| pid.ends_with(ch.slug)) - .map(|ch| vec![format!("radio-paradise:channel:{}:history", 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)]) + .map(|ch| { + vec![format!( + "radio-paradise:channel:{}:liveplaylist", + ch.slug + )] + }) .unwrap_or_default() }; @@ -181,7 +194,10 @@ impl RadioParadiseSource { match response.json::().await { Ok(json) => { // Parse metadata from JSON and create an Item - let title = json["title"].as_str().unwrap_or("Unknown Title").to_string(); + let title = json["title"] + .as_str() + .unwrap_or("Unknown Title") + .to_string(); let artist = json["artist"].as_str().map(|s| s.to_string()); let album = json["album"].as_str().map(|s| s.to_string()); let year = json["year"].as_u64().map(|y| y as u32); @@ -201,10 +217,12 @@ impl RadioParadiseSource { .or_else(|| json["duration"].as_f64()) .map(|secs| { let total_secs = secs as u64; - format!("{}:{:02}:{:02}", + format!( + "{}:{:02}:{:02}", total_secs / 3600, (total_secs % 3600) / 60, - total_secs % 60) + total_secs % 60 + ) }); // Create the item with current metadata @@ -254,6 +272,42 @@ impl RadioParadiseSource { format!("radio-paradise-live-{}", slug) } + #[cfg(feature = "playlist")] + async fn wait_for_live_playlist_ready(&self, slug: &str) -> 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 start = Instant::now(); + loop { + match reader.remaining().await { + Ok(count) if count >= LIVE_PLAYLIST_MIN_READY_ITEMS => return Ok(()), + Ok(_) => {} + Err(e) => { + return Err(MusicSourceError::BrowseError(format!( + "Failed to inspect live playlist {}: {}", + playlist_id, e + ))); + } + } + + if start.elapsed() >= LIVE_PLAYLIST_READY_TIMEOUT { + tracing::warn!( + "Timeout waiting for live playlist {} to reach {} items", + playlist_id, + LIVE_PLAYLIST_MIN_READY_ITEMS + ); + return Ok(()); + } + + tokio::time::sleep(LIVE_PLAYLIST_READY_POLL).await; + } + } + /// Get channel descriptor by slug fn get_channel_by_slug(slug: &str) -> Option<&'static ChannelDescriptor> { ALL_CHANNELS.iter().find(|ch| ch.slug == slug) @@ -380,7 +434,10 @@ impl RadioParadiseSource { /// Build a history container with accurate child count from playlist #[cfg(feature = "playlist")] - async fn build_history_container_with_count(&self, descriptor: &ChannelDescriptor) -> Container { + async fn build_history_container_with_count( + &self, + descriptor: &ChannelDescriptor, + ) -> Container { let mut container = self.build_history_container(descriptor); // Try to get actual count from playlist @@ -466,32 +523,48 @@ impl RadioParadiseSource { _offset: usize, count: usize, ) -> Result> { + #[cfg(all(feature = "playlist", feature = "pmoaudio"))] + if let Some(descriptor) = Self::get_channel_by_slug(slug) { + if let Some(manager) = crate::stream_channel::get_global_channel_manager() { + if let Err(e) = manager.prefetch_until_horizon(descriptor.id).await { + tracing::warn!( + "Failed to prefetch live playlist for {}: {}", + descriptor.slug, + e + ); + } + } + } + + #[cfg(feature = "playlist")] + if let Err(e) = self.wait_for_live_playlist_ready(slug).await { + tracing::warn!( + "Failed to wait for live playlist readiness on {}: {}", + slug, + e + ); + } + 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)) + 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 - )) + 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 - ); + 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); @@ -519,10 +592,7 @@ impl RadioParadiseSource { #[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 - ); + let expected_id = format!("radio-paradise:channel:{}:liveplaylist:track:{}", slug, pk); for item in items { if item.id == expected_id { return Ok(item); @@ -533,7 +603,6 @@ impl RadioParadiseSource { pk ))) } - } /// Types of object IDs in the Radio Paradise source @@ -619,7 +688,8 @@ impl MusicSource for RadioParadiseSource { #[cfg(feature = "playlist")] { - let history_container = self.build_history_container_with_count(descriptor).await; + let history_container = + self.build_history_container_with_count(descriptor).await; let items = self.get_history_items(&slug, 0, 100).await?; Ok(BrowseResult::Mixed { containers: vec![history_container], @@ -821,14 +891,14 @@ impl MusicSource for RadioParadiseSource { 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:{}:history:track:{}", slug, pk2); - item.parent_id = - format!("radio-paradise:channel:{}:history", slug); + item.id = format!( + "radio-paradise:channel:{}:history:track:{}", + slug, pk2 + ); + item.parent_id = format!("radio-paradise:channel:{}:history", slug); if resource.url.starts_with('/') { - resource.url = - format!("{}{}", self.base_url, resource.url); + resource.url = format!("{}{}", self.base_url, resource.url); } } } @@ -839,7 +909,8 @@ impl MusicSource for RadioParadiseSource { } // Find the item matching this pk in the item ID - let expected_id = format!("radio-paradise:channel:{}:history:track:{}", slug, pk); + let expected_id = + format!("radio-paradise:channel:{}:history:track:{}", slug, pk); for item in adjusted { if item.id == expected_id { return Ok(item); @@ -887,10 +958,8 @@ impl MusicSource for RadioParadiseSource { "radio-paradise:channel:{}:liveplaylist:track:{}", slug, pk2 ); - item.parent_id = format!( - "radio-paradise:channel:{}:liveplaylist", - slug - ); + item.parent_id = + format!("radio-paradise:channel:{}:liveplaylist", slug); if resource.url.starts_with('/') { resource.url = format!("{}{}", self.base_url, resource.url); @@ -910,10 +979,8 @@ impl MusicSource for RadioParadiseSource { item.album_art = Some(self.default_cover_url()); } - let expected_id = format!( - "radio-paradise:channel:{}:liveplaylist:track:{}", - slug, pk - ); + let expected_id = + format!("radio-paradise:channel:{}:liveplaylist:track:{}", slug, pk); if item.id == expected_id { return Ok(item); } diff --git a/pmoparadise/src/stream_channel.rs b/pmoparadise/src/stream_channel.rs index 1b053985..28123b60 100644 --- a/pmoparadise/src/stream_channel.rs +++ b/pmoparadise/src/stream_channel.rs @@ -21,12 +21,13 @@ use crate::{ models::{Block, EventId}, playlist_feeder::RadioParadisePlaylistFeeder, }; -use anyhow::{anyhow, Result}; +use anyhow::{anyhow, Context as AnyhowContext, Result}; +use once_cell::sync::OnceCell; use pmoaudio::{AudioError, AudioPipelineNode}; use pmoaudio_ext::{ FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, OggFlacStreamHandle, - PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, TrackBoundaryCoverNode, - StreamingSinkOptions, + PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, StreamingSinkOptions, + TrackBoundaryCoverNode, }; use pmoaudiocache::{get_audio_cache, Cache as AudioCache}; use pmocovers::{get_cover_cache, Cache as CoverCache}; @@ -203,9 +204,14 @@ impl ParadiseStreamChannel { // Propager server_base_url dans les options pour que les encoders injectent les covers du cache let mut config = config; if let Some(ref base) = config.server_base_url { - config.flac_options = - config.flac_options.clone().with_server_base_url(Some(base.clone())); - config.ogg_options = config.ogg_options.clone().with_server_base_url(Some(base.clone())); + config.flac_options = config + .flac_options + .clone() + .with_server_base_url(Some(base.clone())); + config.ogg_options = config + .ogg_options + .clone() + .with_server_base_url(Some(base.clone())); } let cover_cache = cover_cache .or_else(|| history.as_ref().map(|opts| opts.cover_cache.clone())) @@ -312,6 +318,7 @@ impl ParadiseStreamChannel { activity_notify: Notify::new(), stop_token, current_block: Mutex::new(None), + prefetch_lock: Mutex::new(()), }); let pipeline_state = state.clone(); @@ -496,6 +503,12 @@ impl Drop for ParadiseStreamChannel { const MAX_BLOCK_LEAD: Duration = Duration::from_secs(3600); const BLOCK_LEAD_CHECK_CHUNK: Duration = Duration::from_secs(300); +const LIVE_PREFETCH_MIN_TRACKS: usize = 5; +const LIVE_PREFETCH_TIMEOUT: Duration = Duration::from_secs(10); +const LIVE_PREFETCH_POLL_INTERVAL: Duration = Duration::from_millis(200); +const LIVE_PREFETCH_MAX_BLOCKS: usize = 4; + +static GLOBAL_CHANNEL_MANAGER: OnceCell> = OnceCell::new(); struct ChannelState { descriptor: ChannelDescriptor, @@ -510,6 +523,7 @@ struct ChannelState { activity_notify: Notify, stop_token: CancellationToken, current_block: Mutex>, + prefetch_lock: Mutex<()>, } impl ChannelState { @@ -580,6 +594,66 @@ impl ChannelState { } } + fn live_playlist_id(&self) -> String { + format!("radio-paradise-live-{}", self.descriptor.slug) + } + + async fn prefetch_until_horizon(&self) -> Result<()> { + let _guard = self.prefetch_lock.lock().await; + let playlist_id = self.live_playlist_id(); + let manager = PlaylistManager::get(); + let reader = manager + .get_read_handle(&playlist_id) + .await + .with_context(|| format!("Failed to get live playlist {}", playlist_id))?; + let start = Instant::now(); + let mut next_event: Option = None; + let mut attempts = 0usize; + + loop { + let available = reader + .remaining() + .await + .with_context(|| format!("Failed to inspect playlist {}", playlist_id))?; + if available >= LIVE_PREFETCH_MIN_TRACKS { + return Ok(()); + } + + if start.elapsed() >= LIVE_PREFETCH_TIMEOUT { + warn!( + "Prefetch timeout for channel {} ({} tracks available)", + self.descriptor.display_name, available + ); + return Ok(()); + } + + if attempts >= LIVE_PREFETCH_MAX_BLOCKS { + warn!( + "Prefetch block limit reached for channel {} ({} tracks available)", + self.descriptor.display_name, available + ); + return Ok(()); + } + + match self.client.get_block(next_event).await { + Ok(block) => { + attempts += 1; + next_event = Some(block.end_event); + self.feeder.push_block_id(block.event).await; + } + Err(e) => { + warn!( + "Failed to fetch block during prefetch for channel {}: {}", + self.descriptor.display_name, e + ); + return Ok(()); + } + } + + tokio::time::sleep(LIVE_PREFETCH_POLL_INTERVAL).await; + } + } + async fn set_current_block(&self, event_id: EventId) { let mut guard = self.current_block.lock().await; *guard = Some(event_id); @@ -866,12 +940,7 @@ impl ParadiseChannelManager { ); let channel = match tokio::time::timeout( Duration::from_secs(20), - ParadiseStreamChannel::new( - descriptor, - config, - cover_cache.clone(), - history_opts, - ), + ParadiseStreamChannel::new(descriptor, config, cover_cache.clone(), history_opts), ) .await { @@ -918,4 +987,25 @@ impl ParadiseChannelManager { pub fn iter(&self) -> impl Iterator> { self.channels.values() } + + pub async fn prefetch_until_horizon(&self, channel_id: u8) -> Result<()> { + let channel = self + .get(channel_id) + .ok_or_else(|| anyhow!("Unknown channel id {}", channel_id))?; + channel.prefetch_until_horizon().await + } +} + +pub fn register_global_channel_manager(manager: Arc) { + let _ = GLOBAL_CHANNEL_MANAGER.set(Arc::downgrade(&manager)); +} + +pub fn get_global_channel_manager() -> Option> { + GLOBAL_CHANNEL_MANAGER.get().and_then(|weak| weak.upgrade()) +} + +impl ParadiseStreamChannel { + pub async fn prefetch_until_horizon(&self) -> Result<()> { + self.state.prefetch_until_horizon().await + } } diff --git a/pmoplaylist/src/handle/read.rs b/pmoplaylist/src/handle/read.rs index 1ca7e85a..b1909f92 100644 --- a/pmoplaylist/src/handle/read.rs +++ b/pmoplaylist/src/handle/read.rs @@ -222,7 +222,12 @@ impl ReadHandle { let resource = meta.to_didl_resource(url).await; // Récupérer les métadonnées pour construire l'Item DIDL - let title = meta.get_title().await.ok().flatten().unwrap_or_else(|| "Unknown".to_string()); + let title = meta + .get_title() + .await + .ok() + .flatten() + .unwrap_or_else(|| "Unknown".to_string()); let artist = meta.get_artist().await.ok().flatten(); let album = meta.get_album().await.ok().flatten(); let genre = meta.get_genre().await.ok().flatten(); diff --git a/pmoplaylist/src/handle/write.rs b/pmoplaylist/src/handle/write.rs index 3104e9e3..2a561511 100644 --- a/pmoplaylist/src/handle/write.rs +++ b/pmoplaylist/src/handle/write.rs @@ -222,6 +222,16 @@ impl WriteHandle { Ok(()) } + /// Vérifie si la playlist contient déjà un pk + pub async fn contains_pk(&self, cache_pk: &str) -> Result { + if !self.playlist.is_alive() { + return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); + } + + let core = self.playlist.core.read().await; + Ok(core.tracks.iter().any(|record| record.cache_pk == cache_pk)) + } + /// Change le TTL par défaut pub async fn set_default_ttl(&self, ttl: Option) -> Result<()> { if !self.playlist.is_alive() { diff --git a/pmoplaylist/src/lib.rs b/pmoplaylist/src/lib.rs index ac0e69cf..090038d6 100644 --- a/pmoplaylist/src/lib.rs +++ b/pmoplaylist/src/lib.rs @@ -47,13 +47,13 @@ mod error; mod handle; mod manager; -mod persistence; -mod playlist; -mod track; -#[cfg(feature = "pmoserver")] -mod sse; #[cfg(feature = "pmoserver")] pub mod openapi; +mod persistence; +mod playlist; +#[cfg(feature = "pmoserver")] +mod sse; +mod track; #[cfg(feature = "pmoconfig")] mod config_ext; @@ -62,10 +62,10 @@ mod config_ext; pub use error::{Error, Result}; pub use handle::{ReadHandle, WriteHandle}; pub use manager::{register_audio_cache, PlaylistManager, PlaylistManager as Manager}; -pub use manager::{PlaylistEvent, PlaylistEventEnvelope, PlaylistEventKind, subscribe_events}; -pub use track::PlaylistTrack; +pub use manager::{subscribe_events, PlaylistEvent, PlaylistEventEnvelope, PlaylistEventKind}; #[cfg(feature = "pmoserver")] pub use sse::playlist_events_router; +pub use track::PlaylistTrack; #[cfg(feature = "pmoconfig")] pub use config_ext::PlaylistConfigExt; diff --git a/pmoplaylist/src/manager.rs b/pmoplaylist/src/manager.rs index dd7253cc..753a6582 100644 --- a/pmoplaylist/src/manager.rs +++ b/pmoplaylist/src/manager.rs @@ -5,15 +5,18 @@ use crate::persistence::PersistenceManager; use crate::playlist::core::PlaylistConfig; use crate::playlist::Playlist; use crate::Result; -use pmocache::{CacheBroadcastEvent, CacheSubscription}; use once_cell::sync::OnceCell; +use pmocache::{CacheBroadcastEvent, CacheSubscription}; use std::collections::HashMap; use std::path::PathBuf; -use std::sync::{Arc, atomic::{AtomicU64, Ordering}}; -use std::time::Duration; -use tokio::sync::RwLock; -use tokio::sync::broadcast; use std::sync::RwLock as StdRwLock; +use std::sync::{ + atomic::{AtomicU64, Ordering}, + Arc, +}; +use std::time::Duration; +use tokio::sync::broadcast; +use tokio::sync::RwLock; /// Singleton PlaylistManager static PLAYLIST_MANAGER: OnceCell = OnceCell::new(); @@ -181,14 +184,16 @@ impl PlaylistManager { /// Notifie tous les callbacks qu'une playlist a changé. pub(crate) fn notify_playlist_changed(&self, id: &str) { - self.notify_playlist_event( - id, - PlaylistEventKind::Updated, - ); + self.notify_playlist_event(id, PlaylistEventKind::Updated); } /// Notifie les callbacks qu'un morceau a été joué pour une playlist donnée. - pub(crate) fn notify_playlist_track_played(&self, playlist_id: &str, cache_pk: &str, qualifier: &str) { + pub(crate) fn notify_playlist_track_played( + &self, + playlist_id: &str, + cache_pk: &str, + qualifier: &str, + ) { self.notify_playlist_event( playlist_id, PlaylistEventKind::TrackPlayed { diff --git a/pmoplaylist/src/sse.rs b/pmoplaylist/src/sse.rs index dc15ae60..b4492a6d 100644 --- a/pmoplaylist/src/sse.rs +++ b/pmoplaylist/src/sse.rs @@ -3,14 +3,14 @@ //! Route type : `GET /api/playlists/events?playlist_id=foo` use crate::{subscribe_events, PlaylistEventKind}; +#[cfg(feature = "pmoserver")] +use async_stream::stream; use axum::{ extract::Query, response::sse::{Event, KeepAlive, Sse}, response::IntoResponse, Router, }; -#[cfg(feature = "pmoserver")] -use async_stream::stream; use serde::{Deserialize, Serialize}; #[cfg(feature = "pmoserver")] use tokio_stream::StreamExt; diff --git a/pmoserver/src/logs/mod.rs b/pmoserver/src/logs/mod.rs index c9b21faa..cfb82fab 100644 --- a/pmoserver/src/logs/mod.rs +++ b/pmoserver/src/logs/mod.rs @@ -23,8 +23,8 @@ use serde::{Deserialize, Serialize}; use tokio::sync::broadcast; use tracing::Level; use tracing_subscriber::{ - filter::EnvFilter, filter::LevelFilter, layer::SubscriberExt, reload, util::SubscriberInitExt, - Registry, + Registry, filter::EnvFilter, filter::LevelFilter, layer::SubscriberExt, reload, + util::SubscriberInitExt, }; /// Représente une entrée de log @@ -66,10 +66,7 @@ impl LogState { if let Err(e) = self.reload_handle.write().unwrap().reload(filter) { eprintln!("❌ Failed to reload log level filter: {}", e); } else { - eprintln!( - "✅ Log level filter reloaded successfully to: {:?}", - level - ); + eprintln!("✅ Log level filter reloaded successfully to: {:?}", level); } } @@ -282,8 +279,10 @@ pub fn init_logging() -> LogState { } else { match EnvFilter::try_new(trimmed) { Ok(filter) => { - let level_hint = - filter.max_level_hint().and_then(levelfilter_to_level).unwrap_or(Level::TRACE); + let level_hint = filter + .max_level_hint() + .and_then(levelfilter_to_level) + .unwrap_or(Level::TRACE); (filter, level_hint, format!("RUST_LOG ({})", trimmed)) } Err(e) => { diff --git a/pmoserver/src/server.rs b/pmoserver/src/server.rs index 3c72be15..f041bdef 100644 --- a/pmoserver/src/server.rs +++ b/pmoserver/src/server.rs @@ -241,12 +241,8 @@ 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 + 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, diff --git a/pmosource/src/api.rs b/pmosource/src/api.rs index 35ba42bb..9beda562 100644 --- a/pmosource/src/api.rs +++ b/pmosource/src/api.rs @@ -955,7 +955,9 @@ async fn stream_source_item_metadata( } }); - Sse::new(stream).keep_alive(KeepAlive::default()).into_response() + Sse::new(stream) + .keep_alive(KeepAlive::default()) + .into_response() } None => ( StatusCode::NOT_FOUND, diff --git a/pmoupnp/src/actions/argument_methods.rs b/pmoupnp/src/actions/argument_methods.rs index 78c956d5..a21ebea7 100644 --- a/pmoupnp/src/actions/argument_methods.rs +++ b/pmoupnp/src/actions/argument_methods.rs @@ -17,7 +17,10 @@ impl UpnpTyped for Argument { impl UpnpObject for Argument { fn to_xml_element(&self) -> Element { // Compat: retourne le premier argument (utile si consommé isolément) - self.to_xml_elements().into_iter().next().unwrap_or_else(|| Element::new("argument")) + self.to_xml_elements() + .into_iter() + .next() + .unwrap_or_else(|| Element::new("argument")) } } diff --git a/pmoupnp/src/upnp_server.rs b/pmoupnp/src/upnp_server.rs index d0ea21f2..f6431ddc 100644 --- a/pmoupnp/src/upnp_server.rs +++ b/pmoupnp/src/upnp_server.rs @@ -318,12 +318,14 @@ impl UpnpServerExt for Server { // API playlists (SSE + OpenAPI) #[cfg(feature = "server")] { - use pmoplaylist::{playlist_events_router, openapi::ApiDoc}; + use pmoplaylist::{openapi::ApiDoc, playlist_events_router}; // SSE /api/playlists/events - self.add_router("/api/playlists", playlist_events_router()).await; + self.add_router("/api/playlists", playlist_events_router()) + .await; // OpenAPI pour playlists let openapi = ApiDoc::openapi(); - self.add_openapi(axum::Router::new(), openapi, "playlists").await; + self.add_openapi(axum::Router::new(), openapi, "playlists") + .await; } // Enregistrer le cache dans le registre global