From c81a4651d604577b03d861b2eb95034b7c3332e4 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sun, 16 Nov 2025 07:59:09 +0100 Subject: [PATCH] Simplification de la gestion des channels. --- Cargo.lock | 1 + pmoaudio-ext/src/sources/playlist_source.rs | 94 ++- pmomediaserver/Cargo.toml | 2 + pmomediaserver/src/paradise_streaming.rs | 4 + pmoparadise/.pmomusic/config.yaml | 12 - pmoparadise/ARCHITECTURE.md | 165 +++++ pmoparadise/Cargo.toml | 2 +- pmoparadise/src/lib.rs | 8 +- pmoparadise/src/models.rs | 12 + pmoparadise/src/playlist_feeder.rs | 200 ++++++ pmoparadise/src/stream_channel.rs | 204 +++--- pmoparadise/src/stream_channel_old.rs | 697 ++++++++++++++++++++ pmoplaylist/src/handle/write.rs | 28 + 13 files changed, 1327 insertions(+), 102 deletions(-) delete mode 100644 pmoparadise/.pmomusic/config.yaml create mode 100644 pmoparadise/ARCHITECTURE.md create mode 100644 pmoparadise/src/playlist_feeder.rs create mode 100644 pmoparadise/src/stream_channel_old.rs diff --git a/Cargo.lock b/Cargo.lock index dc62c18b..319c9689 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3221,6 +3221,7 @@ dependencies = [ "pmocovers", "pmodidl", "pmoparadise", + "pmoplaylist", "pmoqobuz", "pmoserver", "pmosource", diff --git a/pmoaudio-ext/src/sources/playlist_source.rs b/pmoaudio-ext/src/sources/playlist_source.rs index 960b6830..3c8f863b 100644 --- a/pmoaudio-ext/src/sources/playlist_source.rs +++ b/pmoaudio-ext/src/sources/playlist_source.rs @@ -57,12 +57,50 @@ //! # } //! ``` //! +//! # Historique des morceaux joués +//! +//! Utilisez `PlaylistSource::with_history()` pour créer une source qui transfère +//! automatiquement les morceaux joués vers une playlist historique : +//! +//! ```rust,no_run +//! use pmoaudio_ext::PlaylistSource; +//! use pmoplaylist::PlaylistManager; +//! use pmoaudiocache::cache::new_cache; +//! use std::sync::Arc; +//! +//! # async fn example() -> Result<(), Box> { +//! let manager = PlaylistManager::get(); +//! let cache = Arc::new(new_cache("./cache", 500)?); +//! +//! // Playlist live (consommée par la source) +//! let live_read = manager.get_read_handle("radio-live").await?; +//! +//! // Playlist historique (capacité 200 morceaux) +//! let history_write = manager.create_persistent_playlist("radio-history".into()).await?; +//! history_write.set_capacity(Some(200)).await?; +//! +//! // Créer la source avec historique +//! let source = PlaylistSource::with_history( +//! live_read, +//! cache, +//! Arc::new(history_write) +//! ); +//! +//! // Les morceaux joués seront automatiquement ajoutés à "radio-history" +//! # Ok(()) +//! # } +//! ``` +//! +//! **Note** : L'historique utilise `push()` sans TTL. Les morceaux restent dans l'historique +//! jusqu'à ce que la capacité maximale soit atteinte (FIFO). +//! //! # Comportement //! //! - **Polling** : Si la playlist est vide, attend `poll_interval_ms` avant de réessayer //! - **TrackBoundary** : Émet un marqueur avec metadata entre chaque piste //! - **Erreurs** : Si un fichier est inaccessible, émet un `Error` marker et continue //! - **Arrêt** : Via `CancellationToken`, émet `EndOfStream` avant de terminer +//! - **Historique** : Si configuré, ajoute chaque piste jouée à la playlist historique //! //! # Synchronisation //! @@ -98,6 +136,7 @@ pub struct PlaylistSourceLogic { cache: Arc, chunk_frames: usize, poll_interval_ms: u64, + history_playlist: Option>, } impl PlaylistSourceLogic { @@ -112,8 +151,14 @@ impl PlaylistSourceLogic { cache, chunk_frames, poll_interval_ms, + history_playlist: None, } } + + /// Enregistre une playlist historique pour sauvegarder les morceaux joués + pub fn set_history_playlist(&mut self, history: Arc) { + self.history_playlist = Some(history); + } } #[async_trait::async_trait] @@ -233,7 +278,7 @@ 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(); - if let Err(e) = decode_and_emit_track( + match decode_and_emit_track( &file_path, self.chunk_frames, &output, @@ -243,10 +288,28 @@ impl NodeLogic for PlaylistSourceLogic { ) .await { - tracing::error!("PlaylistSourceLogic: error decoding track: {}", e); - let error_marker = AudioSegment::new_error(0, 0.0, format!("Decode error: {}", e)); - send_to_children!(error_marker); - // Continue vers la piste suivante + Ok(()) => { + // 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 { + tracing::warn!( + "PlaylistSourceLogic: failed to add track to history: {}", + e + ); + } else { + tracing::debug!( + "PlaylistSourceLogic: added track {} to history", + cache_pk + ); + } + } + } + Err(e) => { + tracing::error!("PlaylistSourceLogic: error decoding track: {}", e); + let error_marker = AudioSegment::new_error(0, 0.0, format!("Decode error: {}", e)); + send_to_children!(error_marker); + // Continue vers la piste suivante + } } // Boucler pour la piste suivante (pas d'EndOfStream entre pistes !) @@ -613,6 +676,27 @@ impl PlaylistSource { inner: Node::new_source(logic), } } + + /// Crée une nouvelle source avec playlist historique + /// + /// * `playlist_handle` - Handle de lecture sur la playlist live + /// * `cache` - Cache audio contenant les fichiers + /// * `history_playlist` - Handle d'écriture pour l'historique des morceaux joués + /// + /// Après avoir joué chaque morceau, il sera automatiquement ajouté à la playlist historique. + /// La playlist historique utilise push() sans TTL, donc les morceaux y restent jusqu'à + /// ce que la capacité maximale soit atteinte (FIFO). + pub fn with_history( + playlist_handle: ReadHandle, + cache: Arc, + history_playlist: Arc, + ) -> Self { + let mut logic = PlaylistSourceLogic::new(playlist_handle, cache, 0, 100); + logic.set_history_playlist(history_playlist); + Self { + inner: Node::new_source(logic), + } + } } #[async_trait::async_trait] diff --git a/pmomediaserver/Cargo.toml b/pmomediaserver/Cargo.toml index 88d7af76..e25a2196 100644 --- a/pmomediaserver/Cargo.toml +++ b/pmomediaserver/Cargo.toml @@ -28,6 +28,7 @@ pmoconfig = { path = "../pmoconfig", optional = true } anyhow = { version = "1.0", optional = true } pmoaudiocache = { path = "../pmoaudiocache", optional = true } pmocovers = { path = "../pmocovers", optional = true } +pmoplaylist = { path = "../pmoplaylist", optional = true } tokio-util = { version = "0.7", features = ["io"], optional = true } [features] @@ -44,6 +45,7 @@ paradise = [ "dep:anyhow", "dep:pmoaudiocache", "dep:pmocovers", + "dep:pmoplaylist", "dep:tokio-util" ] # Feature pour activer l'API REST de Radio Paradise (en plus de la source UPnP) diff --git a/pmomediaserver/src/paradise_streaming.rs b/pmomediaserver/src/paradise_streaming.rs index 6ef2a28e..823c40ae 100644 --- a/pmomediaserver/src/paradise_streaming.rs +++ b/pmomediaserver/src/paradise_streaming.rs @@ -16,6 +16,7 @@ use axum::{ 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 pmoplaylist::register_audio_cache as register_playlist_audio_cache; use std::sync::Arc; use tokio_util::io::ReaderStream; use tracing::{error, info}; @@ -81,6 +82,8 @@ impl ParadiseStreamingExt for pmoserver::Server { let audio_cache = match get_audio_cache() { Some(cache) => { info!(" ✅ Using existing audio cache singleton"); + // S'assurer qu'il est aussi enregistré dans le playlist manager + register_playlist_audio_cache(cache.clone()); cache } None => { @@ -90,6 +93,7 @@ impl ParadiseStreamingExt for pmoserver::Server { .await .context("Failed to initialize audio cache")?; register_audio_cache(cache.clone()); + register_playlist_audio_cache(cache.clone()); cache } }; diff --git a/pmoparadise/.pmomusic/config.yaml b/pmoparadise/.pmomusic/config.yaml deleted file mode 100644 index ae069315..00000000 --- a/pmoparadise/.pmomusic/config.yaml +++ /dev/null @@ -1,12 +0,0 @@ -host: - http_port: '8080' - cover_cache: - directory: cache_covers - size: 2000 - audio_cache: - directory: cache_audio - size: 500 - logger: - buffer_capacity: 200 - enable_console: true - min_level: INFO diff --git a/pmoparadise/ARCHITECTURE.md b/pmoparadise/ARCHITECTURE.md new file mode 100644 index 00000000..f7dd4f67 --- /dev/null +++ b/pmoparadise/ARCHITECTURE.md @@ -0,0 +1,165 @@ +# Radio Paradise - Architecture + +## Vue d'ensemble + +Cette crate fournit deux architectures pour accéder à Radio Paradise : + +1. **RadioParadiseStreamSource** (legacy) : Télécharge les blocs FLAC entiers et découpe manuellement +2. **RadioParadisePlaylistFeeder** (recommandé) : Utilise les URLs gapless individuelles + système de playlist + +## RadioParadisePlaylistFeeder (Architecture simplifiée) + +### Principe + +Au lieu de télécharger un gros bloc FLAC contenant plusieurs chansons et de calculer manuellement les bornes de chaque chanson, cette architecture : + +1. Récupère le bloc via l'API `get_block` +2. Filtre les chansons : garde uniquement celles où `sched_time_millis + duration >= now()` +3. Télécharge chaque chanson individuellement via son `gapless_url` +4. Stocke les métadonnées (titre, artiste, album, cover) dans le cache audio +5. Push les PKs dans une playlist avec TTL calculé = `sched_end - now()` +6. La playlist est consommée par `PlaylistSource` qui produit le flux audio + +### Avantages + +- **Simplicité** : Pas de calcul de bornes, pas de découpe manuelle +- **Précision** : Chaque fichier FLAC = une chanson exactement +- **Réutilisabilité** : Utilise l'infrastructure existante (pmoplaylist, pmoaudiocache, PlaylistSource) +- **TTL automatique** : Les chansons expirées sont automatiquement retirées de la playlist + +### Exemple d'utilisation + +```rust +use pmoparadise::{RadioParadiseClient, RadioParadisePlaylistFeeder}; +use pmoaudiocache::cache::new_cache; +use pmocovers::cache::new_cache as new_covers_cache; +use pmoaudio_ext::PlaylistSource; +use std::sync::Arc; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + // Créer les caches + let audio_cache = Arc::new(new_cache("./cache/audio", 500)?); + let covers_cache = Arc::new(new_covers_cache("./cache/covers", 500)?); + + // Créer le client Radio Paradise + let client = RadioParadiseClient::new().await?; + + // Créer le feeder (retourne feeder + read_handle) + let (feeder, read_handle) = RadioParadisePlaylistFeeder::new( + client.clone(), + audio_cache.clone(), + covers_cache.clone(), + "rp-live".to_string(), + Some("radio-paradise".to_string()), + ).await?; + + // Lancer le feeder dans une tâche + let feeder = Arc::new(feeder); + let feeder_clone = feeder.clone(); + tokio::spawn(async move { + if let Err(e) = feeder_clone.run().await { + tracing::error!("Feeder error: {}", e); + } + }); + + // Enqueue le bloc actuel + let now_playing = client.now_playing().await?; + feeder.push_block_id(now_playing.block.event); + + // Créer la source audio depuis la playlist + let playlist_source = PlaylistSource::new(read_handle, audio_cache); + + // Utiliser playlist_source dans un pipeline pmoaudio... + + Ok(()) +} +``` + +## Radio Paradise API - Référence des URLs + +### URLs d'artistes + +**Format** : `https://radioparadise.com/music/artist/{artist_id}` +**Format alternatif** : `https://radioparadise.com/music/artist/{artist_id}/{Artist_Name}` + +Le champ `artist_id` est disponible dans `song.song_credit_list[].artist_id`. + +**Exemples** : +- Sting (ID 4247) : https://radioparadise.com/music/artist/4247 +- Pink Martini (ID 3718) : https://radioparadise.com/music/artist/3718/Pink_Martini + +### URLs de chansons + +**Format** : `https://legacy.radioparadise.com/rp3.php?file=songinfo&name=Music&song_id={song_id}` + +Le champ `song_id` est disponible dans `song.song_id`. + +### URLs gapless (FLAC individuels) + +**Format** : Fourni directement par l'API dans `song.gapless_url` + +**Exemple** : `https://audio-geo.radioparadise.com/chan/1/x/1065/4/g/1065-3.flac` + +Ces URLs pointent vers des fichiers FLAC contenant **une seule chanson**, permettant un téléchargement et un traitement simplifiés. + +### Timestamps (`sched_time_millis`) + +Tous les timestamps de l'API Radio Paradise sont en **UTC** (Unix timestamp en millisecondes). + +**Exemple** : +```json +"sched_time_millis": 1763272707000 // 2025-11-16 06:16:09 UTC +``` + +Pour calculer la fin de diffusion d'une chanson : +```rust +let sched_end = song.sched_time_millis + song.duration; +let is_still_playing = sched_end >= now_ms; +``` + +## Notes d'implémentation future + +Ces URLs peuvent être utilisées pour : +- **Enrichir les métadonnées** avec les biographies d'artistes (scraping des pages artistes) +- **Récupérer les paroles** (via l'API ou scraping) +- **Afficher l'historique de diffusion** par chanson +- **Lier vers les pages communautaires** Radio Paradise pour ratings/commentaires +- **Intégration MusicBrainz/Discogs** : utiliser `asin` ou rechercher par artiste+titre+album + +## Structure des données + +### Block + +Un bloc Radio Paradise contient : +- `event` : ID de début du bloc +- `end_event` : ID de fin (= event du bloc suivant) +- `length` : Durée totale en millisecondes +- `url` : URL du bloc FLAC complet (legacy) +- `song` : Map des chansons indexées par position ("0", "1", "2", ...) + +### Song + +Chaque chanson contient : +- **Métadonnées** : `title`, `artist`, `album`, `year`, `rating` +- **Timing** : `elapsed` (position dans le bloc), `duration`, `sched_time_millis` +- **Identifiants** : `song_id`, `audio_id`, `event` +- **Covers** : `cover`, `cover_large`, `cover_medium`, `cover_small` +- **Streaming** : `gapless_url` (⭐ nouveau, recommandé) +- **Artiste** : `artist_id` (pour construire les URLs) + +### Filtrage des chansons + +Pour éviter de télécharger des chansons déjà terminées : + +```rust +let now_ms = SystemTime::now() + .duration_since(UNIX_EPOCH)? + .as_millis() as u64; + +for (idx, song) in block.songs_ordered() { + if song.is_still_playing(now_ms) { + // Télécharger et ajouter à la playlist + } +} +``` diff --git a/pmoparadise/Cargo.toml b/pmoparadise/Cargo.toml index 0e747ff9..98fbe2d9 100644 --- a/pmoparadise/Cargo.toml +++ b/pmoparadise/Cargo.toml @@ -89,7 +89,7 @@ pmoconfig = ["dep:pmoconfig"] # Feature cache (deprecated - toujours actif maintenant) cache = [] # Active le support pmoaudio node (RadioParadiseStreamSource) -pmoaudio = ["dep:pmoaudio", "dep:pmoflac", "dep:pmometadata", "dep:futures-util"] +pmoaudio = ["dep:pmoaudio", "dep:pmoflac", "dep:pmometadata", "dep:futures-util", "dep:pmoaudio-ext"] # Active le support complet avec playlist (pour les exemples avancés) full = ["pmoaudio", "dep:pmoaudio-ext", "pmoconfig", "pmoserver"] diff --git a/pmoparadise/src/lib.rs b/pmoparadise/src/lib.rs index f7896c1f..bca8da94 100644 --- a/pmoparadise/src/lib.rs +++ b/pmoparadise/src/lib.rs @@ -231,6 +231,9 @@ pub mod radio_paradise_stream_source; #[cfg(feature = "pmoaudio")] pub mod stream_channel; +#[cfg(feature = "pmoaudio")] +pub mod playlist_feeder; + // Re-exports for convenience pub use client::{ClientBuilder, RadioParadiseClient}; pub use error::{Error, Result}; @@ -238,7 +241,10 @@ pub use models::{Block, DurationMs, EventId, NowPlaying, Song}; pub use source::RadioParadiseSource; #[cfg(feature = "pmoaudio")] -pub use radio_paradise_stream_source::{RadioParadiseStreamSource, END_OF_BLOCKS_SIGNAL}; +pub use radio_paradise_stream_source::RadioParadiseStreamSource; + +#[cfg(feature = "pmoaudio")] +pub use playlist_feeder::{RadioParadisePlaylistFeeder, END_OF_BLOCKS_SIGNAL}; #[cfg(feature = "pmoaudio")] pub use stream_channel::{ diff --git a/pmoparadise/src/models.rs b/pmoparadise/src/models.rs index aa76ca0d..621ee64b 100644 --- a/pmoparadise/src/models.rs +++ b/pmoparadise/src/models.rs @@ -197,6 +197,18 @@ impl Song { pub fn contains_timestamp(&self, timestamp_ms: DurationMs) -> bool { timestamp_ms >= self.elapsed && timestamp_ms < self.end_time_ms() } + + /// Calcule le timestamp de fin de diffusion (sched_time + duration) + pub fn sched_end_time_ms(&self) -> Option { + self.sched_time_millis.map(|start| start + self.duration) + } + + /// Vérifie si la chanson est encore en lecture ou à venir + pub fn is_still_playing(&self, now_ms: u64) -> bool { + self.sched_end_time_ms() + .map(|end| end >= now_ms) + .unwrap_or(false) + } } /// Image information diff --git a/pmoparadise/src/playlist_feeder.rs b/pmoparadise/src/playlist_feeder.rs new file mode 100644 index 00000000..f02e371e --- /dev/null +++ b/pmoparadise/src/playlist_feeder.rs @@ -0,0 +1,200 @@ +//! RadioParadisePlaylistFeeder - Télécharge et alimente une playlist à partir des blocs RP +//! +//! Architecture simplifiée utilisant les URLs gapless individuelles au lieu du bloc FLAC entier. + +use crate::{client::RadioParadiseClient, models::EventId}; +use pmoaudiocache::Cache as AudioCache; +use pmocovers::Cache as CoversCache; +use pmoplaylist::{PlaylistManager, ReadHandle, WriteHandle}; +use std::{ + collections::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; + +/// Feeder qui télécharge les blocs RP et alimente une playlist +pub struct RadioParadisePlaylistFeeder { + client: RadioParadiseClient, + audio_cache: Arc, + covers_cache: Arc, + playlist_handle: Arc, + block_queue: Arc>>, + notify: Arc, + collection: Option, +} + +impl RadioParadisePlaylistFeeder { + /// Crée un nouveau feeder et retourne (feeder, read_handle) + pub async fn new( + client: RadioParadiseClient, + audio_cache: Arc, + covers_cache: Arc, + playlist_id: String, + collection: Option, + ) -> Result<(Self, ReadHandle)> { + let manager = PlaylistManager::get(); + let write_handle = manager.create_persistent_playlist(playlist_id.clone()).await?; + let read_handle = manager.get_read_handle(&playlist_id).await?; + + Ok(( + Self { + client, + audio_cache, + covers_cache, + playlist_handle: Arc::new(write_handle), + block_queue: Arc::new(tokio::sync::Mutex::new(VecDeque::new())), + notify: Arc::new(Notify::new()), + collection, + }, + read_handle, + )) + } + + /// Enqueue un bloc pour traitement + pub async fn push_block_id(&self, event_id: EventId) { + { + let mut queue = self.block_queue.lock().await; + queue.push_back(event_id); + } + self.notify.notify_one(); + } + + /// Boucle principale de traitement (à exécuter dans une tâche tokio) + pub async fn run(self: Arc) -> Result<()> { + loop { + // Attendre un bloc + let event_id = loop { + { + 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"); + return Ok(()); + } + break id; + } + } + self.notify.notified().await; + }; + + // Traiter le bloc + if let Err(e) = self.process_block(event_id).await { + tracing::error!("RadioParadisePlaylistFeeder: Failed to process block {}: {}", event_id, e); + } + } + } + + /// Traite un bloc : fetch, filtre, download, push playlist + async fn process_block(&self, event_id: EventId) -> Result<()> { + tracing::info!("RadioParadisePlaylistFeeder: Processing block {}", event_id); + + // 1. Fetch le bloc + 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; + + // 3. Filtrer les chansons encore en lecture ou à venir + let songs = block.songs_ordered(); + let mut processed = 0; + + for (idx, song) in songs { + 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) + ); + continue; + } + + // 4. Télécharger la chanson + 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 + ); + + let pk = self.audio_cache + .add_from_url(gapless_url, self.collection.as_deref()) + .await?; + + // 5. Sauvegarder les métadonnées + self.save_metadata(&pk, song, &block).await?; + + // 6. Calculer le TTL + 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); + + // 7. Push dans la playlist avec TTL + self.playlist_handle.push_with_ttl(pk.clone(), ttl).await?; + + tracing::info!( + "RadioParadisePlaylistFeeder: Added {} to playlist (pk={}, ttl={}s)", + song.title, pk, ttl.as_secs() + ); + + processed += 1; + } + + tracing::info!( + "RadioParadisePlaylistFeeder: Processed block {} - added {} songs to playlist", + event_id, processed + ); + + Ok(()) + } + + /// Sauvegarde les métadonnées dans le cache audio + async fn save_metadata( + &self, + pk: &str, + song: &crate::models::Song, + block: &crate::models::Block, + ) -> Result<()> { + use pmoaudiocache::AudioTrackMetadataExt; + + let metadata = self.audio_cache.track_metadata(pk); + let mut meta = metadata.write().await; + + // Métadonnées de base + meta.set_title(Some(song.title.clone())).await?; + meta.set_artist(Some(song.artist.clone())).await?; + if let Some(ref album) = song.album { + meta.set_album(Some(album.clone())).await?; + } + if let Some(year) = song.year { + meta.set_year(Some(year)).await?; + } + + // Cover + if let Some(ref cover_large) = song.cover_large { + if let Some(cover_url) = block.cover_url(cover_large) { + 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 { + Ok(cover_pk) => { + meta.set_cover_pk(Some(cover_pk)).await?; + tracing::debug!("RadioParadisePlaylistFeeder: Cached cover for {}", song.title); + } + Err(e) => { + tracing::warn!("RadioParadisePlaylistFeeder: Failed to cache cover: {}", e); + } + } + } + } + + Ok(()) + } +} diff --git a/pmoparadise/src/stream_channel.rs b/pmoparadise/src/stream_channel.rs index 7e7b49b0..b004ed5b 100644 --- a/pmoparadise/src/stream_channel.rs +++ b/pmoparadise/src/stream_channel.rs @@ -1,3 +1,9 @@ +//! Version simplifiée de stream_channel.rs utilisant RadioParadisePlaylistFeeder + PlaylistSource +//! +//! Cette version remplace l'architecture complexe RadioParadiseStreamSource par : +//! - RadioParadisePlaylistFeeder : télécharge les URLs gapless et alimente une playlist +//! - PlaylistSource::with_history() : lit la playlist et gère l'historique automatiquement + use std::{ collections::HashMap, pin::Pin, @@ -12,19 +18,19 @@ use std::{ use crate::{ channels::{ChannelDescriptor, ParadiseChannelKind, ALL_CHANNELS}, client::RadioParadiseClient, - radio_paradise_stream_source::RadioParadiseStreamSource, + playlist_feeder::RadioParadisePlaylistFeeder, }; use anyhow::{anyhow, Result}; -use pmoaudio::{nodes::DEFAULT_CHANNEL_SIZE, AudioPipelineNode}; +use pmoaudio::AudioPipelineNode; use pmoaudio_ext::{ - FlacCacheSink, FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, - OggFlacStreamHandle, PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, + FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, OggFlacStreamHandle, + PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, TrackBoundaryCoverNode, }; use pmoaudiocache::Cache as AudioCache; use pmocovers::Cache as CoverCache; use pmoflac::EncoderOptions; -use pmoplaylist::WriteHandle; +use pmoplaylist::PlaylistManager; use thiserror::Error; use tokio::io::{AsyncRead, ReadBuf}; use tokio::sync::Notify; @@ -52,9 +58,9 @@ pub struct ParadiseHistoryOptions { pub audio_cache: Arc, pub cover_cache: Arc, pub playlist_id: String, - pub playlist_writer: WriteHandle, pub collection: Option, pub replay_max_lead_seconds: f64, + pub max_history_tracks: Option, } /// Builder pratique pour configurer automatiquement les playlists historiques. @@ -87,19 +93,6 @@ impl ParadiseHistoryBuilder { descriptor: &ChannelDescriptor, ) -> Result { let playlist_id = format!("{}-{}", self.playlist_prefix, descriptor.slug); - let manager = pmoplaylist::PlaylistManager(); - let writer = manager - .get_persistent_write_handle(playlist_id.clone()) - .await?; - - if let Some(prefix) = &self.playlist_title_prefix { - let title = format!("{} - {}", prefix, descriptor.display_name); - writer.set_title(title).await?; - } - - if let Some(capacity) = self.max_history_tracks { - writer.set_capacity(Some(capacity)).await?; - } let collection = self .collection_prefix @@ -110,19 +103,13 @@ impl ParadiseHistoryBuilder { audio_cache: self.audio_cache.clone(), cover_cache: self.cover_cache.clone(), playlist_id, - playlist_writer: writer, collection, replay_max_lead_seconds: self.replay_max_lead_seconds, + max_history_tracks: self.max_history_tracks, }) } } -struct HistoryState { - playlist_id: String, - audio_cache: Arc, - replay_max_lead_seconds: f64, -} - #[cfg(feature = "pmoconfig")] impl ParadiseStreamChannelConfig { pub fn from_config(cfg: &pmoconfig::Config, channel: ParadiseChannelKind) -> Self { @@ -169,26 +156,73 @@ impl ParadiseStreamChannelConfig { } /// Stream complet (FLAC pur + OGG-FLAC) pour un canal Radio Paradise. +/// +/// Version simplifiée utilisant RadioParadisePlaylistFeeder + PlaylistSource pub struct ParadiseStreamChannel { descriptor: ChannelDescriptor, state: Arc, pipeline_handle: JoinHandle<()>, feeder_handle: JoinHandle<()>, - history: Option, } impl ParadiseStreamChannel { /// Crée un canal avec client déjà configuré. - pub fn with_client( + pub async fn with_client( descriptor: ChannelDescriptor, client: RadioParadiseClient, config: ParadiseStreamChannelConfig, cover_cache: Option>, history: Option, - ) -> Self { - let mut source = RadioParadiseStreamSource::new(client.clone()); - let block_handle = source.block_handle(); + ) -> Result { + let manager = PlaylistManager::get(); + // 1. Créer la playlist live pour ce canal + let live_playlist_id = format!("radio-paradise-live-{}", descriptor.slug); + let (feeder, live_read) = if let Some(ref history_opts) = history { + RadioParadisePlaylistFeeder::new( + client.clone(), + history_opts.audio_cache.clone(), + history_opts.cover_cache.clone(), + live_playlist_id.clone(), + history_opts.collection.clone(), + ) + .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)")); + }; + + let feeder = Arc::new(feeder); + + // 2. Créer/récupérer la playlist historique si activée + let history_write = if let Some(ref history_opts) = history { + let write = manager + .get_persistent_write_handle(history_opts.playlist_id.clone()) + .await?; + + // Configurer la capacité + if let Some(capacity) = history_opts.max_history_tracks { + write.set_capacity(Some(capacity)).await?; + } + + // Configurer le titre + let title = format!("Radio Paradise History - {}", descriptor.display_name); + write.set_title(title).await?; + + Some(Arc::new(write)) + } else { + None + }; + + // 3. Créer la source playlist avec historique + let audio_cache = history.as_ref().unwrap().audio_cache.clone(); + let mut source = if let Some(history_write) = history_write.clone() { + PlaylistSource::with_history(live_read, audio_cache.clone(), history_write) + } else { + PlaylistSource::new(live_read, audio_cache.clone()) + }; + + // 4. Créer les sinks de broadcast (FLAC + OGG) let (flac_sink, stream_handle) = StreamingFlacSink::with_max_broadcast_lead( EncoderOptions::default(), 16, @@ -204,33 +238,7 @@ impl ParadiseStreamChannel { downstream_children.push(Box::new(flac_sink)); downstream_children.push(Box::new(ogg_sink)); - let mut history_state = None; - - if let Some(history_opts) = history { - let ParadiseHistoryOptions { - audio_cache, - cover_cache, - playlist_id, - playlist_writer, - collection, - replay_max_lead_seconds, - } = history_opts; - let mut cache_sink = FlacCacheSink::with_config( - audio_cache.clone(), - cover_cache, - DEFAULT_CHANNEL_SIZE, - EncoderOptions::default(), - collection, - ); - cache_sink.register_playlist(playlist_writer); - downstream_children.push(Box::new(cache_sink)); - history_state = Some(HistoryState { - playlist_id, - audio_cache, - replay_max_lead_seconds, - }); - } - + // 5. Optionnel : ajouter le nœud de cache de covers if let Some(cache) = cover_cache { let mut cover_node = TrackBoundaryCoverNode::new(cache); for child in downstream_children { @@ -242,9 +250,11 @@ impl ParadiseStreamChannel { source.register(child); } } + stream_handle.set_auto_stop(false); ogg_handle.set_auto_stop(false); + // 6. Lancer le pipeline audio let stop_token = CancellationToken::new(); let pipeline_stop = stop_token.clone(); let pipeline_handle = tokio::spawn(async move { @@ -264,26 +274,36 @@ impl ParadiseStreamChannel { descriptor, config, client, - block_handle, + feeder: feeder.clone(), stream_handle, ogg_handle, + history_playlist_id: history.map(|h| h.playlist_id), + history_audio_cache: history_write.map(|_| audio_cache), active_clients: AtomicUsize::new(0), activity_notify: Notify::new(), stop_token, }); + // 7. Lancer le feeder qui traite les blocs + let feeder_runner = feeder.clone(); + tokio::spawn(async move { + if let Err(e) = feeder_runner.run().await { + error!("RadioParadisePlaylistFeeder error: {}", e); + } + }); + + // 8. Lancer le scheduler qui enqueue les blocs let feeder_state = state.clone(); let feeder_handle = tokio::spawn(async move { feeder_state.run_scheduler().await; }); - Self { + Ok(Self { descriptor, state, pipeline_handle, feeder_handle, - history: history_state, - } + }) } /// Crée un canal en construisant automatiquement le client pour ce descriptor. @@ -297,13 +317,13 @@ impl ParadiseStreamChannel { .channel(descriptor.id) .build() .await?; - Ok(Self::with_client( + Self::with_client( descriptor, client, config, cover_cache, history, - )) + ).await } /// S'abonne au flux FLAC pur. @@ -346,32 +366,40 @@ impl ParadiseStreamChannel { &self, client_id: &str, ) -> Result { - let history = self - .history + let history_id = self + .state + .history_playlist_id .as_ref() .ok_or(HistoryStreamError::HistoryDisabled)?; + + let audio_cache = self + .state + .history_audio_cache + .as_ref() + .ok_or(HistoryStreamError::HistoryDisabled)?; + tracing::info!( "Starting historical FLAC replay for channel {} (client_id={})", self.descriptor.display_name, client_id ); - let reader = pmoplaylist::PlaylistManager() - .get_read_handle(&history.playlist_id) + let reader = pmoplaylist::PlaylistManager::get() + .get_read_handle(history_id) .await .map_err(|e| HistoryStreamError::Playlist(e.to_string()))?; - let mut source = PlaylistSource::new(reader, history.audio_cache.clone()); + + let mut source = PlaylistSource::new(reader, audio_cache.clone()); let (flac_sink, handle) = StreamingFlacSink::with_max_broadcast_lead( EncoderOptions::default(), 16, - history.replay_max_lead_seconds, + self.state.config.max_lead_seconds, ); source.register(Box::new(flac_sink)); let stop_token = CancellationToken::new(); - let mut pipeline_source = source; let stop_clone = stop_token.clone(); let pipeline = tokio::spawn(async move { - let _ = Box::new(pipeline_source).run(stop_clone).await; + let _ = Box::new(source).run(stop_clone).await; }); let stream = handle.subscribe_flac(); Ok(HistoryFlacStream::new(stream, stop_token, pipeline)) @@ -382,32 +410,40 @@ impl ParadiseStreamChannel { &self, client_id: &str, ) -> Result { - let history = self - .history + let history_id = self + .state + .history_playlist_id .as_ref() .ok_or(HistoryStreamError::HistoryDisabled)?; + + let audio_cache = self + .state + .history_audio_cache + .as_ref() + .ok_or(HistoryStreamError::HistoryDisabled)?; + tracing::info!( "Starting historical OGG replay for channel {} (client_id={})", self.descriptor.display_name, client_id ); - let reader = pmoplaylist::PlaylistManager() - .get_read_handle(&history.playlist_id) + let reader = pmoplaylist::PlaylistManager::get() + .get_read_handle(history_id) .await .map_err(|e| HistoryStreamError::Playlist(e.to_string()))?; - let mut source = PlaylistSource::new(reader, history.audio_cache.clone()); + + let mut source = PlaylistSource::new(reader, audio_cache.clone()); let (ogg_sink, handle) = StreamingOggFlacSink::with_max_broadcast_lead( EncoderOptions::default(), 16, - history.replay_max_lead_seconds, + self.state.config.max_lead_seconds, ); source.register(Box::new(ogg_sink)); let stop_token = CancellationToken::new(); - let mut pipeline_source = source; let stop_clone = stop_token.clone(); let pipeline = tokio::spawn(async move { - let _ = Box::new(pipeline_source).run(stop_clone).await; + let _ = Box::new(source).run(stop_clone).await; }); let stream = handle.subscribe(); Ok(HistoryOggStream::new(stream, stop_token, pipeline)) @@ -426,9 +462,11 @@ struct ChannelState { descriptor: ChannelDescriptor, config: ParadiseStreamChannelConfig, client: RadioParadiseClient, - block_handle: crate::radio_paradise_stream_source::BlockQueueHandle, + feeder: Arc, stream_handle: StreamHandle, ogg_handle: OggFlacStreamHandle, + history_playlist_id: Option, + history_audio_cache: Option>, active_clients: AtomicUsize, activity_notify: Notify, stop_token: CancellationToken, @@ -472,7 +510,7 @@ impl ChannelState { "Channel {} streaming block {}", self.descriptor.display_name, block.event ); - self.block_handle.enqueue(block.event); + self.feeder.push_block_id(block.event).await; let mut next_event = block.end_event; loop { @@ -486,7 +524,7 @@ impl ChannelState { match self.client.get_block(Some(next_event)).await { Ok(next_block) => { - self.block_handle.enqueue(next_block.event); + self.feeder.push_block_id(next_block.event).await; next_event = next_block.end_event; backoff = Duration::from_secs(5); } diff --git a/pmoparadise/src/stream_channel_old.rs b/pmoparadise/src/stream_channel_old.rs new file mode 100644 index 00000000..7e7b49b0 --- /dev/null +++ b/pmoparadise/src/stream_channel_old.rs @@ -0,0 +1,697 @@ +use std::{ + collections::HashMap, + pin::Pin, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, + task::{Context, Poll}, + time::Duration, +}; + +use crate::{ + channels::{ChannelDescriptor, ParadiseChannelKind, ALL_CHANNELS}, + client::RadioParadiseClient, + radio_paradise_stream_source::RadioParadiseStreamSource, +}; +use anyhow::{anyhow, Result}; +use pmoaudio::{nodes::DEFAULT_CHANNEL_SIZE, AudioPipelineNode}; +use pmoaudio_ext::{ + FlacCacheSink, FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, + OggFlacStreamHandle, PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, + TrackBoundaryCoverNode, +}; +use pmoaudiocache::Cache as AudioCache; +use pmocovers::Cache as CoverCache; +use pmoflac::EncoderOptions; +use pmoplaylist::WriteHandle; +use thiserror::Error; +use tokio::io::{AsyncRead, ReadBuf}; +use tokio::sync::Notify; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; +use tracing::{error, info, warn}; + +/// Configuration pour un canal Radio Paradise. +#[derive(Clone, Debug)] +pub struct ParadiseStreamChannelConfig { + /// Durée maximale (en secondes) d'avance acceptée par le broadcast. + pub max_lead_seconds: f64, +} + +impl Default for ParadiseStreamChannelConfig { + fn default() -> Self { + Self { + max_lead_seconds: 1.0, + } + } +} + +/// Options pour activer l'archivage/historique d'un canal. +pub struct ParadiseHistoryOptions { + pub audio_cache: Arc, + pub cover_cache: Arc, + pub playlist_id: String, + pub playlist_writer: WriteHandle, + pub collection: Option, + pub replay_max_lead_seconds: f64, +} + +/// Builder pratique pour configurer automatiquement les playlists historiques. +#[derive(Clone)] +pub struct ParadiseHistoryBuilder { + pub audio_cache: Arc, + pub cover_cache: Arc, + pub playlist_prefix: String, + pub playlist_title_prefix: Option, + pub max_history_tracks: Option, + pub collection_prefix: Option, + pub replay_max_lead_seconds: f64, +} + +impl ParadiseHistoryBuilder { + pub fn new(audio_cache: Arc, cover_cache: Arc) -> Self { + Self { + audio_cache, + cover_cache, + playlist_prefix: "radio-paradise-history".into(), + playlist_title_prefix: Some("Radio Paradise History".into()), + max_history_tracks: Some(500), + collection_prefix: Some("radio-paradise".into()), + replay_max_lead_seconds: 1.0, + } + } + + pub async fn build_for_channel( + &self, + descriptor: &ChannelDescriptor, + ) -> Result { + let playlist_id = format!("{}-{}", self.playlist_prefix, descriptor.slug); + let manager = pmoplaylist::PlaylistManager(); + let writer = manager + .get_persistent_write_handle(playlist_id.clone()) + .await?; + + if let Some(prefix) = &self.playlist_title_prefix { + let title = format!("{} - {}", prefix, descriptor.display_name); + writer.set_title(title).await?; + } + + if let Some(capacity) = self.max_history_tracks { + writer.set_capacity(Some(capacity)).await?; + } + + let collection = self + .collection_prefix + .as_ref() + .map(|prefix| format!("{}-{}", prefix, descriptor.slug)); + + Ok(ParadiseHistoryOptions { + audio_cache: self.audio_cache.clone(), + cover_cache: self.cover_cache.clone(), + playlist_id, + playlist_writer: writer, + collection, + replay_max_lead_seconds: self.replay_max_lead_seconds, + }) + } +} + +struct HistoryState { + playlist_id: String, + audio_cache: Arc, + replay_max_lead_seconds: f64, +} + +#[cfg(feature = "pmoconfig")] +impl ParadiseStreamChannelConfig { + pub fn from_config(cfg: &pmoconfig::Config, channel: ParadiseChannelKind) -> Self { + use serde_yaml::Value; + let path = [ + "sources", + "radio_paradise", + "channels", + channel.slug(), + "max_lead_seconds", + ]; + match cfg.get_value(&path) { + Ok(Value::Number(num)) => { + if let Some(v) = num.as_f64() { + Self { + max_lead_seconds: v.max(0.1), + } + } else { + let default = Self::default(); + let _ = + cfg.set_value(&path, Value::String(default.max_lead_seconds.to_string())); + default + } + } + Ok(Value::String(s)) => { + if let Ok(v) = s.parse::() { + Self { + max_lead_seconds: v.max(0.1), + } + } else { + let default = Self::default(); + let _ = + cfg.set_value(&path, Value::String(default.max_lead_seconds.to_string())); + default + } + } + _ => { + let default = Self::default(); + let _ = cfg.set_value(&path, Value::String(default.max_lead_seconds.to_string())); + default + } + } + } +} + +/// Stream complet (FLAC pur + OGG-FLAC) pour un canal Radio Paradise. +pub struct ParadiseStreamChannel { + descriptor: ChannelDescriptor, + state: Arc, + pipeline_handle: JoinHandle<()>, + feeder_handle: JoinHandle<()>, + history: Option, +} + +impl ParadiseStreamChannel { + /// Crée un canal avec client déjà configuré. + pub fn with_client( + descriptor: ChannelDescriptor, + client: RadioParadiseClient, + config: ParadiseStreamChannelConfig, + cover_cache: Option>, + history: Option, + ) -> Self { + let mut source = RadioParadiseStreamSource::new(client.clone()); + let block_handle = source.block_handle(); + + let (flac_sink, stream_handle) = StreamingFlacSink::with_max_broadcast_lead( + EncoderOptions::default(), + 16, + config.max_lead_seconds, + ); + let (ogg_sink, ogg_handle) = StreamingOggFlacSink::with_max_broadcast_lead( + EncoderOptions::default(), + 16, + config.max_lead_seconds, + ); + + let mut downstream_children: Vec> = Vec::new(); + downstream_children.push(Box::new(flac_sink)); + downstream_children.push(Box::new(ogg_sink)); + + let mut history_state = None; + + if let Some(history_opts) = history { + let ParadiseHistoryOptions { + audio_cache, + cover_cache, + playlist_id, + playlist_writer, + collection, + replay_max_lead_seconds, + } = history_opts; + let mut cache_sink = FlacCacheSink::with_config( + audio_cache.clone(), + cover_cache, + DEFAULT_CHANNEL_SIZE, + EncoderOptions::default(), + collection, + ); + cache_sink.register_playlist(playlist_writer); + downstream_children.push(Box::new(cache_sink)); + history_state = Some(HistoryState { + playlist_id, + audio_cache, + replay_max_lead_seconds, + }); + } + + if let Some(cache) = cover_cache { + let mut cover_node = TrackBoundaryCoverNode::new(cache); + for child in downstream_children { + cover_node.register(child); + } + source.register(Box::new(cover_node)); + } else { + for child in downstream_children { + source.register(child); + } + } + stream_handle.set_auto_stop(false); + ogg_handle.set_auto_stop(false); + + let stop_token = CancellationToken::new(); + let pipeline_stop = stop_token.clone(); + let pipeline_handle = tokio::spawn(async move { + info!( + "RadioParadise stream pipeline started for channel {}", + descriptor.display_name + ); + if let Err(e) = Box::new(source).run(pipeline_stop).await { + error!( + "Pipeline error for channel {}: {}", + descriptor.display_name, e + ); + } + }); + + let state = Arc::new(ChannelState { + descriptor, + config, + client, + block_handle, + stream_handle, + ogg_handle, + active_clients: AtomicUsize::new(0), + activity_notify: Notify::new(), + stop_token, + }); + + let feeder_state = state.clone(); + let feeder_handle = tokio::spawn(async move { + feeder_state.run_scheduler().await; + }); + + Self { + descriptor, + state, + pipeline_handle, + feeder_handle, + history: history_state, + } + } + + /// Crée un canal en construisant automatiquement le client pour ce descriptor. + pub async fn new( + descriptor: ChannelDescriptor, + config: ParadiseStreamChannelConfig, + cover_cache: Option>, + history: Option, + ) -> Result { + let client = RadioParadiseClient::builder() + .channel(descriptor.id) + .build() + .await?; + Ok(Self::with_client( + descriptor, + client, + config, + cover_cache, + history, + )) + } + + /// S'abonne au flux FLAC pur. + pub fn subscribe_flac(&self) -> ChannelFlacStream { + self.state.on_client_added(); + let inner = self.state.stream_handle.subscribe_flac(); + ChannelFlacStream::new(inner, self.state.clone()) + } + + /// S'abonne au flux FLAC + ICY metadata. + pub fn subscribe_icy(&self) -> ChannelIcyStream { + self.state.on_client_added(); + let inner = self.state.stream_handle.subscribe_icy(); + ChannelIcyStream::new(inner, self.state.clone()) + } + + /// S'abonne au flux OGG-FLAC. + pub fn subscribe_ogg(&self) -> ChannelOggStream { + self.state.on_client_added(); + let inner = self.state.ogg_handle.subscribe(); + ChannelOggStream::new(inner, self.state.clone()) + } + + /// Snapshot des métadonnées actuelles. + pub async fn metadata(&self) -> MetadataSnapshot { + self.state.stream_handle.get_metadata().await + } + + /// Nombre de clients actifs. + pub fn active_clients(&self) -> usize { + self.state.active_clients.load(Ordering::SeqCst) + } + + pub fn descriptor(&self) -> ChannelDescriptor { + self.descriptor + } + + /// Lance un pipeline dédié pour rejouer l'historique (FLAC pur) pour un client. + pub async fn stream_history_flac( + &self, + client_id: &str, + ) -> Result { + let history = self + .history + .as_ref() + .ok_or(HistoryStreamError::HistoryDisabled)?; + tracing::info!( + "Starting historical FLAC replay for channel {} (client_id={})", + self.descriptor.display_name, + client_id + ); + + let reader = pmoplaylist::PlaylistManager() + .get_read_handle(&history.playlist_id) + .await + .map_err(|e| HistoryStreamError::Playlist(e.to_string()))?; + let mut source = PlaylistSource::new(reader, history.audio_cache.clone()); + let (flac_sink, handle) = StreamingFlacSink::with_max_broadcast_lead( + EncoderOptions::default(), + 16, + history.replay_max_lead_seconds, + ); + source.register(Box::new(flac_sink)); + let stop_token = CancellationToken::new(); + let mut pipeline_source = source; + let stop_clone = stop_token.clone(); + let pipeline = tokio::spawn(async move { + let _ = Box::new(pipeline_source).run(stop_clone).await; + }); + let stream = handle.subscribe_flac(); + Ok(HistoryFlacStream::new(stream, stop_token, pipeline)) + } + + /// Lance un pipeline dédié pour rejouer l'historique (OGG-FLAC) pour un client. + pub async fn stream_history_ogg( + &self, + client_id: &str, + ) -> Result { + let history = self + .history + .as_ref() + .ok_or(HistoryStreamError::HistoryDisabled)?; + tracing::info!( + "Starting historical OGG replay for channel {} (client_id={})", + self.descriptor.display_name, + client_id + ); + + let reader = pmoplaylist::PlaylistManager() + .get_read_handle(&history.playlist_id) + .await + .map_err(|e| HistoryStreamError::Playlist(e.to_string()))?; + let mut source = PlaylistSource::new(reader, history.audio_cache.clone()); + let (ogg_sink, handle) = StreamingOggFlacSink::with_max_broadcast_lead( + EncoderOptions::default(), + 16, + history.replay_max_lead_seconds, + ); + source.register(Box::new(ogg_sink)); + let stop_token = CancellationToken::new(); + let mut pipeline_source = source; + let stop_clone = stop_token.clone(); + let pipeline = tokio::spawn(async move { + let _ = Box::new(pipeline_source).run(stop_clone).await; + }); + let stream = handle.subscribe(); + Ok(HistoryOggStream::new(stream, stop_token, pipeline)) + } +} + +impl Drop for ParadiseStreamChannel { + fn drop(&mut self) { + self.state.stop_token.cancel(); + self.pipeline_handle.abort(); + self.feeder_handle.abort(); + } +} + +struct ChannelState { + descriptor: ChannelDescriptor, + config: ParadiseStreamChannelConfig, + client: RadioParadiseClient, + block_handle: crate::radio_paradise_stream_source::BlockQueueHandle, + stream_handle: StreamHandle, + ogg_handle: OggFlacStreamHandle, + active_clients: AtomicUsize, + activity_notify: Notify, + stop_token: CancellationToken, +} + +impl ChannelState { + fn on_client_added(&self) { + if self.active_clients.fetch_add(1, Ordering::SeqCst) == 0 { + self.activity_notify.notify_one(); + } + } + + fn on_client_removed(&self) { + self.active_clients.fetch_sub(1, Ordering::SeqCst); + } + + async fn wait_for_clients(&self) -> bool { + while self.active_clients.load(Ordering::SeqCst) == 0 { + tokio::select! { + _ = self.stop_token.cancelled() => return false, + _ = self.activity_notify.notified() => {}, + } + } + true + } + + async fn run_scheduler(self: Arc) { + let mut backoff = Duration::from_secs(5); + loop { + if self.stop_token.is_cancelled() { + break; + } + + if !self.wait_for_clients().await { + break; + } + + match self.client.get_block(None).await { + Ok(block) => { + info!( + "Channel {} streaming block {}", + self.descriptor.display_name, block.event + ); + self.block_handle.enqueue(block.event); + let mut next_event = block.end_event; + + loop { + if self.stop_token.is_cancelled() { + return; + } + + if self.active_clients.load(Ordering::SeqCst) == 0 { + break; + } + + match self.client.get_block(Some(next_event)).await { + Ok(next_block) => { + self.block_handle.enqueue(next_block.event); + next_event = next_block.end_event; + backoff = Duration::from_secs(5); + } + Err(e) => { + warn!( + "Failed to fetch next block for channel {}: {}", + self.descriptor.display_name, e + ); + tokio::select! { + _ = self.stop_token.cancelled() => return, + _ = tokio::time::sleep(backoff) => {}, + } + backoff = (backoff * 2).min(Duration::from_secs(60)); + } + } + } + } + Err(e) => { + warn!( + "Failed to fetch current block for channel {}: {}", + self.descriptor.display_name, e + ); + tokio::select! { + _ = self.stop_token.cancelled() => break, + _ = tokio::time::sleep(backoff) => {}, + } + backoff = (backoff * 2).min(Duration::from_secs(60)); + } + } + } + } +} + +macro_rules! wrap_stream { + ($name:ident, $inner:ty) => { + pub struct $name { + inner: $inner, + state: Arc, + } + + impl $name { + fn new(inner: $inner, state: Arc) -> Self { + Self { inner, state } + } + } + + impl AsyncRead for $name { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } + } + + impl Drop for $name { + fn drop(&mut self) { + self.state.on_client_removed(); + } + } + }; +} + +wrap_stream!(ChannelFlacStream, FlacClientStream); +wrap_stream!(ChannelIcyStream, IcyClientStream); +wrap_stream!(ChannelOggStream, OggFlacClientStream); + +#[derive(Debug, Error)] +pub enum HistoryStreamError { + #[error("history replay not enabled for this channel")] + HistoryDisabled, + #[error("playlist error: {0}")] + Playlist(String), +} + +pub struct HistoryFlacStream { + inner: FlacClientStream, + stop_token: CancellationToken, + pipeline: Option>, +} + +impl HistoryFlacStream { + fn new( + inner: FlacClientStream, + stop_token: CancellationToken, + pipeline: JoinHandle<()>, + ) -> Self { + Self { + inner, + stop_token, + pipeline: Some(pipeline), + } + } +} + +impl AsyncRead for HistoryFlacStream { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +impl Unpin for HistoryFlacStream {} + +impl Drop for HistoryFlacStream { + fn drop(&mut self) { + self.stop_token.cancel(); + if let Some(handle) = self.pipeline.take() { + handle.abort(); + } + } +} + +pub struct HistoryOggStream { + inner: OggFlacClientStream, + stop_token: CancellationToken, + pipeline: Option>, +} + +impl HistoryOggStream { + fn new( + inner: OggFlacClientStream, + stop_token: CancellationToken, + pipeline: JoinHandle<()>, + ) -> Self { + Self { + inner, + stop_token, + pipeline: Some(pipeline), + } + } +} + +impl AsyncRead for HistoryOggStream { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +impl Unpin for HistoryOggStream {} + +impl Drop for HistoryOggStream { + fn drop(&mut self) { + self.stop_token.cancel(); + if let Some(handle) = self.pipeline.take() { + handle.abort(); + } + } +} + +/// Gestionnaire multi-canaux. +pub struct ParadiseChannelManager { + channels: HashMap>, +} + +impl ParadiseChannelManager { + pub fn new(channels: HashMap>) -> Self { + Self { channels } + } + + pub async fn with_defaults_with_cover_cache( + cover_cache: Option>, + history_builder: Option, + ) -> Result { + let mut map = HashMap::new(); + for descriptor in ALL_CHANNELS.iter().copied() { + let history_opts = if let Some(builder) = &history_builder { + Some( + builder + .build_for_channel(&descriptor) + .await + .map_err(|e| anyhow!("Failed to init history playlist: {}", e))?, + ) + } else { + None + }; + let channel = ParadiseStreamChannel::new( + descriptor, + ParadiseStreamChannelConfig::default(), + cover_cache.clone(), + history_opts, + ) + .await?; + map.insert(descriptor.id, Arc::new(channel)); + } + Ok(Self { channels: map }) + } + + pub async fn with_defaults() -> Result { + Self::with_defaults_with_cover_cache(None, None).await + } + + pub fn get(&self, id: u8) -> Option> { + self.channels.get(&id).cloned() + } + + pub fn iter(&self) -> impl Iterator> { + self.channels.values() + } +} diff --git a/pmoplaylist/src/handle/write.rs b/pmoplaylist/src/handle/write.rs index 48feb2f5..366016d1 100644 --- a/pmoplaylist/src/handle/write.rs +++ b/pmoplaylist/src/handle/write.rs @@ -50,6 +50,34 @@ impl WriteHandle { Ok(()) } + /// Ajoute un morceau avec un TTL personnalisé + pub async fn push_with_ttl(&self, cache_pk: String, ttl: Duration) -> Result<()> { + if !self.playlist.is_alive() { + return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); + } + + // Vérifier que le pk existe dans le cache + let cache = crate::manager::audio_cache()?; + if !cache.is_valid_pk(&cache_pk).await { + return Err(crate::Error::CacheEntryNotFound(cache_pk)); + } + + // Ajouter à la playlist avec TTL + let record = Record::with_ttl(cache_pk, ttl); + let mut core = self.playlist.core.write().await; + core.push(record); + drop(core); + + self.playlist.touch().await; + + // Sauvegarder si persistante + if self.playlist.persistent { + self.save_to_db().await?; + } + + Ok(()) + } + /// Ajoute plusieurs morceaux de manière atomique pub async fn push_set(&self, cache_pks: Vec) -> Result<()> { if !self.playlist.is_alive() {