Merge pull request #26 from coissac/claude/fix-play-and-cache-streaming-011CUsMBxH4fsgoadgkiPdoK
Claude/fix play and cache streaming 011 c us m bx h4fsgoadgki pdo k
This commit is contained in:
coissac
GitHub
485
OPTIMIZATION_PREBUFFER_TO_PLAYLIST.md
Normal file
485
OPTIMIZATION_PREBUFFER_TO_PLAYLIST.md
Normal file
@@ -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<AudioSegment>,
|
||||
rx: &mut mpsc::Receiver<Arc<AudioSegment>>, // ← &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<AudioSegment>,
|
||||
rx: mpsc::Receiver<Arc<AudioSegment>>, // ownership!
|
||||
pcm_tx: mpsc::Sender<Vec<u8>>,
|
||||
bits_per_sample: u8,
|
||||
expected_rate: u32,
|
||||
stop_token: CancellationToken,
|
||||
) -> Result<(u64, u64, f64, StopReason, mpsc::Receiver<Arc<AudioSegment>>), 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<R, F>(
|
||||
&self,
|
||||
source_uri: Option<&str>,
|
||||
reader: R,
|
||||
length: Option<u64>,
|
||||
collection: Option<&str>,
|
||||
on_prebuffer: F, // ← nouveau callback
|
||||
) -> Result<String>
|
||||
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<AudioSegment>,
|
||||
mut rx: mpsc::Receiver<Arc<AudioSegment>>,
|
||||
pcm_tx: mpsc::Sender<Vec<u8>>,
|
||||
bits_per_sample: u8,
|
||||
expected_rate: u32,
|
||||
stop_token: CancellationToken,
|
||||
) -> Result<(u64, u64, f64, StopReason, mpsc::Receiver<Arc<AudioSegment>>), 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é"
|
||||
@@ -86,16 +86,22 @@ impl NodeLogic for FlacCacheSinkLogic {
|
||||
_output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
|
||||
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(());
|
||||
}
|
||||
};
|
||||
@@ -125,12 +131,14 @@ 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 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");
|
||||
|
||||
// Ingérer le FLAC progressivement dans le cache
|
||||
// add_from_reader lance l'ingestion en arrière-plan et retourne dès que
|
||||
@@ -138,6 +146,7 @@ impl NodeLogic for FlacCacheSinkLogic {
|
||||
// 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,
|
||||
@@ -156,13 +165,15 @@ impl NodeLogic for FlacCacheSinkLogic {
|
||||
);
|
||||
|
||||
// Attendre les deux tâches en parallèle
|
||||
tracing::debug!("FlacCacheSink: Waiting for cache and pump to complete");
|
||||
let (cache_result, pump_result) = tokio::join!(cache_future, pump_future);
|
||||
|
||||
tracing::debug!("FlacCacheSink: tokio::join! completed, checking results");
|
||||
let pk = cache_result.map_err(|e| {
|
||||
AudioError::ProcessingError(format!("Failed to add to cache: {}", e))
|
||||
})?;
|
||||
|
||||
tracing::debug!("Track added to cache with pk {}, prebuffer complete", pk);
|
||||
tracing::debug!("FlacCacheSink: Track added to cache with pk {}, prebuffer complete", pk);
|
||||
|
||||
let (_chunks, _samples, _duration_sec, stop_reason) = pump_result?;
|
||||
|
||||
|
||||
@@ -172,11 +172,70 @@ fn chunk_to_f32_interleaved(chunk: &AudioChunk) -> Vec<f32> {
|
||||
// ═══════════════════════════════════════════════════════════════════════════
|
||||
|
||||
/// 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<Arc<AudioSegment>>,
|
||||
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 {
|
||||
|
||||
@@ -387,22 +387,23 @@ impl<C: CacheConfig> Cache<C> {
|
||||
where
|
||||
R: AsyncRead + Send + Unpin + 'static,
|
||||
{
|
||||
// 1. Lire jusqu'à 1024 octets (ou ce qui est disponible)
|
||||
let header = crate::download::peek_reader_header(&mut reader, 1024)
|
||||
// 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 peek reader header: {}", e))?;
|
||||
.map_err(|e| anyhow!("Failed to read header bytes: {}", e))?;
|
||||
|
||||
// 2. Calculer le pk en utilisant au plus les 512 derniers octets
|
||||
// Ceci évite les collisions pour les fichiers avec headers identiques (ex: FLAC)
|
||||
// tout en fonctionnant pour les petits fichiers (images < 512 octets)
|
||||
// 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() > 512 {
|
||||
// Fichier >= 512 octets: utiliser les octets 512+ (au plus 512 octets)
|
||||
let pk_bytes = if header.len() >= 1024 {
|
||||
// Gros fichier (>= 1024 octets): skip les 512 premiers (header FLAC)
|
||||
&header[512..]
|
||||
} else {
|
||||
// Petit fichier < 512 octets: utiliser tout le contenu
|
||||
// Petit fichier (< 1024 octets): utiliser TOUT le contenu
|
||||
&header[..]
|
||||
};
|
||||
crate::cache_trait::pk_from_content_header(pk_bytes)
|
||||
|
||||
@@ -138,9 +138,54 @@ pub trait FileCache<C: CacheConfig>: 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
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<R>(reader: &mut R, size: usize) -> Result<Vec<u8>, 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)
|
||||
}
|
||||
|
||||
@@ -50,8 +50,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
|
||||
// Récupérer les arguments
|
||||
let args: Vec<String> = env::args().collect();
|
||||
if args.len() != 2 {
|
||||
eprintln!("Usage: {} <channel_id>", args[0]);
|
||||
if args.len() < 2 {
|
||||
eprintln!("Usage: {} <channel_id> [--null-audio]", args[0]);
|
||||
eprintln!();
|
||||
eprintln!("Downloads a Radio Paradise block, caches it, and plays it simultaneously.");
|
||||
eprintln!();
|
||||
@@ -60,6 +60,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
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 +74,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
}
|
||||
};
|
||||
|
||||
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
|
||||
@@ -188,7 +196,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
tracing::debug!("PlaylistSource created");
|
||||
|
||||
// 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
|
||||
|
||||
@@ -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<mpsc::Sender<Arc<AudioSegment>>>,
|
||||
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
|
||||
|
||||
97
pmoupnp/src/cache_registry.rs
Normal file
97
pmoupnp/src/cache_registry.rs
Normal file
@@ -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<Arc<CoverCache>> {
|
||||
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<Arc<AudioCache>> {
|
||||
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<usize>) -> anyhow::Result<String> {
|
||||
// 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<String> {
|
||||
// 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))
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user