on retravaille les sources et pmoparadise en particulier
This commit is contained in:
@@ -151,31 +151,17 @@ async fn register_paradise(Json(params): Json<ParadiseParams>) -> impl IntoRespo
|
||||
};
|
||||
|
||||
// Créer et enregistrer la source depuis le registry
|
||||
let source = if let Some(capacity) = params.fifo_capacity {
|
||||
match RadioParadiseSource::from_registry(client, capacity) {
|
||||
Ok(s) => Arc::new(s),
|
||||
Err(e) => {
|
||||
return (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(ErrorResponse {
|
||||
error: format!("Failed to create source: {}", e),
|
||||
}),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
}
|
||||
} else {
|
||||
match RadioParadiseSource::from_registry_default(client) {
|
||||
Ok(s) => Arc::new(s),
|
||||
Err(e) => {
|
||||
return (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(ErrorResponse {
|
||||
error: format!("Failed to create source: {}", e),
|
||||
}),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
// Note: params.fifo_capacity is currently not used by from_registry
|
||||
let source = match RadioParadiseSource::from_registry(client) {
|
||||
Ok(s) => Arc::new(s),
|
||||
Err(e) => {
|
||||
return (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(ErrorResponse {
|
||||
error: format!("Failed to create source: {}", e),
|
||||
}),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@@ -48,8 +48,10 @@ url = "2.5"
|
||||
# Audio decoding/encoding
|
||||
symphonia = { version = "0.5", features = ["all"] }
|
||||
|
||||
# Audio decoding - claxon for FLAC streaming
|
||||
claxon = "0.4"
|
||||
|
||||
# Per-track feature dependencies
|
||||
claxon = { version = "0.4", optional = true }
|
||||
hound = { version = "3.5", optional = true }
|
||||
tempfile = { version = "3.8", optional = true }
|
||||
|
||||
@@ -78,8 +80,8 @@ axum = { version = "0.8.4", optional = true }
|
||||
default = ["metadata-only"]
|
||||
# Mode métadonnées seules (pas de décodage FLAC)
|
||||
metadata-only = []
|
||||
# Active le décodage FLAC par-track
|
||||
per-track = ["dep:claxon", "dep:hound", "dep:tempfile"]
|
||||
# Active l'extraction par-track (WAV export, etc.)
|
||||
per-track = ["dep:hound", "dep:tempfile"]
|
||||
# Active l'API REST pmoserver
|
||||
pmoserver = ["dep:pmoserver", "dep:utoipa", "dep:axum", "server"]
|
||||
# Active le media server UPnP (includes pmoserver)
|
||||
|
||||
610
pmoparadise/STREAMING_IMPLEMENTATION_GUIDE.md
Normal file
610
pmoparadise/STREAMING_IMPLEMENTATION_GUIDE.md
Normal file
@@ -0,0 +1,610 @@
|
||||
# Guide d'Implémentation - Streaming Progressif avec Claxon
|
||||
|
||||
## Objectif
|
||||
Transformer le worker pour qu'il décode le FLAC en streaming au fur et à mesure du téléchargement HTTP, afin d'envoyer le premier morceau au cache en **~6-8 secondes** au lieu de 12-16 secondes.
|
||||
|
||||
---
|
||||
|
||||
## Vue d'Ensemble de l'Architecture
|
||||
|
||||
### Architecture Actuelle (LENTE - 12-16s)
|
||||
```
|
||||
HTTP Request → Télécharger TOUT le block (75-100 MB) → Symphonia (Cursor)
|
||||
↓
|
||||
Décoder TOUT en PCM
|
||||
↓
|
||||
Pour chaque morceau:
|
||||
- Découper PCM
|
||||
- Encoder FLAC
|
||||
- Envoyer au cache
|
||||
```
|
||||
|
||||
**Problème** : On attend le téléchargement complet avant de commencer quoi que ce soit.
|
||||
|
||||
### Architecture Cible (RAPIDE - 6-8s)
|
||||
```
|
||||
HTTP Stream → StreamReader (adapt async → sync)
|
||||
↓
|
||||
claxon::FlacReader (lit frame par frame SANS Seek)
|
||||
↓
|
||||
Accumule PCM dans buffer
|
||||
↓
|
||||
Dès que buffer.samples >= durée_morceau_1:
|
||||
- Découper buffer
|
||||
- Encoder FLAC
|
||||
- Envoyer au cache (morceau 1 disponible!)
|
||||
↓
|
||||
Continue streaming pour morceaux 2, 3, ...
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Étape 1 : Créer AsyncReadAdapter
|
||||
|
||||
### But
|
||||
Convertir `Stream<Item = Result<Bytes>>` (async) en `impl Read` (sync) pour claxon.
|
||||
|
||||
### Localisation
|
||||
Ajouter au début de `paradise/worker.rs`, après les imports.
|
||||
|
||||
### Code Complet
|
||||
|
||||
```rust
|
||||
use std::pin::Pin;
|
||||
use std::task::{Context, Poll};
|
||||
use futures::stream::Stream;
|
||||
use std::io::{self, Read};
|
||||
use std::collections::VecDeque;
|
||||
use tokio::runtime::Handle;
|
||||
use bytes::Bytes;
|
||||
|
||||
/// Adapte un Stream async en impl Read synchrone
|
||||
///
|
||||
/// Utilise le runtime tokio courant pour bloquer sur le stream async.
|
||||
/// ATTENTION: Doit être appelé depuis un contexte tokio (spawn_blocking).
|
||||
struct AsyncReadAdapter {
|
||||
stream: Pin<Box<dyn Stream<Item = Result<Bytes, reqwest::Error>> + Send>>,
|
||||
buffer: VecDeque<u8>,
|
||||
runtime: Handle,
|
||||
done: bool,
|
||||
}
|
||||
|
||||
impl AsyncReadAdapter {
|
||||
fn new(stream: Pin<Box<dyn Stream<Item = Result<Bytes, reqwest::Error>> + Send>>) -> Self {
|
||||
Self {
|
||||
stream,
|
||||
buffer: VecDeque::new(),
|
||||
runtime: Handle::current(),
|
||||
done: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn fill_buffer(&mut self) -> io::Result<()> {
|
||||
if self.done {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Bloquer pour récupérer le prochain chunk du stream
|
||||
let next_chunk = self.runtime.block_on(async {
|
||||
use futures::StreamExt;
|
||||
self.stream.next().await
|
||||
});
|
||||
|
||||
match next_chunk {
|
||||
Some(Ok(bytes)) => {
|
||||
self.buffer.extend(bytes.iter());
|
||||
Ok(())
|
||||
}
|
||||
Some(Err(e)) => {
|
||||
self.done = true;
|
||||
Err(io::Error::new(io::ErrorKind::Other, e))
|
||||
}
|
||||
None => {
|
||||
self.done = true;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Read for AsyncReadAdapter {
|
||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
// Si buffer vide et stream pas terminé, remplir
|
||||
while self.buffer.is_empty() && !self.done {
|
||||
self.fill_buffer()?;
|
||||
}
|
||||
|
||||
// Copier du buffer vers buf
|
||||
let to_read = buf.len().min(self.buffer.len());
|
||||
for i in 0..to_read {
|
||||
buf[i] = self.buffer.pop_front().unwrap();
|
||||
}
|
||||
|
||||
Ok(to_read)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Pièges à Éviter
|
||||
|
||||
1. **Context Tokio** : `AsyncReadAdapter` DOIT être créé dans un contexte tokio (utilisez `tokio::task::spawn_blocking`)
|
||||
2. **Deadlock** : Ne jamais appeler depuis le même thread qui exécute le stream
|
||||
3. **Buffer Size** : VecDeque peut grossir - surveiller la mémoire
|
||||
|
||||
---
|
||||
|
||||
## Étape 2 : Remplacer process_block()
|
||||
|
||||
### Architecture de la Nouvelle Fonction
|
||||
|
||||
```rust
|
||||
async fn process_block(&mut self, block: Block) -> Result<()> {
|
||||
// 1. Vérifications habituelles
|
||||
if self.is_recent_block(block.event) { ... }
|
||||
|
||||
// 2. Lancer le streaming HTTP
|
||||
let block_url = Url::parse(&block.url)?;
|
||||
let http_stream = self.client.stream_block(&block_url).await?;
|
||||
|
||||
// 3. Spawn un thread bloquant pour le décodage
|
||||
let songs_ordered = block.songs_ordered();
|
||||
let channel_id = self.descriptor.id;
|
||||
let sample_rate = 44100; // Sera détecté par claxon
|
||||
|
||||
let tracks = tokio::task::spawn_blocking(move || {
|
||||
decode_and_split_streaming(
|
||||
http_stream,
|
||||
songs_ordered,
|
||||
sample_rate
|
||||
)
|
||||
}).await??;
|
||||
|
||||
// 4. Pour chaque track décodé, envoyer au cache
|
||||
for (track_pcm, song_index, song) in tracks {
|
||||
let entry = self.process_song_from_pcm(
|
||||
&block,
|
||||
song_index,
|
||||
song,
|
||||
track_pcm
|
||||
).await?;
|
||||
|
||||
self.playlist.push_active(entry).await;
|
||||
}
|
||||
|
||||
// 5. Mise à jour
|
||||
self.record_processed_block(block.event);
|
||||
self.next_block_hint = Some(block.end_event);
|
||||
Ok(())
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Étape 3 : Fonction de Décodage Streaming
|
||||
|
||||
### Pseudo-Code Détaillé
|
||||
|
||||
```rust
|
||||
fn decode_and_split_streaming(
|
||||
http_stream: BlockStream, // Le stream de client.stream_block()
|
||||
songs: Vec<(usize, &Song)>,
|
||||
expected_sample_rate: u32,
|
||||
) -> Result<Vec<(Vec<i32>, usize, Song)>> {
|
||||
// 1. Convertir BlockStream en AsyncReadAdapter
|
||||
let adapter = AsyncReadAdapter::new(http_stream.into_inner());
|
||||
let buffered = std::io::BufReader::new(adapter);
|
||||
|
||||
// 2. Créer le FlacReader de claxon
|
||||
let mut reader = claxon::FlacReader::new(buffered)
|
||||
.map_err(|e| anyhow!("Failed to create FLAC reader: {e}"))?;
|
||||
|
||||
let streaminfo = reader.streaminfo();
|
||||
let channels = streaminfo.channels as usize;
|
||||
let sample_rate = streaminfo.sample_rate;
|
||||
let bits_per_sample = streaminfo.bits_per_sample;
|
||||
|
||||
// 3. Buffer PCM accumulé
|
||||
let mut accumulated_samples: Vec<i32> = Vec::new();
|
||||
let mut current_frame = 0; // Nombre de frames PCM lues
|
||||
let mut tracks = Vec::new();
|
||||
let mut next_song_idx = 0;
|
||||
|
||||
// 4. Lire frame par frame
|
||||
loop {
|
||||
// Lire une frame FLAC
|
||||
let frame = match reader.read_next_or_eof(/* buffer */) {
|
||||
Ok(Some(frame_data)) => frame_data,
|
||||
Ok(None) => break, // EOF
|
||||
Err(e) => return Err(anyhow!("FLAC decode error: {e}")),
|
||||
};
|
||||
|
||||
// Convertir frame en i32 et accumuler
|
||||
// NOTE: claxon retourne des samples par canal, il faut entrelacer
|
||||
let samples_in_frame = frame.len() / channels;
|
||||
for sample_idx in 0..samples_in_frame {
|
||||
for ch in 0..channels {
|
||||
let sample = frame[ch * samples_in_frame + sample_idx];
|
||||
// Normaliser selon bits_per_sample
|
||||
let normalized = normalize_sample(sample, bits_per_sample);
|
||||
accumulated_samples.push(normalized);
|
||||
}
|
||||
}
|
||||
|
||||
current_frame += samples_in_frame;
|
||||
|
||||
// 5. Vérifier si on a atteint la fin du morceau courant
|
||||
if next_song_idx < songs.len() {
|
||||
let (song_index, song) = &songs[next_song_idx];
|
||||
let song_end_frame = if next_song_idx + 1 < songs.len() {
|
||||
// Fin = début du prochain morceau
|
||||
ms_to_frames(songs[next_song_idx + 1].1.elapsed, sample_rate)
|
||||
} else {
|
||||
// Dernier morceau = fin du block
|
||||
usize::MAX // On prendra tout jusqu'à la fin
|
||||
};
|
||||
|
||||
if current_frame >= song_end_frame {
|
||||
// 6. Découper le buffer
|
||||
let song_start_frame = ms_to_frames(song.elapsed, sample_rate);
|
||||
let start_sample = song_start_frame * channels;
|
||||
let end_sample = song_end_frame * channels;
|
||||
|
||||
let track_samples = accumulated_samples[start_sample..end_sample.min(accumulated_samples.len())]
|
||||
.to_vec();
|
||||
|
||||
tracks.push((track_samples, *song_index, (*song).clone()));
|
||||
|
||||
next_song_idx += 1;
|
||||
|
||||
// IMPORTANT: Premier morceau envoyé ici!
|
||||
// Les suivants continueront pendant que le premier est traité
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 7. Traiter le dernier morceau si nécessaire
|
||||
if next_song_idx < songs.len() {
|
||||
let (song_index, song) = &songs[next_song_idx];
|
||||
let song_start_frame = ms_to_frames(song.elapsed, sample_rate);
|
||||
let start_sample = song_start_frame * channels;
|
||||
let track_samples = accumulated_samples[start_sample..].to_vec();
|
||||
tracks.push((track_samples, *song_index, (*song).clone()));
|
||||
}
|
||||
|
||||
Ok(tracks)
|
||||
}
|
||||
|
||||
fn normalize_sample(sample: i32, bits_per_sample: u32) -> i32 {
|
||||
match bits_per_sample {
|
||||
0..=16 => sample << 16, // Shift to 32-bit range
|
||||
17..=24 => sample << 8,
|
||||
_ => sample,
|
||||
}
|
||||
}
|
||||
|
||||
fn ms_to_frames(ms: u64, sample_rate: u32) -> usize {
|
||||
((ms as u128 * sample_rate as u128) / 1000) as usize
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Étape 4 : Adapter process_song
|
||||
|
||||
### Nouvelle Signature
|
||||
|
||||
```rust
|
||||
async fn process_song_from_pcm(
|
||||
&self,
|
||||
block: &Block,
|
||||
song_index: usize,
|
||||
song: &Song,
|
||||
track_samples: Vec<i32>, // PCM déjà découpé
|
||||
) -> Result<Arc<PlaylistEntry>>
|
||||
```
|
||||
|
||||
### Changements
|
||||
|
||||
1. **Supprimer** le découpage (déjà fait dans decode_and_split_streaming)
|
||||
2. **Garder** l'encodage FLAC
|
||||
3. **Garder** le cache audio/cover
|
||||
4. **Garder** la création de PlaylistEntry
|
||||
|
||||
```rust
|
||||
async fn process_song_from_pcm(
|
||||
&self,
|
||||
block: &Block,
|
||||
song_index: usize,
|
||||
song: &Song,
|
||||
track_samples: Vec<i32>,
|
||||
) -> Result<Arc<PlaylistEntry>> {
|
||||
// 1. Encoder PCM → FLAC (déjà existant)
|
||||
let flac_bytes = encode_samples_to_flac(
|
||||
track_samples,
|
||||
2, // channels - TODO: passer en paramètre
|
||||
44100, // sample_rate - TODO: passer en paramètre
|
||||
16, // bits - TODO: passer en paramètre
|
||||
).await?;
|
||||
|
||||
// 2. Calculer track_id
|
||||
let track_id = self.compute_track_id(&flac_bytes);
|
||||
let placeholder_uri = format!("{}#{}", block.url, song_index);
|
||||
|
||||
// 3. Cache cover (inchangé)
|
||||
let cover_pk = self.cache_cover(block, song).await?;
|
||||
|
||||
// 4. Cache audio (inchangé)
|
||||
let flac_len = flac_bytes.len() as u64;
|
||||
let reader = StreamReader::new(stream::iter(vec![Ok::<_, std::io::Error>(
|
||||
Bytes::from(flac_bytes)
|
||||
)]));
|
||||
let audio_pk = self.cache_manager
|
||||
.cache_audio_from_reader(&track_id, reader, Some(flac_len))
|
||||
.await?;
|
||||
|
||||
// 5. Metadata (inchangé)
|
||||
let metadata = TrackMetadata {
|
||||
original_uri: placeholder_uri,
|
||||
cached_audio_pk: Some(audio_pk.clone()),
|
||||
cached_cover_pk: cover_pk,
|
||||
};
|
||||
self.cache_manager.update_metadata(track_id.clone(), metadata).await;
|
||||
|
||||
// 6. Créer PlaylistEntry (inchangé)
|
||||
let duration_ms = song.duration;
|
||||
let file_path = self.cache_manager.audio_file_path(&audio_pk).await;
|
||||
let entry = Arc::new(PlaylistEntry::new(
|
||||
track_id,
|
||||
self.descriptor.id,
|
||||
Arc::new(song.clone()),
|
||||
Utc::now(),
|
||||
duration_ms,
|
||||
Some(audio_pk),
|
||||
file_path,
|
||||
self.active_clients,
|
||||
));
|
||||
|
||||
Ok(entry)
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Étape 5 : API de claxon
|
||||
|
||||
### Documentation Claxon
|
||||
|
||||
```rust
|
||||
// Créer un reader
|
||||
let mut reader = claxon::FlacReader::new(buffered_reader)?;
|
||||
|
||||
// Obtenir les infos du stream
|
||||
let info = reader.streaminfo();
|
||||
// info.channels: u32
|
||||
// info.sample_rate: u32
|
||||
// info.bits_per_sample: u32
|
||||
// info.samples: Option<u64> (peut être None pour streams)
|
||||
|
||||
// Lire des samples
|
||||
// Option 1: Frame par frame (recommandé pour streaming)
|
||||
let mut samples = vec![0i32; info.channels as usize * 4096];
|
||||
loop {
|
||||
match reader.read_next_or_eof(samples.as_mut_slice()) {
|
||||
Ok(Some(n)) => {
|
||||
// n samples lus, entrelacer si nécessaire
|
||||
}
|
||||
Ok(None) => break, // EOF
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
}
|
||||
|
||||
// Option 2: Iterator (plus simple mais moins contrôle)
|
||||
for sample in reader.samples() {
|
||||
let s = sample?;
|
||||
// Traiter sample par sample
|
||||
}
|
||||
```
|
||||
|
||||
### Entrelacement des Samples
|
||||
|
||||
Claxon retourne les samples **par canal** :
|
||||
```
|
||||
Buffer claxon: [L0, L1, L2, ..., Ln, R0, R1, R2, ..., Rn]
|
||||
```
|
||||
|
||||
Il faut entrelacer pour PCM standard :
|
||||
```
|
||||
Buffer PCM: [L0, R0, L1, R1, L2, R2, ..., Ln, Rn]
|
||||
```
|
||||
|
||||
```rust
|
||||
fn interleave_samples(frame: &[i32], channels: usize) -> Vec<i32> {
|
||||
let samples_per_channel = frame.len() / channels;
|
||||
let mut interleaved = Vec::with_capacity(frame.len());
|
||||
|
||||
for i in 0..samples_per_channel {
|
||||
for ch in 0..channels {
|
||||
interleaved.push(frame[ch * samples_per_channel + i]);
|
||||
}
|
||||
}
|
||||
|
||||
interleaved
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Étape 6 : Gestion d'Erreurs
|
||||
|
||||
### Erreurs Potentielles
|
||||
|
||||
1. **Stream HTTP interrompu** : Gérer les EOF prématurés
|
||||
2. **Mauvais timing** : Vérifier que `song.elapsed` < durée totale
|
||||
3. **Corruption FLAC** : claxon peut échouer sur frames corrompues
|
||||
|
||||
### Pattern de Gestion
|
||||
|
||||
```rust
|
||||
match reader.read_next_or_eof(buffer) {
|
||||
Ok(Some(n)) => {
|
||||
// Traiter n samples
|
||||
}
|
||||
Ok(None) => {
|
||||
// EOF normal
|
||||
break;
|
||||
}
|
||||
Err(claxon::Error::FormatError(msg)) => {
|
||||
// Frame corrompue, continuer ou abandonner?
|
||||
warn!("FLAC format error: {}", msg);
|
||||
continue; // Ou break selon la criticité
|
||||
}
|
||||
Err(e) => {
|
||||
// Erreur fatale
|
||||
return Err(anyhow!("FLAC decode error: {}", e));
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Étape 7 : Tests Recommandés
|
||||
|
||||
### Test 1 : AsyncReadAdapter
|
||||
|
||||
```rust
|
||||
#[tokio::test]
|
||||
async fn test_async_read_adapter() {
|
||||
let data = vec![
|
||||
Ok(Bytes::from_static(b"Hello ")),
|
||||
Ok(Bytes::from_static(b"World")),
|
||||
];
|
||||
let stream = futures::stream::iter(data);
|
||||
|
||||
let mut adapter = AsyncReadAdapter::new(Box::pin(stream));
|
||||
let mut buf = [0u8; 11];
|
||||
let n = adapter.read(&mut buf).unwrap();
|
||||
|
||||
assert_eq!(n, 11);
|
||||
assert_eq!(&buf, b"Hello World");
|
||||
}
|
||||
```
|
||||
|
||||
### Test 2 : Décodage d'un Petit FLAC
|
||||
|
||||
Créer un fichier FLAC de test (1 morceau, 10 secondes) et vérifier :
|
||||
1. Le stream est lu progressivement
|
||||
2. Le morceau est correctement découpé
|
||||
3. Le FLAC réencodé est valide
|
||||
|
||||
### Test 3 : Integration Complète
|
||||
|
||||
1. Télécharger un vrai block Radio Paradise
|
||||
2. Chronométrer le temps jusqu'au premier morceau disponible
|
||||
3. Vérifier que les morceaux suivants arrivent bien
|
||||
|
||||
---
|
||||
|
||||
## Étape 8 : Optimisations Futures
|
||||
|
||||
### Buffer Size Tuning
|
||||
|
||||
```rust
|
||||
// Ajuster selon le réseau
|
||||
const STREAM_BUFFER_SIZE: usize = 64 * 1024; // 64 KB
|
||||
```
|
||||
|
||||
### Parallélisation
|
||||
|
||||
Une fois le premier morceau envoyé, les suivants peuvent être traités en parallèle :
|
||||
|
||||
```rust
|
||||
let mut tasks = Vec::new();
|
||||
for (track_pcm, song_index, song) in tracks {
|
||||
let task = tokio::spawn(async move {
|
||||
// Encoder + envoyer au cache
|
||||
});
|
||||
tasks.push(task);
|
||||
}
|
||||
|
||||
// Attendre tous en parallèle
|
||||
futures::future::join_all(tasks).await;
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Pièges Critiques à Éviter
|
||||
|
||||
### 1. Seek dans claxon
|
||||
**ERREUR** : claxon::FlacReader n'a **PAS** de méthode `seek()` !
|
||||
- Ne tentez pas `reader.seek_to(position)` (compile pas)
|
||||
- Le streaming est **séquentiel uniquement**
|
||||
|
||||
### 2. Thread Blocking
|
||||
**ERREUR** : Créer AsyncReadAdapter dans un contexte async
|
||||
```rust
|
||||
// ❌ MAUVAIS
|
||||
async fn foo() {
|
||||
let adapter = AsyncReadAdapter::new(stream); // Deadlock!
|
||||
}
|
||||
|
||||
// ✅ BON
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let adapter = AsyncReadAdapter::new(stream);
|
||||
// ...
|
||||
})
|
||||
```
|
||||
|
||||
### 3. Normalisation des Samples
|
||||
**ERREUR** : Ne pas normaliser selon bits_per_sample
|
||||
- claxon retourne des samples **natifs** (16-bit → i32 avec shift)
|
||||
- flacenc attend des samples dans la plage correcte
|
||||
- **Toujours normaliser** selon les bits réels
|
||||
|
||||
### 4. Accumulation Mémoire
|
||||
**ATTENTION** : `accumulated_samples` peut devenir ÉNORME (100 MB+)
|
||||
- **Solution** : Ne garder que le nécessaire, supprimer les samples déjà traités
|
||||
- Ou: Traiter morceau par morceau sans accumuler tout le block
|
||||
|
||||
---
|
||||
|
||||
## Mesures de Performance Attendues
|
||||
|
||||
### Avant (Architecture Actuelle)
|
||||
- Téléchargement block : 12-16 secondes (75-100 MB @ 50 Mbps)
|
||||
- Premier morceau disponible : **12-16 secondes**
|
||||
|
||||
### Après (Streaming Progressif)
|
||||
- Temps pour 1er morceau (3 min, ~30 MB) : **~6-8 secondes**
|
||||
- Amélioration : **2x plus rapide** ⚡
|
||||
|
||||
### Métriques à Surveiller
|
||||
1. Temps entre `get_block()` et premier `push_active()`
|
||||
2. Débit du stream HTTP (surveiller throttling)
|
||||
3. Utilisation mémoire de `accumulated_samples`
|
||||
|
||||
---
|
||||
|
||||
## Checklist d'Implémentation
|
||||
|
||||
- [ ] Créer `AsyncReadAdapter` avec tests unitaires
|
||||
- [ ] Remplacer `decode_block_audio()` par `decode_and_split_streaming()`
|
||||
- [ ] Adapter `process_block()` pour utiliser streaming
|
||||
- [ ] Créer `process_song_from_pcm()`
|
||||
- [ ] Tester avec un petit FLAC local
|
||||
- [ ] Tester avec un vrai block Radio Paradise
|
||||
- [ ] Mesurer les performances (avant/après)
|
||||
- [ ] Vérifier pas de régression sur la qualité audio
|
||||
- [ ] Vérifier pas de fuite mémoire
|
||||
- [ ] Ajouter logs de debug pour troubleshooting
|
||||
|
||||
---
|
||||
|
||||
## Ressources
|
||||
|
||||
- **claxon docs** : https://docs.rs/claxon/latest/claxon/
|
||||
- **Radio Paradise API** : https://api.radioparadise.com/api
|
||||
- **FLAC spec** : https://xiph.org/flac/format.html
|
||||
|
||||
---
|
||||
|
||||
Bon courage pour l'implémentation ! 🚀
|
||||
@@ -25,45 +25,6 @@ pub const DEFAULT_BLOCK_TIMEOUT_SECS: u64 = 180;
|
||||
pub const DEFAULT_USER_AGENT: &str = "pmoparadise/0.1.0";
|
||||
|
||||
|
||||
fn normalize_cover_base_url(base: &str) -> String {
|
||||
let mut normalized = base.trim().to_string();
|
||||
|
||||
if normalized.is_empty() {
|
||||
return DEFAULT_IMAGE_BASE.to_string();
|
||||
}
|
||||
|
||||
if normalized.starts_with("//") {
|
||||
normalized = format!("https:{}", normalized);
|
||||
} else if !(normalized.starts_with("http://") || normalized.starts_with("https://")) {
|
||||
normalized = format!("https://{}", normalized.trim_start_matches('/'));
|
||||
}
|
||||
|
||||
if !normalized.ends_with('/') {
|
||||
normalized.push('/');
|
||||
}
|
||||
|
||||
normalized
|
||||
}
|
||||
|
||||
fn resolve_cover_with_base(base: &str, cover_path: &str) -> Result<Url> {
|
||||
let cover_path = cover_path.trim();
|
||||
|
||||
if cover_path.starts_with("http://") || cover_path.starts_with("https://") {
|
||||
return Ok(Url::parse(cover_path)?);
|
||||
}
|
||||
|
||||
if cover_path.starts_with("//") {
|
||||
let url = format!("https:{}", cover_path);
|
||||
return Ok(Url::parse(&url)?);
|
||||
}
|
||||
|
||||
let base = normalize_cover_base_url(base);
|
||||
let base_url = Url::parse(&base)?;
|
||||
|
||||
Ok(base_url.join(cover_path)?)
|
||||
}
|
||||
|
||||
|
||||
/// Radio Paradise HTTP client
|
||||
///
|
||||
/// This client provides access to Radio Paradise's streaming API,
|
||||
@@ -89,7 +50,6 @@ pub struct RadioParadiseClient {
|
||||
pub(crate) client: Client,
|
||||
api_base: String,
|
||||
block_base: String,
|
||||
image_base: String,
|
||||
bitrate: Bitrate,
|
||||
channel: u8,
|
||||
pub(crate) request_timeout: Duration,
|
||||
@@ -118,7 +78,6 @@ impl RadioParadiseClient {
|
||||
client,
|
||||
api_base: DEFAULT_API_BASE.to_string(),
|
||||
block_base: DEFAULT_BLOCK_BASE.to_string(),
|
||||
image_base: normalize_cover_base_url(DEFAULT_IMAGE_BASE),
|
||||
bitrate: Bitrate::default(),
|
||||
channel: 0,
|
||||
request_timeout: Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS),
|
||||
@@ -158,13 +117,6 @@ impl RadioParadiseClient {
|
||||
cloned
|
||||
}
|
||||
|
||||
/// Clone the client with an updated channel and bitrate.
|
||||
pub fn clone_with_channel_and_bitrate(&self, channel: u8, bitrate: Bitrate) -> Self {
|
||||
let mut cloned = self.clone_with_channel(channel);
|
||||
cloned.bitrate = bitrate;
|
||||
cloned
|
||||
}
|
||||
|
||||
/// Get a block by event ID
|
||||
///
|
||||
/// If `event` is None, returns the current block.
|
||||
@@ -223,11 +175,14 @@ impl RadioParadiseClient {
|
||||
|
||||
let mut block: Block = response.json().await?;
|
||||
|
||||
// Set image_base if not provided
|
||||
if let Some(ref mut base) = block.image_base {
|
||||
*base = normalize_cover_base_url(base);
|
||||
// Normalize protocol-relative URLs from API (//img.radioparadise.com/)
|
||||
if let Some(ref base) = block.image_base {
|
||||
if base.starts_with("//") {
|
||||
block.image_base = Some(format!("https:{}", base));
|
||||
}
|
||||
} else {
|
||||
block.image_base = Some(self.image_base.clone());
|
||||
// Fallback if API doesn't provide image_base (should never happen)
|
||||
block.image_base = Some(DEFAULT_IMAGE_BASE.to_string());
|
||||
}
|
||||
|
||||
#[cfg(feature = "logging")]
|
||||
@@ -252,28 +207,6 @@ impl RadioParadiseClient {
|
||||
Ok(NowPlaying::from_block(block))
|
||||
}
|
||||
|
||||
/// Get the full URL for a cover image
|
||||
///
|
||||
/// # Arguments
|
||||
///
|
||||
/// * `cover_path` - The cover filename/path from song metadata
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
/// ```no_run
|
||||
/// # use pmoparadise::RadioParadiseClient;
|
||||
/// # #[tokio::main]
|
||||
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
/// let client = RadioParadiseClient::new().await?;
|
||||
/// let url = client.cover_url("B00000I0JF.jpg")?;
|
||||
/// println!("Cover URL: {}", url);
|
||||
/// # Ok(())
|
||||
/// # }
|
||||
/// ```
|
||||
pub fn cover_url(&self, cover_path: &str) -> Result<Url> {
|
||||
resolve_cover_with_base(&self.image_base, cover_path)
|
||||
}
|
||||
|
||||
/// Prefetch metadata for the next block
|
||||
///
|
||||
/// Stores the next block URL internally for seamless transitions.
|
||||
@@ -318,7 +251,6 @@ pub struct ClientBuilder {
|
||||
client: Option<Client>,
|
||||
api_base: String,
|
||||
block_base: String,
|
||||
image_base: String,
|
||||
bitrate: Bitrate,
|
||||
channel: u8,
|
||||
request_timeout: Duration,
|
||||
@@ -333,7 +265,6 @@ impl Default for ClientBuilder {
|
||||
client: None,
|
||||
api_base: DEFAULT_API_BASE.to_string(),
|
||||
block_base: DEFAULT_BLOCK_BASE.to_string(),
|
||||
image_base: DEFAULT_IMAGE_BASE.to_string(),
|
||||
bitrate: Bitrate::default(),
|
||||
channel: 0,
|
||||
request_timeout: Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS),
|
||||
@@ -368,12 +299,6 @@ impl ClientBuilder {
|
||||
self
|
||||
}
|
||||
|
||||
/// Set the image base URL
|
||||
pub fn image_base(mut self, url: impl Into<String>) -> Self {
|
||||
self.image_base = url.into();
|
||||
self
|
||||
}
|
||||
|
||||
/// Set the bitrate/quality level
|
||||
///
|
||||
/// # Example
|
||||
@@ -441,13 +366,11 @@ impl ClientBuilder {
|
||||
} else {
|
||||
self.block_base.clone()
|
||||
};
|
||||
let image_base = normalize_cover_base_url(&self.image_base);
|
||||
|
||||
Ok(RadioParadiseClient {
|
||||
client,
|
||||
api_base: self.api_base,
|
||||
block_base,
|
||||
image_base,
|
||||
bitrate: self.bitrate,
|
||||
channel: self.channel,
|
||||
request_timeout: self.request_timeout,
|
||||
@@ -469,13 +392,4 @@ mod tests {
|
||||
assert_eq!(builder.channel, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_cover_url() {
|
||||
let client = RadioParadiseClient::with_client(Client::new());
|
||||
let url = client.cover_url("test.jpg").unwrap();
|
||||
assert_eq!(
|
||||
url.as_str(),
|
||||
"https://img.radioparadise.com/covers/l/test.jpg"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -114,6 +114,11 @@ pub const ALL_CHANNELS: [ChannelDescriptor; 4] = [
|
||||
ChannelDescriptor::new(ParadiseChannelKind::Eclectic),
|
||||
];
|
||||
|
||||
/// Returns the maximum valid channel ID
|
||||
pub const fn max_channel_id() -> u8 {
|
||||
(ALL_CHANNELS.len() - 1) as u8
|
||||
}
|
||||
|
||||
/// Public handle to interact with a channel.
|
||||
#[derive(Clone)]
|
||||
pub struct ParadiseChannel {
|
||||
|
||||
@@ -19,7 +19,8 @@ mod playlist;
|
||||
mod worker;
|
||||
|
||||
pub use channel::{
|
||||
ChannelDescriptor, ParadiseChannel, ParadiseChannelKind, ParadiseClientStream, ALL_CHANNELS,
|
||||
max_channel_id, ChannelDescriptor, ParadiseChannel, ParadiseChannelKind, ParadiseClientStream,
|
||||
ALL_CHANNELS,
|
||||
};
|
||||
pub use config::{
|
||||
ActivityConfig, ApiConfig, CacheConfig, HistoryConfig, PollingConfig, RadioParadiseConfig,
|
||||
|
||||
@@ -452,14 +452,15 @@ impl WorkerState {
|
||||
|
||||
async fn cache_cover(&self, block: &Block, song: &Song) -> Result<Option<String>> {
|
||||
if let Some(ref cover_path) = song.cover {
|
||||
let cover_url =
|
||||
resolve_cover_url(block.image_base.as_deref(), &self.client, cover_path)
|
||||
.context("Invalid cover URL")?;
|
||||
match self.cache_manager.cache_cover(cover_url.as_str()).await {
|
||||
Ok(pk) => return Ok(Some(pk)),
|
||||
Err(err) => {
|
||||
warn!(channel = self.descriptor.slug, "Cover cache error: {err}");
|
||||
if let Some(cover_url) = block.cover_url(cover_path) {
|
||||
match self.cache_manager.cache_cover(&cover_url).await {
|
||||
Ok(pk) => return Ok(Some(pk)),
|
||||
Err(err) => {
|
||||
warn!(channel = self.descriptor.slug, "Cover cache error: {err}");
|
||||
}
|
||||
}
|
||||
} else {
|
||||
warn!(channel = self.descriptor.slug, "Unable to resolve cover URL for {}", cover_path);
|
||||
}
|
||||
}
|
||||
Ok(None)
|
||||
@@ -729,30 +730,3 @@ async fn encode_samples_to_flac(
|
||||
.await?
|
||||
}
|
||||
|
||||
fn resolve_cover_url(
|
||||
image_base: Option<&str>,
|
||||
client: &RadioParadiseClient,
|
||||
cover: &str,
|
||||
) -> Result<Url> {
|
||||
if cover.starts_with("http://") || cover.starts_with("https://") {
|
||||
return Url::parse(cover).map_err(|e| anyhow!("Invalid cover URL '{cover}': {e}"));
|
||||
}
|
||||
|
||||
if cover.starts_with("//") {
|
||||
let url = format!("https:{cover}");
|
||||
return Url::parse(&url).map_err(|e| anyhow!("Invalid cover URL '{cover}': {e}"));
|
||||
}
|
||||
|
||||
if let Some(base) = image_base {
|
||||
match Url::parse(base).and_then(|base_url| base_url.join(cover)) {
|
||||
Ok(url) => return Ok(url),
|
||||
Err(err) => {
|
||||
debug!("Failed to join cover '{cover}' with base '{base}': {err}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
client
|
||||
.cover_url(cover)
|
||||
.map_err(|e| anyhow!("Invalid cover URL '{cover}': {e}"))
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
//! Ce module fournit un trait d'extension pour ajouter facilement l'API Radio Paradise
|
||||
//! à un serveur pmoserver.
|
||||
|
||||
use crate::paradise::{ParadiseChannel, PlaylistEntry};
|
||||
use crate::paradise::{max_channel_id, ParadiseChannel, PlaylistEntry, ALL_CHANNELS};
|
||||
use crate::{models::Bitrate, Block, NowPlaying, RadioParadiseClient, RadioParadiseSource};
|
||||
use axum::{
|
||||
body::Body,
|
||||
@@ -30,8 +30,6 @@ pub struct RadioParadiseState {
|
||||
source: Arc<RadioParadiseSource>,
|
||||
}
|
||||
|
||||
const MAX_CHANNEL_ID: u8 = 3;
|
||||
|
||||
#[derive(Debug, Default, Deserialize)]
|
||||
#[serde(default)]
|
||||
struct ParadiseQuery {
|
||||
@@ -101,7 +99,7 @@ impl RadioParadiseState {
|
||||
let mut client = base_client;
|
||||
|
||||
if let Some(channel) = params.channel {
|
||||
if channel > MAX_CHANNEL_ID {
|
||||
if channel > max_channel_id() {
|
||||
tracing::warn!("Invalid Radio Paradise channel requested: {}", channel);
|
||||
return Err(StatusCode::BAD_REQUEST);
|
||||
}
|
||||
@@ -120,7 +118,7 @@ impl RadioParadiseState {
|
||||
}
|
||||
|
||||
fn channel_for_id(&self, channel_id: u8) -> Result<Arc<ParadiseChannel>, StatusCode> {
|
||||
if channel_id > MAX_CHANNEL_ID {
|
||||
if channel_id > max_channel_id() {
|
||||
return Err(StatusCode::BAD_REQUEST);
|
||||
}
|
||||
self.source
|
||||
@@ -144,6 +142,16 @@ pub struct ChannelInfo {
|
||||
pub description: String,
|
||||
}
|
||||
|
||||
impl From<&crate::paradise::ChannelDescriptor> for ChannelInfo {
|
||||
fn from(descriptor: &crate::paradise::ChannelDescriptor) -> Self {
|
||||
Self {
|
||||
id: descriptor.id,
|
||||
name: descriptor.display_name.to_string(),
|
||||
description: descriptor.description.to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Réponse avec informations étendues sur le morceau en cours
|
||||
#[derive(Debug, Clone, Serialize, ToSchema)]
|
||||
pub struct NowPlayingResponse {
|
||||
@@ -372,29 +380,7 @@ async fn get_block_by_id(
|
||||
tag = "Radio Paradise"
|
||||
)]
|
||||
async fn get_channels() -> Json<Vec<ChannelInfo>> {
|
||||
let channels = vec![
|
||||
ChannelInfo {
|
||||
id: 0,
|
||||
name: "Main Mix".to_string(),
|
||||
description: "Eclectic mix of rock, world, electronica, and more".to_string(),
|
||||
},
|
||||
ChannelInfo {
|
||||
id: 1,
|
||||
name: "Mellow Mix".to_string(),
|
||||
description: "Mellower, less aggressive music".to_string(),
|
||||
},
|
||||
ChannelInfo {
|
||||
id: 2,
|
||||
name: "Rock Mix".to_string(),
|
||||
description: "Heavier, more guitar-driven music".to_string(),
|
||||
},
|
||||
ChannelInfo {
|
||||
id: 3,
|
||||
name: "World/Etc Mix".to_string(),
|
||||
description: "Global beats and world music".to_string(),
|
||||
},
|
||||
];
|
||||
|
||||
let channels: Vec<ChannelInfo> = ALL_CHANNELS.iter().map(Into::into).collect();
|
||||
Json(channels)
|
||||
}
|
||||
|
||||
|
||||
@@ -75,7 +75,7 @@ impl std::fmt::Debug for RadioParadiseSource {
|
||||
|
||||
impl RadioParadiseSource {
|
||||
#[cfg(feature = "server")]
|
||||
pub fn from_registry(client: RadioParadiseClient, _legacy_capacity: usize) -> Result<Self> {
|
||||
pub fn from_registry(client: RadioParadiseClient) -> Result<Self> {
|
||||
let config = Arc::new(RadioParadiseConfig::load_from_pmoconfig().unwrap_or_default());
|
||||
let history_backend = history_backend_from_config(&config.history).map_err(|e| {
|
||||
MusicSourceError::SourceUnavailable(format!(
|
||||
@@ -114,12 +114,11 @@ impl RadioParadiseSource {
|
||||
|
||||
#[cfg(feature = "server")]
|
||||
pub fn from_registry_default(client: RadioParadiseClient) -> Result<Self> {
|
||||
Self::from_registry(client, 0)
|
||||
Self::from_registry(client)
|
||||
}
|
||||
|
||||
pub fn new(
|
||||
client: RadioParadiseClient,
|
||||
_legacy_capacity: usize,
|
||||
cover_cache: Arc<CoverCache>,
|
||||
audio_cache: Arc<AudioCache>,
|
||||
) -> Self {
|
||||
@@ -166,7 +165,7 @@ impl RadioParadiseSource {
|
||||
cover_cache: Arc<CoverCache>,
|
||||
audio_cache: Arc<AudioCache>,
|
||||
) -> Self {
|
||||
Self::new(client, 0, cover_cache, audio_cache)
|
||||
Self::new(client, cover_cache, audio_cache)
|
||||
}
|
||||
|
||||
pub fn client_for_channel(&self, channel: u8) -> Option<RadioParadiseClient> {
|
||||
|
||||
@@ -4,7 +4,7 @@ use crate::error::{Error, Result};
|
||||
use crate::models::Block;
|
||||
use crate::RadioParadiseClient;
|
||||
use bytes::Bytes;
|
||||
use futures::stream::Stream;
|
||||
use futures::stream::{Stream, StreamExt};
|
||||
use std::pin::Pin;
|
||||
use std::task::{Context, Poll};
|
||||
use url::Url;
|
||||
@@ -121,12 +121,14 @@ impl RadioParadiseClient {
|
||||
self.stream_block(&url).await
|
||||
}
|
||||
|
||||
/// Download a complete block to memory
|
||||
/// Download an entire block as Bytes
|
||||
///
|
||||
/// **Warning**: Blocks can be large (50-100MB for FLAC). Use streaming
|
||||
/// for playback instead of downloading the entire block to memory.
|
||||
/// This downloads the complete block file into memory. For streaming playback,
|
||||
/// use `stream_block()` instead which is more memory efficient.
|
||||
///
|
||||
/// This is useful for the per-track feature which needs random access.
|
||||
/// # Arguments
|
||||
///
|
||||
/// * `block_url` - The URL of the block to download
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
@@ -137,38 +139,24 @@ impl RadioParadiseClient {
|
||||
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
/// let client = RadioParadiseClient::new().await?;
|
||||
/// let block = client.get_block(None).await?;
|
||||
///
|
||||
/// let data = client.download_block(&block.url.parse()?).await?;
|
||||
/// println!("Downloaded {} bytes", data.len());
|
||||
///
|
||||
/// let url = block.url.parse()?;
|
||||
/// let bytes = client.download_block(&url).await?;
|
||||
/// println!("Downloaded {} bytes", bytes.len());
|
||||
/// Ok(())
|
||||
/// }
|
||||
/// ```
|
||||
pub async fn download_block(&self, block_url: &Url) -> Result<Bytes> {
|
||||
#[cfg(feature = "logging")]
|
||||
tracing::debug!("Downloading complete block: {}", block_url);
|
||||
let mut stream = self.stream_block(block_url).await?;
|
||||
let mut data = Vec::new();
|
||||
|
||||
let response = self
|
||||
.client
|
||||
.get(block_url.clone())
|
||||
.timeout(self.block_timeout)
|
||||
.send()
|
||||
.await?;
|
||||
|
||||
if !response.status().is_success() {
|
||||
return Err(Error::other(format!(
|
||||
"Failed to download block: HTTP {}",
|
||||
response.status()
|
||||
)));
|
||||
while let Some(chunk_result) = stream.next().await {
|
||||
let chunk = chunk_result?;
|
||||
data.extend_from_slice(&chunk);
|
||||
}
|
||||
|
||||
let bytes = response.bytes().await?;
|
||||
|
||||
#[cfg(feature = "logging")]
|
||||
tracing::debug!("Downloaded {} bytes", bytes.len());
|
||||
|
||||
Ok(bytes)
|
||||
Ok(Bytes::from(data))
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
Reference in New Issue
Block a user