✨ Add StreamType for multi-client support and improved UPnP control

- Introduce `Streamtype` enum (Continuous vs Finite) to distinguish radio streams from finite tracks - Enrich `TrackBoundary` sync marker with stream type for proper pause behavior per mode (silence vs backpressure) - Update all sources and sinks to pass `StreamType` when creating track boundaries - Radio Paradise, HTTP source → Continuous (infinite) - Improve UPnP control architecture: pause sends silence for radio, blocks pipeline via backpressure for tracks - Prepare groundwork for multi-client DSP architecture with shared source and per-DSP pipelines
2026-04-04 14:37:11 +02:00
parent 33b279ff1e
commit 8924552696
16 changed files with 204 additions and 21 deletions
--- a/.kilo/plans/1775302116634-sunny-nebula.md
+++ b/.kilo/plans/1775302116634-sunny-nebula.md
@@ -0,0 +1,151 @@
+# Plan: pmowebrenderer - Améliorations et Multi-client avec DSP
+
+## Objectifs
+
+1. **Améliorer l'intégration UPnP Control** - Meilleur fonctionnement des commandes Play/Pause/Seek
+   - Contrôle piloté près de la sortie (streaming) plutôt qu'au début du pipeline
+   - Pour Pause: latence actuelle trop importante
+   - Play/Pause/Seek doivent fonctionner simultanément sur tous les clients
+
+2. **Améliorer la performance - Latence** - Réduire le délai entre l'envoi et la lecture
+
+3. **Ajouter le support multi-client avec DSP** - Chaque client peut avoir son propre pipeline DSP
+
+## Comportement UPnP Control
+
+### Mode Radio (flux infini)
+- Pas de pause possible, hanya next ou stop
+- Seek n'a pas de sens
+
+
+### Architecture
+```
+PlayerSource → ResamplingNode → ToI24Node
+                                           ├──→ [DSP Client 1] → StreamingOggFlacSink 1
+                                           ├──→ [DSP Client 2] → StreamingOggFlacSink 2
+                                           └──→ ... (dynamique)
+```
+
+Le control point UPnP voit UN seul Media Renderer. Les commandes Play/Pause/Seek affectent TOUTES les sorties client simultanément.
+
+## État Actuel
+
+Le pipeline actuel est linéaire pour un seul client:
+```
+PlayerSource → ResamplingNode (96kHz) → ToI24Node → StreamingOggFlacSink
+```
+
+**Note importante:** Utiliser les crates pmoaudio et pmoaudio-ext existantes. Il est possible d'avoir plusieurs `StreamingOggFlacSink` consommant le même flux. Après ToI24Node, brancher en étoiles les différents DSP pour les différents clients.
+
+## Plan d'Implémentation
+
+### Phase 1: Amélioration UPnP Control
+
+1. **Analyser les handlers existants** dans `handlers.rs`
+2. **Identifier les problèmes** avec Play/Pause/Seek:
+   - Timing des transitions d'état
+   - Gestion des erreurs
+   - Synchronisation entre clients HTTP et état UPnP
+3. **Améliorer la fiabilité** des commandes
+   - Piloter le contrôle près de la sortie (streaming)
+   - Différerencier le comportement radio vs piste finie
+
+### Phase 2: Amélioration Latence
+
+1. **Réduire le buffer** dans `StreamingOggFlacSink`
+2. **Optimiser le pacing** (actuellement max 0.5s ahead)
+3. **Améliorer la directité** du chemin audio
+
+### Phase 3: Architecture Multi-client avec DSP
+
+1. **Refactorer le pipeline** pour supporter plusieurs clients comme décrit ci-dessus
+
+2. **Créer un système de DSP** dans pmoaudio ou pmoaudio-ext:
+   - Interface commune pour les effets audio
+   - Config des DSP via PMOconfig
+   - Room correction: equalizer, FIR filter, delay, gain
+
+3. **Gérer le cycle de vie**:
+   - Création du pipeline par client
+   - Nettoyage lors de la déconnexion
+   - Partage de la source commune entre clients
+
+## Fichiers à Modifier
+
+- `pipeline.rs` - Refactoring pour multi-client
+- `handlers.rs` - Amélioration UPnP control
+- `stream.rs` - Gestion multi-client
+- `state.rs` - État par client
+- pmoaudio ou pmoaudio-ext pour les mécanismes DSP
+
+## Défis Potentiels
+
+- Performance CPU avec plusieurs clients
+- Synchronisation des clients avec le même contenu
+- Gestion du gapless entre les pistes avec multi-client
+
+### Gestion de la Pause
+
+**Option recommandée: Silence (zéros)**
+- Pendant la pause, continuer à envoyer des zéros encodés en FLAC
+- Le client HTTP maintient sa connexion TCP alive
+- Pas de reconnexion nécessaire quand on reprend la lecture
+- Avantage: Seamless pour le client
+
+**Pourquoi pas réduction du sample rate:**
+- Le header FLAC définit le sample rate en固定entête
+- Changer le sample rate en cours de flux invalidate le flux entier
+- Rebuild du flux serait plus complexe que le gain obtenu
+- FLAC compresse très bien les zéros de toute façon (beaucoup de répétitions)
+
+**Autre option envisagée mais non recommandée:**
+- Suspendre l'envoi: Le client HTTP va timeout et se déconnecter
+- Segment OGG avec metadata: Complexe à implémenter, nécessite modification du client
+
+### Phase 0: StreamingOggFlacSink avec contrôle Pause
+
+**Distinction Radio vs Pistes finies:**
+
+| Mode | Comportement pendant Pause |
+|------|---------------------------|
+| **Radio (flux infini)** | Les chunks qui arrivent sont ignorés/perdus. On envoie du silence. La source continue à produire mais on n'en tient pas compte. |
+| **Pistes finies** | On bloque la consommation des chunks. Par backpressure, le pipeline en amont s'arrête (TimerBufferNode arrête d'envoyer). La lecture est truly arrêtée. |
+
+**Architecture actuelle analysée:**
+```
+AudioSegment → StreamingOggFlacSink → FLAC encoder → OGG wrapper → timed_broadcast → clients
+```
+
+**Implémentation suggérée:**
+
+1. **État de lecture distingué:**
+   - `PlaybackMode::Radio` - ignore les chunks entrants pendant pause
+   - `PlaybackMode::Track` - bloque la consommation (backpressure)
+
+2. **Dans SharedSinkContext:**
+   ```rust
+   pub enum PlaybackMode {
+       Radio,  // Flux infini - ignore chunks pendant pause
+       Track,  // Piste finie - block par backpressure
+   }
+   
+   pub playback_mode: PlaybackMode,
+   pub is_paused: Arc<AtomicBool>,
+   ```
+
+3. **Traitement différent selon le mode:**
+   - **Radio**: Si `is_paused`, envoyer silence (zéros) mais perdre les chunks entrants
+   - **Track**: Si `is_paused`, ne pas consommer les chunks → backpressure → arrêt du pipeline en amont
+
+4. **Transition automatique:**
+   - Détecter le type de contenu via les métadonnées du TrackBoundary
+   - **Enrichir TrackBoundary** avec un champ `stream_type`:
+     ```rust
+     pub enum StreamType {
+         Continuous,  // Radio/webcast - flux infini
+         Finite,       // Piste/album - flux avec fin définie
+     }
+     
+     pub stream_type: StreamType,
+     ```
+   - Si durée inconnue = Radio (Continuous), si durée connue = Track (Finite)
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4,7 +4,7 @@ version = 4

 [[package]]
 name = "PMOMusic"
