diff --git a/Cargo.lock b/Cargo.lock index b4fc4f12..dba3d97f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/INSTALL_LIBSOXR.md b/INSTALL_LIBSOXR.md index a3c4129f..3632ccac 100644 --- a/INSTALL_LIBSOXR.md +++ b/INSTALL_LIBSOXR.md @@ -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 diff --git a/INSTALL_NOTES.md b/INSTALL_NOTES.md index cf2fad54..77d256e6 100644 --- a/INSTALL_NOTES.md +++ b/INSTALL_NOTES.md @@ -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`. --- diff --git a/OPTIMIZATION_PREBUFFER_TO_PLAYLIST.md b/OPTIMIZATION_PREBUFFER_TO_PLAYLIST.md new file mode 100644 index 00000000..cf13eea7 --- /dev/null +++ b/OPTIMIZATION_PREBUFFER_TO_PLAYLIST.md @@ -0,0 +1,485 @@ +# Optimisation: Réduction du délai prebuffer → playlist (19s → 1s) + +## Contexte + +Le système de progressive caching fonctionne correctement, mais il y a un délai non optimal entre le moment où le prebuffer est atteint et le moment où la track est ajoutée à la playlist. + +### État actuel (branche `claude/fix-play-and-cache-streaming-011CUsMBxH4fsgoadgkiPdoK`) + +**Timing mesuré:** +``` +t=0.6s : Prebuffer complete (512KB téléchargés) ✅ +t=19.2s : tokio::join!() complete (pump_future finit) +t=19.2s : Track added to playlist +t=19.7s : Playback starts +``` + +**Délai total: ~19 secondes** + +### Code actuel problématique + +Location: `pmoaudio-ext/src/sinks/flac_cache_sink.rs:167-178` + +```rust +// Exécuter pump et add_from_reader en parallèle +let pump_future = pump_track_segments( + first_segment, + &mut rx, // ← emprunte muablement rx + pcm_tx, + bits_per_sample, + sample_rate, + &stop_token, +); + +// Attendre les deux tâches en parallèle +let (cache_result, pump_result) = tokio::join!(cache_future, pump_future); + +let pk = cache_result.map_err(|e| { + AudioError::ProcessingError(format!("Failed to add to cache: {}", e)) +})?; + +let (_chunks, _samples, _duration_sec, stop_reason) = pump_result?; +``` + +**Le problème:** `tokio::join!()` attend que **LES DEUX** futures se terminent: +- `cache_future` retourne après prebuffer (~0.6s) ✅ +- `pump_future` lit **toute** la première track du RadioParadiseStreamSource (~19s) ⏱️ + +Donc même si le prebuffer est atteint en 0.6s, on attend 19s avant de push à la playlist! + +## Objectif + +Réduire le délai à **~1 seconde** en pushant à la playlist **immédiatement après le prebuffer**, sans attendre que `pump_future` se termine. + +**Timing visé:** +``` +t=0.6s : Prebuffer complete ✅ +t=0.7s : Track added to playlist ← IMMÉDIAT! +t=1.2s : Playback starts ← ~1 seconde! +t=19.2s : pump_future finit en arrière-plan +``` + +## Contraintes techniques + +### 1. Problème du borrow checker + +`pump_future` emprunte muablement `rx`: +```rust +async fn pump_track_segments( + first_segment: Arc, + rx: &mut mpsc::Receiver>, // ← &mut borrow + // ... +) +``` + +On ne peut pas faire: +```rust +tokio::pin!(cache_future); +tokio::pin!(pump_future); // ← pump_future contient un &mut rx + +let pk = cache_future.await; // cache_future termine + +// ❌ ERREUR: on a toujours un borrow mutable de rx dans pump_future +// On ne peut pas continuer à utiliser rx (ou l'objet qui le contient) +playlist_handle.push(pk.clone()).await; + +let result = pump_future.await; // pump_future continue +``` + +Le borrow checker nous empêche d'attendre `cache_future` seul, puis de faire d'autres opérations, puis d'attendre `pump_future`, car `pump_future` garde un borrow mutable de `rx` pendant toute sa durée de vie. + +### 2. Contraintes de l'API + +- `pump_track_segments()` doit lire `rx` pour recevoir les segments du RadioParadiseStreamSource +- Le FlacCacheSinkLogic doit garder ownership de `rx` pour traiter les tracks suivantes +- `pump_future` ne peut pas être spawné dans un tokio::spawn car il retourne un `StopReason` nécessaire pour la logique métier + +## Solutions possibles + +### Solution A: Refactoriser pump_track_segments pour prendre ownership de rx + +**Approche:** +1. Créer `pump_track_segments_owned` qui prend ownership de `rx` +2. Cette fonction retourne `(result, rx)` - elle rend ownership de `rx` +3. Spawner cette future dans tokio::spawn +4. Attendre cache_future seul, push immédiatement +5. Attendre la task spawnée plus tard + +**Signature:** +```rust +async fn pump_track_segments_owned( + first_segment: Arc, + rx: mpsc::Receiver>, // ownership! + pcm_tx: mpsc::Sender>, + bits_per_sample: u8, + expected_rate: u32, + stop_token: CancellationToken, +) -> Result<(u64, u64, f64, StopReason, mpsc::Receiver>), AudioError> +// ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ rend rx +``` + +**Utilisation:** +```rust +let pump_handle = tokio::spawn(pump_track_segments_owned( + first_segment, + rx, // move ownership + pcm_tx, + bits_per_sample, + sample_rate, + stop_token.clone(), +)); + +// Attendre SEULEMENT le prebuffer +let pk = cache_future.await?; + +// Push IMMÉDIATEMENT à la playlist +#[cfg(feature = "playlist")] +if let Some(ref playlist_handle) = self.playlist_handle { + playlist_handle.push(pk.clone()).await?; +} + +// MAINTENANT attendre que pump finisse +let (result, rx_returned) = pump_handle.await.unwrap()?; +rx = rx_returned; // récupérer rx pour la prochaine track +``` + +**Avantages:** +- ✅ Pas de problème de borrow checker +- ✅ Push immédiat après prebuffer +- ✅ Délai réduit à ~1s + +**Inconvénients:** +- ⚠️ Nécessite de modifier la signature de `pump_track_segments` +- ⚠️ Plus complexe (ownership passé puis rendu) + +### Solution B: Utiliser un channel pour signaler le prebuffer + +**Approche:** +1. Créer un oneshot channel `(prebuffer_tx, prebuffer_rx)` +2. `cache_future` envoie le pk via `prebuffer_tx` dès le prebuffer atteint +3. Le code principal attend `prebuffer_rx`, push immédiatement +4. Puis attend `tokio::join!()` normalement + +**Code:** +```rust +let (prebuffer_tx, prebuffer_rx) = tokio::sync::oneshot::channel(); + +let cache_future = async { + let pk = self.cache.add_from_reader(...).await?; + let _ = prebuffer_tx.send(pk.clone()); // Signal prebuffer! + Ok(pk) +}; + +let pump_future = pump_track_segments(...); + +// Spawner les deux en parallèle +let cache_handle = tokio::spawn(cache_future); +let pump_handle = tokio::spawn(pump_future); + +// Attendre SEULEMENT le signal de prebuffer +let pk = prebuffer_rx.await.unwrap(); + +// Push IMMÉDIATEMENT à la playlist +playlist_handle.push(pk.clone()).await?; + +// Puis attendre que tout finisse +let (cache_result, pump_result) = tokio::join!(cache_handle, pump_handle); +``` + +**Avantages:** +- ✅ Pas besoin de changer les signatures +- ✅ Push immédiat après prebuffer + +**Inconvénients:** +- ⚠️ Nécessite de wrapper cache_future pour envoyer le signal +- ⚠️ Ajoute un oneshot channel + +### Solution C: Modifier l'API du cache pour avoir un callback + +**Approche:** +1. Ajouter un paramètre callback à `add_from_reader()` +2. Le cache appelle ce callback dès le prebuffer atteint +3. Le callback push à la playlist + +**Signature:** +```rust +pub async fn add_from_reader_with_callback( + &self, + source_uri: Option<&str>, + reader: R, + length: Option, + collection: Option<&str>, + on_prebuffer: F, // ← nouveau callback +) -> Result +where + R: AsyncRead + Send + Unpin + 'static, + F: FnOnce(String) + Send + 'static, // F reçoit le pk +``` + +**Avantages:** +- ✅ API propre et réutilisable +- ✅ Pas de problème de borrow checker + +**Inconvénients:** +- ⚠️ Nécessite de modifier l'API du cache (impact sur autres parties du code) +- ⚠️ Ajoute de la complexité à l'API + +## Recommandation + +**Je recommande la Solution A** (refactoriser `pump_track_segments_owned`): +- Plus explicite et claire +- Pas d'impact sur l'API du cache +- Ownership bien défini (passage puis retour de rx) +- Testable indépendamment + +## Plan d'implémentation + +### Étape 1: Créer pump_track_segments_owned + +Location: `pmoaudio-ext/src/sinks/flac_cache_sink.rs` + +```rust +/// Pompe les segments pour une seule track (s'arrête au TrackBoundary). +/// +/// Version qui prend ownership de rx pour permettre un await séparé du cache. +/// Retourne rx à la fin pour permettre le traitement des tracks suivantes. +async fn pump_track_segments_owned( + first_segment: Arc, + mut rx: mpsc::Receiver>, + pcm_tx: mpsc::Sender>, + bits_per_sample: u8, + expected_rate: u32, + stop_token: CancellationToken, +) -> Result<(u64, u64, f64, StopReason, mpsc::Receiver>), AudioError> { + let mut chunks = 0u64; + let mut samples = 0u64; + let mut duration_sec = 0.0f64; + + // Traiter le premier segment + if let Some(chunk) = first_segment.as_chunk() { + let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?; + if !pcm_bytes.is_empty() { + if pcm_tx.send(pcm_bytes).await.is_err() { + drop(pcm_tx); + return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed, rx)); + } + chunks += 1; + samples += chunk.len() as u64; + duration_sec += chunk.len() as f64 / expected_rate as f64; + } + } + + // Loop pour le reste des segments... + loop { + let segment = tokio::select! { + result = rx.recv() => { + match result { + Some(seg) => seg, + None => { + drop(pcm_tx); + return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed, rx)); + } + } + } + _ = stop_token.cancelled() => { + drop(pcm_tx); + return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed, rx)); + } + }; + + match &segment.segment { + _AudioSegment::Chunk(chunk) => { + let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?; + if !pcm_bytes.is_empty() { + if pcm_tx.send(pcm_bytes).await.is_err() { + drop(pcm_tx); + return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed, rx)); + } + chunks += 1; + samples += chunk.len() as u64; + duration_sec += chunk.len() as f64 / expected_rate as f64; + } + } + _AudioSegment::Sync(marker) => match &**marker { + SyncMarker::TrackBoundary { metadata, .. } => { + drop(pcm_tx); + return Ok((chunks, samples, duration_sec, StopReason::TrackBoundary(metadata.clone()), rx)); + } + SyncMarker::EndOfStream => { + drop(pcm_tx); + return Ok((chunks, samples, duration_sec, StopReason::EndOfStream, rx)); + } + _ => continue, + }, + } + } +} +``` + +### Étape 2: Modifier FlacCacheSinkLogic::process + +Location: `pmoaudio-ext/src/sinks/flac_cache_sink.rs:~167` + +```rust +let collection_ref = self.collection.as_deref(); +let cache_future = self.cache.add_from_reader( + None, + flac_stream, + None, + collection_ref, +); + +// Spawner pump_future avec ownership de rx +let pump_handle = tokio::spawn(pump_track_segments_owned( + first_segment, + rx, // move ownership! + pcm_tx, + bits_per_sample, + sample_rate, + stop_token.clone(), +)); + +// Attendre SEULEMENT le prebuffer (cache retourne après 512KB) +tracing::debug!("FlacCacheSink: Waiting for cache prebuffer to complete"); +let pk = cache_future.await.map_err(|e| { + AudioError::ProcessingError(format!("Failed to add to cache: {}", e)) +})?; + +tracing::debug!("FlacCacheSink: Prebuffer complete with pk {}, pushing to playlist NOW", pk); + +// Copier les métadonnées AVANT push +if let Some(src_metadata) = track_metadata.clone() { + let dest_metadata = self.cache.track_metadata(&pk); + pmometadata::copy_metadata_into(&src_metadata, &dest_metadata) + .await + .map_err(|e| { + AudioError::ProcessingError(format!("Failed to copy metadata to cache: {}", e)) + })?; +} + +// Push IMMÉDIATEMENT à la playlist (après prebuffer, avant pump complet!) +#[cfg(feature = "playlist")] +if let Some(ref playlist_handle) = self.playlist_handle { + tracing::debug!("FlacCacheSink: Pushing pk {} to playlist", pk); + playlist_handle.push(pk.clone()).await.map_err(|e| { + AudioError::ProcessingError(format!("Failed to add to playlist: {}", e)) + })?; + tracing::debug!("FlacCacheSink: Successfully pushed to playlist"); +} + +// MAINTENANT attendre que pump finisse (il continue en arrière-plan) +tracing::debug!("FlacCacheSink: Waiting for pump to complete"); +let pump_result = pump_handle.await.map_err(|e| { + AudioError::ProcessingError(format!("Pump task panicked: {}", e)) +})?; + +let (_chunks, _samples, _duration_sec, stop_reason, rx_returned) = pump_result?; +rx = rx_returned; // récupérer rx pour la prochaine track +tracing::debug!("FlacCacheSink: Pump completed"); + +// Continuer avec download des covers en arrière-plan... +``` + +### Étape 3: Tester + +```bash +# Nettoyer et rebuild +rm -rf /tmp/pmomusic_test +source setup-env.sh +cargo build --example play_and_cache --features full + +# Tester avec logs de timing +RUST_LOG=debug target/debug/examples/play_and_cache 0 --null-audio 2>&1 | \ + grep -E "Prebuffer complete|Pushing pk.*to playlist|popped track" | \ + head -20 +``` + +**Résultats attendus:** +``` +[TIME_A] FlacCacheSink: Prebuffer complete with pk XXX, pushing to playlist NOW +[TIME_B] FlacCacheSink: Successfully pushed to playlist +[TIME_C] PlaylistSourceLogic: popped track from playlist + +Délai (TIME_C - TIME_A) devrait être < 1 seconde! +``` + +### Étape 4: Valider le comportement + +Vérifier que: +1. ✅ Le prebuffer est atteint rapidement (~0.6s) +2. ✅ Le push à la playlist est immédiat (~0.1s après prebuffer) +3. ✅ La lecture démarre rapidement (~1s total) +4. ✅ Toutes les tracks se suivent correctement +5. ✅ Les completion markers sont créés +6. ✅ Les tracks suivantes fonctionnent (rx est bien récupéré) +7. ✅ Pas de panic ou deadlock + +## Debugging + +### Si le borrow checker proteste + +Vérifier que: +- `pump_track_segments_owned` prend bien ownership de `rx` (pas `&mut`) +- `rx` est bien retourné dans le tuple de retour +- `rx = rx_returned;` récupère bien ownership après await + +### Si les tracks suivantes ne fonctionnent pas + +Vérifier que: +- `rx` est bien réassigné après le pump: `rx = rx_returned;` +- La loop dans `process()` continue correctement avec le nouveau `rx` + +### Si le timing n'est pas amélioré + +Ajouter des logs avec timestamps: +```rust +let start = std::time::Instant::now(); +let pk = cache_future.await?; +tracing::info!("Prebuffer took {:?}", start.elapsed()); + +let start2 = std::time::Instant::now(); +playlist_handle.push(pk.clone()).await?; +tracing::info!("Playlist push took {:?}", start2.elapsed()); +``` + +## Fichiers à modifier + +1. **pmoaudio-ext/src/sinks/flac_cache_sink.rs** + - Ajouter `pump_track_segments_owned()` (~ligne 432) + - Modifier `FlacCacheSinkLogic::process()` (~ligne 167) + +## Tests de régression + +Après l'implémentation, tester: + +```bash +# Test 1: Premier download (cache vide) +rm -rf /tmp/pmomusic_test +target/debug/examples/play_and_cache 0 --null-audio + +# Test 2: Deuxième download (fichier déjà en cache) +# Ne pas supprimer /tmp/pmomusic_test +target/debug/examples/play_and_cache 0 --null-audio + +# Test 3: Download interrompu (Ctrl+C) +target/debug/examples/play_and_cache 0 --null-audio +# Appuyer Ctrl+C après 2 secondes + +# Test 4: Plusieurs tracks consécutives +# Laisser tourner 1 minute pour voir plusieurs tracks +timeout 60 target/debug/examples/play_and_cache 0 --null-audio +``` + +## Métriques de succès + +- ✅ Délai prebuffer → playlist: **< 1 seconde** (actuellement ~19s) +- ✅ Délai prebuffer → lecture: **< 2 secondes** (actuellement ~19.5s) +- ✅ Pas de régression fonctionnelle +- ✅ Toutes les tracks se suivent correctement +- ✅ Les completion markers sont créés + +## Références + +- Branche actuelle: `claude/fix-play-and-cache-streaming-011CUsMBxH4fsgoadgkiPdoK` +- Code de référence: commit `ed0bbfb` (Add FlacCacheSink debug logs - system now works!) +- Issue originale: "play_and_cache n'a pas le comportement souhaité" diff --git a/Readme.md b/Readme.md index 0dc06507..f2785fe9 100644 --- a/Readme.md +++ b/Readme.md @@ -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 diff --git a/analyze_flac.sh b/analyze_flac.sh new file mode 100755 index 00000000..28ef4187 --- /dev/null +++ b/analyze_flac.sh @@ -0,0 +1,31 @@ +#!/bin/bash +# Affiche toutes les stats d'un fichier FLAC + +if [ $# -eq 0 ]; then + echo "Usage: $0 " + exit 1 +fi + +FILE="$1" + +if [ ! -f "$FILE" ]; then + echo "Error: File not found: $FILE" + exit 1 +fi + +echo "=== Analyzing: $(basename "$FILE") ===" +echo "" +echo "--- SoX Statistics ---" +sox "$FILE" -n stat 2>&1 + +echo "" +echo "--- File Info ---" +file "$FILE" + +echo "" +echo "--- FLAC Metadata ---" +metaflac --list "$FILE" 2>/dev/null || echo "metaflac not installed" + +echo "" +echo "--- Audio Integrity Check ---" +flac -t "$FILE" 2>&1 || echo "flac not installed" diff --git a/check_audio_quality.sh b/check_audio_quality.sh new file mode 100755 index 00000000..b87534f6 --- /dev/null +++ b/check_audio_quality.sh @@ -0,0 +1,90 @@ +#!/bin/bash +# Script pour vérifier la qualité audio (détection de clics via le ratio delta) + +CACHE_DIR="${1:-/tmp/pmomusic_test/audio_cache}" +THRESHOLD=10.0 # Ratio Maximum delta / Mean delta acceptable + +if [ ! -d "$CACHE_DIR" ]; then + echo "Error: Cache directory not found: $CACHE_DIR" + exit 1 +fi + +echo "=== Audio Quality Check ===" +echo "Scanning: $CACHE_DIR" +echo "Threshold: Maximum/Mean delta ratio < $THRESHOLD" +echo "" + +# Vérifier que sox est installé +if ! command -v sox &> /dev/null; then + echo "Error: sox is not installed. Install it with: sudo apt install sox" + exit 1 +fi + +count=0 +suspicious=0 +good=0 + +for file in "$CACHE_DIR"/*.orig.flac; do + if [ ! -f "$file" ]; then + echo "No FLAC files found in $CACHE_DIR" + exit 0 + fi + + filename=$(basename "$file") + + # Obtenir les stats delta + stats=$(sox "$file" -n stat 2>&1) + + max_delta=$(echo "$stats" | grep "Maximum delta" | awk '{print $3}') + mean_delta=$(echo "$stats" | grep "Mean delta" | awk '{print $3}') + + if [ -z "$max_delta" ] || [ -z "$mean_delta" ]; then + echo "❌ $filename - Cannot parse stats" + suspicious=$((suspicious + 1)) + count=$((count + 1)) + continue + fi + + # Éviter division par zéro + if (( $(echo "$mean_delta == 0" | bc -l) )); then + echo "❌ $filename - Invalid mean delta (0)" + suspicious=$((suspicious + 1)) + count=$((count + 1)) + continue + fi + + # Calculer le ratio + ratio=$(echo "scale=2; $max_delta / $mean_delta" | bc -l) + + # Comparer au seuil + is_bad=$(echo "$ratio > $THRESHOLD" | bc -l) + + if [ "$is_bad" = "1" ]; then + echo "⚠️ CLICKS DETECTED: $filename" + echo " Max delta: $max_delta, Mean delta: $mean_delta, Ratio: ${ratio}x (threshold: ${THRESHOLD}x)" + suspicious=$((suspicious + 1)) + else + echo "✓ OK: $filename (ratio: ${ratio}x)" + good=$((good + 1)) + fi + + count=$((count + 1)) +done + +echo "" +echo "=== Summary ===" +echo "Total files scanned: $count" +echo "✓ Good quality: $good" +echo "⚠️ Clicks detected: $suspicious" + +if [ $suspicious -gt 0 ]; then + echo "" + echo "⚠️ Warning: $suspicious file(s) have clicks." + echo "These files were likely encoded with the old buffer size (8)." + echo "Delete the cache and re-download to fix: rm -rf $CACHE_DIR/*.flac" + exit 1 +fi + +echo "" +echo "✓ All files are good quality!" +exit 0 diff --git a/detect_clicks.sh b/detect_clicks.sh new file mode 100755 index 00000000..31999d26 --- /dev/null +++ b/detect_clicks.sh @@ -0,0 +1,76 @@ +#!/bin/bash +# Script pour détecter les clics dans les fichiers FLAC du cache + +CACHE_DIR="${1:-/tmp/pmomusic_test/audio_cache}" + +if [ ! -d "$CACHE_DIR" ]; then + echo "Error: Cache directory not found: $CACHE_DIR" + exit 1 +fi + +echo "=== FLAC Click Detection Tool ===" +echo "Scanning: $CACHE_DIR" +echo "" + +# Vérifier que sox est installé +if ! command -v sox &> /dev/null; then + echo "Error: sox is not installed. Install it with: sudo apt install sox" + exit 1 +fi + +count=0 +suspicious=0 + +for file in "$CACHE_DIR"/*.orig.flac; do + if [ ! -f "$file" ]; then + echo "No FLAC files found in $CACHE_DIR" + exit 0 + fi + + filename=$(basename "$file") + echo "Analyzing: $filename" + + # Obtenir toutes les stats + stats=$(sox "$file" -n stat 2>&1) + + # Extraire les valeurs importantes + pk_lev=$(echo "$stats" | grep "Pk lev dB" | awk '{print $4}') + rms_lev=$(echo "$stats" | grep "RMS lev dB" | awk '{print $4}') + crest=$(echo "$stats" | grep "Crest factor" | awk '{print $3}') + + echo " Peak level: ${pk_lev:-N/A} dB" + echo " RMS level: ${rms_lev:-N/A} dB" + echo " Crest factor: ${crest:-N/A} dB" + + # Analyser la variance d'amplitude (détection de clics) + # On compte le nombre de pics au-dessus d'un seuil + peaks=$(sox "$file" -n stats 2>&1 | grep "Maximum amplitude" | awk '{print $3}') + + if [ -n "$peaks" ]; then + # Si le peak est proche de 1.0 (clipping), c'est suspect + is_clipping=$(echo "$peaks > 0.95" | bc -l 2>/dev/null) + if [ "$is_clipping" = "1" ]; then + echo " ⚠️ WARNING: Possible clipping detected!" + suspicious=$((suspicious + 1)) + else + echo " ✓ OK" + fi + else + echo " ✓ OK" + fi + + echo "" + count=$((count + 1)) +done + +echo "=== Summary ===" +echo "Files scanned: $count" +echo "Suspicious files: $suspicious" + +if [ $suspicious -gt 0 ]; then + echo "" + echo "⚠️ Some files may have issues. Listen to them carefully." + exit 1 +fi + +exit 0 diff --git a/pmoaudio-ext/src/sinks/flac_cache_sink.rs b/pmoaudio-ext/src/sinks/flac_cache_sink.rs index f1924a9c..2e6d377c 100755 --- a/pmoaudio-ext/src/sinks/flac_cache_sink.rs +++ b/pmoaudio-ext/src/sinks/flac_cache_sink.rs @@ -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}, @@ -87,16 +86,22 @@ impl NodeLogic for FlacCacheSinkLogic { _output: Vec>>, stop_token: CancellationToken, ) -> Result<(), AudioError> { + tracing::debug!("FlacCacheSink::process() started"); let mut rx = input.expect("FlacCacheSink must have input"); let mut track_number = 0; loop { // Attendre le premier chunk audio pour cette track + tracing::debug!("FlacCacheSink: Waiting for first audio chunk (track_number={})", track_number); let (first_segment, track_metadata) = match wait_for_first_audio_chunk_with_metadata(&mut rx, &stop_token).await { - Ok(result) => result, - Err(_) => { + Ok(result) => { + tracing::debug!("FlacCacheSink: Got first audio chunk"); + result + } + Err(e) => { // Plus d'audio disponible + tracing::debug!("FlacCacheSink: No more audio available: {}", e); return Ok(()); } }; @@ -126,60 +131,120 @@ impl NodeLogic for FlacCacheSinkLogic { options_with_metadata.metadata = track_metadata.clone(); // Créer l'encoder + tracing::debug!("FlacCacheSink: Creating FLAC 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)) })?; + tracing::debug!("FlacCacheSink: FLAC encoder created"); - // 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(); + tracing::debug!("FlacCacheSink: Starting cache ingestion and pump in parallel"); + 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 - let pump_future = pump_track_segments( + // Créer un channel dédié pour dispatcher les chunks vers ce pump + let (track_tx, track_rx) = mpsc::channel::>(16); + + // Lancer le pump en arrière-plan avec son channel dédié + // Cela permet à plusieurs pumps de tourner simultanément (écriture parallèle) + let pump_handle = tokio::spawn(pump_track_segments_from_channel( first_segment, - &mut rx, + track_rx, pcm_tx, bits_per_sample, 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>(()) + )); + + // Dispatcher les segments vers track_tx en parallèle de l'attente du prebuffer + // Utiliser tokio::select! pour éviter le deadlock + let start = std::time::Instant::now(); + tracing::debug!("FlacCacheSink: Starting dispatcher loop with prebuffer wait"); + + // Pin la future pour pouvoir l'utiliser dans select! + tokio::pin!(cache_future); + + // Phase 1: Dispatcher jusqu'à ce que le prebuffer soit terminé + let pk = loop { + tokio::select! { + // Attendre le prebuffer + result = &mut cache_future => { + match result { + Ok(pk) => { + let prebuffer_time = start.elapsed(); + tracing::info!("FlacCacheSink: Prebuffer complete with pk {} in {:?}", pk, prebuffer_time); + break pk; // Sort de la loop pour faire les métadonnées et le push + } + Err(e) => { + return Err(AudioError::ProcessingError(format!("Failed to add to cache: {}", e))); + } + } + } + + // Dispatcher les segments depuis rx vers track_tx + result = rx.recv() => { + match result { + Some(segment) => { + match &segment.segment { + _AudioSegment::Chunk(_) => { + // Dispatcher vers le pump + if track_tx.send(segment).await.is_err() { + // Le pump est mort - erreur fatale + tracing::error!("FlacCacheSink: pump died unexpectedly during prebuffer phase"); + return Err(AudioError::ProcessingError("Pump task died".to_string())); + } + } + _AudioSegment::Sync(marker) => match &**marker { + SyncMarker::TrackBoundary { .. } => { + // TrackBoundary avant fin du prebuffer - track trop courte + tracing::error!("FlacCacheSink: TrackBoundary received before prebuffer complete - track too short"); + return Err(AudioError::ProcessingError("Track too short for prebuffer".to_string())); + } + SyncMarker::EndOfStream => { + tracing::debug!("FlacCacheSink: EndOfStream during prebuffer"); + drop(track_tx); + drop(pump_handle); + return Ok(()); + } + _ => { + // Transmettre les autres syncmarkers au pump + let _ = track_tx.send(segment).await; + } + }, + } + } + None => { + // EOF sur rx + drop(track_tx); + drop(pump_handle); + return Ok(()); + } + } + } + + _ = stop_token.cancelled() => { + drop(track_tx); + drop(pump_handle); + return Ok(()); + } + } }; - // 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?; - - // 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| { - AudioError::ProcessingError(format!("Failed to add to cache: {}", e)) - })?; - + // Phase 2: Prebuffer terminé! Copier les métadonnées et pusher à la playlist // Copier les métadonnées du TrackBoundary dans le cache - if let Some(src_metadata) = track_metadata { + // IMPORTANT: Faire ceci AVANT d'ajouter à la playlist pour que les métadonnées soient disponibles + if let Some(src_metadata) = track_metadata.clone() { let dest_metadata = self.cache.track_metadata(&pk); // Utiliser copy_metadata_into pour copier toutes les métadonnées @@ -221,24 +286,102 @@ impl NodeLogic for FlacCacheSinkLogic { } } - // Ajouter à la playlist si enregistrée + // Push IMMÉDIATEMENT à la playlist (après prebuffer, avant pump complet!) #[cfg(feature = "playlist")] if let Some(ref playlist_handle) = self.playlist_handle { + let push_start = std::time::Instant::now(); + tracing::debug!("FlacCacheSink: Pushing pk {} to playlist", pk); playlist_handle.push(pk.clone()).await.map_err(|e| { AudioError::ProcessingError(format!("Failed to add to playlist: {}", e)) })?; + tracing::info!("FlacCacheSink: Successfully pushed to playlist in {:?}", push_start.elapsed()); } - // Vérifier le stop_reason pour savoir si on continue - match stop_reason { - StopReason::TrackBoundary(_metadata) => { - // Continuer avec la prochaine track - track_number += 1; - continue; - } - StopReason::EndOfStream | StopReason::ChannelClosed => { - // Fin de l'encodage - return Ok(()); + // Phase 3: Continuer à dispatcher jusqu'au TrackBoundary + tracing::debug!("FlacCacheSink: Continuing dispatch until TrackBoundary (pump runs in background)"); + let mut track_tx = Some(track_tx); + let mut pump_handle = Some(pump_handle); + let mut pump_closed = false; + loop { + let segment = tokio::select! { + result = rx.recv() => { + match result { + Some(seg) => seg, + None => { + // EOF sur rx + drop(track_tx); + drop(pump_handle); + return Ok(()); + } + } + } + _ = stop_token.cancelled() => { + drop(track_tx); + drop(pump_handle); + return Ok(()); + } + }; + + match &segment.segment { + _AudioSegment::Chunk(_) => { + // Continuer à dispatcher vers le pump (sauf si déjà fermé) + if !pump_closed { + if let Some(ref tx) = track_tx { + if tx.send(segment).await.is_err() { + // Le pump a fermé son channel - cela peut arriver si le fichier + // était déjà en cache (add_from_reader retourne immédiatement) + tracing::debug!("FlacCacheSink: pump closed track_tx, checking pump status"); + drop(track_tx.take()); + + // Attendre que le pump se termine et vérifier le résultat + if let Some(handle) = pump_handle.take() { + match handle.await { + Ok(Ok(_)) => { + // Le pump s'est terminé proprement (fichier était en cache) + tracing::debug!("FlacCacheSink: pump completed successfully, ignoring remaining chunks until TrackBoundary"); + pump_closed = true; + } + Ok(Err(e)) => { + // Le pump a rencontré une erreur + tracing::error!("FlacCacheSink: pump died with error: {}", e); + return Err(e); + } + Err(e) => { + // Le pump task a paniqué + tracing::error!("FlacCacheSink: pump task panicked: {}", e); + return Err(AudioError::ProcessingError("Pump task panicked".to_string())); + } + } + } + } + } + } + // Si pump_closed, ignorer silencieusement le chunk + } + _AudioSegment::Sync(marker) => match &**marker { + SyncMarker::TrackBoundary { .. } => { + // Nouveau morceau - fermer le pump si pas déjà fermé + tracing::debug!("FlacCacheSink: TrackBoundary received, closing pump"); + drop(track_tx.take()); + drop(pump_handle.take()); + track_number += 1; + break; // Sort de la Phase 3, retour à la loop externe pour next track + } + SyncMarker::EndOfStream => { + tracing::debug!("FlacCacheSink: EndOfStream received"); + drop(track_tx.take()); + drop(pump_handle.take()); + return Ok(()); + } + _ => { + // Transmettre les autres syncmarkers au pump (sauf si fermé) + if !pump_closed { + if let Some(ref tx) = track_tx { + let _ = tx.send(segment).await; + } + } + } + }, } } } @@ -292,7 +435,7 @@ impl FlacCacheSink { encoder_options: EncoderOptions, collection: Option, ) -> Self { - let logic = FlacCacheSinkLogic::new(cache, covers, collection, encoder_options, 8); + let logic = FlacCacheSinkLogic::new(cache, covers, collection, encoder_options, 256); Self { inner: Node::new_with_input(logic, channel_size), } @@ -360,6 +503,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>, + stop_token: &CancellationToken, +) -> Result { + 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, @@ -377,10 +564,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 +610,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; @@ -450,6 +641,82 @@ async fn pump_track_segments( } } +/// Pompe les segments pour une seule track depuis un channel dédié. +/// +/// Cette version permet d'avoir plusieurs pumps en parallèle (pour cache progressif), +/// car chaque pump a son propre channel et ne bloque pas le traitement des tracks suivantes. +async fn pump_track_segments_from_channel( + first_segment: Arc, + mut track_rx: mpsc::Receiver>, + pcm_tx: mpsc::Sender>, + bits_per_sample: u8, + expected_rate: u32, +) -> Result<(u64, u64, f64), AudioError> { + let mut chunks = 0u64; + let mut samples = 0u64; + let mut duration_sec = 0.0f64; + + // Traiter le premier segment + if let Some(chunk) = first_segment.as_chunk() { + let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?; + if !pcm_bytes.is_empty() { + if pcm_tx.send(pcm_bytes).await.is_err() { + drop(pcm_tx); + tracing::debug!("pump_track_segments_from_channel: pcm_tx closed on first segment"); + return Ok((chunks, samples, duration_sec)); + } + chunks += 1; + samples += chunk.len() as u64; + duration_sec += chunk.len() as f64 / expected_rate as f64; + } + } + + // Boucle sur les segments depuis le channel dédié + loop { + let segment = match track_rx.recv().await { + Some(seg) => seg, + None => { + // Channel fermé - la track est terminée (TrackBoundary a été reçu en amont) + drop(pcm_tx); + tracing::debug!("pump_track_segments_from_channel: channel closed, track finished"); + return Ok((chunks, samples, duration_sec)); + } + }; + + match &segment.segment { + _AudioSegment::Chunk(chunk) => { + if chunk.sample_rate() != expected_rate { + return Err(AudioError::ProcessingError(format!( + "FlacCacheSink: inconsistent sample rate ({} vs {})", + chunk.sample_rate(), + expected_rate + ))); + } + + let pcm_bytes = chunk_to_pcm_bytes(&chunk, bits_per_sample)?; + if pcm_bytes.is_empty() { + continue; + } + + if pcm_tx.send(pcm_bytes).await.is_err() { + // Le cache a fermé le channel (erreur ou déjà en cache) + drop(pcm_tx); + tracing::debug!("pump_track_segments_from_channel: pcm_tx closed"); + return Ok((chunks, samples, duration_sec)); + } + + chunks += 1; + samples += chunk.len() as u64; + duration_sec += chunk.len() as f64 / expected_rate as f64; + } + _AudioSegment::Sync(_marker) => { + // Ignorer les syncmarkers - le TrackBoundary est géré en amont + // Le channel sera fermé quand le TrackBoundary est détecté + } + } + } +} + /// Détermine la profondeur de bit d'un chunk audio fn get_chunk_bit_depth(chunk: &AudioChunk) -> u8 { match chunk { diff --git a/pmoaudio-ext/src/sources/playlist_source.rs b/pmoaudio-ext/src/sources/playlist_source.rs index bdb39fd6..5bca30a9 100644 --- a/pmoaudio-ext/src/sources/playlist_source.rs +++ b/pmoaudio-ext/src/sources/playlist_source.rs @@ -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,12 +268,39 @@ 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>], stop_token: &CancellationToken, + cache: &Arc, + cache_pk: &str, ) -> Result<(), AudioError> { + // Attendre que le fichier soit suffisamment gros pour le sniffing + // Le cache progressif permet de commencer la lecture après le prebuffer (512 KB) + loop { + let metadata = tokio::fs::metadata(path) + .await + .map_err(|e| AudioError::IoError(format!("Failed to stat {:?}: {}", path, e)))?; + + let file_size = metadata.len(); + const MIN_FILE_SIZE: u64 = 512 * 1024; // 512 KB (prebuffer size) + + if file_size >= MIN_FILE_SIZE || cache.is_download_complete(cache_pk) { + tracing::trace!("decode_and_emit_track: file ready ({} bytes), starting decode", file_size); + break; + } + + tracing::trace!( + "decode_and_emit_track: file too small ({} bytes), waiting 50ms...", + file_size + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + // Ouvrir et décoder let file = File::open(path) .await @@ -320,9 +351,25 @@ 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() { - break; + + // Si EOF atteint (read == 0) + if read == 0 { + // Vérifier si le fichier est complètement écrit (completion marker existe) + if !cache.is_download_complete(cache_pk) { + // Fichier encore en cours d'écriture - attendre et réessayer + // Retry plus longtemps pour le cache progressif + tracing::trace!("decode_and_emit_track: EOF but file incomplete, waiting 200ms..."); + tokio::time::sleep(Duration::from_millis(200)).await; + continue; // Retry + } + + // Completion marker existe - vraie fin du fichier + tracing::trace!("decode_and_emit_track: EOF and file complete"); + if pending.is_empty() { + break; + } } + if read > 0 { pending.extend_from_slice(&read_buf[..read]); } diff --git a/pmoaudio/src/lib.rs b/pmoaudio/src/lib.rs index 6c0e3edd..dd3c9a6c 100755 --- a/pmoaudio/src/lib.rs +++ b/pmoaudio/src/lib.rs @@ -124,6 +124,7 @@ pub use nodes::{ flac_file_sink::{FlacFileSink, FlacFileSinkStats}, http_source::HttpSource, resampling_node::ResamplingNode, + timer_node::TimerNode, AudioError, AudioNode, TypedAudioNode, }; diff --git a/pmoaudio/src/nodes/audio_sink.rs b/pmoaudio/src/nodes/audio_sink.rs index 37126896..e71798d1 100644 --- a/pmoaudio/src/nodes/audio_sink.rs +++ b/pmoaudio/src/nodes/audio_sink.rs @@ -172,11 +172,70 @@ fn chunk_to_f32_interleaved(chunk: &AudioChunk) -> Vec { // ═══════════════════════════════════════════════════════════════════════════ /// Logique pure de lecture audio via cpal -pub struct AudioSinkLogic {} +pub struct AudioSinkLogic { + use_null_output: bool, +} impl AudioSinkLogic { pub fn new() -> Self { - Self {} + Self { + use_null_output: false, + } + } + + pub fn with_null_output() -> Self { + Self { + use_null_output: true, + } + } + + /// Version null output - consomme les segments sans les jouer + async fn process_null_output( + mut rx: mpsc::Receiver>, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + loop { + let segment = tokio::select! { + result = rx.recv() => { + match result { + Some(seg) => seg, + None => { + tracing::debug!("AudioSinkLogic (null): input channel closed"); + return Ok(()); + } + } + } + _ = stop_token.cancelled() => { + tracing::debug!("AudioSinkLogic (null): cancelled"); + return Ok(()); + } + }; + + // Juste logger les segments sans les jouer + match &segment.segment { + crate::_AudioSegment::Chunk(chunk) => { + tracing::trace!( + "AudioSink (null): consumed chunk with {} frames at {}Hz", + chunk.len(), + chunk.sample_rate() + ); + } + crate::_AudioSegment::Sync(marker) => { + match **marker { + SyncMarker::TrackBoundary { .. } => { + tracing::debug!("AudioSink (null): TrackBoundary received"); + } + SyncMarker::EndOfStream => { + tracing::debug!("AudioSink (null): EndOfStream received"); + return Ok(()); + } + _ => { + tracing::trace!("AudioSink (null): sync marker"); + } + } + } + } + } } } @@ -198,6 +257,12 @@ impl NodeLogic for AudioSinkLogic { tracing::debug!("AudioSinkLogic::process started"); + // Si null output, juste consommer les segments sans jouer + if self.use_null_output { + tracing::debug!("Using null audio output (no playback)"); + return Self::process_null_output(rx, stop_token).await; + } + // Créer le buffer partagé let buffer = Arc::new(Mutex::new(SharedBuffer::new())); let buffer_clone = buffer.clone(); @@ -485,6 +550,14 @@ impl AudioSink { inner: Node::new_with_input(AudioSinkLogic::new(), channel_size), } } + + /// Crée un AudioSink avec null output (pour tests sans carte audio) + /// Consomme les segments audio sans les jouer + pub fn with_null_output() -> Self { + Self { + inner: Node::new_with_input(AudioSinkLogic::with_null_output(), DEFAULT_CHANNEL_SIZE), + } + } } impl Default for AudioSink { diff --git a/pmoaudio/src/nodes/flac_file_sink.rs b/pmoaudio/src/nodes/flac_file_sink.rs index b8f7a27c..0d7783ec 100755 --- a/pmoaudio/src/nodes/flac_file_sink.rs +++ b/pmoaudio/src/nodes/flac_file_sink.rs @@ -133,9 +133,24 @@ impl NodeLogic for FlacFileSinkLogic { AudioError::ProcessingError(format!("Failed to create {:?}: {}", track_path, e)) })?; - // Exécuter pump et copy en parallèle avec tokio::select! en boucle - let pump_future = - pump_track_segments(first_segment, &mut rx, pcm_tx, bits_per_sample, sample_rate, &stop_token); + // Créer un channel dédié pour dispatcher les chunks vers ce pump + let (track_tx, track_rx) = mpsc::channel::>(16); + + // Lancer le pump en arrière-plan avec son channel dédié + // Cela permet à plusieurs pumps de tourner simultanément (cache progressive compliant) + let pump_handle = tokio::spawn(pump_track_segments_from_channel( + first_segment, + track_rx, + pcm_tx, + bits_per_sample, + sample_rate, + )); + + // Dispatcher les segments vers track_tx en parallèle de l'écriture du fichier + // Utiliser tokio::select! pour éviter le deadlock et permettre cache progressif + tracing::debug!("FlacFileSink: Starting dispatcher loop with file write"); + + // Pin la future pour pouvoir l'utiliser dans select! let copy_future = async { let copy_result = tokio::io::copy(&mut flac_stream, &mut output).await; let flush_result = output.flush().await; @@ -150,22 +165,118 @@ impl NodeLogic for FlacFileSinkLogic { .map_err(|e| AudioError::ProcessingError(format!("Encoder failed: {}", e)))?; Ok::<_, AudioError>(()) }; + tokio::pin!(copy_future); - // Attendre les deux tâches en parallèle - let (copy_result, pump_result) = tokio::join!(copy_future, pump_future); - copy_result?; - let stop_reason = pump_result?; + // Phase 1: Dispatcher jusqu'à ce que le fichier soit complètement écrit + let mut copy_done = false; + loop { + tokio::select! { + // Attendre l'écriture du fichier + result = &mut copy_future, if !copy_done => { + result?; + tracing::info!("FlacFileSink: File write complete for track {}", track_number); + copy_done = true; + // Continue dispatching jusqu'au TrackBoundary + } - // Vérifier le stop_reason pour savoir si on continue - match stop_reason { - StopReason::TrackBoundary(_metadata) => { - // Continuer avec la prochaine track - track_number += 1; - continue; - } - StopReason::EndOfStream | StopReason::ChannelClosed | StopReason::Cancelled => { - // Fin de l'encodage - return Ok(()); + // Dispatcher les segments depuis rx vers track_tx + result = rx.recv() => { + match result { + Some(segment) => { + match &segment.segment { + crate::_AudioSegment::Chunk(_) => { + // Dispatcher vers le pump + if track_tx.send(segment).await.is_err() { + // Le pump est mort - erreur fatale + tracing::error!("FlacFileSink: pump died unexpectedly"); + return Err(AudioError::ProcessingError("Pump task died".to_string())); + } + } + crate::_AudioSegment::Sync(marker) => match &**marker { + SyncMarker::TrackBoundary { .. } => { + // Nouveau morceau - fermer le pump et passer au suivant + tracing::debug!("FlacFileSink: TrackBoundary received"); + + // Vérifier que copy est terminé avant de continuer + if !copy_done { + copy_future.await?; + tracing::info!("FlacFileSink: File write complete for track {}", track_number); + } + + drop(track_tx); // Ferme le channel, le pump se termine proprement + drop(pump_handle); + + // Créer le marqueur de complétude + let completion_marker = track_path.with_extension("flac.complete"); + if let Err(e) = tokio::fs::File::create(&completion_marker).await { + tracing::warn!("FlacFileSink: Failed to create completion marker {:?}: {}", completion_marker, e); + } else { + tracing::debug!("FlacFileSink: Created completion marker {:?}", completion_marker); + } + + track_number += 1; + break; // Sort de la Phase 1, retour à la loop externe pour next track + } + SyncMarker::EndOfStream => { + tracing::debug!("FlacFileSink: EndOfStream received"); + + // Vérifier que copy est terminé + if !copy_done { + copy_future.await?; + tracing::info!("FlacFileSink: File write complete for track {}", track_number); + } + + drop(track_tx); + drop(pump_handle); + + // Créer le marqueur de complétude + let completion_marker = track_path.with_extension("flac.complete"); + if let Err(e) = tokio::fs::File::create(&completion_marker).await { + tracing::warn!("FlacFileSink: Failed to create completion marker {:?}: {}", completion_marker, e); + } else { + tracing::debug!("FlacFileSink: Created completion marker {:?}", completion_marker); + } + + return Ok(()); + } + _ => { + // Transmettre les autres syncmarkers au pump + let _ = track_tx.send(segment).await; + } + }, + } + } + None => { + // EOF sur rx + tracing::debug!("FlacFileSink: EOF on rx"); + + // Vérifier que copy est terminé + if !copy_done { + copy_future.await?; + tracing::info!("FlacFileSink: File write complete for track {}", track_number); + } + + drop(track_tx); + drop(pump_handle); + + // Créer le marqueur de complétude + let completion_marker = track_path.with_extension("flac.complete"); + if let Err(e) = tokio::fs::File::create(&completion_marker).await { + tracing::warn!("FlacFileSink: Failed to create completion marker {:?}: {}", completion_marker, e); + } else { + tracing::debug!("FlacFileSink: Created completion marker {:?}", completion_marker); + } + + return Ok(()); + } + } + } + + _ = stop_token.cancelled() => { + drop(track_tx); + drop(pump_handle); + return Ok(()); + } } } } @@ -367,6 +478,71 @@ async fn pump_track_segments( } } +/// Pompe les segments pour une seule track depuis un channel dédié. +/// +/// Cette version permet d'avoir plusieurs pumps en parallèle (cache progressive compliant), +/// car chaque pump a son propre channel et ne bloque pas le traitement des tracks suivantes. +async fn pump_track_segments_from_channel( + first_segment: Arc, + mut track_rx: mpsc::Receiver>, + pcm_tx: mpsc::Sender>, + bits_per_sample: u8, + expected_rate: u32, +) -> Result<(), AudioError> { + // Traiter le premier segment + if let Some(chunk) = first_segment.as_chunk() { + let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?; + if !pcm_bytes.is_empty() { + if pcm_tx.send(pcm_bytes).await.is_err() { + drop(pcm_tx); + tracing::debug!("pump_track_segments_from_channel: pcm_tx closed on first segment"); + return Ok(()); + } + } + } + + // Boucle sur les segments depuis le channel dédié + loop { + let segment = match track_rx.recv().await { + Some(seg) => seg, + None => { + // Channel fermé - la track est terminée (TrackBoundary a été reçu en amont) + drop(pcm_tx); + tracing::debug!("pump_track_segments_from_channel: channel closed, track finished"); + return Ok(()); + } + }; + + match &segment.segment { + crate::_AudioSegment::Chunk(chunk) => { + if chunk.sample_rate() != expected_rate { + return Err(AudioError::ProcessingError(format!( + "FlacFileSink: inconsistent sample rate ({} vs {})", + chunk.sample_rate(), + expected_rate + ))); + } + + let pcm_bytes = chunk_to_pcm_bytes(&chunk, bits_per_sample)?; + if pcm_bytes.is_empty() { + continue; + } + + if pcm_tx.send(pcm_bytes).await.is_err() { + // Le fichier a fermé le channel (erreur) + drop(pcm_tx); + tracing::debug!("pump_track_segments_from_channel: pcm_tx closed"); + return Ok(()); + } + } + crate::_AudioSegment::Sync(_marker) => { + // Ignorer les syncmarkers - le TrackBoundary est géré en amont + // Le channel sera fermé quand le TrackBoundary est détecté + } + } + } +} + /// Détermine la profondeur de bit d'un chunk audio fn get_chunk_bit_depth(chunk: &AudioChunk) -> u8 { match chunk { diff --git a/pmoaudio/src/nodes/mod.rs b/pmoaudio/src/nodes/mod.rs index 9f51198a..877073bf 100755 --- a/pmoaudio/src/nodes/mod.rs +++ b/pmoaudio/src/nodes/mod.rs @@ -25,6 +25,7 @@ pub mod file_source; pub mod flac_file_sink; pub mod http_source; pub mod resampling_node; +pub mod timer_node; // Modules temporairement désactivés /* @@ -36,7 +37,6 @@ pub mod dsp_node; pub mod mpd_sink; pub mod sink_node; pub mod source_node; -pub mod timer_node; pub mod volume_node; */ diff --git a/pmoaudio/src/nodes/timer_node.rs b/pmoaudio/src/nodes/timer_node.rs new file mode 100644 index 00000000..794e3b59 --- /dev/null +++ b/pmoaudio/src/nodes/timer_node.rs @@ -0,0 +1,263 @@ +//! TimerNode - Régule le débit des chunks audio en fonction de leurs timestamps +//! +//! Ce node implémente un pacing temporel pour éviter que les sources rapides +//! saturent les sinks lents. Il tolère une avance configurable (buffer) et +//! attend activement pour maintenir la synchronisation temps réel. +//! +//! # Use Cases +//! +//! - **Progressive caching**: Empêche PlaylistSource de lire plus vite que FlacCacheSink n'écrit +//! - **Rate limiting**: Contrôle le débit de n'importe quel pipeline audio +//! - **Streaming**: Synchronise la production avec la consommation temps réel +//! +//! # Exemple +//! +//! ```no_run +//! use pmoaudio::{PlaylistSource, TimerNode, FlacCacheSink}; +//! +//! let mut source = PlaylistSource::new(reader, cache); +//! let mut timer = TimerNode::new(3.0); // 3s d'avance max +//! let mut sink = FlacCacheSink::new(cache, covers); +//! +//! source.register(Box::new(timer)); +//! timer.register(Box::new(sink)); +//! ``` +//! +//! # Architecture +//! +//! ```text +//! PlaylistSource → TimerNode → FlacCacheSink +//! ↓ ↓ ↓ +//! Lit à fond Régule en Écrit au +//! temps réel bon rythme +//! ``` +//! +//! Le TimerNode: +//! 1. Reçoit des chunks avec timestamps +//! 2. Compare `chunk.timestamp_sec` avec le temps écoulé depuis `TopZeroSync` +//! 3. Si l'avance > `max_lead_time_sec`, attend: `sleep(avance - max_lead_time)` +//! 4. Transmet le chunk aux enfants +//! +//! # Markers Supportés +//! +//! - **TopZeroSync**: Reset le timer de référence (instant zero) +//! - **TrackBoundary**: Passthrough transparent +//! - **Heartbeat**: Passthrough transparent +//! - **EndOfStream**: Passthrough transparent +//! +//! # Performance +//! +//! - **CPU**: Quasi-nul (tokio::time::sleep efficace) +//! - **Latency**: Ajoute `max_lead_time_sec` de buffering +//! - **Memory**: Minimal (pas de buffer de chunks) + +use crate::{ + nodes::{AudioError, TypedAudioNode, DEFAULT_CHANNEL_SIZE}, + pipeline::{AudioPipelineNode, Node, NodeLogic}, + type_constraints::TypeRequirement, + AudioSegment, SyncMarker, _AudioSegment, +}; +use std::sync::Arc; +use tokio::sync::mpsc; +use tokio::time::{Duration, Instant}; +use tokio_util::sync::CancellationToken; + +// ═══════════════════════════════════════════════════════════════════════════ +// TimerNodeLogic - Logique pure de pacing temporel +// ═══════════════════════════════════════════════════════════════════════════ + +/// Logique pure de régulation temporelle +/// +/// Contrôle le débit des chunks audio pour éviter qu'une source rapide +/// sature un sink lent (ex: progressive caching). +pub struct TimerNodeLogic { + /// Avance maximale tolérée en secondes (buffer) + max_lead_time_sec: f64, + /// Instant de référence (reset au TopZeroSync) + start_time: Option, +} + +impl TimerNodeLogic { + pub fn new(max_lead_time_sec: f64) -> Self { + Self { + max_lead_time_sec: max_lead_time_sec.max(0.0), + start_time: None, + } + } +} + +#[async_trait::async_trait] +impl NodeLogic for TimerNodeLogic { + async fn process( + &mut self, + input: Option>>, + output: Vec>>, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + let mut rx = input.expect("TimerNode must have input"); + tracing::debug!( + "TimerNodeLogic::process started (max_lead_time={:.1}s), {} children", + self.max_lead_time_sec, + output.len() + ); + + // Macro helper pour envoyer à tous les enfants + macro_rules! send_to_children { + ($segment:expr) => { + for tx in &output { + tx.send($segment.clone()) + .await + .map_err(|_| AudioError::ChildDied)?; + } + }; + } + + loop { + let segment = tokio::select! { + _ = stop_token.cancelled() => { + tracing::debug!("TimerNodeLogic cancelled"); + break; + } + + result = rx.recv() => { + match result { + Some(seg) => seg, + None => { + tracing::debug!("TimerNodeLogic received EOF"); + break; + } + } + } + }; + + // Traitement selon le type de segment + match &segment.segment { + _AudioSegment::Sync(marker) => { + match &**marker { + SyncMarker::TopZeroSync => { + // Reset le timer de référence + self.start_time = Some(Instant::now()); + tracing::debug!("TimerNodeLogic: TopZeroSync received, timer reset"); + } + _ => { + // Autres markers: passthrough transparent + } + } + send_to_children!(segment); + } + + _AudioSegment::Chunk(_) => { + // Vérifier le pacing seulement si on a un timer de référence + if let Some(start) = self.start_time { + let chunk_timestamp = segment.timestamp_sec; + let elapsed = start.elapsed().as_secs_f64(); + let lead_time = chunk_timestamp - elapsed; + + if lead_time > self.max_lead_time_sec { + // On est trop en avance, attendre + let sleep_duration = lead_time - self.max_lead_time_sec; + tracing::trace!( + "TimerNodeLogic: lead_time={:.3}s > max={:.1}s, sleeping {:.3}s", + lead_time, + self.max_lead_time_sec, + sleep_duration + ); + + tokio::select! { + _ = tokio::time::sleep(Duration::from_secs_f64(sleep_duration)) => {} + _ = stop_token.cancelled() => { + tracing::debug!("TimerNodeLogic cancelled during sleep"); + break; + } + } + } else if lead_time < -0.5 { + // On est en retard de plus de 500ms, log warning + tracing::warn!( + "TimerNodeLogic: lagging behind by {:.3}s (chunk ts={:.3}s, elapsed={:.3}s)", + -lead_time, + chunk_timestamp, + elapsed + ); + } + } else { + // Pas encore de TopZeroSync reçu, passthrough sans pacing + tracing::trace!("TimerNodeLogic: no timer set yet, passthrough"); + } + + send_to_children!(segment); + } + } + } + + tracing::debug!("TimerNodeLogic::process finished"); + Ok(()) + } +} + +// ═══════════════════════════════════════════════════════════════════════════ +// TimerNode - Wrapper utilisant Node +// ═══════════════════════════════════════════════════════════════════════════ + +pub struct TimerNode { + inner: Node, +} + +impl TimerNode { + /// Crée un TimerNode avec une avance maximale tolérée + /// + /// # Arguments + /// + /// * `max_lead_time_sec` - Avance maximale en secondes (ex: 3.0 pour 3s de buffer) + /// + /// # Exemples + /// + /// ```no_run + /// use pmoaudio::TimerNode; + /// + /// // Tolérer 3 secondes d'avance + /// let timer = TimerNode::new(3.0); + /// ``` + pub fn new(max_lead_time_sec: f64) -> Self { + Self::with_channel_size(max_lead_time_sec, DEFAULT_CHANNEL_SIZE) + } + + /// Crée un TimerNode avec une taille de buffer MPSC personnalisée + /// + /// # Arguments + /// + /// * `max_lead_time_sec` - Avance maximale en secondes + /// * `channel_size` - Taille du buffer MPSC (nombre de segments en attente) + pub fn with_channel_size(max_lead_time_sec: f64, channel_size: usize) -> Self { + let logic = TimerNodeLogic::new(max_lead_time_sec); + Self { + inner: Node::new_with_input(logic, channel_size), + } + } +} + +#[async_trait::async_trait] +impl AudioPipelineNode for TimerNode { + fn get_tx(&self) -> Option>> { + self.inner.get_tx() + } + + fn register(&mut self, child: Box) { + self.inner.register(child); + } + + async fn run(self: Box, stop_token: CancellationToken) -> Result<(), AudioError> { + Box::new(self.inner).run(stop_token).await + } +} + +impl TypedAudioNode for TimerNode { + fn input_type(&self) -> Option { + // Accepte n'importe quel type + Some(TypeRequirement::any()) + } + + fn output_type(&self) -> Option { + // Passthrough: produit le même type qu'il consomme + Some(TypeRequirement::any()) + } +} diff --git a/pmoaudiocache/src/cache.rs b/pmoaudiocache/src/cache.rs index 5f168ebb..8c117856 100644 --- a/pmoaudiocache/src/cache.rs +++ b/pmoaudiocache/src/cache.rs @@ -56,6 +56,46 @@ pub fn new_cache(dir: &str, limit: usize) -> Result { 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> { + 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 : diff --git a/pmoaudiocache/tests/test_cache.rs b/pmoaudiocache/tests/test_cache.rs new file mode 100644 index 00000000..e4494ec8 --- /dev/null +++ b/pmoaudiocache/tests/test_cache.rs @@ -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); +} diff --git a/pmocache/Cargo.toml b/pmocache/Cargo.toml index 409df8b1..8cf63f3d 100644 --- a/pmocache/Cargo.toml +++ b/pmocache/Cargo.toml @@ -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"] diff --git a/pmocache/src/cache.rs b/pmocache/src/cache.rs index 879cc8be..4ba635da 100755 --- a/pmocache/src/cache.rs +++ b/pmocache/src/cache.rs @@ -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 { downloads: Arc>>>, /// Factory pour créer des transformers (optionnel) transformer_factory: Option 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, } impl Cache { + /// 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 { + 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> { + 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) -> Result { + // 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 Cache { 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::::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,25 +297,16 @@ impl Cache { 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 - tracing::debug!("File with pk {} already in cache, updating timestamp", pk); - self.db.update_hit(&pk)?; - return Ok(pk); - } + // 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); - return Ok(pk); - } + if let Some(pk) = self.check_ongoing_download(&pk).await? { + return Ok(pk); } // 5. Lancer le téléchargement complet avec transformer @@ -202,20 +324,14 @@ impl Cache { // 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,66 @@ impl Cache { 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) - .await - .map_err(|e| anyhow!("Failed to peek reader header: {}", e))?; + self.add_from_reader_with_pk(source_uri, reader, length, collection, None).await + } - // 2. Calculer le pk basé sur le contenu - let pk = crate::cache_trait::pk_from_content_header(&header); + /// 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( + &self, + source_uri: Option<&str>, + mut reader: R, + length: Option, + collection: Option<&str>, + explicit_pk: Option, + ) -> Result + where + R: AsyncRead + Send + Unpin + 'static, + { + // 1. Lire EXACTEMENT 1024 octets (ou EOF si fichier plus petit) + // Utilise read_exact_or_eof qui boucle jusqu'à avoir tous les octets demandés + let header = crate::download::read_exact_or_eof(&mut reader, 1024) + .await + .map_err(|e| anyhow!("Failed to read header bytes: {}", e))?; + + // 2. Calculer le pk selon la taille du fichier + // - Fichiers >= 1024 octets (FLAC): skip header (512 premiers octets), utilise octets 512-1024 + // - Fichiers < 1024 octets (images, petits fichiers): utilise TOUT le contenu + let pk = if let Some(explicit) = explicit_pk { + explicit + } else { + let pk_bytes = if header.len() >= 1024 { + // Gros fichier (>= 1024 octets): skip les 512 premiers (header FLAC) + &header[512..] + } else { + // Petit fichier (< 1024 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 - tracing::debug!("File with pk {} already in cache, updating timestamp", pk); - self.db.update_hit(&pk)?; - return Ok(pk); - } + // 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); - return Ok(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 +451,8 @@ impl Cache { 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 +630,22 @@ impl Cache { } /// 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 +662,14 @@ impl Cache { 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 +679,20 @@ impl Cache { 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; } } } @@ -559,6 +716,27 @@ impl Cache { downloads.get(pk).cloned() } + /// Vérifie si le téléchargement/ingestion d'un fichier est complètement terminé + /// + /// Cette méthode vérifie l'existence du fichier marker de complétion (.complete) + /// qui est créé uniquement quand le fichier est complètement écrit et fermé. + /// + /// Utile pour différencier: + /// - EOF temporaire : fichier encore en cours d'écriture (retourne false) + /// - EOF réel : fichier complètement écrit (retourne true) + /// + /// # Arguments + /// + /// * `pk` - Clé primaire du fichier + /// + /// # Returns + /// + /// `true` si le fichier est complètement écrit (marker existe), `false` sinon + pub fn is_download_complete(&self, pk: &str) -> bool { + let completion_marker = self.get_completion_marker_path(pk); + completion_marker.exists() + } + /// Retourne la taille actuelle téléchargée (source) /// /// Si le download est en cours, retourne la taille téléchargée. @@ -770,17 +948,10 @@ impl Cache { 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; } } diff --git a/pmocache/src/cache_trait.rs b/pmocache/src/cache_trait.rs index 07909f2d..1036130c 100644 --- a/pmocache/src/cache_trait.rs +++ b/pmocache/src/cache_trait.rs @@ -138,9 +138,54 @@ pub trait FileCache: Send + Sync { /// /// # Returns /// - /// `true` si l'entrée existe en base de données et que le fichier est présent + /// `true` si l'entrée existe en base de données et que le fichier est présent ET: + /// - SOIT le fichier est complet (marker .complete existe) + /// - SOIT le download est en cours (fichier récent sans marker) + /// + /// Ceci permet le progressive caching: les fichiers en cours de download sont acceptés + /// dès que le prebuffer est atteint, sans attendre le marker de completion. fn is_valid_pk(&self, pk: &str) -> bool { - self.get_database().get(pk, false).is_ok() && self.file_path(pk).exists() + if self.get_database().get(pk, false).is_err() { + tracing::debug!("is_valid_pk({}): DB entry not found", pk); + return false; + } + + let file_path = self.file_path(pk); + if !file_path.exists() { + tracing::debug!("is_valid_pk({}): File does not exist", pk); + return false; + } + + // Vérifier d'abord si le marker de completion existe + let completion_marker = file_path.with_extension( + format!("{}.complete", C::file_extension()) + ); + + if completion_marker.exists() { + tracing::debug!("is_valid_pk({}): Completion marker found, file is complete", pk); + return true; + } + + // Pas de marker - vérifier si le download est en cours (fichier récent) + // Un fichier en cours de download aura une modification récente + if let Ok(metadata) = file_path.metadata() { + if let Ok(modified) = metadata.modified() { + if let Ok(elapsed) = modified.elapsed() { + let age_secs = elapsed.as_secs(); + if age_secs < 60 { + tracing::debug!("is_valid_pk({}): No marker but file is recent ({}s), download in progress", pk, age_secs); + return true; + } else { + tracing::debug!("is_valid_pk({}): No marker and file is old ({}s), incomplete download", pk, age_secs); + return false; + } + } + } + } + + // Ne peut pas vérifier le statut - rejeter par sécurité + tracing::debug!("is_valid_pk({}): Could not check file status, rejecting", pk); + false } } diff --git a/pmocache/src/db.rs b/pmocache/src/db.rs index 657a11ae..cf00bf77 100644 --- a/pmocache/src/db.rs +++ b/pmocache/src/db.rs @@ -192,20 +192,24 @@ impl DB { collection: Option<&str>, metadata: Option<&Value>, ) -> rusqlite::Result<()> { - let conn = self.lock_conn("add_with_metadata"); + // Bloc pour limiter la durée du lock + { + let conn = self.lock_conn("add_with_metadata"); - conn.execute( - "INSERT INTO asset (pk, id, collection, hits, last_used) - VALUES (?1, ?2, ?3, 0, ?4) - ON CONFLICT(pk) DO UPDATE SET - id = excluded.id, - collection = excluded.collection, - last_used = excluded.last_used", - params![pk, id, collection, Utc::now().to_rfc3339()], - )?; + conn.execute( + "INSERT INTO asset (pk, id, collection, hits, last_used) + VALUES (?1, ?2, ?3, 0, ?4) + ON CONFLICT(pk) DO UPDATE SET + id = excluded.id, + collection = excluded.collection, + 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", diff --git a/pmocache/src/download.rs b/pmocache/src/download.rs index 636e89b6..aaeb12d6 100644 --- a/pmocache/src/download.rs +++ b/pmocache/src/download.rs @@ -600,3 +600,37 @@ where buffer.truncate(n); Ok(buffer) } + +/// Lit exactement `size` octets du reader, ou jusqu'à EOF. +/// +/// Contrairement à `peek_reader_header`, cette fonction boucle jusqu'à avoir lu +/// exactement `size` octets (ou atteindre EOF). Ceci est crucial pour calculer +/// un pk fiable sur un nombre d'octets précis. +/// +/// # Returns +/// +/// Le buffer contenant exactement `size` octets, ou moins si EOF est atteint +pub async fn read_exact_or_eof(reader: &mut R, size: usize) -> Result, String> +where + R: AsyncRead + Unpin, +{ + let mut buffer = vec![0u8; size]; + let mut total_read = 0; + + while total_read < size { + let n = reader + .read(&mut buffer[total_read..]) + .await + .map_err(|e| format!("Failed to read from stream: {}", e))?; + + if n == 0 { + // EOF atteint + buffer.truncate(total_read); + return Ok(buffer); + } + + total_read += n; + } + + Ok(buffer) +} diff --git a/pmocache/tests/test_cache.rs b/pmocache/tests/test_cache.rs new file mode 100644 index 00000000..f8aa2e69 --- /dev/null +++ b/pmocache/tests/test_cache.rs @@ -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; + +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); +} diff --git a/pmocache/tests/test_db.rs b/pmocache/tests/test_db.rs new file mode 100644 index 00000000..1066e7a2 --- /dev/null +++ b/pmocache/tests/test_db.rs @@ -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)); +} diff --git a/pmocovers/Cargo.toml b/pmocovers/Cargo.toml index eea54228..32db1101 100644 --- a/pmocovers/Cargo.toml +++ b/pmocovers/Cargo.toml @@ -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"] diff --git a/pmocovers/tests/test_cache.rs b/pmocovers/tests/test_cache.rs new file mode 100644 index 00000000..7c95acf2 --- /dev/null +++ b/pmocovers/tests/test_cache.rs @@ -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 { + let img: ImageBuffer, Vec> = 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); +} diff --git a/pmocovers/tests/test_webp.rs b/pmocovers/tests/test_webp.rs new file mode 100644 index 00000000..8bb7b1a2 --- /dev/null +++ b/pmocovers/tests/test_webp.rs @@ -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, Vec> = 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); +} diff --git a/pmoparadise/examples/play_and_cache.rs b/pmoparadise/examples/play_and_cache.rs index 57354fa6..41a72173 100644 --- a/pmoparadise/examples/play_and_cache.rs +++ b/pmoparadise/examples/play_and_cache.rs @@ -4,7 +4,8 @@ //! 1. RadioParadiseStreamSource - Télécharge et décode un bloc FLAC //! 2. FlacCacheSink - Cache chaque piste en FLAC et alimente une playlist //! 3. PlaylistSource - Lit la playlist pendant le téléchargement -//! 4. AudioSink - Joue l'audio sur la sortie standard +//! 4. TimerNode - Régule le débit pour éviter EOF prématurés (progressive cache) +//! 5. AudioSink - Joue l'audio sur la sortie standard //! //! Architecture : //! ```text @@ -12,7 +13,10 @@ //! RadioParadiseStreamSource → FlacCacheSink (avec playlist abonnée) //! //! Pipeline 2 (Playback): -//! PlaylistSource (lit la playlist) → AudioSink (joue l'audio) +//! PlaylistSource → TimerNode (rate limiting) → AudioSink +//! ↓ +//! Prévention EOF +//! (3s max lead) //! ``` //! //! Usage: @@ -22,7 +26,7 @@ //! cargo run --example play_and_cache --features full -- 0 # Main Mix //! cargo run --example play_and_cache --features full -- 2 # Rock Mix -use pmoaudio::{AudioPipelineNode, AudioSink}; +use pmoaudio::{AudioPipelineNode, AudioSink, TimerNode}; use pmoaudio_ext::{FlacCacheSink, PlaylistSource}; use pmoaudiocache::Cache as AudioCache; use pmocovers::Cache as CoverCache; @@ -50,8 +54,8 @@ async fn main() -> Result<(), Box> { // Récupérer les arguments let args: Vec = env::args().collect(); - if args.len() != 2 { - eprintln!("Usage: {} ", args[0]); + if args.len() < 2 { + eprintln!("Usage: {} [--null-audio]", args[0]); eprintln!(); eprintln!("Downloads a Radio Paradise block, caches it, and plays it simultaneously."); eprintln!(); @@ -60,6 +64,9 @@ async fn main() -> Result<(), Box> { eprintln!(" 1 - Mellow Mix (smooth, chilled music)"); eprintln!(" 2 - Rock Mix (classic & modern rock)"); eprintln!(" 3 - World/Etc Mix (global sounds)"); + eprintln!(); + eprintln!("Options:"); + eprintln!(" --null-audio Don't play audio (for testing without audio device)"); std::process::exit(1); } @@ -71,7 +78,12 @@ async fn main() -> Result<(), Box> { } }; + let use_null_audio = args.len() > 2 && args[2] == "--null-audio"; + tracing::info!("Channel ID: {}", channel_id); + if use_null_audio { + tracing::info!("Using null audio output (no playback)"); + } // ═══════════════════════════════════════════════════════════════════════════ // Initialiser les caches et le gestionnaire de playlist @@ -187,13 +199,25 @@ async fn main() -> Result<(), Box> { let mut playlist_source = PlaylistSource::new(reader, audio_cache.clone()); tracing::debug!("PlaylistSource created"); + // Créer le timer node pour réguler le débit (empêche EOF prématurés) + // Tolère 3 secondes d'avance max pour permettre le buffering + let mut timer = TimerNode::new(3.0); + tracing::debug!("TimerNode created (max_lead_time=3.0s)"); + // Créer le sink audio - let audio_sink = AudioSink::new(); + let audio_sink = if use_null_audio { + AudioSink::with_null_output() + } else { + AudioSink::new() + }; tracing::debug!("AudioSink created"); - // Connecter playlist → audio - playlist_source.register(Box::new(audio_sink)); - tracing::info!("Playback pipeline connected: PlaylistSource → AudioSink"); + // Connecter timer → audio (AVANT de mettre timer dans une Box) + timer.register(Box::new(audio_sink)); + + // Connecter playlist → timer + playlist_source.register(Box::new(timer)); + tracing::info!("Playback pipeline connected: PlaylistSource → TimerNode → AudioSink"); // ═══════════════════════════════════════════════════════════════════════════ // Lancer les deux pipelines en parallèle @@ -233,9 +257,9 @@ async fn main() -> Result<(), Box> { }); 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"), diff --git a/pmoparadise/src/radio_paradise_stream_source.rs b/pmoparadise/src/radio_paradise_stream_source.rs index 71c191b6..d7282913 100644 --- a/pmoparadise/src/radio_paradise_stream_source.rs +++ b/pmoparadise/src/radio_paradise_stream_source.rs @@ -86,6 +86,7 @@ impl RadioParadiseStreamSourceLogic { order: &mut u64, ) -> Result<(), AudioError> { // Télécharger le FLAC + tracing::debug!("Sending HTTP GET request for block FLAC"); let response = self.client.client .get(&block.url) .timeout(self.client.block_timeout) @@ -93,6 +94,7 @@ impl RadioParadiseStreamSourceLogic { .await .map_err(|e| AudioError::ProcessingError(format!("Block download failed: {}", e)))?; + tracing::debug!("HTTP response received, status={}", response.status()); if !response.status().is_success() { return Err(AudioError::ProcessingError(format!( "Block download returned status {}", @@ -101,12 +103,15 @@ impl RadioParadiseStreamSourceLogic { } // Créer un stream reader + tracing::debug!("Creating byte stream reader"); let byte_stream = response.bytes_stream().map(|result| { result.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)) }); let stream_reader = StreamReader::new(byte_stream); + tracing::debug!("Stream reader created"); // Décoder le FLAC + tracing::debug!("Decoding FLAC stream..."); let mut decoder = decode_audio_stream(stream_reader) .await .map_err(|e| AudioError::ProcessingError(format!("FLAC decode failed: {}", e)))?; @@ -114,20 +119,25 @@ impl RadioParadiseStreamSourceLogic { let stream_info = decoder.info().clone(); let sample_rate = stream_info.sample_rate; let bits_per_sample = stream_info.bits_per_sample; + tracing::debug!("FLAC decoder initialized: {}Hz, {} bits/sample", sample_rate, bits_per_sample); // Préparer les songs ordonnées pour tracking let songs = block.songs_ordered(); let mut song_index = 0; let mut next_song: Option<(usize, &Song)> = songs.get(0).copied(); let mut total_samples = 0u64; + tracing::debug!("Block has {} songs", songs.len()); // Envoyer TopZeroSync au début du bloc + tracing::debug!("Sending TopZeroSync to {} outputs", output.len()); let top_zero = Arc::new(AudioSegment { order: *order, timestamp_sec: 0.0, segment: pmoaudio::_AudioSegment::Sync(Arc::new(SyncMarker::TopZeroSync)), }); self.send_to_children(output, top_zero).await?; + tracing::debug!("TopZeroSync sent, starting audio chunk loop"); + // Buffer pour lecture let bytes_per_sample = (bits_per_sample / 8) as usize; @@ -393,48 +403,68 @@ impl NodeLogic for RadioParadiseStreamSourceLogic { output: Vec>>, stop_token: CancellationToken, ) -> Result<(), AudioError> { + tracing::debug!("RadioParadiseStreamSource::process() started, block_queue has {} items", self.block_queue.len()); + for (i, event_id) in self.block_queue.iter().enumerate() { + tracing::debug!(" block_queue[{}] = {}", i, event_id); + } + let mut order = 0u64; loop { // Attendre un block ID (timeout court pour une radio) + tracing::debug!("Waiting for block_id from queue (timeout={}s)...", BLOCK_ID_TIMEOUT_SECS); let event_id = match tokio::time::timeout( Duration::from_secs(BLOCK_ID_TIMEOUT_SECS), async { while self.block_queue.is_empty() { + tracing::trace!("block_queue is empty, sleeping..."); tokio::time::sleep(Duration::from_millis(100)).await; if stop_token.is_cancelled() { + tracing::debug!("stop_token cancelled while waiting for block_id"); return None; } } self.block_queue.pop_front() } ).await { - Ok(Some(id)) => id, - Ok(None) => break, // Cancelled + Ok(Some(id)) => { + tracing::debug!("Got event_id {} from queue", id); + id + } + Ok(None) => { + tracing::debug!("Loop cancelled, breaking"); + break; + } // Cancelled Err(_) => { // Timeout - pas de nouveau bloc, on termine + tracing::warn!("Timeout waiting for block_id, breaking"); break; } }; // Vérifier si déjà téléchargé récemment if self.is_recent_block(event_id) { + tracing::debug!("Block {} was recently downloaded, skipping", event_id); continue; } // Récupérer les métadonnées du bloc + tracing::debug!("Fetching block metadata for event_id {}...", event_id); let block = self.client .get_block(Some(event_id)) .await .map_err(|e| AudioError::ProcessingError(format!("Failed to get block: {}", e)))?; + tracing::debug!("Block metadata received: url={}", block.url); // Marquer comme téléchargé self.mark_block_downloaded(event_id); // Télécharger et décoder le bloc + tracing::info!("Starting download and decode for block {}...", event_id); self.download_and_decode_block(&block, &output, &stop_token, &mut order) .await?; + tracing::info!("Finished download and decode for block {}", event_id); } // Envoyer EndOfStream diff --git a/pmoupnp/src/cache_registry.rs b/pmoupnp/src/cache_registry.rs new file mode 100644 index 00000000..ee5a2840 --- /dev/null +++ b/pmoupnp/src/cache_registry.rs @@ -0,0 +1,97 @@ +//! Registre centralisé des caches pour le serveur UPnP (couche de compatibilité) +//! +//! Ce module fournit une couche de compatibilité pour pmosource qui utilise +//! les singletons de pmoaudiocache et pmocovers pour accéder aux caches. + +use pmoaudiocache::Cache as AudioCache; +use pmocache::FileCache; +use pmocovers::Cache as CoverCache; +use std::sync::Arc; + +/// 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> { + pmocovers::get_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> { + pmoaudiocache::get_audio_cache() +} + +/// Construit l'URL complète pour une couverture +/// +/// # Arguments +/// +/// * `pk` - Clé primaire de la couverture +/// * `size` - Taille optionnelle de l'image +/// +/// # Examples +/// +/// ```rust,ignore +/// use pmoupnp::cache_registry::build_cover_url; +/// +/// let url = build_cover_url("abc123", Some(300))?; +/// // url = "http://localhost:8080/covers/images/abc123/300" +/// ``` +pub fn build_cover_url(pk: &str, size: Option) -> anyhow::Result { + // Récupérer l'URL de base depuis la variable d'environnement ou une config + let base_url = std::env::var("PMO_SERVER_URL") + .unwrap_or_else(|_| "http://localhost:8080".to_string()); + + let cache = get_cover_cache() + .ok_or_else(|| anyhow::anyhow!("No registered cover cache"))?; + + let param = match size { + Some(size_) => Some(size_.to_string()), + None => None, + }; + let route = cache.route_for(pk, param.as_deref()); + Ok(format!("{}{}", base_url, route)) +} + +/// Construit l'URL complète pour une piste audio +/// +/// # Arguments +/// +/// * `pk` - Clé primaire de la piste +/// * `param` - Paramètre optionnel (ex: "orig", "stream") +/// +/// # Examples +/// +/// ```rust,ignore +/// use pmoupnp::cache_registry::build_audio_url; +/// +/// let url = build_audio_url("abc123", Some("stream"))?; +/// // url = "http://localhost:8080/audio/tracks/abc123/stream" +/// ``` +pub fn build_audio_url(pk: &str, param: Option<&str>) -> anyhow::Result { + // Récupérer l'URL de base depuis la variable d'environnement ou une config + let base_url = std::env::var("PMO_SERVER_URL") + .unwrap_or_else(|_| "http://localhost:8080".to_string()); + + let cache = get_audio_cache() + .ok_or_else(|| anyhow::anyhow!("No registered audio cache"))?; + + let route = cache.route_for(pk, param); + Ok(format!("{}{}", base_url, route)) +} diff --git a/pmoupnp/src/lib.rs b/pmoupnp/src/lib.rs index 28e0cb93..67af0ac0 100644 --- a/pmoupnp/src/lib.rs +++ b/pmoupnp/src/lib.rs @@ -2,6 +2,7 @@ mod object_set; mod object_trait; pub mod actions; +pub mod cache_registry; pub mod devices; pub mod services; pub mod soap; diff --git a/setup-deps.sh b/setup-deps.sh new file mode 100755 index 00000000..d07de01c --- /dev/null +++ b/setup-deps.sh @@ -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 ""