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

Implémentation du cache de métadonnées avec gestion des TTL et amélioration du streaming

Browse Source 
Cette mise à jour implémente un système de cache de métadonnées avec gestion des TTL pour les données Radio France. Le cache gère les mises à jour automatiques via des tâches de rafraîchissement, et les clients interrogent uniquement le cache pour obtenir des données à jour. Le streaming est amélioré avec une gestion correcte des connexions et des tâches de rafraîchissement qui s'arrêtent automatiquement à la déconnexion du client. Les versions des packages PMOMusic et pmomediaserver sont mises à jour à 0.3.13.
This commit is contained in:
Eric Coissac
2026-01-23 22:41:54 +01:00
parent d3d2f24a6a
commit b5e9bb116f
9 changed files with 178 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
Blackboard/Todo/metadata_RF_cache.md Normal file
View File

@@ -0,0 +1,22 @@
**Tu réaliseras ce travail en appliquant scrupuleusement les règles définies dans [@Rules.md](file:///Users/coissac/Sync/maison/Petite_maisons/src/pmomusic/Blackboard/Rules.md)**
**On ne travaille que dans la Crate PMORadioFrance**
À partir de maintenant, tu ne prends plus en compte ce que tu pensais avant et tu écoutes bien. Et tu construis un plan d'implémentation que je dois valider. Tu arrêtes de prendre des initiatives et de faire des bêtises.
- Tu as des fonctions d'interrogation de l'API Radio France qu'il faut utiliser au minimum. Mais Radio France nous donne des dates d'invalidation des métadonnées. Globalement, on doit gérer une grosse map où les valeurs ont des TTL.
- Quand un client demande une donnée du cache, si le TTL est atteint, il commence par utiliser l'API Radio France, modifie le cache, puis la retourne. Dans le cas contraire, il retourne directement la donnée.
A chaque fois qu'il fait un appel de l'API Radio France pour modifier ses valeurs, Le cache émet un événement ou avertit ses abonnés, comme quoi les données d'un slug particulier ont été modifiées. Comme ça tout le monde peut se synchroniser.
Les clients, par exemple la fonction Browse, n'interrogent que le cache qui a forcément des données à jour. Les métadonnées ne sont jamais stockées hors du cache, on se réfère toujours à elles.
Nous devons maintenant considérer le fonctionnement du control point. Celui-ci est capable de s'abonner à une playlist pour suivre ses modifications. Il ne peut pas s'abonner à un item.
Dans le cas d'une radio, on peut considérer que chaque canal, chaque slug, est en réalité une playlist à un item qu'il faut suivre. Ainsi, le Control Point peut décider de jouer cette playlist en s'abonnant à elle et être tenu au courant des modifications par des événements GENA.
La source Radio France doit donc s'abonner aux événements du Cache. A chaque fois qu'un slug est modifié, elle avertit par un événement Jenna que la playlist à un item qui correspond à ce Slug est modifiée.
Maintenant, il y a le cache des stations. Le cache des stations finalement il ne stock qu'un emboîtement de listes de slug. Ça, normalement, ça ne bouge quasiment pas. On peut dire que une fois par jour, on met à jour ce cache. Les listes de slug ont donc un TTL mais très long.
A chaque browse, on reconstruit un document didl à partir des métadonnées à jour provenant du cache.

2
Cargo.lock generated
View File

@@ -4,7 +4,7 @@ version = 4
[[package]]
name = "PMOMusic"
version = "0.3.11"
version = "0.3.13"
dependencies = [
"axum 0.8.7",
"console-subscriber",

2
PMOMusic/Cargo.toml
View File

@@ -1,6 +1,6 @@
[package]
name = "PMOMusic"
version = "0.3.11"
version = "0.3.13"
edition = "2024"
[dependencies]

15
pmomediaserver/src/sources.rs
View File

@@ -270,13 +270,16 @@ impl SourcesExt for Server {
});
let source = source.with_container_notifier(notifier);
// Enregistrer la source
self.register_music_source(Arc::new(source)).await;
// Enregistrer la source (Arc pour partage avec l'API)
let source_arc = Arc::new(source);
self.register_music_source(source_arc.clone()).await;
// Initialiser les routes API Radio France
self.init_radiofrance().await.map_err(|e| {
SourceInitError::RadioFranceError(format!("Failed to init API routes: {}", e))
})?;
// Initialiser les routes API Radio France avec la source
self.init_radiofrance_with_source(source_arc)
.await
.map_err(|e| {
SourceInitError::RadioFranceError(format!("Failed to init API routes: {}", e))
})?;
tracing::info!("✅ Radio France source registered successfully");

54
pmoradiofrance/src/api_rest.rs
View File

@@ -16,6 +16,7 @@ use axum::{
};
use futures::StreamExt;
use serde_json;
use std::sync::Arc;
// ============ Gestion des erreurs ============
@@ -65,6 +66,7 @@ async fn get_stations(
State(state): State<RadioFranceState>,
) -> Result<Json<StationGroups>, AppError> {
let stations = state
.source
.client
.get_stations()
.await
@@ -81,6 +83,7 @@ async fn get_metadata(
Path(slug): Path<String>,
) -> Result<Json<LiveResponse>, AppError> {
let metadata = state
.source
.client
.get_live_metadata(&slug)
.await
@@ -95,8 +98,23 @@ async fn proxy_stream(
State(state): State<RadioFranceState>,
Path(slug): Path<String>,
) -> Result<Response, AppError> {
// Start metadata refresh when stream is accessed
#[cfg(feature = "logging")]
tracing::info!("Stream proxy accessed for station: {}", slug);
// Spawn refresh task (non-blocking)
let source_clone = Arc::clone(&state.source);
let slug_clone = slug.clone();
tokio::spawn(async move {
if let Err(e) = source_clone.start_metadata_refresh(&slug_clone).await {
#[cfg(feature = "logging")]
tracing::error!("Failed to start metadata refresh for {}: {}", slug_clone, e);
}
});
// Get the stream URL
let stream_url = state
.source
.client
.get_stream_url(&slug)
.await
@@ -116,12 +134,44 @@ async fn proxy_stream(
headers.insert("content-type", "audio/aac".parse().unwrap());
headers.insert("cache-control", "no-cache".parse().unwrap());
// Create streaming body
// Create streaming body with cleanup on disconnect
let stream = response
.bytes_stream()
.map(|chunk| chunk.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)));
let body = Body::from_stream(stream);
// Wrap the stream to detect when client disconnects
let source_for_cleanup = Arc::clone(&state.source);
let slug_for_cleanup = slug.clone();
let monitored_stream =
futures::stream::unfold((stream, false), move |(mut stream, mut done)| {
let source = source_for_cleanup.clone();
let slug = slug_for_cleanup.clone();
async move {
if done {
return None;
}
match stream.next().await {
Some(Ok(chunk)) => Some((Ok(chunk), (stream, false))),
Some(Err(e)) => {
// Error occurred, stop refresh
#[cfg(feature = "logging")]
tracing::info!("Stream error for {}, stopping refresh", slug);
source.stop_metadata_refresh(&slug).await;
Some((Err(e), (stream, true)))
}
None => {
// Stream ended normally, stop refresh
#[cfg(feature = "logging")]
tracing::info!("Stream ended for {}, stopping refresh", slug);
source.stop_metadata_refresh(&slug).await;
None
}
}
}
});
let body = Body::from_stream(monitored_stream);
Ok((headers, body).into_response())
}

18
pmoradiofrance/src/pmoserver_ext.rs
View File

@@ -11,14 +11,12 @@ use crate::stateful_client::RadioFranceStatefulClient;
/// État partagé pour les handlers Radio France
#[derive(Clone)]
pub struct RadioFranceState {
pub client: Arc<RadioFranceStatefulClient>,
pub source: Arc<crate::source::RadioFranceSource>,
}
impl RadioFranceState {
pub fn new(client: RadioFranceStatefulClient) -> Self {
Self {
client: Arc::new(client),
}
pub fn new(source: Arc<crate::source::RadioFranceSource>) -> Self {
Self { source }
}
}
@@ -80,6 +78,16 @@ pub trait RadioFranceExt {
/// server.init_radiofrance().await?;
/// ```
async fn init_radiofrance(&mut self) -> Result<Arc<RadioFranceState>>;
/// Initialise l'extension Radio France avec une source existante
///
/// Cette méthode est similaire à `init_radiofrance()` mais utilise une source
/// déjà créée et enregistrée, permettant de partager la même instance entre
/// le MediaServer UPnP et les routes API REST.
async fn init_radiofrance_with_source(
&mut self,
source: Arc<crate::source::RadioFranceSource>,
) -> Result<Arc<RadioFranceState>>;
}
// L'implémentation du trait sera dans un module séparé (pmoserver_impl.rs)

29
pmoradiofrance/src/pmoserver_impl.rs
View File

@@ -38,14 +38,14 @@ impl RadioFranceExt for Server {
async fn init_radiofrance(&mut self) -> Result<Arc<RadioFranceState>> {
info!("Initializing Radio France API...");
// Créer le client stateful
// Créer une source dédiée pour l'API (sans enregistrement UPnP)
let config = pmoconfig::get_config();
let client = RadioFranceStatefulClient::new(config)
let source = crate::source::RadioFranceSource::new(config)
.await
.map_err(|e| anyhow::anyhow!("Failed to create Radio France client: {}", e))?;
.map_err(|e| anyhow::anyhow!("Failed to create Radio France source: {}", e))?;
// Créer l'état partagé (RadioFranceState est Clone et contient déjà un Arc<client>)
let state = RadioFranceState::new(client);
// Créer l'état partagé
let state = RadioFranceState::new(Arc::new(source));
// Créer et enregistrer le router
let router = create_router(state.clone());
@@ -56,4 +56,23 @@ impl RadioFranceExt for Server {
Ok(Arc::new(state))
}
async fn init_radiofrance_with_source(
&mut self,
source: Arc<crate::source::RadioFranceSource>,
) -> Result<Arc<RadioFranceState>> {
info!("Initializing Radio France API with existing source...");
// Créer l'état partagé (simplement une référence à la source)
let state = RadioFranceState::new(source);
// Créer et enregistrer le router
let router = create_router(state.clone());
self.add_router("/api/radiofrance", router).await;
info!("Radio France API initialized with source");
info!("API endpoints available at /api/radiofrance/*");
Ok(Arc::new(state))
}
}

150
pmoradiofrance/src/source.rs
View File

@@ -33,10 +33,7 @@ pub const RADIOFRANCE_DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/radiofran
/// - Hierarchical organization (standalone, groups, local radios)
pub struct RadioFranceSource {
/// Stateful client with automatic caching
client: RadioFranceStatefulClient,
/// Cache of playlists by station slug (volatile metadata)
playlists: Arc<RwLock<HashMap<String, StationPlaylist>>>,
pub(crate) client: RadioFranceStatefulClient,
/// Background tasks for metadata refresh
refresh_handles: Arc<RwLock<HashMap<String, JoinHandle<()>>>>,
@@ -83,7 +80,6 @@ impl RadioFranceSource {
Ok(Self {
client,
playlists: Arc::new(RwLock::new(HashMap::new())),
refresh_handles: Arc::new(RwLock::new(HashMap::new())),
#[cfg(feature = "cache")]
cover_cache: None,
@@ -151,7 +147,7 @@ impl RadioFranceSource {
}
/// Start metadata refresh task for a station
async fn start_metadata_refresh(&self, station_slug: &str) -> Result<()> {
pub async fn start_metadata_refresh(&self, station_slug: &str) -> Result<()> {
let mut handles = self.refresh_handles.write().await;
// If already running, do nothing
@@ -160,69 +156,54 @@ impl RadioFranceSource {
}
let client = self.client.clone();
let playlists = self.playlists.clone();
let slug = station_slug.to_string();
let update_id = self.update_id.clone();
let last_change = self.last_change.clone();
let container_notifier = self.container_notifier.clone();
#[cfg(feature = "cache")]
let cover_cache = self.cover_cache.clone();
let server_base_url = self.server_base_url.clone();
let handle = tokio::spawn(async move {
loop {
match client.get_live_metadata(&slug).await {
// Force refresh metadata (bypass cache to get fresh data)
match client.refresh_live_metadata(&slug).await {
Ok(metadata) => {
let delay = std::time::Duration::from_millis(metadata.delay_to_refresh);
// Update the playlist metadata
#[cfg(feature = "cache")]
#[cfg(feature = "logging")]
{
let mut pls = playlists.write().await;
if let Some(playlist) = pls.get_mut(&slug) {
let _: Result<()> = playlist
.update_metadata(
&metadata,
cover_cache.as_ref(),
server_base_url.as_deref(),
)
.await;
// Update change tracking
*update_id.write().await = update_id.read().await.wrapping_add(1);
*last_change.write().await = Some(SystemTime::now());
// Notify UPnP ContentDirectory of the change
if let Some(ref notifier) = container_notifier {
// Notify the station's stream item container
let container_id = format!("radiofrance:{}", slug);
notifier(&[container_id]);
}
}
let artist = metadata
.now
.song
.as_ref()
.and_then(|s| {
if s.interpreters.is_empty() {
None
} else {
Some(s.artists_display())
}
})
.unwrap_or_else(|| "".to_string());
tracing::debug!(
"Refreshed metadata for {}: title='{}' artist='{}' delay={}ms",
slug,
metadata.now.first_line.title.as_deref().unwrap_or(""),
artist,
metadata.delay_to_refresh
);
}
#[cfg(not(feature = "cache"))]
{
let mut pls = playlists.write().await;
if let Some(playlist) = pls.get_mut(&slug) {
let _: Result<()> = playlist.update_metadata_no_cache(
&metadata,
server_base_url.as_deref(),
);
// Update change tracking
*update_id.write().await = update_id.read().await.wrapping_add(1);
*last_change.write().await = Some(SystemTime::now());
// Update change tracking
*update_id.write().await = update_id.read().await.wrapping_add(1);
*last_change.write().await = Some(SystemTime::now());
// Notify UPnP ContentDirectory of the change
if let Some(ref notifier) = container_notifier {
// Notify the station's stream item container
let container_id = format!("radiofrance:{}", slug);
// Notify UPnP ContentDirectory of the change
if let Some(ref notifier) = container_notifier {
// Notify the station's stream item container
let container_id = format!("radiofrance:{}", slug);
notifier(&[container_id]);
}
}
#[cfg(feature = "logging")]
tracing::debug!("Notifying UPnP container update: {}", container_id);
notifier(&[container_id]);
}
tokio::time::sleep(delay).await;
@@ -246,7 +227,7 @@ impl RadioFranceSource {
}
/// Stop metadata refresh task for a station
async fn stop_metadata_refresh(&self, station_slug: &str) {
pub async fn stop_metadata_refresh(&self, station_slug: &str) {
let mut handles = self.refresh_handles.write().await;
if let Some(handle) = handles.remove(station_slug) {
handle.abort();
@@ -394,25 +375,13 @@ impl RadioFranceSource {
station.slug
);
let playlists = self.playlists.read().await;
// If we already have this station in cache, use it
if let Some(existing) = playlists.get(&station.slug) {
#[cfg(feature = "logging")]
tracing::debug!("Using cached item for: {}", station.slug);
return Ok(existing.stream_item.clone());
}
// Release read lock before fetching metadata
drop(playlists);
// Fetch metadata from API
// Fetch metadata from API (cached by RadioFranceStatefulClient)
let metadata = self.client.get_live_metadata(&station.slug).await?;
// Create playlist with metadata
// Build item from live metadata (no caching here, rely on client cache)
#[cfg(feature = "cache")]
let playlist = StationPlaylist::from_live_metadata(
station.clone(),
let item = StationPlaylist::build_item_from_metadata(
station,
&metadata,
self.cover_cache.as_ref(),
self.server_base_url.as_deref(),
@@ -420,17 +389,12 @@ impl RadioFranceSource {
.await?;
#[cfg(not(feature = "cache"))]
let playlist = StationPlaylist::from_live_metadata_no_cache(
station.clone(),
let item = StationPlaylist::build_item_from_metadata_sync(
station,
&metadata,
self.server_base_url.as_deref(),
)?;
// Cache it
let mut playlists_write = self.playlists.write().await;
playlists_write.insert(station.slug.clone(), playlist.clone());
drop(playlists_write);
// Note: We don't start metadata refresh here to avoid blocking during browse.
// Refresh will be started in resolve_uri() when the stream is actually played.
@@ -438,11 +402,11 @@ impl RadioFranceSource {
tracing::debug!(
"Built item for {}: {} resources, album_art: {:?}",
station.slug,
playlist.stream_item.resources.len(),
playlist.stream_item.album_art.is_some()
item.resources.len(),
item.album_art.is_some()
);
Ok(playlist.stream_item)
Ok(item)
}
}
@@ -535,22 +499,18 @@ impl MusicSource for RadioFranceSource {
// Build items for this group only (main + webradios)
let group_id = format!("radiofrance:group:{}", slug);
let mut main_item = self
.build_station_item(&group.main)
.await
.map_err(|e| MusicSourceError::BrowseError(e.to_string()))?;
// Fix parent_id to point to the group container
main_item.parent_id = group_id.clone();
let mut items = vec![main_item];
// Paralléliser les fetches pour éviter les timeouts
let mut futures = vec![self.build_station_item(&group.main)];
for webradio in &group.webradios {
let mut item = self
.build_station_item(webradio)
.await
.map_err(|e| MusicSourceError::BrowseError(e.to_string()))?;
futures.push(self.build_station_item(webradio));
}
// Fix parent_id to point to the group container
let results = futures::future::join_all(futures).await;
let mut items = Vec::new();
for result in results {
let mut item =
result.map_err(|e| MusicSourceError::BrowseError(e.to_string()))?;
item.parent_id = group_id.clone();
items.push(item);
}

2
version.txt
View File

@@ -1 +1 @@
0.3.11
0.3.13