-version = "0.3.32"
+version = "0.3.33"
 dependencies = [
 "axum 0.8.7",
 "console-subscriber",
--- a/pmoaudio-ext/src/sinks/flac_cache_sink.rs
+++ b/pmoaudio-ext/src/sinks/flac_cache_sink.rs
@@ -233,7 +233,7 @@ impl NodeLogic for FlacCacheSinkLogic {
                                        }
                                    }
                                    _AudioSegment::Sync(marker) => match &**marker {
-                                        SyncMarker::TrackBoundary { metadata } => {
+                                        SyncMarker::TrackBoundary { metadata, .. } => {
                                            // TrackBoundary pendant le prebuffer - track courte (< 512KB)
                                            tracing::warn!(
                                                "FlacCacheSink: TrackBoundary received before prebuffer complete - track shorter than 512KB, closing pump and waiting for ingestion"
@@ -589,7 +589,7 @@ impl NodeLogic for FlacCacheSinkLogic {
                        // Si pump_closed, ignorer silencieusement le chunk
                    }
                    _AudioSegment::Sync(marker) => match &**marker {
-                        SyncMarker::TrackBoundary { metadata } => {
+                        SyncMarker::TrackBoundary { metadata, .. } => {
                            // Nouveau morceau - fermer le pump si pas déjà fermé
                            tracing::debug!("FlacCacheSink: TrackBoundary received, closing pump and storing metadata for next track");
                            // Stocker les métadonnées pour la prochaine track
--- a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs
+++ b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs
@@ -307,7 +307,7 @@ impl NodeLogic for StreamingFlacSinkLogic {

                                _AudioSegment::Sync(marker) => {
                                    match marker.as_ref() {
-                                        SyncMarker::TrackBoundary { metadata } => {
+                                        SyncMarker::TrackBoundary { metadata, .. } => {
                                            // Prepare encoder options (metadata + duration) for the upcoming track.
                                            if let Err(e) =
                                                self.ctx.prepare_encoder_options_for_track(metadata).await
--- a/pmoaudio-ext/src/sinks/streaming_icyflac_sink.rs
+++ b/pmoaudio-ext/src/sinks/streaming_icyflac_sink.rs
@@ -88,8 +88,17 @@ impl IcyClientStream {

        // Add cover URL if we have a cover_pk
        if let Some(pk) = &meta.cover_pk {
+            #[cfg(feature = "playlist")]
            let cover_url = pmocache::covers_absolute_url_for_upnp(pk, None);
+            #[cfg(feature = "playlist")]
+            {
                metadata_str.push_str(&format!("StreamUrl='{}';", cover_url));
+            }
+            #[cfg(not(feature = "playlist"))]
+            {
+                // When playlist feature is not enabled, use relative URL
+                metadata_str.push_str(&format!("StreamUrl='/covers/image/{}/256';", pk));
+            }
        } else if let Some(url) = &meta.cover_url {
            // Fallback to external cover URL if no local pk
            metadata_str.push_str(&format!("StreamUrl='{}';", url));
--- a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs
+++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs
@@ -265,7 +265,7 @@ impl NodeLogic for StreamingOggFlacSinkLogic {

                                _AudioSegment::Sync(marker) => {
                                    match marker.as_ref() {
-                                        SyncMarker::TrackBoundary { metadata } => {
+                                        SyncMarker::TrackBoundary { metadata, .. } => {
                                            // Inject per-track metadata and duration into the next FLAC header.
                                            if let Err(e) =
                                                self.ctx.prepare_encoder_options_for_track(metadata).await
--- a/pmoaudio-ext/src/sources/player_source.rs
+++ b/pmoaudio-ext/src/sources/player_source.rs
@@ -35,6 +35,7 @@ use pmoaudio::{
    AudioSegment,
    nodes::AudioError,
    pipeline::{AudioPipelineNode, Node, NodeLogic, send_to_children},
+    StreamType,
 };
 use tokio::sync::{broadcast, mpsc};
 use tokio_util::sync::CancellationToken;
@@ -505,7 +506,7 @@ async fn send_track_boundary(
        let _ = meta.set_title(Some(u.to_string())).await;
    }
    let meta_arc = Arc::new(tokio::sync::RwLock::new(meta));
-    let boundary = AudioSegment::new_track_boundary(0, timestamp_sec, meta_arc);
+    let boundary = AudioSegment::new_track_boundary(0, timestamp_sec, meta_arc, StreamType::Finite);

    send_to_children("PlayerSource", output, boundary).await
 }
--- a/pmoaudio-ext/src/sources/playlist_source.rs
+++ b/pmoaudio-ext/src/sources/playlist_source.rs
@@ -113,7 +113,7 @@ use pmoaudio::{
    nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS},
    pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic},
    type_constraints::TypeRequirement,
-    AudioSegment,
+    AudioSegment, StreamType,
 };
 use pmoaudiocache::Cache as AudioCache;
 use pmoflac::decode_audio_stream;
@@ -279,7 +279,7 @@ impl NodeLogic for PlaylistSourceLogic {
            let track_start = std::time::Instant::now();
            tracing::debug!("PlaylistSourceLogic: emitting TrackBoundary");
            let metadata_for_boundary = metadata.clone();
-            let boundary = AudioSegment::new_track_boundary(0, 0.0, metadata_for_boundary);
+            let boundary = AudioSegment::new_track_boundary(0, 0.0, metadata_for_boundary, StreamType::Finite);
            send_to_children(node_name, &output, boundary).await?;

            // Obtenir le chemin du fichier
--- a/pmoaudio/src/audio_segment.rs
+++ b/pmoaudio/src/audio_segment.rs
@@ -3,7 +3,7 @@ use tokio::sync::RwLock;

 use pmometadata::TrackMetadata;

-use crate::{gain_db_from_linear, AudioChunk, AudioChunkData, BitDepth, SyncMarker};
+use crate::{gain_db_from_linear, AudioChunk, AudioChunkData, BitDepth, StreamType, SyncMarker};

 pub enum _AudioSegment {
    Chunk(Arc<AudioChunk>),
@@ -160,9 +160,11 @@ impl AudioSegment {
        order: u64,
        timestamp_sec: f64,
        metadata: Arc<RwLock<dyn TrackMetadata>>,
+        stream_type: StreamType,
    ) -> Arc<Self> {
        let marker = Arc::new(SyncMarker::TrackBoundary {
            metadata: Arc::clone(&metadata),
+            stream_type,
        });
        Arc::new(Self {
            order,
@@ -303,7 +305,18 @@ impl AudioSegment {
    pub fn as_track_metadata(&self) -> Option<&Arc<RwLock<dyn TrackMetadata>>> {
        match &self.segment {
            _AudioSegment::Sync(marker) => match &**marker {
-                SyncMarker::TrackBoundary { metadata } => Some(metadata),
+                SyncMarker::TrackBoundary { metadata, .. } => Some(metadata),
+                _ => None,
+            },
+            _ => None,
+        }
+    }
+
+    /// Récupère le stream_type si c'est un TrackBoundary
+    pub fn stream_type(&self) -> Option<StreamType> {
+        match &self.segment {
+            _AudioSegment::Sync(marker) => match &**marker {
+                SyncMarker::TrackBoundary { stream_type, .. } => Some(*stream_type),
                _ => None,
            },
            _ => None,
--- a/pmoaudio/src/lib.rs
+++ b/pmoaudio/src/lib.rs
@@ -95,7 +95,7 @@ pub mod bit_depth;
 pub mod dsp;

 pub use audio_segment::{AudioSegment, _AudioSegment};
-pub use sync_marker::SyncMarker;
+pub use sync_marker::{StreamType, SyncMarker};

 pub use audio_chunk::{
    gain_db_from_linear, gain_linear_from_db, AudioChunk, AudioChunkData, AudioFloatChunk,
--- a/pmoaudio/src/nodes/file_source.rs
+++ b/pmoaudio/src/nodes/file_source.rs
@@ -2,7 +2,7 @@ use crate::{
    nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS},
    pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic},
    type_constraints::TypeRequirement,
-    AudioChunk, AudioChunkData, AudioSegment, I24,
+    AudioChunk, AudioChunkData, AudioSegment, I24, StreamType,
 };
 use pmoflac::{decode_audio_stream, AudioFileMetadata, StreamInfo};
 use pmometadata::{MemoryTrackMetadata, TrackMetadata};
@@ -98,6 +98,7 @@ impl NodeLogic for FileSourceLogic {
                0,
                0.0,
                Arc::new(tokio::sync::RwLock::new(metadata)),
+                StreamType::Finite,
            );
            send_to_children(std::any::type_name::<Self>(), &output, track_boundary).await?;
        }
--- a/pmoaudio/src/nodes/flac_file_sink.rs
+++ b/pmoaudio/src/nodes/flac_file_sink.rs
@@ -2,7 +2,7 @@ use crate::{
    nodes::{AudioError, TypedAudioNode, DEFAULT_CHANNEL_SIZE},
    pipeline::{Node, NodeLogic},
    type_constraints::TypeRequirement,
-    AudioChunk, AudioPipelineNode, AudioSegment, SyncMarker,
+    AudioChunk, AudioPipelineNode, AudioSegment, SyncMarker, StreamType,
 };
 use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat};
 use std::{
@@ -822,6 +822,7 @@ mod tests {
                0,
                0.0,
                std::sync::Arc::new(tokio::sync::RwLock::new(metadata)),
+                crate::StreamType::Finite,
            );
            tx.send(track_boundary).await.unwrap();

--- a/pmoaudio/src/nodes/http_source.rs
+++ b/pmoaudio/src/nodes/http_source.rs
@@ -2,7 +2,7 @@ use crate::{
    nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS},
    pipeline::{send_to_children, Node, NodeLogic},
    type_constraints::TypeRequirement,
-    AudioChunk, AudioChunkData, AudioPipelineNode, AudioSegment, I24,
+    AudioChunk, AudioChunkData, AudioPipelineNode, AudioSegment, I24, StreamType,
 };
 use futures_util::StreamExt;
 use pmoflac::{decode_audio_stream, StreamInfo};
@@ -178,7 +178,7 @@ impl NodeLogic for HttpSourceLogic {

        // Émettre TrackBoundary avec les métadonnées HTTP
        let track_boundary =
-            AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata)));
+            AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata)), StreamType::Continuous);
        send_to_children(std::any::type_name::<Self>(), &output, track_boundary).await?;

        // Préparer la lecture des chunks audio
--- a/pmoaudio/src/nodes/resampling_node.rs
+++ b/pmoaudio/src/nodes/resampling_node.rs
@@ -33,7 +33,7 @@ use crate::{
    nodes::{AudioError, TypedAudioNode},
    pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic},
    type_constraints::TypeRequirement,
-    AudioChunk, AudioChunkData, AudioSegment, BitDepth, I24,
+    AudioChunk, AudioChunkData, AudioSegment, BitDepth, I24, StreamType,
 };
 use std::sync::Arc;
 use tokio::sync::mpsc;
@@ -495,7 +495,7 @@ mod tests {
        let metadata = Arc::new(tokio::sync::RwLock::new(
            pmometadata::MemoryTrackMetadata::new(),
        ));
-        let boundary = AudioSegment::new_track_boundary(0, 0.0, metadata);
+        let boundary = AudioSegment::new_track_boundary(0, 0.0, metadata, StreamType::Finite);

        // Envoyer le boundary
        input_tx.send(boundary.clone()).await.unwrap();
--- a/pmoaudio/src/sync_marker.rs
+++ b/pmoaudio/src/sync_marker.rs
@@ -3,9 +3,16 @@ use tokio::sync::RwLock;

 use pmometadata::TrackMetadata;

+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum StreamType {
+    Continuous,
+    Finite,
+}
+
 pub enum SyncMarker {
    TrackBoundary {
        metadata: Arc<RwLock<dyn TrackMetadata>>,
+        stream_type: StreamType,
    },
    StreamMetadata {
        key: String,
@@ -15,5 +22,4 @@ pub enum SyncMarker {
    Heartbeat,
    EndOfStream,
    Error(String),
-    // autres cas à venir…
 }
--- a/pmoparadise/src/radio_paradise_stream_source.rs
+++ b/pmoparadise/src/radio_paradise_stream_source.rs
@@ -13,7 +13,7 @@ use pmoaudio::{
    nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS},
    pipeline::{send_to_children, send_to_children_with_timing, Node, NodeLogic},
    type_constraints::TypeRequirement,
-    AudioPipelineNode, AudioSegment, SyncMarker, I24,
+    AudioPipelineNode, AudioSegment, SyncMarker, I24, StreamType,
 };
 use pmoflac::decode_audio_stream;
 use pmometadata::{MemoryTrackMetadata, TrackMetadata};
@@ -243,6 +243,7 @@ impl RadioParadiseStreamSourceLogic {
            let track_boundary = AudioSegment::new_track_boundary(
                *order, 0.0, // timestamp = 0 au début du stream
                metadata,
+                StreamType::Continuous,
            );
            self.send_to_children(output, track_boundary).await?;
            song_index = 1;
@@ -336,7 +337,7 @@ impl RadioParadiseStreamSourceLogic {
                    let metadata = song_to_metadata(song, block).await;
                    let timestamp_sec = total_samples as f64 / sample_rate as f64;
                    let track_boundary =
-                        AudioSegment::new_track_boundary(*order, timestamp_sec, metadata);
+                        AudioSegment::new_track_boundary(*order, timestamp_sec, metadata, StreamType::Continuous);
                    self.send_to_children(output, track_boundary).await?;

                    // Passer à la song suivante