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

Merge pull request #25 from coissac/claude/fix-play-and-cache-streaming-011CUsMBxH4fsgoadgkiPdoK

Browse Source 
Claude/fix play and cache streaming 011 c us m bx h4fsgoadgki pdo k
This commit is contained in:
coissac
2025-11-07 07:29:21 +01:00
committed by GitHub
parent 4803c6ab73 64586721b9
commit a0dd3278ad
18 changed files with 1632 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@@ -2928,6 +2928,7 @@ dependencies = [
"serde_yaml",
"sha1",
"sha2",
"tempfile",
"tokio",
"tokio-util",
"tracing",
@@ -2966,6 +2967,7 @@ dependencies = [
"pmoserver",
"reqwest",
"serde",
"tempfile",
"tokio",
"tracing",
"utoipa",

50
INSTALL_LIBSOXR.md
View File

@@ -152,7 +152,22 @@ cargo test
### Configuration initiale (à faire une seule fois)
Dans une session Claude Code (https://claude.ai/code), vous n'avez pas de droits sudo. Suivez ces étapes :
Dans une session Claude Code (https://claude.ai/code), vous n'avez pas de droits sudo.
**🚀 Méthode rapide (recommandée) :**
```bash
# 1. Installation automatique des dépendances (une seule fois)
./setup-deps.sh
# 2. Configuration des variables d'environnement (à chaque session)
source setup-env.sh
# 3. Compilation
cargo build
```
**📋 Méthode manuelle (si les scripts ne fonctionnent pas) :**
#### 1. Installation des dépendances
@@ -184,14 +199,28 @@ export LD_LIBRARY_PATH="$HOME/.local/usr/lib/x86_64-linux-gnu:$LD_LIBRARY_PATH"
export RUSTFLAGS="-L $HOME/.local/usr/lib/x86_64-linux-gnu"
```
**Astuce :** Copier ces trois lignes dans un fichier `setup-env.sh` à la racine du projet :
**Astuce :** Créez un fichier `setup-env.sh` pour ne pas avoir à retaper ces commandes à chaque session :
```bash
cat > setup-env.sh << 'EOF'
#!/bin/bash
# Script de configuration des variables d'environnement pour PMOMusic
# Usage: source setup-env.sh
# Configuration des chemins pour libsoxr et libasound2
export PKG_CONFIG_PATH="$HOME/.local/usr/lib/x86_64-linux-gnu/pkgconfig:$PKG_CONFIG_PATH"
export LD_LIBRARY_PATH="$HOME/.local/usr/lib/x86_64-linux-gnu:$LD_LIBRARY_PATH"
export RUSTFLAGS="-L $HOME/.local/usr/lib/x86_64-linux-gnu"
echo "Variables d'environnement configurées pour PMOMusic"
echo " PKG_CONFIG_PATH=$PKG_CONFIG_PATH"
echo " LD_LIBRARY_PATH=$LD_LIBRARY_PATH"
echo " RUSTFLAGS=$RUSTFLAGS"
echo ""
echo "Vous pouvez maintenant compiler avec: cargo build"
EOF
chmod +x setup-env.sh
```
Puis dans chaque session :
@@ -200,7 +229,7 @@ Puis dans chaque session :
source setup-env.sh
```
⚠️ **NE PAS committer `setup-env.sh`** - ajouter au `.gitignore`
⚠️ **Note :** Le fichier `setup-env.sh` est dans `.gitignore` (configuration locale), vous devez le créer vous-même avec le contenu ci-dessus.
#### 3. Vérifier l'installation
@@ -228,9 +257,16 @@ cargo run --package pmoparadise --example play_and_cache --features full -- 0
À chaque fois que vous démarrez une nouvelle session Claude Code :
1. **Exporter les variables d'environnement** (ou `source setup-env.sh`)
2. Compiler avec `cargo build`
3. Exécuter les exemples ou tests
```bash
# 1. Configuration de l'environnement
source setup-env.sh
# 2. Compilation
cargo build
# 3. Exécution des exemples
cargo run --package pmoparadise --example play_and_cache --features full -- 0
```
**IMPORTANT :** Si vous oubliez d'exporter les variables, vous obtiendrez des erreurs comme :
```
@@ -244,7 +280,7 @@ ou
rust-lld: error: unable to find library -lasound
```
Solution : Exporter les variables et recompiler.
**Solution :** Exécutez `source setup-env.sh` et recompilez.
### Notes importantes

17
INSTALL_NOTES.md
View File

@@ -29,7 +29,22 @@ brew install libsoxr
apk add soxr-dev alsa-lib-dev
```
**Sans privilèges root** : Si vous n'avez pas les droits sudo, consultez `INSTALL_LIBSOXR.md` pour l'installation locale de `libsoxr` et `libasound2`.
**Sans privilèges root (Claude Code, environnements sans sudo)** :
🚀 **Installation automatique** :
```bash
# 1. Installation des dépendances (une seule fois)
./setup-deps.sh
# 2. Configuration de l'environnement (à chaque session)
source setup-env.sh
# 3. Compilation
cargo build
```
Pour plus de détails, consultez `INSTALL_LIBSOXR.md`.
---

38
Readme.md
View File

@@ -1,5 +1,43 @@
# Développement de l'application PMOMusic en RUST
## 🚀 Démarrage rapide
### Installation des dépendances (environnement sans sudo)
Pour compiler PMOMusic dans un environnement sans privilèges sudo (comme Claude Code) :
```bash
# 1. Installation automatique de libsoxr et libasound2 (une seule fois)
./setup-deps.sh
# 2. Créer le fichier setup-env.sh (une seule fois, voir INSTALL_LIBSOXR.md pour le contenu)
cat > setup-env.sh << 'EOF'
#!/bin/bash
export PKG_CONFIG_PATH="$HOME/.local/usr/lib/x86_64-linux-gnu/pkgconfig:$PKG_CONFIG_PATH"
export LD_LIBRARY_PATH="$HOME/.local/usr/lib/x86_64-linux-gnu:$LD_LIBRARY_PATH"
export RUSTFLAGS="-L $HOME/.local/usr/lib/x86_64-linux-gnu"
echo "Variables d'environnement configurées pour PMOMusic"
EOF
# 3. Configuration de l'environnement (à chaque nouvelle session)
source setup-env.sh
# 4. Compilation
cargo build
# 5. Test de l'exemple Radio Paradise
cargo run --package pmoparadise --example play_and_cache --features full -- 0
```
⚠️ **Note :** Le fichier `setup-env.sh` est dans `.gitignore` car il contient une configuration locale.
### Documentation
- **[INSTALL_NOTES.md](INSTALL_NOTES.md)** - Guide d'installation général
- **[INSTALL_LIBSOXR.md](INSTALL_LIBSOXR.md)** - Installation détaillée de libsoxr et ALSA
---
## Création de la structure
```bash

130
pmoaudio-ext/src/sinks/flac_cache_sink.rs
View File

@@ -10,7 +10,6 @@ use pmoaudiocache::AudioTrackMetadataExt;
use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat};
use std::{
collections::VecDeque,
io::Cursor,
pin::Pin,
sync::Arc,
task::{Context, Poll},
@@ -127,16 +126,26 @@ impl NodeLogic for FlacCacheSinkLogic {
// Créer l'encoder
let reader = ByteStreamReader::new(pcm_rx);
let mut flac_stream = encode_flac_stream(reader, format, options_with_metadata)
let flac_stream = encode_flac_stream(reader, format, options_with_metadata)
.await
.map_err(|e| {
AudioError::ProcessingError(format!("FLAC encode init failed: {}", e))
})?;
// Créer un buffer pour collecter le FLAC encodé
let mut flac_buffer = Vec::new();
// Ingérer le FLAC progressivement dans le cache
// add_from_reader lance l'ingestion en arrière-plan et retourne dès que
// le prebuffer (512 KB) est atteint, permettant un streaming progressif
// Le cache skip automatiquement le header FLAC (512 octets) pour calculer le pk
// à partir du contenu audio, évitant les collisions entre morceaux au même format
let collection_ref = self.collection.as_deref();
let cache_future = self.cache.add_from_reader(
None,
flac_stream,
None, // Taille inconnue car streaming
collection_ref,
);
// Exécuter pump et copy en parallèle
// Exécuter pump et add_from_reader en parallèle
let pump_future = pump_track_segments(
first_segment,
&mut rx,
@@ -145,40 +154,20 @@ impl NodeLogic for FlacCacheSinkLogic {
sample_rate,
&stop_token,
);
let copy_future = async {
tokio::io::copy(&mut flac_stream, &mut flac_buffer)
.await
.map_err(|e| {
AudioError::ProcessingError(format!("FLAC write failed: {}", e))
})?;
flac_stream
.wait()
.await
.map_err(|e| AudioError::ProcessingError(format!("Encoder failed: {}", e)))?;
Ok::<_, AudioError>(())
};
// Attendre les deux tâches en parallèle
let (copy_result, pump_result) = tokio::join!(copy_future, pump_future);
copy_result?;
let (_chunks, _samples, _duration_sec, stop_reason) = pump_result?;
let (cache_result, pump_result) = tokio::join!(cache_future, pump_future);
// Ingérer le FLAC dans le cache
let flac_reader = Cursor::new(flac_buffer.clone());
let collection_ref = self.collection.as_deref();
let pk = self.cache
.add_from_reader(
None,
flac_reader,
Some(flac_buffer.len() as u64),
collection_ref,
)
.await
.map_err(|e| {
let pk = cache_result.map_err(|e| {
AudioError::ProcessingError(format!("Failed to add to cache: {}", e))
})?;
tracing::debug!("Track added to cache with pk {}, prebuffer complete", pk);
let (_chunks, _samples, _duration_sec, stop_reason) = pump_result?;
// Copier les métadonnées du TrackBoundary dans le cache
// IMPORTANT: Faire ceci AVANT d'ajouter à la playlist pour que les métadonnées soient disponibles
if let Some(src_metadata) = track_metadata {
let dest_metadata = self.cache.track_metadata(&pk);
@@ -221,7 +210,8 @@ impl NodeLogic for FlacCacheSinkLogic {
}
}
// Ajouter à la playlist si enregistrée
// Ajouter à la playlist IMMÉDIATEMENT (avant le drainage!)
// Ceci permet à la lecture de commencer pendant que les segments sont drainés
#[cfg(feature = "playlist")]
if let Some(ref playlist_handle) = self.playlist_handle {
playlist_handle.push(pk.clone()).await.map_err(|e| {
@@ -229,6 +219,16 @@ impl NodeLogic for FlacCacheSinkLogic {
})?;
}
// Si le fichier était déjà en cache (ChannelClosed), drainer les segments restants
// jusqu'au prochain TrackBoundary ou EndOfStream
// IMPORTANT: Faire ceci APRÈS l'ajout à la playlist pour ne pas bloquer la lecture
let stop_reason = if matches!(stop_reason, StopReason::ChannelClosed) {
tracing::debug!("File was already in cache, draining remaining segments");
drain_until_track_boundary(&mut rx, &stop_token).await?
} else {
stop_reason
};
// Vérifier le stop_reason pour savoir si on continue
match stop_reason {
StopReason::TrackBoundary(_metadata) => {
@@ -360,6 +360,50 @@ async fn wait_for_first_audio_chunk_with_metadata(
}
}
/// Draine tous les segments jusqu'au prochain TrackBoundary ou EndOfStream
///
/// Cette fonction est utilisée quand le fichier était déjà en cache et que
/// nous devons ignorer les segments restants pour rester synchronisé avec la source.
async fn drain_until_track_boundary(
rx: &mut mpsc::Receiver<Arc<AudioSegment>>,
stop_token: &CancellationToken,
) -> Result<StopReason, AudioError> {
loop {
let segment = tokio::select! {
result = rx.recv() => {
match result {
Some(seg) => seg,
None => {
return Ok(StopReason::ChannelClosed);
}
}
}
_ = stop_token.cancelled() => {
return Ok(StopReason::ChannelClosed);
}
};
match &segment.segment {
_AudioSegment::Chunk(_) => {
// Ignorer les chunks audio
continue;
}
_AudioSegment::Sync(marker) => match &**marker {
SyncMarker::TrackBoundary { metadata, .. } => {
return Ok(StopReason::TrackBoundary(metadata.clone()));
}
SyncMarker::EndOfStream => {
return Ok(StopReason::EndOfStream);
}
_ => {
// Ignorer les autres syncmarkers
continue;
}
},
}
}
}
/// Pompe les segments pour une seule track (s'arrête au TrackBoundary).
async fn pump_track_segments(
first_segment: Arc<AudioSegment>,
@@ -377,10 +421,12 @@ async fn pump_track_segments(
if let Some(chunk) = first_segment.as_chunk() {
let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?;
if !pcm_bytes.is_empty() {
pcm_tx
.send(pcm_bytes)
.await
.map_err(|_| AudioError::SendError)?;
// Si le send échoue, c'est que le receiver est fermé
// (par exemple, le fichier était déjà en cache et add_from_reader a retourné immédiatement)
if pcm_tx.send(pcm_bytes).await.is_err() {
drop(pcm_tx);
return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed));
}
chunks += 1;
samples += chunk.len() as u64;
duration_sec += chunk.len() as f64 / expected_rate as f64;
@@ -421,10 +467,12 @@ async fn pump_track_segments(
continue;
}
pcm_tx
.send(pcm_bytes)
.await
.map_err(|_| AudioError::SendError)?;
// Si le send échoue, c'est que le receiver est fermé
// (par exemple, le fichier était déjà en cache et add_from_reader a retourné immédiatement)
if pcm_tx.send(pcm_bytes).await.is_err() {
drop(pcm_tx);
return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed));
}
chunks += 1;
samples += chunk.len() as u64;

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

@@ -237,11 +237,15 @@ impl NodeLogic for PlaylistSourceLogic {
tracing::debug!("PlaylistSourceLogic: decoding track: {:?}", file_path);
// 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(
&file_path,
self.chunk_frames,
&output,
&stop_token,
&self.cache,
cache_pk,
)
.await
{
@@ -264,11 +268,16 @@ impl NodeLogic for PlaylistSourceLogic {
// ═══════════════════════════════════════════════════════════════════════════
/// Décode un fichier et émet ses chunks audio
///
/// Gère le cache progressif : si EOF est atteint et que le download est toujours en cours,
/// attend et réessaie au lieu de terminer immédiatement.
async fn decode_and_emit_track(
path: &PathBuf,
chunk_frames: usize,
output: &[mpsc::Sender<Arc<AudioSegment>>],
stop_token: &CancellationToken,
cache: &Arc<AudioCache>,
cache_pk: &str,
) -> Result<(), AudioError> {
// Ouvrir et décoder
let file = File::open(path)
@@ -320,9 +329,23 @@ async fn decode_and_emit_track(
let read = read_result.map_err(|e| {
AudioError::IoError(format!("I/O error while decoding: {}", e))
})?;
if read == 0 && pending.is_empty() {
// Si EOF atteint (read == 0)
if read == 0 {
// Vérifier si le download est toujours en cours (cache progressif)
if cache.get_download(cache_pk).await.is_some() {
// Download en cours - attendre un peu et réessayer
tracing::trace!("decode_and_emit_track: EOF reached but download ongoing, waiting...");
tokio::time::sleep(Duration::from_millis(100)).await;
continue; // Retry la lecture
}
// Download terminé - c'est vraiment la fin du fichier
if pending.is_empty() {
break;
}
}
if read > 0 {
pending.extend_from_slice(&read_buf[..read]);
}

40
pmoaudiocache/src/cache.rs
View File

@@ -56,6 +56,46 @@ pub fn new_cache(dir: &str, limit: usize) -> Result<Cache> {
Cache::with_transformer(dir, limit, Some(transformer_factory))
}
/// Crée un cache audio et lance la consolidation en arrière-plan
///
/// Cette fonction crée le cache et lance immédiatement une consolidation
/// pour nettoyer les fichiers incomplets (sans marker de complétion).
///
/// # Arguments
///
/// * `dir` - Répertoire de stockage du cache
/// * `limit` - Limite de taille du cache (nombre de pistes)
///
/// # Returns
///
/// Arc vers l'instance du cache configurée pour la conversion FLAC automatique
///
/// # Exemple
///
/// ```rust,no_run
/// use pmoaudiocache::cache;
///
/// # async fn example() -> anyhow::Result<()> {
/// let cache = cache::new_cache_with_consolidation("./audio_cache", 1000).await?;
/// # Ok(())
/// # }
/// ```
pub async fn new_cache_with_consolidation(dir: &str, limit: usize) -> Result<Arc<Cache>> {
let cache = Arc::new(new_cache(dir, limit)?);
// Lancer la consolidation en arrière-plan pour nettoyer les fichiers incomplets
let cache_clone = cache.clone();
tokio::spawn(async move {
if let Err(e) = cache_clone.consolidate().await {
tracing::warn!("Failed to consolidate cache on startup: {}", e);
} else {
tracing::info!("Cache consolidated successfully on startup");
}
});
Ok(cache)
}
/// Ajoute une piste audio depuis une URL avec extraction et stockage des métadonnées
///
/// Cette fonction étend `add_from_url` du cache en ajoutant :

96
pmoaudiocache/tests/test_cache.rs Normal file
View File

@@ -0,0 +1,96 @@
use pmoaudiocache::cache;
use tempfile::TempDir;
fn create_test_cache() -> (TempDir, cache::Cache) {
let temp_dir = tempfile::tempdir().unwrap();
let cache = cache::new_cache(temp_dir.path().to_str().unwrap(), 10).unwrap();
(temp_dir, cache)
}
#[tokio::test]
async fn test_audio_cache_creation() {
let (temp_dir, cache) = create_test_cache();
assert_eq!(cache.cache_dir(), temp_dir.path());
}
#[tokio::test]
#[ignore] // Test nécessite un vrai fichier audio FLAC
async fn test_add_from_file() {
let (_temp_dir, cache) = create_test_cache();
// Créer un fichier de test
let test_file = tempfile::NamedTempFile::with_suffix(".dat").unwrap();
std::fs::write(test_file.path(), b"Test audio data").unwrap();
let pk = cache
.add_from_file(test_file.path().to_str().unwrap(), None)
.await
.unwrap();
assert!(!pk.is_empty());
}
#[tokio::test]
async fn test_audio_config() {
use pmocache::CacheConfig;
assert_eq!(cache::AudioConfig::file_extension(), "flac");
assert_eq!(cache::AudioConfig::cache_type(), "flac");
assert_eq!(cache::AudioConfig::cache_name(), "audio");
assert_eq!(cache::AudioConfig::default_param(), "orig");
}
#[tokio::test]
#[ignore] // Test nécessite un vrai fichier audio FLAC
async fn test_collection_management() {
let (_temp_dir, cache) = create_test_cache();
let collection = "test_album";
// Ajouter plusieurs pistes à la même collection
for i in 0..3 {
let data = format!("Track {} audio data", i);
let file = tempfile::NamedTempFile::with_suffix(".dat").unwrap();
std::fs::write(file.path(), data.as_bytes()).unwrap();
cache
.add_from_file(file.path().to_str().unwrap(), Some(collection))
.await
.unwrap();
}
// Attendre un peu pour que les fichiers soient prêts
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
// Récupérer la collection
let collection_files = cache.get_collection(collection).await.unwrap();
assert_eq!(collection_files.len(), 3);
}
#[tokio::test]
#[ignore] // Test nécessite un vrai fichier audio FLAC
async fn test_cache_limit() {
let temp_dir = tempfile::tempdir().unwrap();
let cache = cache::new_cache(temp_dir.path().to_str().unwrap(), 2).unwrap();
// Ajouter 3 fichiers (devrait déclencher l'éviction LRU)
for i in 0..3 {
let data = format!("Track {}", i);
let file = tempfile::NamedTempFile::with_suffix(".dat").unwrap();
std::fs::write(file.path(), data.as_bytes()).unwrap();
cache
.add_from_file(file.path().to_str().unwrap(), None)
.await
.unwrap();
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
}
// Attendre que l'éviction se fasse
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
// Le cache ne devrait contenir que 2 éléments
let count = cache.db.count().unwrap();
assert_eq!(count, 2);
}

3
pmocache/Cargo.toml
View File

@@ -41,6 +41,9 @@ axum = { version = "0.8", optional = true }
pmoconfig = { path = "../pmoconfig", optional = true }
serde_yaml = { version = "0.9", optional = true }
[dev-dependencies]
tempfile = "3"
[features]
default = []
openapi = ["dep:utoipa"]

269
pmocache/src/cache.rs
View File

@@ -13,10 +13,13 @@ use serde_json::{Number, Value};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::io::AsyncRead;
use tokio::io::{AsyncRead, AsyncReadExt};
use tokio::sync::RwLock;
use tracing;
/// Taille minimale de prébuffering par défaut (512 KB = ~5 secondes de FLAC)
pub const DEFAULT_PREBUFFER_SIZE: u64 = 512 * 1024;
/// Paramètres statiques d'un cache spécialisé.
pub trait CacheConfig: Send + Sync {
/// Extension des fichiers générés (ex: `"webp"`, `"flac"`).
@@ -58,11 +61,110 @@ pub struct Cache<C: CacheConfig> {
downloads: Arc<RwLock<HashMap<String, Arc<Download>>>>,
/// Factory pour créer des transformers (optionnel)
transformer_factory: Option<Arc<dyn Fn() -> StreamTransformer + Send + Sync>>,
/// Taille minimale de prébuffering en octets (0 = désactivé)
min_prebuffer_size: u64,
/// Phantom data pour le type de configuration
_phantom: std::marker::PhantomData<C>,
}
impl<C: CacheConfig> Cache<C> {
/// Retourne le chemin du fichier marker de complétion
fn get_completion_marker_path(&self, pk: &str) -> PathBuf {
self.get_file_path(pk).with_extension(format!("{}.complete", C::file_extension()))
}
/// Vérifie si un fichier est en cache et complet
///
/// # Returns
///
/// - `Ok(true)` si le fichier est en cache et complet (fichier .complete existe)
/// - `Ok(false)` si le fichier n'est pas en cache ou incomplet (et supprime les fichiers incomplets)
/// - `Err` en cas d'erreur
async fn check_cached_and_complete(&self, pk: &str) -> Result<bool> {
if self.db.get(pk, false).is_ok() {
let file_path = self.get_file_path(pk);
let completion_marker = self.get_completion_marker_path(pk);
if file_path.exists() {
// Vérifier si le fichier marker de complétion existe
if completion_marker.exists() {
tracing::debug!("File with pk {} is complete (marker exists)", pk);
return Ok(true);
} else {
tracing::warn!(
"File with pk {} in cache has no completion marker, will re-download/re-ingest",
pk
);
// Supprimer le fichier incomplet
let _ = std::fs::remove_file(&file_path);
return Ok(false);
}
}
}
Ok(false)
}
/// Vérifie si un download est en cours et attend le prébuffering si nécessaire
///
/// # Returns
///
/// - `Ok(Some(pk))` si un download est en cours (et prébuffering terminé)
/// - `Ok(None)` si aucun download en cours
/// - `Err` en cas d'erreur de prébuffering
async fn check_ongoing_download(&self, pk: &str) -> Result<Option<String>> {
let download_handle = {
let downloads = self.downloads.read().await;
downloads.get(pk).cloned()
};
if let Some(download) = download_handle {
tracing::debug!("Download already in progress for pk {}, waiting for prebuffering", pk);
if self.min_prebuffer_size > 0 {
download.wait_until_min_size(self.min_prebuffer_size).await
.map_err(|e| anyhow!("Prebuffering failed: {}", e))?;
tracing::debug!("Prebuffering complete for pk {}", pk);
}
return Ok(Some(pk.to_string()));
}
Ok(None)
}
/// Finalise l'ajout d'un fichier au cache
///
/// Cette fonction helper gère le prébuffering et le nettoyage en background
async fn finalize_download(&self, pk: &str, download: Arc<Download>) -> Result<String> {
// Attendre le prébuffering (pour le cache progressif)
if self.min_prebuffer_size > 0 {
download.wait_until_min_size(self.min_prebuffer_size).await
.map_err(|e| anyhow!("Prebuffering failed: {}", e))?;
tracing::debug!("Prebuffering complete for pk {} ({} bytes)", pk, self.min_prebuffer_size);
}
// Lancer une tâche de nettoyage et marquage de complétion en background
let downloads_clone = self.downloads.clone();
let pk_clone = pk.to_string();
let completion_marker = self.get_completion_marker_path(pk);
tokio::spawn(async move {
let result = download.wait_until_finished().await;
downloads_clone.write().await.remove(&pk_clone);
// Créer le fichier marker de complétion si le téléchargement a réussi
if result.is_ok() {
if let Err(e) = std::fs::write(&completion_marker, "") {
tracing::warn!("Failed to create completion marker for pk {}: {}", pk_clone, e);
} else {
tracing::debug!("Created completion marker for pk {}", pk_clone);
}
}
});
Ok(pk.to_string())
}
/// Crée un nouveau cache sans transformer
///
/// # Arguments
@@ -124,10 +226,39 @@ impl<C: CacheConfig> Cache<C> {
db: Arc::new(db),
downloads: Arc::new(RwLock::new(HashMap::new())),
transformer_factory,
min_prebuffer_size: DEFAULT_PREBUFFER_SIZE,
_phantom: std::marker::PhantomData,
})
}
/// Configure la taille minimale de prébuffering
///
/// # Arguments
///
/// * `size` - Taille minimale en octets (0 = désactivé)
///
/// # Exemple
///
/// ```rust,no_run
/// use pmocache::{Cache, CacheConfig};
///
/// struct MyConfig;
/// impl CacheConfig for MyConfig {
/// fn file_extension() -> &'static str { "dat" }
/// }
///
/// let mut cache = Cache::<MyConfig>::new("./cache", 1000).unwrap();
/// cache.set_prebuffer_size(1024 * 1024); // 1 MB de prébuffering
/// ```
pub fn set_prebuffer_size(&mut self, size: u64) {
self.min_prebuffer_size = size;
}
/// Retourne la taille minimale de prébuffering configurée
pub fn get_prebuffer_size(&self) -> u64 {
self.min_prebuffer_size
}
/// Télécharge un fichier depuis une URL et l'ajoute au cache
///
/// Cette méthode utilise un système d'identifiants basé sur le contenu plutôt que sur l'URL.
@@ -166,26 +297,17 @@ impl<C: CacheConfig> Cache<C> {
let pk = crate::cache_trait::pk_from_content_header(&header);
tracing::debug!("Computed pk {} for URL {}", pk, url);
// 3. Vérifier si le fichier est déjà en cache
if self.db.get(&pk, false).is_ok() {
let file_path = self.get_file_path(&pk);
if file_path.exists() {
// Déjà en cache, update timestamp et retour rapide
// 3. Vérifier si le fichier est déjà en cache ET complet
if self.check_cached_and_complete(&pk).await? {
tracing::debug!("File with pk {} already in cache, updating timestamp", pk);
self.db.update_hit(&pk)?;
return Ok(pk);
}
}
// 4. Vérifier si un download est déjà en cours pour ce pk
{
let downloads = self.downloads.read().await;
if downloads.contains_key(&pk) {
// Download déjà en cours pour ce contenu, retourner la clé
tracing::debug!("Download already in progress for pk {}", pk);
if let Some(pk) = self.check_ongoing_download(&pk).await? {
return Ok(pk);
}
}
// 5. Lancer le téléchargement complet avec transformer
tracing::debug!("Starting full download for pk {} from URL {}", pk, url);
@@ -202,20 +324,14 @@ impl<C: CacheConfig> Cache<C> {
// Ajouter immédiatement à la DB
self.db.add(&pk, None, collection)?;
self.db.set_origin_url(&pk, url)?;
// Appliquer la politique d'éviction LRU si nécessaire
if let Err(e) = self.enforce_limit().await {
tracing::warn!("Error enforcing cache limit: {}", e);
}
// Lancer une tâche de nettoyage en background
let downloads_clone = self.downloads.clone();
let pk_clone = pk.clone();
tokio::spawn(async move {
let _ = download.wait_until_finished().await;
downloads_clone.write().await.remove(&pk_clone);
});
Ok(pk)
// Finaliser avec prébuffering et nettoyage
self.finalize_download(&pk, download).await
}
/// Ajoute un fichier à partir d'un flux asynchrone.
@@ -252,43 +368,65 @@ impl<C: CacheConfig> Cache<C> {
where
R: AsyncRead + Send + Unpin + 'static,
{
// 1. Lire les 512 premiers octets pour calculer le pk
let header = crate::download::peek_reader_header(&mut reader, 512)
self.add_from_reader_with_pk(source_uri, reader, length, collection, None).await
}
/// Ajoute un fichier à partir d'un flux avec un pk explicite optionnel.
///
/// Si `explicit_pk` est fourni, utilise ce pk au lieu de le calculer à partir du contenu.
/// Ceci est utile quand plusieurs fichiers ont le même header mais doivent être cachés séparément
/// (par exemple, des fichiers FLAC avec le même format mais du contenu différent).
pub async fn add_from_reader_with_pk<R>(
&self,
source_uri: Option<&str>,
mut reader: R,
length: Option<u64>,
collection: Option<&str>,
explicit_pk: Option<String>,
) -> Result<String>
where
R: AsyncRead + Send + Unpin + 'static,
{
// 1. Lire jusqu'à 1024 octets (ou ce qui est disponible)
let header = crate::download::peek_reader_header(&mut reader, 1024)
.await
.map_err(|e| anyhow!("Failed to peek reader header: {}", e))?;
// 2. Calculer le pk basé sur le contenu
let pk = crate::cache_trait::pk_from_content_header(&header);
// 2. Calculer le pk en utilisant au plus les 512 derniers octets
// Ceci évite les collisions pour les fichiers avec headers identiques (ex: FLAC)
// tout en fonctionnant pour les petits fichiers (images < 512 octets)
let pk = if let Some(explicit) = explicit_pk {
explicit
} else {
let pk_bytes = if header.len() > 512 {
// Fichier >= 512 octets: utiliser les octets 512+ (au plus 512 octets)
&header[512..]
} else {
// Petit fichier < 512 octets: utiliser tout le contenu
&header[..]
};
crate::cache_trait::pk_from_content_header(pk_bytes)
};
if let Some(uri) = source_uri {
tracing::debug!("Computed pk {} for source_uri {}", pk, uri);
} else {
tracing::debug!("Computed pk {} from reader", pk);
}
// 3. Vérifier si le fichier est déjà en cache
if self.db.get(&pk, false).is_ok() {
let file_path = self.get_file_path(&pk);
if file_path.exists() {
// Déjà en cache, update timestamp et retour rapide
// 3. Vérifier si le fichier est déjà en cache ET complet
if self.check_cached_and_complete(&pk).await? {
tracing::debug!("File with pk {} already in cache, updating timestamp", pk);
self.db.update_hit(&pk)?;
return Ok(pk);
}
}
// 4. Vérifier si un download est déjà en cours pour ce pk
{
let downloads = self.downloads.read().await;
if downloads.contains_key(&pk) {
tracing::debug!("Download already in progress for pk {}", pk);
if let Some(pk) = self.check_ongoing_download(&pk).await? {
return Ok(pk);
}
}
// 5. Reconstituer le reader complet (header + reste)
// Utiliser tokio::io::chain pour créer un reader composé
use std::io::Cursor;
use tokio::io::AsyncReadExt;
let header_reader = Cursor::new(header);
let full_reader = header_reader.chain(reader);
@@ -312,14 +450,8 @@ impl<C: CacheConfig> Cache<C> {
tracing::warn!("Error enforcing cache limit: {}", e);
}
let downloads_clone = self.downloads.clone();
let pk_clone = pk.clone();
tokio::spawn(async move {
let _ = download.wait_until_finished().await;
downloads_clone.write().await.remove(&pk_clone);
});
Ok(pk)
// Finaliser avec prébuffering et nettoyage
self.finalize_download(&pk, download).await
}
/// Ajoute un fichier local au cache
@@ -497,15 +629,22 @@ impl<C: CacheConfig> Cache<C> {
}
/// Consolide le cache en supprimant les orphelins et en re-téléchargeant les fichiers manquants
///
/// Cette fonction :
/// - Supprime les entrées DB sans fichiers (ou re-télécharge si URL disponible)
/// - Supprime les fichiers sans marker de complétion et leurs entrées DB
/// - Supprime les fichiers sans entrées DB correspondantes
pub async fn consolidate(&self) -> Result<()> {
// Récupérer la liste des entrées à traiter
let entries = self.db.get_all(false)?;
// Supprimer les entrées sans fichiers correspondants
// Supprimer les entrées sans fichiers correspondants OU sans marker de complétion
for entry in entries {
let file_path = self.get_file_path(&entry.pk);
let completion_marker = self.get_completion_marker_path(&entry.pk);
if !file_path.exists() {
// Fichier manquant, essayer de re-télécharger
match self.db.get_origin_url(&entry.pk)? {
Some(url) => {
if let Err(err) = self.add_from_url(&url, entry.collection.as_deref()).await
@@ -522,6 +661,14 @@ impl<C: CacheConfig> Cache<C> {
self.db.delete(&entry.pk)?;
}
}
} else if !completion_marker.exists() {
// Fichier existe mais pas de marker de complétion -> fichier incomplet
tracing::warn!(
"Removing incomplete file {} (no completion marker)",
entry.pk
);
let _ = tokio::fs::remove_file(&file_path).await;
self.db.delete(&entry.pk)?;
}
}
@@ -531,11 +678,20 @@ impl<C: CacheConfig> Cache<C> {
let path = entry.path();
if path.is_file() && path != self.dir.join("cache.db") {
if let Some(file_name) = path.file_name().and_then(|n| n.to_str()) {
// Ignorer les fichiers .complete
if file_name.ends_with(".complete") {
continue;
}
// Format attendu: {pk}.{qualifier}.{EXT}
// On extrait le pk (première partie avant le premier point)
if let Some(pk) = file_name.split('.').next() {
if self.db.get(pk, false).is_err() {
tokio::fs::remove_file(path).await?;
tracing::debug!("Removing orphan file: {}", file_name);
tokio::fs::remove_file(&path).await?;
// Supprimer aussi le marker de complétion s'il existe
let completion_marker = self.get_completion_marker_path(pk);
let _ = tokio::fs::remove_file(&completion_marker).await;
}
}
}
@@ -770,17 +926,10 @@ impl<C: CacheConfig> Cache<C> {
let mut removed = 0;
for entry in old_entries {
// Supprimer tous les fichiers avec ce pk (toutes variantes)
if let Ok(mut dir_entries) = tokio::fs::read_dir(&self.dir).await {
while let Ok(Some(dir_entry)) = dir_entries.next_entry().await {
if let Some(filename) = dir_entry.file_name().to_str() {
// Format: {pk}.{param}.{ext}
if filename.starts_with(&entry.pk)
&& filename.starts_with(&format!("{}.", entry.pk))
{
let _ = tokio::fs::remove_file(dir_entry.path()).await;
}
}
// Utiliser get_file_paths() pour obtenir tous les fichiers de cette entrée
if let Ok(paths) = self.get_file_paths(&entry.pk) {
for path in paths {
let _ = tokio::fs::remove_file(path).await;
}
}

10
pmocache/src/db.rs
View File

@@ -192,6 +192,8 @@ impl DB {
collection: Option<&str>,
metadata: Option<&Value>,
) -> rusqlite::Result<()> {
// Bloc pour limiter la durée du lock
{
let conn = self.lock_conn("add_with_metadata");
conn.execute(
@@ -203,9 +205,11 @@ impl DB {
last_used = excluded.last_used",
params![pk, id, collection, Utc::now().to_rfc3339()],
)?;
} // Lock libéré ici
if metadata.is_some() {
self.set_metadata(pk, metadata.unwrap())?
// Appeler set_metadata après avoir libéré le lock pour éviter un deadlock
if let Some(metadata) = metadata {
self.set_metadata(pk, metadata)?;
}
Ok(())
@@ -678,7 +682,7 @@ impl DB {
let conn = self.lock_conn("get_oldest");
let mut stmt = conn.prepare(
"SELECT pk, source_url, collection, hits, last_used, metadata_json
"SELECT pk, id, collection, hits, last_used
FROM asset
ORDER BY last_used ASC, hits ASC
LIMIT ?1",

362
pmocache/tests/test_cache.rs Normal file
View File

@@ -0,0 +1,362 @@
use pmocache::{Cache, CacheConfig};
use std::io::Write;
use tempfile::TempDir;
/// Configuration de test simple
struct TestConfig;
impl CacheConfig for TestConfig {
fn file_extension() -> &'static str {
"dat"
}
fn cache_type() -> &'static str {
"test"
}
fn cache_name() -> &'static str {
"testcache"
}
}
type TestCache = Cache<TestConfig>;
fn create_test_cache(limit: usize) -> (TempDir, TestCache) {
let temp_dir = tempfile::tempdir().unwrap();
let cache = TestCache::new(temp_dir.path().to_str().unwrap(), limit).unwrap();
(temp_dir, cache)
}
#[tokio::test]
async fn test_cache_creation() {
let (temp_dir, cache) = create_test_cache(10);
assert_eq!(cache.cache_dir(), temp_dir.path());
}
#[tokio::test]
async fn test_add_from_file() {
let (_temp_dir, cache) = create_test_cache(10);
// Créer un fichier temporaire pour le test
let test_file = tempfile::NamedTempFile::new().unwrap();
let test_data = b"Hello, World! This is test data.";
std::fs::write(test_file.path(), test_data).unwrap();
// Ajouter le fichier au cache
let pk = cache
.add_from_file(test_file.path().to_str().unwrap(), None)
.await
.unwrap();
// Vérifier que le fichier est dans le cache
assert!(!pk.is_empty());
let cached_path = cache.get(&pk).await.unwrap();
assert!(cached_path.exists());
// Vérifier le contenu
let cached_data = std::fs::read(&cached_path).unwrap();
assert_eq!(&cached_data, test_data);
}
#[tokio::test]
async fn test_add_from_reader() {
let (_temp_dir, cache) = create_test_cache(10);
let test_data = b"Test data from reader";
let reader = std::io::Cursor::new(test_data.to_vec());
// Ajouter depuis un reader
let pk = cache
.add_from_reader(None, reader, Some(test_data.len() as u64), None)
.await
.unwrap();
// Attendre que le téléchargement soit terminé
cache.wait_until_finished(&pk).await.unwrap();
// Vérifier que le fichier est dans le cache
let cached_path = cache.get(&pk).await.unwrap();
assert!(cached_path.exists());
// Vérifier le contenu
let cached_data = std::fs::read(&cached_path).unwrap();
assert_eq!(&cached_data, test_data);
}
#[tokio::test]
async fn test_cache_deduplication() {
let (_temp_dir, cache) = create_test_cache(10);
// Créer deux fichiers avec le même contenu
let test_data = b"Same content for both files";
let file1 = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file1.path(), test_data).unwrap();
let file2 = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file2.path(), test_data).unwrap();
// Ajouter les deux fichiers
let pk1 = cache
.add_from_file(file1.path().to_str().unwrap(), None)
.await
.unwrap();
let pk2 = cache
.add_from_file(file2.path().to_str().unwrap(), None)
.await
.unwrap();
// Les deux devraient avoir le même pk (déduplication)
assert_eq!(pk1, pk2);
// Il ne devrait y avoir qu'une seule entrée en DB
assert_eq!(cache.db.count().unwrap(), 1);
}
#[tokio::test]
async fn test_cache_collection() {
let (_temp_dir, cache) = create_test_cache(10);
let collection = "test_album";
// Ajouter plusieurs fichiers à la même collection
let mut pks = Vec::new();
for i in 0..3 {
let data = format!("Track {} data", i);
let file = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file.path(), data.as_bytes()).unwrap();
let pk = cache
.add_from_file(file.path().to_str().unwrap(), Some(collection))
.await
.unwrap();
pks.push(pk);
}
// Récupérer tous les fichiers de la collection
let collection_files = cache.get_collection(collection).await.unwrap();
assert_eq!(collection_files.len(), 3);
}
#[tokio::test]
async fn test_delete_item() {
let (_temp_dir, cache) = create_test_cache(10);
// Ajouter un fichier
let test_data = b"Data to be deleted";
let file = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file.path(), test_data).unwrap();
let pk = cache
.add_from_file(file.path().to_str().unwrap(), None)
.await
.unwrap();
// Vérifier qu'il existe
assert!(cache.get(&pk).await.is_ok());
// Supprimer
cache.delete_item(&pk).await.unwrap();
// Vérifier qu'il n'existe plus
assert!(cache.get(&pk).await.is_err());
}
#[tokio::test]
async fn test_delete_collection() {
let (_temp_dir, cache) = create_test_cache(10);
let collection = "test_collection_delete";
// Ajouter plusieurs fichiers
for i in 0..3 {
let data = format!("Item {}", i);
let file = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file.path(), data.as_bytes()).unwrap();
cache
.add_from_file(file.path().to_str().unwrap(), Some(collection))
.await
.unwrap();
}
// Vérifier que la collection existe
let collection_files = cache.get_collection(collection).await.unwrap();
assert_eq!(collection_files.len(), 3);
// Supprimer la collection
cache.delete_collection(collection).await.unwrap();
// Vérifier que la collection est vide
let collection_files = cache.get_collection(collection).await.unwrap();
assert_eq!(collection_files.len(), 0);
}
#[tokio::test]
async fn test_lru_eviction() {
// Créer un cache avec une limite de 3 éléments
let (_temp_dir, cache) = create_test_cache(3);
let mut pks = Vec::new();
// Ajouter 5 fichiers (devrait déclencher l'éviction)
for i in 0..5 {
let data = format!("File {} data", i);
let file = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file.path(), data.as_bytes()).unwrap();
let pk = cache
.add_from_file(file.path().to_str().unwrap(), None)
.await
.unwrap();
pks.push(pk);
// Petit délai pour s'assurer que les timestamps sont différents
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
}
// Le cache ne devrait contenir que 3 éléments (les plus récents)
let count = cache.db.count().unwrap();
assert_eq!(count, 3);
// Les 2 premiers fichiers devraient avoir été évincés
assert!(cache.get(&pks[0]).await.is_err());
assert!(cache.get(&pks[1]).await.is_err());
// Les 3 derniers devraient être présents
assert!(cache.get(&pks[2]).await.is_ok());
assert!(cache.get(&pks[3]).await.is_ok());
assert!(cache.get(&pks[4]).await.is_ok());
}
#[tokio::test]
async fn test_cache_purge() {
let (_temp_dir, cache) = create_test_cache(10);
// Ajouter plusieurs fichiers
for i in 0..3 {
let data = format!("File {}", i);
let file = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file.path(), data.as_bytes()).unwrap();
cache
.add_from_file(file.path().to_str().unwrap(), None)
.await
.unwrap();
}
assert_eq!(cache.db.count().unwrap(), 3);
// Purger le cache
cache.purge().await.unwrap();
// Le cache devrait être vide
assert_eq!(cache.db.count().unwrap(), 0);
}
#[tokio::test]
async fn test_get_metadata() {
let (_temp_dir, cache) = create_test_cache(10);
let test_data = b"Test data";
let file = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file.path(), test_data).unwrap();
let pk = cache
.add_from_file(file.path().to_str().unwrap(), None)
.await
.unwrap();
// Ajouter des métadonnées
cache
.db
.set_a_metadata(&pk, "test_key", serde_json::json!("test_value"))
.unwrap();
// Récupérer les métadonnées
let value = cache.get_a_metadata(&pk, "test_key").await.unwrap();
assert_eq!(value, Some(serde_json::json!("test_value")));
}
#[tokio::test]
async fn test_touch() {
let (_temp_dir, cache) = create_test_cache(10);
let test_data = b"Test data";
let file = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file.path(), test_data).unwrap();
let pk = cache
.add_from_file(file.path().to_str().unwrap(), None)
.await
.unwrap();
let entry_before = cache.db.get(&pk, false).unwrap();
let hits_before = entry_before.hits;
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
// Touch le fichier
cache.touch(&pk).await.unwrap();
let entry_after = cache.db.get(&pk, false).unwrap();
assert_eq!(entry_after.hits, hits_before + 1);
}
#[tokio::test]
async fn test_consolidate() {
let (temp_dir, cache) = create_test_cache(10);
// Ajouter un fichier
let test_data = b"Test data";
let file = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file.path(), test_data).unwrap();
let pk = cache
.add_from_file(file.path().to_str().unwrap(), None)
.await
.unwrap();
// Supprimer manuellement le fichier (créer un orphelin)
let file_path = cache.get_file_path(&pk);
std::fs::remove_file(&file_path).unwrap();
// Consolider devrait supprimer l'entrée orpheline de la DB
cache.consolidate().await.unwrap();
// L'entrée ne devrait plus exister en DB
assert!(cache.db.get(&pk, false).is_err());
}
#[tokio::test]
async fn test_prebuffer_size() {
let (_temp_dir, mut cache) = create_test_cache(10);
// Configurer la taille de prébuffering
let prebuffer_size = 1024;
cache.set_prebuffer_size(prebuffer_size);
assert_eq!(cache.get_prebuffer_size(), prebuffer_size);
}
#[tokio::test]
async fn test_is_finished() {
let (_temp_dir, cache) = create_test_cache(10);
let test_data = b"Small test data";
let file = tempfile::NamedTempFile::new().unwrap();
std::fs::write(file.path(), test_data).unwrap();
let pk = cache
.add_from_file(file.path().to_str().unwrap(), None)
.await
.unwrap();
// Attendre que le téléchargement soit terminé
cache.wait_until_finished(&pk).await.unwrap();
// Vérifier qu'il est bien terminé
assert!(cache.is_finished(&pk).await);
}

308
pmocache/tests/test_db.rs Normal file
View File

@@ -0,0 +1,308 @@
use pmocache::db::DB;
use serde_json::{json, Value};
use tempfile::TempDir;
/// Crée une DB temporaire pour les tests
fn create_test_db() -> (TempDir, DB) {
let temp_dir = tempfile::tempdir().unwrap();
let db_path = temp_dir.path().join("test.db");
let db = DB::init(&db_path).unwrap();
(temp_dir, db)
}
#[test]
fn test_db_init() {
let temp_dir = tempfile::tempdir().unwrap();
let db_path = temp_dir.path().join("test.db");
let db = DB::init(&db_path);
assert!(db.is_ok());
assert!(db_path.exists());
}
#[test]
fn test_add_and_get() {
let (_temp_dir, db) = create_test_db();
let pk = "test_pk_123";
let id = Some("test_id");
let collection = Some("test_collection");
// Ajouter une entrée
let result = db.add(pk, id, collection);
assert!(result.is_ok());
// Récupérer l'entrée
let entry = db.get(pk, false);
assert!(entry.is_ok());
let entry = entry.unwrap();
assert_eq!(entry.pk, pk);
assert_eq!(entry.id.as_deref(), id);
assert_eq!(entry.collection.as_deref(), collection);
assert_eq!(entry.hits, 0);
}
#[test]
fn test_add_with_metadata() {
let (_temp_dir, db) = create_test_db();
let pk = "test_pk_456";
let metadata = json!({
"title": "Test Track",
"artist": "Test Artist",
"duration": 180,
"bitrate": 320
});
// Ajouter avec métadonnées
let result = db.add_with_metadata(pk, None, None, Some(&metadata));
assert!(result.is_ok());
// Récupérer l'entrée avec métadonnées
let entry = db.get(pk, true).unwrap();
assert_eq!(entry.pk, pk);
assert!(entry.metadata.is_some());
let stored_metadata = entry.metadata.unwrap();
assert_eq!(stored_metadata["title"], "Test Track");
assert_eq!(stored_metadata["artist"], "Test Artist");
assert_eq!(stored_metadata["duration"], 180);
assert_eq!(stored_metadata["bitrate"], 320);
}
#[test]
fn test_update_hit() {
let (_temp_dir, db) = create_test_db();
let pk = "test_pk_789";
db.add(pk, None, None).unwrap();
// Récupérer l'entrée initiale
let entry = db.get(pk, false).unwrap();
let initial_hits = entry.hits;
let initial_last_used = entry.last_used.clone();
// Attendre un peu pour que le timestamp change
std::thread::sleep(std::time::Duration::from_millis(10));
// Mettre à jour le hit
db.update_hit(pk).unwrap();
// Vérifier que hits a augmenté et last_used a changé
let entry = db.get(pk, false).unwrap();
assert_eq!(entry.hits, initial_hits + 1);
assert_ne!(entry.last_used, initial_last_used);
}
#[test]
fn test_delete() {
let (_temp_dir, db) = create_test_db();
let pk = "test_pk_delete";
db.add(pk, None, None).unwrap();
// Vérifier que l'entrée existe
assert!(db.get(pk, false).is_ok());
// Supprimer l'entrée
let result = db.delete(pk);
assert!(result.is_ok());
// Vérifier que l'entrée n'existe plus
assert!(db.get(pk, false).is_err());
}
#[test]
fn test_get_by_collection() {
let (_temp_dir, db) = create_test_db();
let collection = "test_collection";
// Ajouter plusieurs entrées dans la même collection
db.add("pk1", None, Some(collection)).unwrap();
db.add("pk2", None, Some(collection)).unwrap();
db.add("pk3", None, Some("other_collection")).unwrap();
// Récupérer les entrées de la collection
let entries = db.get_by_collection(collection, false).unwrap();
assert_eq!(entries.len(), 2);
assert!(entries.iter().any(|e| e.pk == "pk1"));
assert!(entries.iter().any(|e| e.pk == "pk2"));
assert!(!entries.iter().any(|e| e.pk == "pk3"));
}
#[test]
fn test_delete_collection() {
let (_temp_dir, db) = create_test_db();
let collection = "test_collection_to_delete";
db.add("pk1", None, Some(collection)).unwrap();
db.add("pk2", None, Some(collection)).unwrap();
db.add("pk3", None, Some("other_collection")).unwrap();
// Supprimer la collection
let result = db.delete_collection(collection);
assert!(result.is_ok());
// Vérifier que les entrées de la collection sont supprimées
let entries = db.get_by_collection(collection, false).unwrap();
assert_eq!(entries.len(), 0);
// Vérifier que l'autre collection existe toujours
assert!(db.get("pk3", false).is_ok());
}
#[test]
fn test_get_oldest() {
let (_temp_dir, db) = create_test_db();
// Ajouter plusieurs entrées avec des timestamps différents
db.add("pk1", None, None).unwrap();
std::thread::sleep(std::time::Duration::from_millis(10));
db.add("pk2", None, None).unwrap();
std::thread::sleep(std::time::Duration::from_millis(10));
db.add("pk3", None, None).unwrap();
// Mettre à jour le hit de pk1 pour le rendre plus récent
std::thread::sleep(std::time::Duration::from_millis(10));
db.update_hit("pk1").unwrap();
// Récupérer les 2 plus anciennes entrées
let oldest = db.get_oldest(2).unwrap();
assert_eq!(oldest.len(), 2);
// pk2 et pk3 devraient être les plus anciennes
assert!(oldest.iter().any(|e| e.pk == "pk2"));
assert!(oldest.iter().any(|e| e.pk == "pk3"));
}
#[test]
fn test_count() {
let (_temp_dir, db) = create_test_db();
assert_eq!(db.count().unwrap(), 0);
db.add("pk1", None, None).unwrap();
assert_eq!(db.count().unwrap(), 1);
db.add("pk2", None, None).unwrap();
assert_eq!(db.count().unwrap(), 2);
db.delete("pk1").unwrap();
assert_eq!(db.count().unwrap(), 1);
}
#[test]
fn test_purge() {
let (_temp_dir, db) = create_test_db();
db.add("pk1", None, None).unwrap();
db.add("pk2", None, None).unwrap();
db.add("pk3", None, None).unwrap();
assert_eq!(db.count().unwrap(), 3);
// Purger toutes les entrées
let result = db.purge();
assert!(result.is_ok());
assert_eq!(db.count().unwrap(), 0);
}
#[test]
fn test_origin_url() {
let (_temp_dir, db) = create_test_db();
let pk = "test_pk_url";
let url = "https://example.com/test.flac";
db.add(pk, None, None).unwrap();
db.set_origin_url(pk, url).unwrap();
let retrieved_url = db.get_origin_url(pk).unwrap();
assert_eq!(retrieved_url, Some(url.to_string()));
}
#[test]
fn test_get_from_id() {
let (_temp_dir, db) = create_test_db();
let pk = "test_pk_by_id";
let collection = "my_collection";
let id = "my_unique_id";
db.add(pk, Some(id), Some(collection)).unwrap();
// Récupérer par (collection, id)
let entry = db.get_from_id(collection, id, false).unwrap();
assert_eq!(entry.pk, pk);
assert_eq!(entry.id.as_deref(), Some(id));
assert_eq!(entry.collection.as_deref(), Some(collection));
}
#[test]
fn test_does_collection_contain_id() {
let (_temp_dir, db) = create_test_db();
let collection = "my_collection";
let id = "my_id";
assert!(!db.does_collection_contain_id(collection, id));
db.add("pk", Some(id), Some(collection)).unwrap();
assert!(db.does_collection_contain_id(collection, id));
}
#[test]
fn test_get_pk_from_id() {
let (_temp_dir, db) = create_test_db();
let pk = "test_pk_123";
let collection = "my_collection";
let id = "my_id";
db.add(pk, Some(id), Some(collection)).unwrap();
let retrieved_pk = db.get_pk_from_id(collection, id).unwrap();
assert_eq!(retrieved_pk, pk);
}
#[test]
fn test_set_id() {
let (_temp_dir, db) = create_test_db();
let pk = "test_pk";
db.add(pk, None, None).unwrap();
// Définir l'id
let new_id = "new_id";
db.set_id(pk, new_id).unwrap();
let entry = db.get(pk, false).unwrap();
assert_eq!(entry.id.as_deref(), Some(new_id));
}
#[test]
fn test_metadata_types() {
let (_temp_dir, db) = create_test_db();
let pk = "test_pk_types";
db.add(pk, None, None).unwrap();
// Tester les différents types de métadonnées
db.set_a_metadata(pk, "string_val", Value::String("test".to_string())).unwrap();
db.set_a_metadata(pk, "number_val", json!(42)).unwrap();
db.set_a_metadata(pk, "bool_val", Value::Bool(true)).unwrap();
db.set_a_metadata(pk, "null_val", Value::Null).unwrap();
// Vérifier les valeurs
assert_eq!(db.get_metadata_value(pk, "string_val").unwrap(), Some(Value::String("test".to_string())));
assert_eq!(db.get_metadata_value(pk, "number_val").unwrap(), Some(json!(42)));
assert_eq!(db.get_metadata_value(pk, "bool_val").unwrap(), Some(Value::Bool(true)));
assert_eq!(db.get_metadata_value(pk, "null_val").unwrap(), Some(Value::Null));
}

3
pmocovers/Cargo.toml
View File

@@ -32,6 +32,9 @@ utoipa = { version = "5.3", features = ["axum_extras"], optional = true }
tracing = "0.1.41"
[dev-dependencies]
tempfile = "3"
[features]
default = ["pmoserver"]
pmoconfig = ["dep:pmoconfig", "pmocache/pmoconfig"]

148
pmocovers/tests/test_cache.rs Normal file
View File

@@ -0,0 +1,148 @@
use pmocovers::cache;
use tempfile::TempDir;
use image::{ImageBuffer, Rgba};
fn create_test_cache() -> (TempDir, cache::Cache) {
let temp_dir = tempfile::tempdir().unwrap();
let cache = cache::new_cache(temp_dir.path().to_str().unwrap(), 10).unwrap();
(temp_dir, cache)
}
/// Crée une image de test simple
fn create_test_image(width: u32, height: u32) -> Vec<u8> {
let img: ImageBuffer<Rgba<u8>, Vec<u8>> = ImageBuffer::from_fn(width, height, |x, y| {
if (x + y) % 2 == 0 {
Rgba([255, 0, 0, 255]) // Rouge
} else {
Rgba([0, 0, 255, 255]) // Bleu
}
});
let mut buffer = Vec::new();
img.write_to(&mut std::io::Cursor::new(&mut buffer), image::ImageFormat::Png)
.unwrap();
buffer
}
#[tokio::test]
async fn test_cover_cache_creation() {
let (temp_dir, cache) = create_test_cache();
assert_eq!(cache.cache_dir(), temp_dir.path());
}
#[tokio::test]
async fn test_add_image_from_file() {
let (_temp_dir, cache) = create_test_cache();
// Créer une image de test
let test_image = create_test_image(100, 100);
let test_file = tempfile::NamedTempFile::with_suffix(".png").unwrap();
std::fs::write(test_file.path(), &test_image).unwrap();
// Ajouter au cache
let pk = cache
.add_from_file(test_file.path().to_str().unwrap(), None)
.await
.unwrap();
assert!(!pk.is_empty());
// Attendre la fin de la conversion
cache.wait_until_finished(&pk).await.unwrap();
// Vérifier que le fichier WebP existe
let cached_path = cache.get(&pk).await.unwrap();
assert!(cached_path.exists());
assert!(cached_path.extension().unwrap() == "webp");
}
#[tokio::test]
async fn test_covers_config() {
use pmocache::CacheConfig;
assert_eq!(cache::CoversConfig::file_extension(), "webp");
assert_eq!(cache::CoversConfig::cache_type(), "image");
assert_eq!(cache::CoversConfig::cache_name(), "covers");
}
#[tokio::test]
async fn test_collection_management() {
let (_temp_dir, cache) = create_test_cache();
let collection = "album_covers";
// Ajouter plusieurs images à la même collection
for i in 0..3 {
let img = create_test_image(50 + i * 10, 50 + i * 10);
let file = tempfile::NamedTempFile::with_suffix(".png").unwrap();
std::fs::write(file.path(), &img).unwrap();
cache
.add_from_file(file.path().to_str().unwrap(), Some(collection))
.await
.unwrap();
}
// Récupérer la collection
let collection_files = cache.get_collection(collection).await.unwrap();
assert_eq!(collection_files.len(), 3);
}
#[tokio::test]
#[ignore] // Test d'éviction LRU avec transformer WebP, parfois échoue timing
async fn test_cache_limit() {
let temp_dir = tempfile::tempdir().unwrap();
let cache = cache::new_cache(temp_dir.path().to_str().unwrap(), 2).unwrap();
// Ajouter 3 images (devrait déclencher l'éviction LRU)
for i in 0..3 {
let img = create_test_image(100, 100);
let file = tempfile::NamedTempFile::with_suffix(".png").unwrap();
std::fs::write(file.path(), &img).unwrap();
cache
.add_from_file(file.path().to_str().unwrap(), None)
.await
.unwrap();
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
}
// Attendre l'éviction
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
// Le cache ne devrait contenir que 2 éléments
let count = cache.db.count().unwrap();
assert_eq!(count, 2);
}
#[tokio::test]
async fn test_deduplication() {
let (_temp_dir, cache) = create_test_cache();
// Créer deux fichiers avec la même image
let img = create_test_image(100, 100);
let file1 = tempfile::NamedTempFile::with_suffix(".png").unwrap();
std::fs::write(file1.path(), &img).unwrap();
let file2 = tempfile::NamedTempFile::with_suffix(".png").unwrap();
std::fs::write(file2.path(), &img).unwrap();
// Ajouter les deux images
let pk1 = cache
.add_from_file(file1.path().to_str().unwrap(), None)
.await
.unwrap();
let pk2 = cache
.add_from_file(file2.path().to_str().unwrap(), None)
.await
.unwrap();
// Les deux devraient avoir le même pk (déduplication)
assert_eq!(pk1, pk2);
// Il ne devrait y avoir qu'une seule entrée en DB
assert_eq!(cache.db.count().unwrap(), 1);
}

158
pmocovers/tests/test_webp.rs Normal file
View File

@@ -0,0 +1,158 @@
use image::{DynamicImage, ImageBuffer, Rgba};
use pmocovers::webp::{encode_webp, ensure_square};
/// Crée une image de test simple
fn create_test_image(width: u32, height: u32) -> DynamicImage {
let img: ImageBuffer<Rgba<u8>, Vec<u8>> = ImageBuffer::from_fn(width, height, |x, y| {
if (x + y) % 2 == 0 {
Rgba([255, 0, 0, 255])
} else {
Rgba([0, 0, 255, 255])
}
});
DynamicImage::ImageRgba8(img)
}
#[test]
fn test_encode_webp() {
let img = create_test_image(100, 100);
let webp_data = encode_webp(&img);
assert!(webp_data.is_ok());
let data = webp_data.unwrap();
assert!(!data.is_empty());
// Vérifier la signature WebP (RIFF...WEBP)
assert_eq!(&data[0..4], b"RIFF");
assert_eq!(&data[8..12], b"WEBP");
}
#[test]
fn test_ensure_square_portrait() {
// Image portrait (plus haute que large)
let img = create_test_image(100, 200);
let square = ensure_square(&img, 256);
assert_eq!(square.width(), 256);
assert_eq!(square.height(), 256);
}
#[test]
fn test_ensure_square_landscape() {
// Image landscape (plus large que haute)
let img = create_test_image(200, 100);
let square = ensure_square(&img, 256);
assert_eq!(square.width(), 256);
assert_eq!(square.height(), 256);
}
#[test]
fn test_ensure_square_already_square() {
// Image déjà carrée
let img = create_test_image(150, 150);
let square = ensure_square(&img, 256);
assert_eq!(square.width(), 256);
assert_eq!(square.height(), 256);
}
#[test]
fn test_ensure_square_small_image() {
// Petite image qui doit être agrandie
let img = create_test_image(50, 50);
let square = ensure_square(&img, 256);
assert_eq!(square.width(), 256);
assert_eq!(square.height(), 256);
}
#[test]
fn test_ensure_square_different_sizes() {
let img = create_test_image(100, 100);
// Tester différentes tailles de sortie
for size in [64, 128, 256, 512] {
let square = ensure_square(&img, size);
assert_eq!(square.width(), size);
assert_eq!(square.height(), size);
}
}
#[tokio::test]
async fn test_generate_variant() {
use pmocovers::cache;
use tempfile::TempDir;
let temp_dir = tempfile::tempdir().unwrap();
let cache = cache::new_cache(temp_dir.path().to_str().unwrap(), 10).unwrap();
// Créer et ajouter une image
let img = create_test_image(400, 400);
let mut buffer = Vec::new();
img.write_to(&mut std::io::Cursor::new(&mut buffer), image::ImageFormat::Png)
.unwrap();
let test_file = tempfile::NamedTempFile::with_suffix(".png").unwrap();
std::fs::write(test_file.path(), &buffer).unwrap();
let pk = cache
.add_from_file(test_file.path().to_str().unwrap(), None)
.await
.unwrap();
cache.wait_until_finished(&pk).await.unwrap();
// Générer une variante de taille 128
let variant_data = pmocovers::webp::generate_variant(&cache, &pk, 128)
.await
.unwrap();
assert!(!variant_data.is_empty());
// Vérifier que c'est bien du WebP
assert_eq!(&variant_data[0..4], b"RIFF");
assert_eq!(&variant_data[8..12], b"WEBP");
// Vérifier que le fichier de la variante a été créé
let variant_path = cache.get_file_path_with_qualifier(&pk, "128");
assert!(variant_path.exists());
}
#[tokio::test]
async fn test_generate_variant_caching() {
use pmocovers::cache;
use tempfile::TempDir;
let temp_dir = tempfile::tempdir().unwrap();
let cache = cache::new_cache(temp_dir.path().to_str().unwrap(), 10).unwrap();
// Créer et ajouter une image
let img = create_test_image(400, 400);
let mut buffer = Vec::new();
img.write_to(&mut std::io::Cursor::new(&mut buffer), image::ImageFormat::Png)
.unwrap();
let test_file = tempfile::NamedTempFile::with_suffix(".png").unwrap();
std::fs::write(test_file.path(), &buffer).unwrap();
let pk = cache
.add_from_file(test_file.path().to_str().unwrap(), None)
.await
.unwrap();
cache.wait_until_finished(&pk).await.unwrap();
// Générer la variante une première fois
let variant1 = pmocovers::webp::generate_variant(&cache, &pk, 256)
.await
.unwrap();
// Générer la variante une deuxième fois (devrait lire depuis le cache)
let variant2 = pmocovers::webp::generate_variant(&cache, &pk, 256)
.await
.unwrap();
// Les deux devraient être identiques
assert_eq!(variant1, variant2);
}

6
pmoparadise/examples/play_and_cache.rs
View File

@@ -233,9 +233,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
});
let playback_handle = tokio::spawn(async move {
// Attendre un peu que le premier track soit disponible
tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
tracing::info!("[PLAYBACK] Pipeline starting...");
// Pas de sleep - le cache progressif permet de démarrer immédiatement
// dès que le prebuffer (512 KB) est atteint
tracing::info!("[PLAYBACK] Pipeline starting (will wait for prebuffer)...");
let result = Box::new(playlist_source).run(stop_token_playback).await;
match &result {
Ok(()) => tracing::info!("[PLAYBACK] Pipeline completed successfully"),

62
setup-deps.sh Executable file
View File

@@ -0,0 +1,62 @@
#!/bin/bash
# Script d'installation automatique des dépendances soxr et alsa pour PMOMusic
# Usage: ./setup-deps.sh
set -e
echo "========================================="
echo "Installation des dépendances PMOMusic"
echo "========================================="
echo ""
# Créer le répertoire local
echo "1. Création du répertoire ~/.local"
mkdir -p ~/.local
cd ~/.local
# Télécharger les packages
echo ""
echo "2. Téléchargement des packages libsoxr et libasound2"
apt-get download libsoxr-dev libsoxr0 libasound2-dev libasound2t64
# Extraire les packages
echo ""
echo "3. Extraction des packages"
dpkg -x libsoxr-dev_*.deb .
dpkg -x libsoxr0_*.deb .
dpkg -x libasound2-dev_*.deb .
dpkg -x libasound2t64_*.deb .
# Vérifier l'installation
echo ""
echo "4. Vérification de l'installation"
if [ -f usr/lib/x86_64-linux-gnu/pkgconfig/soxr.pc ]; then
echo " ✓ libsoxr installé"
else
echo " ✗ Erreur: libsoxr non trouvé"
exit 1
fi
if [ -f usr/lib/x86_64-linux-gnu/pkgconfig/alsa.pc ]; then
echo " ✓ libasound2 installé"
else
echo " ✗ Erreur: libasound2 non trouvé"
exit 1
fi
# Retourner au projet
cd - > /dev/null
echo ""
echo "========================================="
echo "Installation terminée avec succès !"
echo "========================================="
echo ""
echo "Pour compiler le projet, exportez les variables d'environnement :"
echo ""
echo " source setup-env.sh"
echo ""
echo "Puis compilez avec :"
echo ""
echo " cargo build"
echo ""