eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Simplification de la gestion des channels.

Browse Source
This commit is contained in:
Eric Coissac
2025-11-16 07:59:09 +01:00
parent 97a383c079
commit c81a4651d6
13 changed files with 1327 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -3221,6 +3221,7 @@ dependencies = [
"pmocovers",
"pmodidl",
"pmoparadise",
"pmoplaylist",
"pmoqobuz",
"pmoserver",
"pmosource",

94
pmoaudio-ext/src/sources/playlist_source.rs
View File

@@ -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<dyn std::error::Error>> {
//! 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<AudioCache>,
chunk_frames: usize,
poll_interval_ms: u64,
history_playlist: Option<Arc<pmoplaylist::WriteHandle>>,
}
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<pmoplaylist::WriteHandle>) {
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<AudioCache>,
history_playlist: Arc<pmoplaylist::WriteHandle>,
) -> 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]

2
pmomediaserver/Cargo.toml
View File

@@ -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)

4
pmomediaserver/src/paradise_streaming.rs
View File

@@ -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
}
};

12
pmoparadise/.pmomusic/config.yaml
View File

@@ -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

165
pmoparadise/ARCHITECTURE.md Normal file
View File

@@ -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
}
}
```

2
pmoparadise/Cargo.toml
View File

@@ -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"]

8
pmoparadise/src/lib.rs
View File

@@ -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::{

12
pmoparadise/src/models.rs
View File

@@ -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<u64> {
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

200
pmoparadise/src/playlist_feeder.rs Normal file
View File

@@ -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<AudioCache>,
covers_cache: Arc<CoversCache>,
playlist_handle: Arc<WriteHandle>,
block_queue: Arc<tokio::sync::Mutex<VecDeque<EventId>>>,
notify: Arc<Notify>,
collection: Option<String>,
}
impl RadioParadisePlaylistFeeder {
/// Crée un nouveau feeder et retourne (feeder, read_handle)
pub async fn new(
client: RadioParadiseClient,
audio_cache: Arc<AudioCache>,
covers_cache: Arc<CoversCache>,
playlist_id: String,
collection: Option<String>,
) -> 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<Self>) -> 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(())
}
}

204
pmoparadise/src/stream_channel.rs
View File

@@ -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<AudioCache>,
pub cover_cache: Arc<CoverCache>,
pub playlist_id: String,
pub playlist_writer: WriteHandle,
pub collection: Option<String>,
pub replay_max_lead_seconds: f64,
pub max_history_tracks: Option<usize>,
}
/// Builder pratique pour configurer automatiquement les playlists historiques.
@@ -87,19 +93,6 @@ impl ParadiseHistoryBuilder {
descriptor: &ChannelDescriptor,
) -> Result<ParadiseHistoryOptions, pmoplaylist::Error> {
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<AudioCache>,
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<ChannelState>,
pipeline_handle: JoinHandle<()>,
feeder_handle: JoinHandle<()>,
history: Option<HistoryState>,
}
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<Arc<CoverCache>>,
history: Option<ParadiseHistoryOptions>,
) -> Self {
let mut source = RadioParadiseStreamSource::new(client.clone());
let block_handle = source.block_handle();
) -> Result<Self> {
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<HistoryFlacStream, HistoryStreamError> {
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<HistoryOggStream, HistoryStreamError> {
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<RadioParadisePlaylistFeeder>,
stream_handle: StreamHandle,
ogg_handle: OggFlacStreamHandle,
history_playlist_id: Option<String>,
history_audio_cache: Option<Arc<AudioCache>>,
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);
}

697
pmoparadise/src/stream_channel_old.rs Normal file
View File

@@ -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<AudioCache>,
pub cover_cache: Arc<CoverCache>,
pub playlist_id: String,
pub playlist_writer: WriteHandle,
pub collection: Option<String>,
pub replay_max_lead_seconds: f64,
}
/// Builder pratique pour configurer automatiquement les playlists historiques.
#[derive(Clone)]
pub struct ParadiseHistoryBuilder {
pub audio_cache: Arc<AudioCache>,
pub cover_cache: Arc<CoverCache>,
pub playlist_prefix: String,
pub playlist_title_prefix: Option<String>,
pub max_history_tracks: Option<usize>,
pub collection_prefix: Option<String>,
pub replay_max_lead_seconds: f64,
}
impl ParadiseHistoryBuilder {
pub fn new(audio_cache: Arc<AudioCache>, cover_cache: Arc<CoverCache>) -> 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<ParadiseHistoryOptions, pmoplaylist::Error> {
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<AudioCache>,
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::<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
}
}
_ => {
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<ChannelState>,
pipeline_handle: JoinHandle<()>,
feeder_handle: JoinHandle<()>,
history: Option<HistoryState>,
}
impl ParadiseStreamChannel {
/// Crée un canal avec client déjà configuré.
pub fn with_client(
descriptor: ChannelDescriptor,
client: RadioParadiseClient,
config: ParadiseStreamChannelConfig,
cover_cache: Option<Arc<CoverCache>>,
history: Option<ParadiseHistoryOptions>,
) -> 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<Box<dyn AudioPipelineNode>> = 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<Arc<CoverCache>>,
history: Option<ParadiseHistoryOptions>,
) -> Result<Self> {
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<HistoryFlacStream, HistoryStreamError> {
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<HistoryOggStream, HistoryStreamError> {
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<Self>) {
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<ChannelState>,
}
impl $name {
fn new(inner: $inner, state: Arc<ChannelState>) -> Self {
Self { inner, state }
}
}
impl AsyncRead for $name {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
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<JoinHandle<()>>,
}
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<std::io::Result<()>> {
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<JoinHandle<()>>,
}
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<std::io::Result<()>> {
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<u8, Arc<ParadiseStreamChannel>>,
}
impl ParadiseChannelManager {
pub fn new(channels: HashMap<u8, Arc<ParadiseStreamChannel>>) -> Self {
Self { channels }
}
pub async fn with_defaults_with_cover_cache(
cover_cache: Option<Arc<CoverCache>>,
history_builder: Option<ParadiseHistoryBuilder>,
) -> Result<Self> {
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> {
Self::with_defaults_with_cover_cache(None, None).await
}
pub fn get(&self, id: u8) -> Option<Arc<ParadiseStreamChannel>> {
self.channels.get(&id).cloned()
}
pub fn iter(&self) -> impl Iterator<Item = &Arc<ParadiseStreamChannel>> {
self.channels.values()
}
}

28
pmoplaylist/src/handle/write.rs
View File

@@ -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<String>) -> Result<()> {
if !self.playlist.is_alive() {