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

Refactoring des pmosource

Browse Source
This commit is contained in:
Eric Coissac
2025-10-17 23:45:01 +02:00
parent b9e998fa8a
commit e0a4eff6dd
14 changed files with 865 additions and 703 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
Cargo.lock generated
View File

@@ -2485,6 +2485,7 @@ dependencies = [
name = "pmoupnp" name = "pmoupnp"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"anyhow",
"axum", "axum",
"base64", "base64",
"bevy_reflect", "bevy_reflect",
@@ -2493,7 +2494,10 @@ dependencies = [
"hex", "hex",
"once_cell", "once_cell",
"parking_lot", "parking_lot",
"pmoaudiocache",
"pmocache",
"pmoconfig", "pmoconfig",
"pmocovers",
"pmodidl", "pmodidl",
"pmoserver", "pmoserver",
"pmoutils", "pmoutils",
@@ -2505,6 +2509,7 @@ dependencies = [
"tokio", "tokio",
"tracing", "tracing",
"url", "url",
"utoipa",
"uuid", "uuid",
"xmltree", "xmltree",
] ]

70
pmoaudiocache/src/lib.rs
View File

@@ -134,12 +134,6 @@ pub mod cache;
pub mod metadata; pub mod metadata;
pub mod flac; pub mod flac;
#[cfg(feature = "pmoserver")]
mod pmoserver_ext;
#[cfg(feature = "pmoserver")]
mod pmoserver_impl;
#[cfg(feature = "pmoserver")] #[cfg(feature = "pmoserver")]
pub mod openapi; pub mod openapi;
@@ -148,7 +142,67 @@ pub use cache::{Cache, AudioConfig, new_cache, add_with_metadata_extraction, get
pub use metadata::AudioMetadata; pub use metadata::AudioMetadata;
#[cfg(feature = "pmoserver")] #[cfg(feature = "pmoserver")]
pub use pmoserver_ext::AudioCacheExt; pub use openapi::ApiDoc;
// ============================================================================
// Extension pmoserver (inline comme pmocovers)
// ============================================================================
/// Trait pour étendre un serveur HTTP avec des fonctionnalités de cache audio.
#[cfg(feature = "pmoserver")]
pub trait AudioCacheExt {
/// Initialise le cache audio et enregistre les routes HTTP.
///
/// # Arguments
///
/// * `cache_dir` - Répertoire de stockage du cache
/// * `limit` - Limite de taille du cache (en nombre de pistes)
///
/// # Returns
///
/// * `Arc<Cache>` - Instance partagée du cache
async fn init_audio_cache(&mut self, cache_dir: &str, limit: usize) -> anyhow::Result<std::sync::Arc<Cache>>;
/// Initialise le cache audio avec la configuration par défaut.
///
/// Utilise automatiquement les paramètres de `pmoconfig::Config`.
async fn init_audio_cache_configured(&mut self) -> anyhow::Result<std::sync::Arc<Cache>>;
}
#[cfg(feature = "pmoserver")] #[cfg(feature = "pmoserver")]
pub use openapi::ApiDoc; use pmocache::pmoserver_ext::{create_file_router, create_api_router};
#[cfg(feature = "pmoserver")]
use std::sync::Arc;
#[cfg(feature = "pmoserver")]
use utoipa::OpenApi;
#[cfg(feature = "pmoserver")]
impl AudioCacheExt for pmoserver::Server {
async fn init_audio_cache(&mut self, cache_dir: &str, limit: usize) -> anyhow::Result<Arc<Cache>> {
let base_url = self.info().base_url;
let cache = Arc::new(crate::cache::new_cache(cache_dir, limit, &base_url)?);
// Router de fichiers pour servir les pistes FLAC
// Routes: GET /audio/tracks/{pk} et GET /audio/tracks/{pk}/{param}
let file_router = create_file_router(
cache.clone(),
"audio/flac" // Content-Type
);
self.add_router("/", file_router).await;
// API REST générique (pmocache)
// Routes: GET/POST/DELETE /api/audio, etc.
let api_router = create_api_router(cache.clone());
let openapi = crate::ApiDoc::openapi();
self.add_openapi(api_router, openapi, "audio").await;
Ok(cache)
}
async fn init_audio_cache_configured(&mut self) -> anyhow::Result<Arc<Cache>> {
let config = pmoconfig::get_config();
let cache_dir = config.get_audio_cache_dir()?;
let limit = config.get_audio_cache_size()?;
self.init_audio_cache(&cache_dir, limit).await
}
}

26
pmoaudiocache/src/pmoserver_ext.rs
View File

@@ -1,26 +0,0 @@
#[cfg(feature = "pmoserver")]
use crate::Cache;
/// Trait pour étendre un serveur HTTP avec des fonctionnalités de cache audio.
///
/// Ce trait permet à `pmoaudiocache` d'ajouter des méthodes d'extension sur des types
/// de serveurs externes (comme `pmoserver::Server`) sans que ces crates dépendent de `pmoaudiocache`.
#[cfg(feature = "pmoserver")]
pub trait AudioCacheExt {
/// Initialise le cache audio et enregistre les routes HTTP.
///
/// # Arguments
///
/// * `cache_dir` - Répertoire de stockage du cache
/// * `limit` - Limite de taille du cache (en nombre de pistes)
///
/// # Returns
///
/// * `Arc<Cache>` - Instance partagée du cache
async fn init_audio_cache(&mut self, cache_dir: &str, limit: usize) -> anyhow::Result<std::sync::Arc<Cache>>;
/// Initialise le cache audio avec la configuration par défaut.
///
/// Utilise automatiquement les paramètres de `pmoconfig::Config`.
async fn init_audio_cache_configured(&mut self) -> anyhow::Result<std::sync::Arc<Cache>>;
}

41
pmoaudiocache/src/pmoserver_impl.rs
View File

@@ -1,41 +0,0 @@
//! Implémentation du trait AudioCacheExt pour pmoserver::Server
#[cfg(feature = "pmoserver")]
use crate::{AudioCacheExt, Cache};
#[cfg(feature = "pmoserver")]
use pmocache::pmoserver_ext::{create_file_router, create_api_router};
#[cfg(feature = "pmoserver")]
use std::sync::Arc;
#[cfg(feature = "pmoserver")]
use utoipa::OpenApi;
#[cfg(feature = "pmoserver")]
impl AudioCacheExt for pmoserver::Server {
async fn init_audio_cache(&mut self, cache_dir: &str, limit: usize) -> anyhow::Result<Arc<Cache>> {
let base_url = self.info().base_url;
let cache = Arc::new(crate::cache::new_cache(cache_dir, limit, &base_url)?);
// Router de fichiers pour servir les pistes FLAC
// Routes: GET /audio/tracks/{pk} et GET /audio/tracks/{pk}/{param}
let file_router = create_file_router(
cache.clone(),
"audio/flac" // Content-Type
);
self.add_router("/", file_router).await;
// API REST générique (pmocache)
// Routes: GET/POST/DELETE /api/audio, etc.
let api_router = create_api_router(cache.clone());
let openapi = crate::ApiDoc::openapi();
self.add_openapi(api_router, openapi, "audio").await;
Ok(cache)
}
async fn init_audio_cache_configured(&mut self) -> anyhow::Result<Arc<Cache>> {
let config = pmoconfig::get_config();
let cache_dir = config.get_audio_cache_dir()?;
let limit = config.get_audio_cache_size()?;
self.init_audio_cache(&cache_dir, limit).await
}
}

16
pmoparadise/Cargo.toml
View File

@@ -28,8 +28,8 @@ anyhow = "1.0"
bytes = "1.5" bytes = "1.5"
futures = "0.3" futures = "0.3"
# Logging (optionnel) # Logging
tracing = { version = "0.1", optional = true } tracing = "0.1"
# URL manipulation # URL manipulation
url = "2.5" url = "2.5"
@@ -51,9 +51,9 @@ pmosource = { path = "../pmosource" }
# Playlist management for FIFO support # Playlist management for FIFO support
pmoplaylist = { path = "../pmoplaylist" } pmoplaylist = { path = "../pmoplaylist" }
# Cache support # Cache support (OBLIGATOIRE - architecture refactorisée)
pmocovers = { path = "../pmocovers", optional = true } pmocovers = { path = "../pmocovers" }
pmoaudiocache = { path = "../pmoaudiocache", optional = true } pmoaudiocache = { path = "../pmoaudiocache" }
[features] [features]
default = ["metadata-only"] default = ["metadata-only"]
@@ -61,12 +61,10 @@ default = ["metadata-only"]
metadata-only = [] metadata-only = []
# Active le décodage FLAC par-track # Active le décodage FLAC par-track
per-track = ["dep:claxon", "dep:hound", "dep:tempfile"] per-track = ["dep:claxon", "dep:hound", "dep:tempfile"]
# Active le logging
logging = ["dep:tracing"]
# Active le media server UPnP # Active le media server UPnP
mediaserver = ["dep:pmoupnp", "dep:pmoserver", "dep:pmodidl", "dep:uuid"] mediaserver = ["dep:pmoupnp", "dep:pmoserver", "dep:pmodidl", "dep:uuid"]
# Active le cache d'images et audio # Feature cache (deprecated - toujours actif maintenant)
cache = ["dep:pmocovers", "dep:pmoaudiocache", "logging"] cache = []
[dev-dependencies] [dev-dependencies]
# Tests # Tests

507
pmoparadise/src/source.rs
View File

@@ -6,17 +6,13 @@
use crate::client::RadioParadiseClient; use crate::client::RadioParadiseClient;
use crate::models::{Block, Song}; use crate::models::{Block, Song};
use pmosource::{async_trait, pmodidl, BrowseResult, MusicSource, MusicSourceError, Result}; use pmosource::{async_trait, pmodidl, BrowseResult, MusicSource, MusicSourceError, Result};
use pmosource::SourceCacheManager;
use pmoaudiocache::{AudioMetadata, Cache as AudioCache};
use pmocovers::Cache as CoverCache;
use pmodidl::{Container, Item, Resource}; use pmodidl::{Container, Item, Resource};
use pmoplaylist::{FifoPlaylist, Track}; use pmoplaylist::{FifoPlaylist, Track};
use std::collections::HashMap;
use std::sync::Arc; use std::sync::Arc;
use std::time::SystemTime; use std::time::SystemTime;
use tokio::sync::RwLock;
#[cfg(feature = "cache")]
use pmocovers::Cache as CoverCache;
#[cfg(feature = "cache")]
use pmoaudiocache::{AudioCache, AudioMetadata};
/// Default image for Radio Paradise (300x300 WebP, embedded in binary) /// Default image for Radio Paradise (300x300 WebP, embedded in binary)
const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp"); const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp");
@@ -67,53 +63,37 @@ struct RadioParadiseSourceInner {
/// FIFO playlist for dynamic track management /// FIFO playlist for dynamic track management
playlist: FifoPlaylist, playlist: FifoPlaylist,
/// Cache server base URL for URI resolution /// Cache manager (centralisé)
cache_base_url: String, cache_manager: SourceCacheManager,
/// Track metadata cache (track_id -> (original_uri, cached_pk, block_event)) /// Blocks cache pour retrouver les métadonnées originales
track_cache: RwLock<HashMap<String, TrackMetadata>>, /// (track_id -> (block, song_index))
blocks: tokio::sync::RwLock<std::collections::HashMap<String, (Arc<Block>, usize)>>,
/// Cover image cache (optional)
#[cfg(feature = "cache")]
cover_cache: Option<Arc<CoverCache>>,
/// Audio cache (optional)
#[cfg(feature = "cache")]
audio_cache: Option<Arc<AudioCache>>,
}
#[derive(Debug, Clone)]
struct TrackMetadata {
original_uri: String,
cached_pk: Option<String>,
block: Arc<Block>,
song_index: usize,
#[cfg(feature = "cache")]
cached_audio_pk: Option<String>,
#[cfg(feature = "cache")]
cached_cover_pk: Option<String>,
} }
impl std::fmt::Debug for RadioParadiseSource { impl std::fmt::Debug for RadioParadiseSource {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RadioParadiseSource") f.debug_struct("RadioParadiseSource")
.field("cache_base_url", &self.inner.cache_base_url)
.finish() .finish()
} }
} }
impl RadioParadiseSource { impl RadioParadiseSource {
/// Create a new Radio Paradise source /// Create a new Radio Paradise source with caches
/// ///
/// # Arguments /// # Arguments
/// ///
/// * `client` - Radio Paradise API client /// * `client` - Radio Paradise API client
/// * `cache_base_url` - Base URL for the cache server (e.g., "http://localhost:8080") /// * `cache_base_url` - Base URL for the cache server (e.g., "http://localhost:8080")
/// * `fifo_capacity` - Maximum number of tracks in the FIFO /// * `fifo_capacity` - Maximum number of tracks in the FIFO
/// * `cover_cache` - Cover image cache (required)
/// * `audio_cache` - Audio cache (required)
pub fn new( pub fn new(
client: RadioParadiseClient, client: RadioParadiseClient,
cache_base_url: impl Into<String>, cache_base_url: impl Into<String>,
fifo_capacity: usize, fifo_capacity: usize,
cover_cache: Arc<CoverCache>,
audio_cache: Arc<AudioCache>,
) -> Self { ) -> Self {
let playlist = FifoPlaylist::new( let playlist = FifoPlaylist::new(
"radio-paradise".to_string(), "radio-paradise".to_string(),
@@ -122,59 +102,32 @@ impl RadioParadiseSource {
DEFAULT_IMAGE, DEFAULT_IMAGE,
); );
let cache_base_url = cache_base_url.into();
let cache_manager = SourceCacheManager::new(
cache_base_url.clone(),
"radio-paradise".to_string(),
cover_cache,
audio_cache,
);
Self { Self {
inner: Arc::new(RadioParadiseSourceInner { inner: Arc::new(RadioParadiseSourceInner {
client, client,
playlist, playlist,
cache_base_url: cache_base_url.into(), cache_manager,
track_cache: RwLock::new(HashMap::new()), blocks: tokio::sync::RwLock::new(std::collections::HashMap::new()),
#[cfg(feature = "cache")]
cover_cache: None,
#[cfg(feature = "cache")]
audio_cache: None,
}), }),
} }
} }
/// Create with default FIFO capacity /// Create with default FIFO capacity
pub fn new_default(client: RadioParadiseClient, cache_base_url: impl Into<String>) -> Self { pub fn new_default(
Self::new(client, cache_base_url, DEFAULT_FIFO_CAPACITY)
}
/// Create a new Radio Paradise source with caching support
///
/// # Arguments
///
/// * `client` - Radio Paradise API client
/// * `cache_base_url` - Base URL for the cache server (e.g., "http://localhost:8080")
/// * `fifo_capacity` - Maximum number of tracks in the FIFO
/// * `cover_cache` - Optional cover image cache
/// * `audio_cache` - Optional audio cache
#[cfg(feature = "cache")]
pub fn new_with_cache(
client: RadioParadiseClient, client: RadioParadiseClient,
cache_base_url: impl Into<String>, cache_base_url: impl Into<String>,
fifo_capacity: usize, cover_cache: Arc<CoverCache>,
cover_cache: Option<Arc<CoverCache>>, audio_cache: Arc<AudioCache>,
audio_cache: Option<Arc<AudioCache>>,
) -> Self { ) -> Self {
let playlist = FifoPlaylist::new( Self::new(client, cache_base_url, DEFAULT_FIFO_CAPACITY, cover_cache, audio_cache)
"radio-paradise".to_string(),
"Radio Paradise".to_string(),
fifo_capacity,
DEFAULT_IMAGE,
);
Self {
inner: Arc::new(RadioParadiseSourceInner {
client,
playlist,
cache_base_url: cache_base_url.into(),
track_cache: RwLock::new(HashMap::new()),
cover_cache,
audio_cache,
}),
}
} }
/// Add a track from a Radio Paradise song and block /// Add a track from a Radio Paradise song and block
@@ -199,30 +152,24 @@ impl RadioParadiseSource {
track = track.with_duration((song.duration / 1000) as u32); track = track.with_duration((song.duration / 1000) as u32);
} }
// Cache cover image and add to track // 1. Cache cover via le manager
#[cfg(feature = "cache")] let cached_cover_pk = if let Some(ref image_base) = block.image_base {
let cached_cover_pk = if let Some(ref cover_cache) = self.inner.cover_cache { if let Some(ref cover) = song.cover {
if let Some(ref image_base) = block.image_base { let image_url = format!("{}{}", image_base, cover);
if let Some(ref cover) = song.cover {
let image_url = format!("{}{}", image_base, cover);
// Cache the cover image asynchronously match self.inner.cache_manager.cache_cover(&image_url).await {
match cover_cache.add_from_url(&image_url).await { Ok(pk) => {
Ok(pk) => { // Use the cached cover URL
// Use the cached cover URL let cached_url = self.inner.cache_manager.cover_url(&pk, None);
let cached_url = format!("{}/covers/images/{}", self.inner.cache_base_url, pk); track = track.with_image(cached_url);
track = track.with_image(cached_url); Some(pk)
Some(pk) }
} Err(e) => {
Err(e) => { tracing::warn!("Failed to cache cover image {}: {}", image_url, e);
tracing::warn!("Failed to cache cover image {}: {}", image_url, e); // Fall back to original URL
// Fall back to original URL track = track.with_image(image_url);
track = track.with_image(image_url); None
None
}
} }
} else {
None
} }
} else { } else {
None None
@@ -231,100 +178,68 @@ impl RadioParadiseSource {
None None
}; };
// If no cache, add original cover image // 2. Cache audio via le manager (métadonnées pour compatibilité)
#[cfg(not(feature = "cache"))] let metadata = AudioMetadata {
if let Some(ref image_base) = block.image_base { title: Some(song.title.clone()),
if let Some(ref cover) = song.cover { artist: if !song.artist.is_empty() {
let image_url = format!("{}{}", image_base, cover); Some(song.artist.clone())
track = track.with_image(image_url); } else {
} None
} },
album: if !song.album.is_empty() {
// Cache audio asynchronously (in background) Some(song.album.clone())
#[cfg(feature = "cache")] } else {
let cached_audio_pk = if let Some(ref audio_cache) = self.inner.audio_cache { None
// Prepare metadata for the audio cache },
let metadata = AudioMetadata { duration_secs: if song.duration > 0 {
title: Some(song.title.clone()), Some((song.duration / 1000) as u64)
artist: if !song.artist.is_empty() { } else {
Some(song.artist.clone()) None
} else { },
None year: None,
}, track_number: None,
album: if !song.album.is_empty() { track_total: None,
Some(song.album.clone()) disc_number: None,
} else { disc_total: None,
None genre: None,
}, sample_rate: None,
duration_secs: if song.duration > 0 { channels: None,
Some((song.duration / 1000) as u64) bitrate: None,
} else {
None
},
year: None,
track_number: None,
track_total: None,
disc_number: None,
disc_total: None,
genre: None,
sample_rate: None,
channels: None,
bitrate: None,
};
// Cache the audio asynchronously
match audio_cache.add_from_url(&block.url, Some(metadata)).await {
Ok((pk, _)) => {
tracing::info!("Successfully cached audio for track {}: {}", track_id, pk);
Some(pk)
}
Err(e) => {
tracing::warn!("Failed to cache audio for track {}: {}", track_id, e);
None
}
}
} else {
None
}; };
// Store metadata let cached_audio_pk = match self.inner.cache_manager.cache_audio(&block.url, Some(metadata)).await {
Ok(pk) => {
tracing::info!("Successfully cached audio for track {}: {}", track_id, pk);
Some(pk)
}
Err(e) => {
tracing::warn!("Failed to cache audio for track {}: {}", track_id, e);
None
}
};
// 3. Store metadata in the cache manager
self.inner.cache_manager.update_metadata(
track_id.clone(),
pmosource::TrackMetadata {
original_uri: block.url.clone(),
cached_audio_pk,
cached_cover_pk,
}
).await;
// 4. Store block for later retrieval
{ {
let mut cache = self.inner.track_cache.write().await; let mut blocks = self.inner.blocks.write().await;
cache.insert( blocks.insert(track_id.clone(), (block.clone(), song_index));
track_id.clone(),
TrackMetadata {
original_uri: block.url.clone(),
cached_pk: None,
block: block.clone(),
song_index,
#[cfg(feature = "cache")]
cached_audio_pk,
#[cfg(feature = "cache")]
cached_cover_pk,
},
);
} }
// Add to FIFO // 5. Add to FIFO
self.inner.playlist.append_track(track).await; self.inner.playlist.append_track(track).await;
Ok(()) Ok(())
} }
/// Mark a track as cached
///
/// Call this after successfully caching a track's audio via pmoaudiocache.
pub async fn cache_track(&self, track_id: &str, cache_pk: String) -> Result<()> {
let mut cache = self.inner.track_cache.write().await;
if let Some(metadata) = cache.get_mut(track_id) {
metadata.cached_pk = Some(cache_pk);
Ok(())
} else {
Err(MusicSourceError::ObjectNotFound(track_id.to_string()))
}
}
/// Convert a pmoplaylist::Track to pmodidl::Item /// Convert a pmoplaylist::Track to pmodidl::Item
fn track_to_item(&self, track: &Track) -> Item { fn track_to_item(&self, track: &Track) -> Item {
let duration_str = track.duration.map(|d| { let duration_str = track.duration.map(|d| {
@@ -398,25 +313,8 @@ impl MusicSource for RadioParadiseSource {
} }
async fn resolve_uri(&self, object_id: &str) -> Result<String> { async fn resolve_uri(&self, object_id: &str) -> Result<String> {
let cache = self.inner.track_cache.read().await; // Delegate to cache manager
self.inner.cache_manager.resolve_uri(object_id).await
if let Some(metadata) = cache.get(object_id) {
// Priority 1: Use cached audio if available
#[cfg(feature = "cache")]
if let Some(ref pk) = metadata.cached_audio_pk {
return Ok(format!("{}/audio/tracks/{}/stream", self.inner.cache_base_url, pk));
}
// Priority 2: Use legacy cached_pk (for backward compatibility)
if let Some(ref pk) = metadata.cached_pk {
return Ok(format!("{}/audio/cache/{}", self.inner.cache_base_url, pk));
}
// Priority 3: Return original block URI (not cached yet)
Ok(metadata.original_uri.clone())
} else {
Err(MusicSourceError::ObjectNotFound(object_id.to_string()))
}
} }
fn supports_fifo(&self) -> bool { fn supports_fifo(&self) -> bool {
@@ -471,11 +369,10 @@ impl MusicSource for RadioParadiseSource {
async fn remove_oldest(&self) -> Result<Option<Item>> { async fn remove_oldest(&self) -> Result<Option<Item>> {
if let Some(track) = self.inner.playlist.remove_oldest().await { if let Some(track) = self.inner.playlist.remove_oldest().await {
// Remove from cache // Remove from caches
{ self.inner.cache_manager.remove_track(&track.id).await;
let mut cache = self.inner.track_cache.write().await; let mut blocks = self.inner.blocks.write().await;
cache.remove(&track.id); blocks.remove(&track.id);
}
Ok(Some(self.track_to_item(&track))) Ok(Some(self.track_to_item(&track)))
} else { } else {
@@ -567,112 +464,56 @@ impl MusicSource for RadioParadiseSource {
} }
async fn get_cache_status(&self, object_id: &str) -> Result<pmosource::CacheStatus> { async fn get_cache_status(&self, object_id: &str) -> Result<pmosource::CacheStatus> {
use pmosource::CacheStatus; // Delegate to cache manager
self.inner.cache_manager.get_cache_status(object_id).await
let cache = self.inner.track_cache.read().await;
if let Some(metadata) = cache.get(object_id) {
#[cfg(feature = "cache")]
{
if let Some(ref audio_cache) = self.inner.audio_cache {
if let Some(ref pk) = metadata.cached_audio_pk {
// Check if the cached file exists and get its size
if let Ok(Some(info)) = audio_cache.get_info(pk).await {
return Ok(CacheStatus::Cached {
size_bytes: info.size_bytes,
});
}
}
}
}
// Check legacy cached_pk for backward compatibility
if metadata.cached_pk.is_some() {
// We don't have size info for legacy cache
return Ok(CacheStatus::Cached { size_bytes: 0 });
}
}
Ok(CacheStatus::NotCached)
} }
async fn cache_item(&self, object_id: &str) -> Result<pmosource::CacheStatus> { async fn cache_item(&self, object_id: &str) -> Result<pmosource::CacheStatus> {
#[cfg(not(feature = "cache"))] use pmosource::CacheStatus;
{
let _ = object_id; // Check if already cached
return Err(MusicSourceError::NotSupported("Caching not enabled".to_string())); let status = self.inner.cache_manager.get_cache_status(object_id).await?;
if matches!(status, CacheStatus::Cached { .. }) {
return Ok(status);
} }
#[cfg(feature = "cache")] // Get metadata and block info
{ let metadata = self.inner.cache_manager.get_metadata(object_id).await
use pmosource::CacheStatus; .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
// Get the track metadata let blocks = self.inner.blocks.read().await;
let cache = self.inner.track_cache.read().await; let (block, song_index) = blocks.get(object_id)
let metadata = cache .ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
.get(object_id) let song = block.get_song(*song_index)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))? .ok_or_else(|| MusicSourceError::ObjectNotFound(format!("Song {} not found", object_id)))?;
.clone();
drop(cache);
// If already cached, return status // Prepare metadata
if metadata.cached_audio_pk.is_some() { let audio_metadata = AudioMetadata {
return self.get_cache_status(object_id).await; title: Some(song.title.clone()),
artist: if !song.artist.is_empty() { Some(song.artist.clone()) } else { None },
album: if !song.album.is_empty() { Some(song.album.clone()) } else { None },
duration_secs: if song.duration > 0 { Some((song.duration / 1000) as u64) } else { None },
year: None,
track_number: None,
track_total: None,
disc_number: None,
disc_total: None,
genre: None,
sample_rate: None,
channels: None,
bitrate: None,
};
// Cache via manager
match self.inner.cache_manager.cache_audio(&metadata.original_uri, Some(audio_metadata)).await {
Ok(pk) => {
// Update metadata with new pk
let mut updated = metadata;
updated.cached_audio_pk = Some(pk);
self.inner.cache_manager.update_metadata(object_id.to_string(), updated).await;
self.get_cache_status(object_id).await
} }
Err(e) => Ok(CacheStatus::Failed { error: e.to_string() }),
// Cache it now
if let Some(ref audio_cache) = self.inner.audio_cache {
let song = &metadata.block.songs[metadata.song_index];
let audio_metadata = pmoaudiocache::AudioMetadata {
title: Some(song.title.clone()),
artist: if !song.artist.is_empty() {
Some(song.artist.clone())
} else {
None
},
album: if !song.album.is_empty() {
Some(song.album.clone())
} else {
None
},
duration_secs: if song.duration > 0 {
Some((song.duration / 1000) as u64)
} else {
None
},
year: None,
track_number: None,
track_total: None,
disc_number: None,
disc_total: None,
genre: None,
sample_rate: None,
channels: None,
bitrate: None,
};
match audio_cache
.add_from_url(&metadata.original_uri, Some(audio_metadata))
.await
{
Ok((pk, _)) => {
// Update the metadata
let mut cache = self.inner.track_cache.write().await;
if let Some(meta) = cache.get_mut(object_id) {
meta.cached_audio_pk = Some(pk);
}
return self.get_cache_status(object_id).await;
}
Err(e) => {
return Ok(CacheStatus::Failed {
error: e.to_string(),
});
}
}
}
Ok(CacheStatus::NotCached)
} }
} }
@@ -701,26 +542,13 @@ impl MusicSource for RadioParadiseSource {
} }
async fn statistics(&self) -> Result<pmosource::SourceStatistics> { async fn statistics(&self) -> Result<pmosource::SourceStatistics> {
let mut stats = pmosource::SourceStatistics::default(); let cache_stats = self.inner.cache_manager.statistics().await;
// Total items in FIFO Ok(pmosource::SourceStatistics {
stats.total_items = Some(self.inner.playlist.len().await); total_items: Some(self.inner.playlist.len().await),
cached_items: Some(cache_stats.cached_tracks),
// Cache statistics ..Default::default()
#[cfg(feature = "cache")] })
{
let cache = self.inner.track_cache.read().await;
let cached_count = cache.values().filter(|m| m.cached_audio_pk.is_some()).count();
stats.cached_items = Some(cached_count);
if let Some(ref audio_cache) = self.inner.audio_cache {
if let Ok(cache_stats) = audio_cache.statistics().await {
stats.cache_size_bytes = Some(cache_stats.total_size_bytes);
}
}
}
Ok(stats)
} }
} }
@@ -728,10 +556,37 @@ impl MusicSource for RadioParadiseSource {
mod tests { mod tests {
use super::*; use super::*;
// Helper to create test caches (requires actual directories in tests)
async fn create_test_caches() -> (Arc<CoverCache>, Arc<AudioCache>) {
let temp_dir = std::env::temp_dir();
let cover_dir = temp_dir.join("test_covers");
let audio_dir = temp_dir.join("test_audio");
std::fs::create_dir_all(&cover_dir).ok();
std::fs::create_dir_all(&audio_dir).ok();
let cover_cache = Arc::new(
pmocovers::Cache::new(cover_dir.to_str().unwrap(), 100, "http://localhost:8080")
.await.unwrap()
);
let audio_cache = Arc::new(
pmoaudiocache::new_cache(audio_dir.to_str().unwrap(), 100, "http://localhost:8080")
.unwrap()
);
(cover_cache, audio_cache)
}
#[tokio::test] #[tokio::test]
async fn test_source_info() { async fn test_source_info() {
let client = RadioParadiseClient::with_client(reqwest::Client::new()); let client = RadioParadiseClient::with_client(reqwest::Client::new());
let source = RadioParadiseSource::new_default(client, "http://localhost:8080"); let (cover_cache, audio_cache) = create_test_caches().await;
let source = RadioParadiseSource::new_default(
client,
"http://localhost:8080",
cover_cache,
audio_cache
);
assert_eq!(source.name(), "Radio Paradise"); assert_eq!(source.name(), "Radio Paradise");
assert_eq!(source.id(), "radio-paradise"); assert_eq!(source.id(), "radio-paradise");
@@ -752,7 +607,13 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn test_fifo_operations() { async fn test_fifo_operations() {
let client = RadioParadiseClient::with_client(reqwest::Client::new()); let client = RadioParadiseClient::with_client(reqwest::Client::new());
let source = RadioParadiseSource::new_default(client, "http://localhost:8080"); let (cover_cache, audio_cache) = create_test_caches().await;
let source = RadioParadiseSource::new_default(
client,
"http://localhost:8080",
cover_cache,
audio_cache
);
// Initially empty // Initially empty
let items = source.get_items(0, 10).await.unwrap(); let items = source.get_items(0, 10).await.unwrap();

14
pmoqobuz/Cargo.toml
View File

@@ -34,11 +34,11 @@ chrono = { version = "0.4", features = ["serde"] }
# Configuration # Configuration
pmoconfig = { path = "../pmoconfig" } pmoconfig = { path = "../pmoconfig" }
# Intégration avec pmocovers pour le cache d'images # Intégration avec pmocovers pour le cache d'images (OBLIGATOIRE)
pmocovers = { path = "../pmocovers", optional = true } pmocovers = { path = "../pmocovers" }
# Intégration avec pmoaudiocache pour le cache audio # Intégration avec pmoaudiocache pour le cache audio (OBLIGATOIRE)
pmoaudiocache = { path = "../pmoaudiocache", optional = true } pmoaudiocache = { path = "../pmoaudiocache" }
# Intégration avec pmodidl pour l'export DIDL # Intégration avec pmodidl pour l'export DIDL
pmodidl = { path = "../pmodidl" } pmodidl = { path = "../pmodidl" }
@@ -57,10 +57,8 @@ pmosource = { path = "../pmosource" }
default = [] default = []
# Feature pour activer les extensions pmoserver # Feature pour activer les extensions pmoserver
pmoserver = ["dep:pmoserver", "dep:axum", "dep:utoipa"] pmoserver = ["dep:pmoserver", "dep:axum", "dep:utoipa"]
# Feature pour activer le cache d'images via pmocovers # Feature cache (deprecated - toujours actif maintenant)
covers = ["dep:pmocovers"] cache = []
# Feature pour activer le cache complet (images + audio)
cache = ["dep:pmocovers", "dep:pmoaudiocache"]
[dev-dependencies] [dev-dependencies]
# Tests # Tests

366
pmoqobuz/src/source.rs
View File

@@ -7,16 +7,12 @@ use crate::client::QobuzClient;
use crate::didl::ToDIDL; use crate::didl::ToDIDL;
use crate::models::Track; use crate::models::Track;
use pmosource::{async_trait, BrowseResult, MusicSource, MusicSourceError, Result}; use pmosource::{async_trait, BrowseResult, MusicSource, MusicSourceError, Result};
use pmosource::SourceCacheManager;
use pmoaudiocache::{AudioMetadata, Cache as AudioCache};
use pmocovers::Cache as CoverCache;
use pmodidl::{Container, Item}; use pmodidl::{Container, Item};
use std::collections::HashMap;
use std::sync::Arc; use std::sync::Arc;
use std::time::SystemTime; use std::time::SystemTime;
use tokio::sync::RwLock;
#[cfg(feature = "cache")]
use pmocovers::{Cache as CoverCache, ImageCacheExt};
#[cfg(feature = "cache")]
use pmoaudiocache::{AudioCache, AudioMetadata};
/// Default image for Qobuz (300x300 WebP, embedded in binary) /// Default image for Qobuz (300x300 WebP, embedded in binary)
const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp"); const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp");
@@ -71,32 +67,12 @@ struct QobuzSourceInner {
/// Qobuz API client /// Qobuz API client
client: QobuzClient, client: QobuzClient,
/// Cache server base URL for URI resolution /// Cache manager (centralisé)
cache_base_url: String, cache_manager: SourceCacheManager,
/// Track metadata cache (track_id -> TrackMetadata)
track_cache: RwLock<HashMap<String, TrackMetadata>>,
/// Cover image cache (optional)
#[cfg(feature = "cache")]
cover_cache: Option<Arc<CoverCache>>,
/// Audio cache (optional)
#[cfg(feature = "cache")]
audio_cache: Option<Arc<AudioCache>>,
/// Update tracking /// Update tracking
update_counter: RwLock<u32>, update_counter: tokio::sync::RwLock<u32>,
last_change: RwLock<SystemTime>, last_change: tokio::sync::RwLock<SystemTime>,
}
#[derive(Debug, Clone)]
struct TrackMetadata {
original_uri: String,
#[cfg(feature = "cache")]
cached_audio_pk: Option<String>,
#[cfg(feature = "cache")]
cached_cover_pk: Option<String>,
} }
impl std::fmt::Debug for QobuzSource { impl std::fmt::Debug for QobuzSource {
@@ -106,89 +82,34 @@ impl std::fmt::Debug for QobuzSource {
} }
impl QobuzSource { impl QobuzSource {
/// Create a new Qobuz source /// Create a new Qobuz source with caches
/// ///
/// # Arguments /// # Arguments
/// ///
/// * `client` - Authenticated Qobuz API client /// * `client` - Authenticated Qobuz API client
/// * `cache_base_url` - Base URL for the cache server (e.g., "http://localhost:8080") /// * `cache_base_url` - Base URL for the cache server (e.g., "http://localhost:8080")
/// /// * `cover_cache` - Cover image cache (required)
/// # Examples /// * `audio_cache` - Audio cache (required)
/// pub fn new(
/// ```no_run
/// use pmoqobuz::{QobuzSource, QobuzClient};
///
/// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let client = QobuzClient::from_config().await?;
/// let source = QobuzSource::new(client, "http://localhost:8080");
/// Ok(())
/// }
/// ```
pub fn new(client: QobuzClient, cache_base_url: impl Into<String>) -> Self {
Self {
inner: Arc::new(QobuzSourceInner {
client,
cache_base_url: cache_base_url.into(),
track_cache: RwLock::new(HashMap::new()),
#[cfg(feature = "cache")]
cover_cache: None,
#[cfg(feature = "cache")]
audio_cache: None,
update_counter: RwLock::new(0),
last_change: RwLock::new(SystemTime::now()),
}),
}
}
/// Create a new Qobuz source with caching support
///
/// # Arguments
///
/// * `client` - Authenticated Qobuz API client
/// * `cache_base_url` - Base URL for the cache server (e.g., "http://localhost:8080")
/// * `cover_cache` - Optional cover image cache
/// * `audio_cache` - Optional audio cache
///
/// # Examples
///
/// ```no_run
/// use pmoqobuz::{QobuzSource, QobuzClient};
/// use pmocovers::Cache as CoverCache;
/// use pmoaudiocache::AudioCache;
/// use std::sync::Arc;
///
/// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let client = QobuzClient::from_config().await?;
/// let cover_cache = Arc::new(CoverCache::new("/tmp/qobuz-covers").await?);
/// let audio_cache = Arc::new(AudioCache::new("/tmp/qobuz-audio").await?);
///
/// let source = QobuzSource::new_with_cache(
/// client,
/// "http://localhost:8080",
/// Some(cover_cache),
/// Some(audio_cache),
/// );
/// Ok(())
/// }
/// ```
#[cfg(feature = "cache")]
pub fn new_with_cache(
client: QobuzClient, client: QobuzClient,
cache_base_url: impl Into<String>, cache_base_url: impl Into<String>,
cover_cache: Option<Arc<CoverCache>>, cover_cache: Arc<CoverCache>,
audio_cache: Option<Arc<AudioCache>>, audio_cache: Arc<AudioCache>,
) -> Self { ) -> Self {
let cache_base_url = cache_base_url.into();
let cache_manager = SourceCacheManager::new(
cache_base_url.clone(),
"qobuz".to_string(),
cover_cache,
audio_cache,
);
Self { Self {
inner: Arc::new(QobuzSourceInner { inner: Arc::new(QobuzSourceInner {
client, client,
cache_base_url: cache_base_url.into(), cache_manager,
track_cache: RwLock::new(HashMap::new()), update_counter: tokio::sync::RwLock::new(0),
cover_cache, last_change: tokio::sync::RwLock::new(SystemTime::now()),
audio_cache,
update_counter: RwLock::new(0),
last_change: RwLock::new(SystemTime::now()),
}), }),
} }
} }
@@ -198,114 +119,56 @@ impl QobuzSource {
&self.inner.client &self.inner.client
} }
/// Add a track from Qobuz with optional caching /// Add a track from Qobuz with caching
/// ///
/// This method is used to add a Qobuz track to the internal cache, /// This method downloads and caches both cover art and audio data.
/// downloading and caching both cover art and audio data if caching is enabled.
///
/// # Arguments
///
/// * `track` - The Qobuz track to add
///
/// # Returns
///
/// Returns the track ID that was used for caching.
pub async fn add_track(&self, track: &Track) -> Result<String> { pub async fn add_track(&self, track: &Track) -> Result<String> {
let track_id = format!("qobuz://track/{}", track.id); let track_id = format!("qobuz://track/{}", track.id);
// Get streaming URL // Get streaming URL
let stream_url = self let stream_url = self.inner.client.get_stream_url(&track.id).await
.inner
.client
.get_stream_url(&track.id)
.await
.map_err(|e| MusicSourceError::UriResolutionError(e.to_string()))?; .map_err(|e| MusicSourceError::UriResolutionError(e.to_string()))?;
// Cache cover image // 1. Cache cover via manager
#[cfg(feature = "cache")] let cached_cover_pk = if let Some(ref album) = track.album {
let cached_cover_pk = if let Some(ref cover_cache) = self.inner.cover_cache { if let Some(ref image_url) = album.image {
if let Some(ref album) = track.album { self.inner.cache_manager.cache_cover(image_url).await.ok()
if let Some(ref image_url) = album.image { } else { None }
match cover_cache.add_image_from_url(image_url).await { } else { None };
Ok(pk) => {
tracing::info!("Successfully cached cover for track {}: {}", track_id, pk); // 2. Prepare rich metadata from Qobuz track
Some(pk) let metadata = AudioMetadata {
} title: Some(track.title.clone()),
Err(e) => { artist: track.performer.as_ref().map(|p| p.name.clone()),
tracing::warn!("Failed to cache cover image {}: {}", image_url, e); album: track.album.as_ref().map(|a| a.title.clone()),
None duration_secs: Some(track.duration as u64),
} year: track.album.as_ref().and_then(|a| {
} a.release_date.as_ref().and_then(|d| d.split('-').next()?.parse().ok())
} else { }),
None track_number: Some(track.track_number),
} track_total: track.album.as_ref().and_then(|a| a.tracks_count),
} else { disc_number: Some(track.media_number),
None disc_total: None,
} genre: track.album.as_ref().and_then(|a| {
} else { if !a.genres.is_empty() { Some(a.genres.join(", ")) } else { None }
None }),
sample_rate: track.sample_rate,
channels: track.channels,
bitrate: None,
}; };
// Cache audio asynchronously // 3. Cache audio via manager
#[cfg(feature = "cache")] let cached_audio_pk = self.inner.cache_manager.cache_audio(&stream_url, Some(metadata)).await.ok();
let cached_audio_pk = if let Some(ref audio_cache) = self.inner.audio_cache {
// Prepare rich metadata from Qobuz track
let metadata = AudioMetadata {
title: Some(track.title.clone()),
artist: track.performer.as_ref().map(|p| p.name.clone()),
album: track.album.as_ref().map(|a| a.title.clone()),
duration_secs: Some(track.duration as u64),
year: track.album.as_ref().and_then(|a| {
a.release_date.as_ref().and_then(|d| {
// Parse year from ISO date (e.g., "2023-01-15")
d.split('-').next()?.parse().ok()
})
}),
track_number: Some(track.track_number),
track_total: track.album.as_ref().and_then(|a| a.tracks_count),
disc_number: Some(track.media_number),
disc_total: None,
genre: track.album.as_ref().and_then(|a| {
if !a.genres.is_empty() {
Some(a.genres.join(", "))
} else {
None
}
}),
sample_rate: track.sample_rate,
channels: track.channels,
bitrate: None, // Qobuz doesn't provide bitrate directly
};
// Cache the audio asynchronously // 4. Store metadata
match audio_cache.add_from_url(&stream_url, Some(metadata)).await { self.inner.cache_manager.update_metadata(
Ok((pk, _)) => { track_id.clone(),
tracing::info!("Successfully cached audio for track {}: {}", track_id, pk); pmosource::TrackMetadata {
Some(pk) original_uri: stream_url,
} cached_audio_pk,
Err(e) => { cached_cover_pk,
tracing::warn!("Failed to cache audio for track {}: {}", track_id, e);
None
}
} }
} else { ).await;
None
};
// Store metadata
{
let mut cache = self.inner.track_cache.write().await;
cache.insert(
track_id.clone(),
TrackMetadata {
original_uri: stream_url,
#[cfg(feature = "cache")]
cached_audio_pk,
#[cfg(feature = "cache")]
cached_cover_pk,
},
);
}
Ok(track_id) Ok(track_id)
} }
@@ -484,33 +347,15 @@ impl MusicSource for QobuzSource {
} }
async fn resolve_uri(&self, object_id: &str) -> Result<String> { async fn resolve_uri(&self, object_id: &str) -> Result<String> {
// Check if we have cached metadata for this track // Try cache manager first
let cache = self.inner.track_cache.read().await; if let Ok(uri) = self.inner.cache_manager.resolve_uri(object_id).await {
return Ok(uri);
if let Some(metadata) = cache.get(object_id) {
// Priority 1: Use cached audio if available
#[cfg(feature = "cache")]
if let Some(ref pk) = metadata.cached_audio_pk {
return Ok(format!("{}/audio/tracks/{}/stream", self.inner.cache_base_url, pk));
}
// Priority 2: Return original stream URI (already fetched)
return Ok(metadata.original_uri.clone());
} }
// If not in cache, extract track ID and get streaming URL from Qobuz // If not cached, extract track ID and get streaming URL from Qobuz
// Object IDs for tracks follow pattern: "qobuz://track/{id}" let track_id = object_id.strip_prefix("qobuz://track/").unwrap_or(object_id);
let track_id = if let Some(id) = object_id.strip_prefix("qobuz://track/") {
id
} else {
object_id
};
// Get streaming URL from Qobuz self.inner.client.get_stream_url(track_id).await
self.inner
.client
.get_stream_url(track_id)
.await
.map_err(|e| MusicSourceError::UriResolutionError(e.to_string())) .map_err(|e| MusicSourceError::UriResolutionError(e.to_string()))
} }
@@ -662,60 +507,22 @@ impl MusicSource for QobuzSource {
} }
async fn get_cache_status(&self, object_id: &str) -> Result<pmosource::CacheStatus> { async fn get_cache_status(&self, object_id: &str) -> Result<pmosource::CacheStatus> {
use pmosource::CacheStatus; self.inner.cache_manager.get_cache_status(object_id).await
let cache = self.inner.track_cache.read().await;
if let Some(metadata) = cache.get(object_id) {
#[cfg(feature = "cache")]
{
if let Some(ref _audio_cache) = self.inner.audio_cache {
if let Some(ref _pk) = metadata.cached_audio_pk {
// TODO: AudioCache doesn't have get_info method yet
// For now, just return that it's cached without size info
return Ok(CacheStatus::Cached {
size_bytes: 0,
});
}
}
}
}
Ok(CacheStatus::NotCached)
} }
async fn cache_item(&self, object_id: &str) -> Result<pmosource::CacheStatus> { async fn cache_item(&self, object_id: &str) -> Result<pmosource::CacheStatus> {
#[cfg(not(feature = "cache"))] // Extract track ID
{ let track_id = object_id.strip_prefix("qobuz://track/").unwrap_or(object_id);
let _ = object_id;
return Err(MusicSourceError::NotSupported("Caching not enabled".to_string()));
}
#[cfg(feature = "cache")] // Get track details from Qobuz
{ let track = self.inner.client.get_track(track_id).await
use pmosource::CacheStatus; .map_err(|e| MusicSourceError::BrowseError(e.to_string()))?;
// Extract track ID // Add track to cache (via manager)
let track_id = if let Some(id) = object_id.strip_prefix("qobuz://track/") { let cached_id = self.add_track(&track).await?;
id
} else {
object_id
};
// Get track details // Return the cache status
let track = self self.get_cache_status(&cached_id).await
.inner
.client
.get_track(track_id)
.await
.map_err(|e| MusicSourceError::BrowseError(e.to_string()))?;
// Add track to cache
let cached_id = self.add_track(&track).await?;
// Return the cache status
self.get_cache_status(&cached_id).await
}
} }
async fn add_favorite(&self, object_id: &str) -> Result<()> { async fn add_favorite(&self, object_id: &str) -> Result<()> {
@@ -942,18 +749,9 @@ impl MusicSource for QobuzSource {
stats.total_items = Some(tracks.len()); stats.total_items = Some(tracks.len());
} }
// Get cache statistics // Get cache statistics from manager
#[cfg(feature = "cache")] let cache_stats = self.inner.cache_manager.statistics().await;
{ stats.cached_items = Some(cache_stats.cached_tracks);
let cache = self.inner.track_cache.read().await;
stats.cached_items = Some(cache.len());
// TODO: AudioCache doesn't have statistics method yet
// For now, just count cached items
if let Some(ref _audio_cache) = self.inner.audio_cache {
// stats.cache_size_bytes will remain None
}
}
Ok(stats) Ok(stats)
} }

242
pmosource/src/cache.rs Normal file
View File

@@ -0,0 +1,242 @@
//! Gestion du cache pour les sources musicales
//!
//! Ce module fournit `SourceCacheManager` qui permet aux sources
//! d'utiliser les caches centralisés du serveur.
//!
//! ## Architecture
//!
//! Les caches (couvertures et audio) sont centralisés au niveau du serveur UPnP.
//! Chaque source utilise ces caches partagés avec sa propre collection.
//!
//! ```text
//! UpnpServer
//! ├─ CoverCache (partagé)
//! │ ├─ collection: "radio-paradise"
//! │ └─ collection: "qobuz"
//! └─ AudioCache (partagé)
//! ├─ collection: "radio-paradise"
//! └─ collection: "qobuz"
//! ```
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use pmocovers::Cache as CoverCache;
use pmoaudiocache::{Cache as AudioCache, AudioMetadata};
use crate::{MusicSourceError, Result, CacheStatus};
/// Métadonnées d'une piste en cache
#[derive(Debug, Clone)]
pub struct TrackMetadata {
/// URI originale de la piste
pub original_uri: String,
/// Clé primaire du fichier audio en cache
pub cached_audio_pk: Option<String>,
/// Clé primaire de la couverture en cache
pub cached_cover_pk: Option<String>,
}
/// Manager centralisé pour gérer le cache d'une source
///
/// Utilise les caches centralisés du serveur avec la collection de la source.
/// Chaque source a son propre `SourceCacheManager` mais partage les mêmes
/// caches (cover et audio) avec les autres sources.
pub struct SourceCacheManager {
/// Métadonnées des pistes (track_id → metadata)
track_cache: RwLock<HashMap<String, TrackMetadata>>,
/// URL de base du serveur
cache_base_url: String,
/// ID de collection pour cette source (ex: "radio-paradise", "qobuz")
collection_id: String,
/// Référence au cache de couvertures centralisé
cover_cache: Arc<CoverCache>,
/// Référence au cache audio centralisé
audio_cache: Arc<AudioCache>,
}
impl SourceCacheManager {
/// Créer un nouveau manager
///
/// # Arguments
///
/// * `cache_base_url` - URL de base du serveur
/// * `collection_id` - ID de collection (source ID)
/// * `cover_cache` - Cache de couvertures centralisé
/// * `audio_cache` - Cache audio centralisé
pub fn new(
cache_base_url: String,
collection_id: String,
cover_cache: Arc<CoverCache>,
audio_cache: Arc<AudioCache>,
) -> Self {
Self {
track_cache: RwLock::new(HashMap::new()),
cache_base_url,
collection_id,
cover_cache,
audio_cache,
}
}
/// Résoudre l'URI d'une piste (priorité au cache)
///
/// Retourne l'URI du fichier audio en cache si disponible,
/// sinon l'URI originale.
pub async fn resolve_uri(&self, object_id: &str) -> Result<String> {
let cache = self.track_cache.read().await;
if let Some(metadata) = cache.get(object_id) {
if let Some(ref pk) = metadata.cached_audio_pk {
return Ok(format!("{}/audio/tracks/{}/stream", self.cache_base_url, pk));
}
return Ok(metadata.original_uri.clone());
}
Err(MusicSourceError::ObjectNotFound(object_id.to_string()))
}
/// Obtenir le statut du cache pour une piste
pub async fn get_cache_status(&self, object_id: &str) -> Result<CacheStatus> {
let cache = self.track_cache.read().await;
if let Some(metadata) = cache.get(object_id) {
if let Some(ref pk) = metadata.cached_audio_pk {
// TODO: Ajouter get_info() à AudioCache
// Pour l'instant, on retourne juste Cached sans taille
return Ok(CacheStatus::Cached { size_bytes: 0 });
}
}
Ok(CacheStatus::NotCached)
}
/// Cacher une couverture depuis une URL
///
/// Utilise la collection de cette source pour organiser les images.
///
/// # Returns
///
/// La clé primaire (pk) de l'image dans le cache
pub async fn cache_cover(&self, url: &str) -> Result<String> {
self.cover_cache
.add_from_url(url, Some(&self.collection_id))
.await
.map_err(|e| MusicSourceError::CacheError(e.to_string()))
}
/// Obtenir l'URL d'une couverture en cache
///
/// # Arguments
///
/// * `pk` - Clé primaire de l'image dans le cache
/// * `size` - Taille optionnelle (génère une variante si spécifiée)
///
/// # Returns
///
/// L'URL complète de l'image
pub fn cover_url(&self, pk: &str, size: Option<usize>) -> String {
if let Some(s) = size {
format!("{}/covers/images/{}/{}", self.cache_base_url, pk, s)
} else {
format!("{}/covers/images/{}", self.cache_base_url, pk)
}
}
/// Cacher une piste audio depuis une URL
///
/// Utilise la collection de cette source pour organiser les pistes.
///
/// # Arguments
///
/// * `url` - URL source de la piste
/// * `_metadata` - Métadonnées audio optionnelles (unused, kept for API compatibility)
///
/// # Returns
///
/// La clé primaire (pk) de la piste dans le cache
pub async fn cache_audio(&self, url: &str, _metadata: Option<AudioMetadata>)
-> Result<String> {
// Note: Les métadonnées seront extraites automatiquement par le cache audio
// lors de la conversion FLAC
let pk = self.audio_cache
.add_from_url(url, Some(&self.collection_id))
.await
.map_err(|e| MusicSourceError::CacheError(e.to_string()))?;
Ok(pk)
}
/// Mettre à jour les métadonnées d'une piste
///
/// Enregistre ou met à jour les métadonnées de cache pour une piste.
pub async fn update_metadata(&self, track_id: String, metadata: TrackMetadata) {
let mut cache = self.track_cache.write().await;
cache.insert(track_id, metadata);
}
/// Récupérer les métadonnées d'une piste
pub async fn get_metadata(&self, track_id: &str) -> Option<TrackMetadata> {
let cache = self.track_cache.read().await;
cache.get(track_id).cloned()
}
/// Supprimer une piste du cache
pub async fn remove_track(&self, track_id: &str) {
let mut cache = self.track_cache.write().await;
cache.remove(track_id);
}
/// Obtenir l'ID de collection
pub fn collection_id(&self) -> &str {
&self.collection_id
}
/// Obtenir les statistiques du cache pour cette source
pub async fn statistics(&self) -> CacheStatistics {
let cache = self.track_cache.read().await;
let cached_count = cache.values()
.filter(|m| m.cached_audio_pk.is_some())
.count();
CacheStatistics {
total_tracks: cache.len(),
cached_tracks: cached_count,
collection_id: self.collection_id.clone(),
}
}
}
/// Statistiques du cache pour une source
#[derive(Debug, Clone)]
pub struct CacheStatistics {
/// Nombre total de pistes connues
pub total_tracks: usize,
/// Nombre de pistes en cache
pub cached_tracks: usize,
/// ID de collection
pub collection_id: String,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_track_metadata() {
let metadata = TrackMetadata {
original_uri: "http://example.com/track.flac".to_string(),
cached_audio_pk: Some("abc123".to_string()),
cached_cover_pk: Some("def456".to_string()),
};
assert_eq!(metadata.original_uri, "http://example.com/track.flac");
assert_eq!(metadata.cached_audio_pk, Some("abc123".to_string()));
}
}

5
pmosource/src/lib.rs
View File

@@ -35,6 +35,8 @@
//! server.register_music_source(Arc::new(my_source)).await; //! server.register_music_source(Arc::new(my_source)).await;
//! ``` //! ```
pub mod cache;
use pmodidl::{Container, Item}; use pmodidl::{Container, Item};
use std::fmt::Debug; use std::fmt::Debug;
use std::time::SystemTime; use std::time::SystemTime;
@@ -904,6 +906,9 @@ pub use async_trait::async_trait;
pub use pmodidl; pub use pmodidl;
pub use pmoplaylist; pub use pmoplaylist;
// Re-export cache types
pub use cache::{TrackMetadata, SourceCacheManager, CacheStatistics};
// Server extension modules (feature-gated) // Server extension modules (feature-gated)
#[cfg(feature = "server")] #[cfg(feature = "server")]
pub mod pmoserver_ext; pub mod pmoserver_ext;

5
pmoupnp/Cargo.toml
View File

@@ -8,12 +8,16 @@ pmoconfig = { path = "../pmoconfig" }
pmodidl = { path = "../pmodidl"} pmodidl = { path = "../pmodidl"}
pmoutils = { path = "../pmoutils" } pmoutils = { path = "../pmoutils" }
pmoserver = { path = "../pmoserver" } pmoserver = { path = "../pmoserver" }
pmocovers = { path = "../pmocovers", features = ["pmoserver"] }
pmoaudiocache = { path = "../pmoaudiocache", features = ["pmoserver"] }
pmocache = { path = "../pmocache" }
url = "2.5.7" url = "2.5.7"
uuid = "1.18.1" uuid = "1.18.1"
hex = "0.4.3" hex = "0.4.3"
base64 = "0.22.1" base64 = "0.22.1"
thiserror = "2.0.16" thiserror = "2.0.16"
anyhow = "1.0"
xmltree = "0.11.0" xmltree = "0.11.0"
axum = "0.8.4" axum = "0.8.4"
tokio = { version = "1.35", features = ["rt-multi-thread", "macros", "sync"] } tokio = { version = "1.35", features = ["rt-multi-thread", "macros", "sync"] }
@@ -27,3 +31,4 @@ tracing = "0.1"
bevy_reflect = "0.17.1" bevy_reflect = "0.17.1"
bevy_reflect_derive = "0.17.1" bevy_reflect_derive = "0.17.1"
reqwest = "0.12.23" reqwest = "0.12.23"
utoipa = { version = "5.3", features = ["axum_extras"] }

112
pmoupnp/src/cache_registry.rs Normal file
View File

@@ -0,0 +1,112 @@
//! Registre centralisé des caches pour le serveur UPnP
//!
//! Ce module gère les caches partagés entre toutes les sources musicales :
//! - Cache de couvertures d'albums (WebP)
//! - Cache de pistes audio (FLAC)
//!
//! Les caches supportent les collections, permettant à chaque source
//! d'avoir sa propre collection dans le cache partagé.
use std::sync::Arc;
use once_cell::sync::Lazy;
use std::sync::RwLock;
use pmocovers::Cache as CoverCache;
use pmoaudiocache::Cache as AudioCache;
/// Registre global des caches
///
/// Contient les instances partagées des caches de couvertures et audio.
/// Ces caches sont uniques et partagés entre toutes les sources musicales.
pub struct CacheRegistry {
/// Cache de couvertures (WebP)
cover_cache: Option<Arc<CoverCache>>,
/// Cache audio (FLAC)
audio_cache: Option<Arc<AudioCache>>,
}
impl CacheRegistry {
/// Créer un nouveau registre vide
pub fn new() -> Self {
Self {
cover_cache: None,
audio_cache: None,
}
}
/// Enregistrer le cache de couvertures
pub fn set_cover_cache(&mut self, cache: Arc<CoverCache>) {
self.cover_cache = Some(cache);
}
/// Récupérer le cache de couvertures
pub fn cover_cache(&self) -> Option<Arc<CoverCache>> {
self.cover_cache.clone()
}
/// Enregistrer le cache audio
pub fn set_audio_cache(&mut self, cache: Arc<AudioCache>) {
self.audio_cache = Some(cache);
}
/// Récupérer le cache audio
pub fn audio_cache(&self) -> Option<Arc<AudioCache>> {
self.audio_cache.clone()
}
}
impl Default for CacheRegistry {
fn default() -> Self {
Self::new()
}
}
/// Registre global thread-safe
///
/// Utilise Lazy pour une initialisation paresseuse et RwLock pour le partage entre threads.
/// Permet aux handlers et aux sources d'accéder aux caches depuis n'importe où.
pub(crate) static CACHE_REGISTRY: Lazy<RwLock<CacheRegistry>> = Lazy::new(|| {
RwLock::new(CacheRegistry::new())
});
/// Accès global au cache de couvertures
///
/// # Examples
///
/// ```rust,ignore
/// use pmoupnp::cache_registry::get_cover_cache;
///
/// if let Some(cache) = get_cover_cache() {
/// let pk = cache.add_from_url("http://example.com/cover.jpg").await?;
/// }
/// ```
pub fn get_cover_cache() -> Option<Arc<CoverCache>> {
CACHE_REGISTRY.read().unwrap().cover_cache()
}
/// Accès global au cache audio
///
/// # Examples
///
/// ```rust,ignore
/// use pmoupnp::cache_registry::get_audio_cache;
///
/// if let Some(cache) = get_audio_cache() {
/// let (pk, _) = cache.add_from_url("http://example.com/track.flac", None).await?;
/// }
/// ```
pub fn get_audio_cache() -> Option<Arc<AudioCache>> {
CACHE_REGISTRY.read().unwrap().audio_cache()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_cache_registry_empty() {
let registry = CacheRegistry::new();
assert!(registry.cover_cache().is_none());
assert!(registry.audio_cache().is_none());
}
}

6
pmoupnp/src/lib.rs
View File

@@ -1,6 +1,7 @@
mod object_trait; mod object_trait;
mod object_set; mod object_set;
pub mod cache_registry;
pub mod upnp_server; pub mod upnp_server;
pub mod upnp_api; pub mod upnp_api;
pub mod actions; pub mod actions;
@@ -12,15 +13,12 @@ pub mod state_variables;
pub mod value_ranges; pub mod value_ranges;
pub mod variable_types; pub mod variable_types;
use std::{collections::HashMap, sync::Arc}; use std::{collections::HashMap, sync::Arc};
use std::sync::RwLock; use std::sync::RwLock;
pub use crate::object_trait::*; pub use crate::object_trait::*;
pub use crate::upnp_server::UpnpServerExt; pub use crate::upnp_server::UpnpServerExt;
pub use crate::cache_registry::{get_cover_cache, get_audio_cache};
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct UpnpObjectType { pub struct UpnpObjectType {

153
pmoupnp/src/upnp_server.rs
View File

@@ -23,10 +23,15 @@ use std::sync::RwLock;
use once_cell::sync::Lazy; use once_cell::sync::Lazy;
use pmoserver::Server; use pmoserver::Server;
use utoipa::OpenApi;
use crate::devices::errors::DeviceError; use crate::devices::errors::DeviceError;
use crate::devices::{Device, DeviceInstance, DeviceRegistry}; use crate::devices::{Device, DeviceInstance, DeviceRegistry};
use crate::UpnpModel; use crate::UpnpModel;
use crate::cache_registry::CACHE_REGISTRY;
use pmocovers::Cache as CoverCache;
use pmoaudiocache::Cache as AudioCache;
/// Registre de devices global et thread-safe. /// Registre de devices global et thread-safe.
/// ///
@@ -70,6 +75,8 @@ static DEVICE_REGISTRY: Lazy<RwLock<DeviceRegistry>> = Lazy::new(|| {
/// let devices = server.device_registry().list_devices(); /// let devices = server.device_registry().list_devices();
/// ``` /// ```
pub trait UpnpServerExt { pub trait UpnpServerExt {
// ========= Device Management (existant) =========
/// Enregistre un device UPnP et toutes ses URLs. /// Enregistre un device UPnP et toutes ses URLs.
/// ///
/// # Arguments /// # Arguments
@@ -89,6 +96,57 @@ pub trait UpnpServerExt {
/// Récupère un device par son UDN. /// Récupère un device par son UDN.
fn get_device(&self, udn: &str) -> Option<Arc<DeviceInstance>>; fn get_device(&self, udn: &str) -> Option<Arc<DeviceInstance>>;
// ========= Cache Management (NOUVEAU) =========
/// Initialiser le cache de couvertures centralisé
///
/// Crée le cache et enregistre les routes HTTP.
/// Toutes les sources musicales utiliseront ce cache partagé.
///
/// # Arguments
///
/// * `cache_dir` - Répertoire de stockage
/// * `limit` - Limite de taille (nombre d'images)
///
/// # Returns
///
/// Instance partagée du cache
async fn init_cover_cache(&mut self, cache_dir: &str, limit: usize)
-> Result<Arc<CoverCache>, anyhow::Error>;
/// Initialiser le cache audio centralisé
///
/// Crée le cache et enregistre les routes HTTP.
/// Toutes les sources musicales utiliseront ce cache partagé.
///
/// # Arguments
///
/// * `cache_dir` - Répertoire de stockage
/// * `limit` - Limite de taille (nombre de pistes)
///
/// # Returns
///
/// Instance partagée du cache
async fn init_audio_cache(&mut self, cache_dir: &str, limit: usize)
-> Result<Arc<AudioCache>, anyhow::Error>;
/// Initialiser les caches depuis la configuration
///
/// Utilise pmoconfig pour charger les paramètres et initialiser
/// automatiquement les deux caches.
///
/// # Returns
///
/// Tuple (cache de couvertures, cache audio)
async fn init_caches(&mut self)
-> Result<(Arc<CoverCache>, Arc<AudioCache>), anyhow::Error>;
/// Récupérer le cache de couvertures
fn cover_cache(&self) -> Option<Arc<CoverCache>>;
/// Récupérer le cache audio
fn audio_cache(&self) -> Option<Arc<AudioCache>>;
} }
// Implémentation du trait UpnpServer pour pmoserver::Server // Implémentation du trait UpnpServer pour pmoserver::Server
@@ -120,6 +178,101 @@ impl UpnpServerExt for Server {
fn get_device(&self, udn: &str) -> Option<Arc<DeviceInstance>> { fn get_device(&self, udn: &str) -> Option<Arc<DeviceInstance>> {
DEVICE_REGISTRY.read().unwrap().get_device(udn) DEVICE_REGISTRY.read().unwrap().get_device(udn)
} }
// ========= Cache Management Implementation =========
async fn init_cover_cache(&mut self, cache_dir: &str, limit: usize)
-> Result<Arc<CoverCache>, anyhow::Error> {
use pmocovers::new_cache;
use pmocache::pmoserver_ext::{create_file_router_with_generator, create_api_router};
let base_url = self.info().base_url;
let cache = Arc::new(new_cache(cache_dir, limit, &base_url)?);
// Routes de fichiers avec génération de variantes
// Routes: GET /covers/image/{pk} et GET /covers/image/{pk}/{size}
let variant_generator: pmocache::pmoserver_ext::ParamGenerator<pmocovers::CoversConfig> =
Arc::new(|cache, pk, param| {
Box::pin(async move {
// Si le param est numérique, c'est une taille de variante
if let Ok(size) = param.parse::<usize>() {
match pmocovers::webp::generate_variant(&cache, &pk, size).await {
Ok(data) => return Some(data),
Err(e) => {
tracing::warn!("Cannot generate variant {}x{} for {}: {}", size, size, pk, e);
return None;
}
}
}
None
})
});
let file_router = create_file_router_with_generator(
cache.clone(),
"image/webp",
Some(variant_generator)
);
self.add_router("/", file_router).await;
// API REST générique (pmocache)
let api_router = create_api_router(cache.clone());
let openapi = pmocovers::ApiDoc::openapi();
self.add_openapi(api_router, openapi, "covers").await;
// Enregistrer dans le registre global
CACHE_REGISTRY.write().unwrap().set_cover_cache(cache.clone());
Ok(cache)
}
async fn init_audio_cache(&mut self, cache_dir: &str, limit: usize)
-> Result<Arc<AudioCache>, anyhow::Error> {
use pmoaudiocache::new_cache;
use pmocache::pmoserver_ext::{create_file_router, create_api_router};
let base_url = self.info().base_url;
let cache = Arc::new(new_cache(cache_dir, limit, &base_url)?);
// Routes de fichiers pour servir les pistes FLAC
let file_router = create_file_router(cache.clone(), "audio/flac");
self.add_router("/", file_router).await;
// API REST générique (pmocache)
let api_router = create_api_router(cache.clone());
let openapi = pmoaudiocache::ApiDoc::openapi();
self.add_openapi(api_router, openapi, "audio").await;
// Enregistrer dans le registre global
CACHE_REGISTRY.write().unwrap().set_audio_cache(cache.clone());
Ok(cache)
}
async fn init_caches(&mut self)
-> Result<(Arc<CoverCache>, Arc<AudioCache>), anyhow::Error> {
let config = pmoconfig::get_config();
let cover_cache = self.init_cover_cache(
&config.get_cover_cache_dir()?,
config.get_cover_cache_size()?
).await?;
let audio_cache = self.init_audio_cache(
&config.get_audio_cache_dir()?,
config.get_audio_cache_size()?
).await?;
Ok((cover_cache, audio_cache))
}
fn cover_cache(&self) -> Option<Arc<CoverCache>> {
crate::cache_registry::get_cover_cache()
}
fn audio_cache(&self) -> Option<Arc<AudioCache>> {
crate::cache_registry::get_audio_cache()
}
} }
/// Fonctions helper pour accéder au registre depuis les handlers. /// Fonctions helper pour accéder au registre depuis les handlers.