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()