From e33484b80279003467d82c513830d2f0933833bf Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sun, 1 Mar 2026 22:46:00 +0100 Subject: [PATCH] Refactor OGG-FLAC streaming to support multi-client broadcast Migrate from single-client direct OGG-FLAC sink to multi-client broadcast streaming sink. - Replace DirectOggFlacHandle with OggFlacStreamHandle for multi-client support - Update sink implementation to use StreamingOggFlacSink with proper backpressure - Change connection model from 'connect()' to 'subscribe()' - Adjust stream handling to support independent client streams - Update stream endpoint to always respond with 200 chunked instead of range requests - Add tracing for stream lifecycle events - Reduce OGG channel capacity to strict backpressure (1) - Add Drop implementation for stream cleanup - Improve logging and error handling for concurrent access This change enables multiple simultaneous clients to connect to the same audio stream without interfering with each other, while maintaining proper TCP backpressure and stream lifecycle management. --- .../src/sinks/direct_ogg_flac_sink.rs | 21 +++++++++++---- pmowebrenderer/src/pipeline.rs | 17 ++++++------ pmowebrenderer/src/registry.rs | 12 ++++----- pmowebrenderer/src/stream.rs | 27 +++++++------------ 4 files changed, 39 insertions(+), 38 deletions(-) diff --git a/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs index 92b4409b..efef08fd 100644 --- a/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs @@ -50,9 +50,9 @@ pub const DIRECT_OGG_FLAC_SAMPLE_RATE: u32 = 96_000; pub const DIRECT_OGG_FLAC_CHANNELS: u8 = 2; pub const DIRECT_OGG_FLAC_BITS_PER_SAMPLE: u8 = 24; -/// Capacité du canal OGG → HTTP (en chunks de ~8 KB). -/// Fournit un petit buffer sans casser la backpressure TCP. -const OGG_CHANNEL_CAPACITY: usize = 8; +/// Capacité du canal OGG → HTTP. +/// Capacité 1 : backpressure stricte, l'encodeur ne produit pas en avance. +const OGG_CHANNEL_CAPACITY: usize = 1; // ─── Types partagés ─────────────────────────────────────────────────────────── @@ -79,7 +79,7 @@ impl DirectOggFlacHandle { /// Retourne le stream OGG-FLAC pour le handler HTTP. /// Lance le premier encodeur au premier appel. pub fn get_stream(&self) -> DirectOggFlacStream { - debug!("DirectOggFlacHandle::get_stream()"); + tracing::info!(target: "pmoaudio_ext::stream_connect", "get_stream() called — new HTTP client connecting"); self.client_notify.notify_one(); DirectOggFlacStream { ogg_rx: self.ogg_rx.clone(), @@ -100,6 +100,12 @@ pub struct DirectOggFlacStream { buffer: VecDeque, } +impl Drop for DirectOggFlacStream { + fn drop(&mut self) { + tracing::info!(target: "pmoaudio_ext::stream_connect", "DirectOggFlacStream dropped — HTTP client disconnected"); + } +} + impl AsyncRead for DirectOggFlacStream { fn poll_read( mut self: Pin<&mut Self>, @@ -118,6 +124,7 @@ impl AsyncRead for DirectOggFlacStream { let mut guard = match self.ogg_rx.try_lock() { Ok(g) => g, Err(_) => { + tracing::warn!(target: "pmoaudio_ext::stream_connect", "DirectOggFlacStream: try_lock() FAILED — concurrent access detected!"); cx.waker().wake_by_ref(); return Poll::Pending; } @@ -126,6 +133,7 @@ impl AsyncRead for DirectOggFlacStream { match guard.poll_recv(cx) { Poll::Ready(Some(bytes)) => { drop(guard); + tracing::info!(target: "pmoaudio_ext::stream_connect", "DirectOggFlacStream: sending {} bytes to HTTP client", bytes.len()); let to_copy = bytes.len().min(buf.remaining()); buf.put_slice(&bytes[..to_copy]); if to_copy < bytes.len() { @@ -133,7 +141,10 @@ impl AsyncRead for DirectOggFlacStream { } Poll::Ready(Ok(())) } - Poll::Ready(None) => Poll::Ready(Ok(())), // EOF + Poll::Ready(None) => { + tracing::info!(target: "pmoaudio_ext::stream_connect", "DirectOggFlacStream: ogg_rx closed → EOF sent to HTTP client"); + Poll::Ready(Ok(())) + } Poll::Pending => Poll::Pending, } } diff --git a/pmowebrenderer/src/pipeline.rs b/pmowebrenderer/src/pipeline.rs index d8f36f61..37b27dfc 100644 --- a/pmowebrenderer/src/pipeline.rs +++ b/pmowebrenderer/src/pipeline.rs @@ -10,8 +10,7 @@ use std::sync::Arc; use pmoaudio::{AudioSegment, ResamplingNode, ToI24Node}; use pmometadata::{MemoryTrackMetadata, TrackMetadata}; use pmoaudio_ext::sinks::{ - DirectOggFlacHandle, DirectOggFlacSink, - DIRECT_OGG_FLAC_SAMPLE_RATE, + OggFlacStreamHandle, StreamingOggFlacSink, }; use pmoaudio_ext::UriSource; use pmoflac::EncoderOptions; @@ -59,10 +58,10 @@ impl PipelineHandle { /// Pipeline audio complet pour une instance WebRenderer. /// /// Créé au `POST /register`. Le flux OGG-FLAC est accessible via `flac_handle` -/// (mono-client, chaque `connect()` crée un nouveau flux avec backpressure TCP). +/// (multi-client broadcast, chaque `subscribe()` crée un flux indépendant). pub struct InstancePipeline { - /// Handle vers le sink OGG-FLAC — clonable, connect() crée un nouveau flux. - pub flac_handle: DirectOggFlacHandle, + /// Handle vers le sink OGG-FLAC — clonable, subscribe() crée un flux indépendant par client. + pub flac_handle: OggFlacStreamHandle, pub pipeline_handle: PipelineHandle, } @@ -78,18 +77,18 @@ impl InstancePipeline { let stop_token = CancellationToken::new(); let (control_tx, control_rx) = mpsc::channel::(32); - // Chaîne de traitement : ResamplingNode(96kHz) → ToI24Node → DirectFlacSink - // La backpressure remonte naturellement depuis le pipe duplex jusqu'à la source. + // Chaîne de traitement : ResamplingNode(96kHz) → ToI24Node → StreamingOggFlacSink + // Le broadcast pacé à 0.5s max d'avance, chaque subscribe() est indépendant. use pmoaudio::pipeline::AudioPipelineNode; - let (sink, flac_handle) = DirectOggFlacSink::new(EncoderOptions::default()); + let (sink, flac_handle) = StreamingOggFlacSink::new(EncoderOptions::default(), 24); // Nœud de conversion de profondeur : tout type entier → I24 let mut to_i24 = ToI24Node::new(); to_i24.register(sink.boxed()); // Nœud de rééchantillonnage : n'importe quel sample rate → 96 kHz - let mut resampler = ResamplingNode::new(DIRECT_OGG_FLAC_SAMPLE_RATE); + let mut resampler = ResamplingNode::new(96_000); resampler.register(to_i24.boxed()); // Le tx d'entrée du resampler est le point d'entrée du pipeline diff --git a/pmowebrenderer/src/registry.rs b/pmowebrenderer/src/registry.rs index 7cb187b7..cbc4b6be 100644 --- a/pmowebrenderer/src/registry.rs +++ b/pmowebrenderer/src/registry.rs @@ -29,8 +29,8 @@ pub struct WebRendererInstance { pub udn: String, pub device_instance: Arc, pub state: SharedState, - /// Handle vers le sink OGG-FLAC — clonable, connect() crée un nouveau flux avec backpressure. - pub flac_handle: pmoaudio_ext::sinks::DirectOggFlacHandle, + /// Handle vers le sink OGG-FLAC — clonable, subscribe() crée un flux indépendant par client. + pub flac_handle: pmoaudio_ext::sinks::OggFlacStreamHandle, pub pipeline: PipelineHandle, pub created_at: SystemTime, } @@ -116,16 +116,16 @@ impl RendererRegistry { Ok((stream_url, udn)) } - /// Retourne un DirectOggFlacStream pour l'endpoint /stream. - /// Chaque appel retourne un wrapper sur le même reader persistant. + /// Retourne un OggFlacClientStream indépendant pour l'endpoint /stream. + /// Chaque appel crée un nouveau subscriber broadcast — safe pour connexions simultanées. pub fn get_stream( &self, instance_id: &str, - ) -> Option { + ) -> Option { self.instances .read() .get(instance_id) - .map(|i| i.flac_handle.get_stream()) + .map(|i| i.flac_handle.subscribe()) } /// Retourne le PipelineHandle par UDN (pour les handlers UPnP) diff --git a/pmowebrenderer/src/stream.rs b/pmowebrenderer/src/stream.rs index 2e759ebb..97cf511d 100644 --- a/pmowebrenderer/src/stream.rs +++ b/pmowebrenderer/src/stream.rs @@ -2,16 +2,15 @@ //! //! Sert le flux OGG-FLAC d'une instance WebRenderer. //! -//! Safari fait systématiquement une requête Range: bytes=0-1 avant de jouer. -//! On répond 206 avec 2 octets factices pour satisfaire la sonde, -//! puis la vraie requête (sans Range) reçoit le stream persistant. +//! Safari envoie parfois Range: bytes=0-N avant de jouer. +//! On ignore ce header et on répond toujours 200 chunked (flux live infini). use axum::{ body::Body, extract::{Path, State}, http::{ HeaderMap, StatusCode, - header::{CACHE_CONTROL, CONNECTION, CONTENT_TYPE, CONTENT_RANGE, CONTENT_LENGTH, ACCEPT_RANGES}, + header::{CACHE_CONTROL, CONNECTION, CONTENT_TYPE, TRANSFER_ENCODING}, }, response::{IntoResponse, Response}, }; @@ -29,20 +28,12 @@ pub async fn stream_handler( ) -> impl IntoResponse { info!(instance_id = %instance_id, "FLAC stream client connecting"); - // Détecter la sonde Range: bytes=0-1 de Safari + // Ignorer le header Range — flux live infini, non seekable. + // On ne répond jamais 206 ni 416 : toujours 200 chunked. + // Safari (et d'autres clients) envoient parfois Range: bytes=0-N ; + // répondre 416 ou 206 leur fait croire à une ressource finie. if let Some(range) = headers.get("range") { - if range.as_bytes() == b"bytes=0-1" { - info!(instance_id = %instance_id, "Safari range probe — responding 206"); - return Response::builder() - .status(StatusCode::PARTIAL_CONTENT) - .header(CONTENT_TYPE, "audio/ogg; codecs=flac") - .header(ACCEPT_RANGES, "bytes") - .header(CONTENT_RANGE, "bytes 0-1/*") - .header(CONTENT_LENGTH, "2") - .body(Body::from(vec![0u8, 0u8])) - .unwrap() - .into_response(); - } + info!(instance_id = %instance_id, "Range header ignored (live stream): {:?}", range); } let stream = match registry.get_stream(&instance_id) { @@ -63,7 +54,7 @@ pub async fn stream_handler( .header(CONTENT_TYPE, "audio/ogg; codecs=flac") .header(CACHE_CONTROL, "no-store, no-transform") .header(CONNECTION, "keep-alive") - .header(ACCEPT_RANGES, "bytes") + .header(TRANSFER_ENCODING, "chunked") .header("X-Content-Type-Options", "nosniff") .body(Body::from_stream(ReaderStream::new(stream))) .unwrap()