From 6fe2f6be954f1cd3995b89e7059d50fcb6036114 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Thu, 26 Feb 2026 20:41:57 +0100 Subject: [PATCH] Refactor audio pipeline for multi-client streaming and improved error handling Refactor the audio pipeline to support multi-client streaming with new OggFlacStreamHandle and StreamingOggFlacSink. - Replace DirectOggFlacSink with StreamingOggFlacSink in pipeline - Update documentation and comments to reflect multi-client support - Add detailed warning logs for buffer underruns and client disconnections - Remove TimerBufferNode from pipeline as it's no longer needed - Update connect() calls to subscribe() for new streaming behavior This change enables multiple clients to subscribe to the same audio stream, with each subscription getting a live feed from the current point in time. --- .../src/sinks/direct_ogg_flac_sink.rs | 16 +++++++++-- pmoaudio/src/nodes/timer_buffer_node.rs | 8 ++++++ pmowebrenderer/src/pipeline.rs | 28 +++++++++---------- pmowebrenderer/src/registry.rs | 8 +++--- pmowebrenderer/src/stream.rs | 2 +- 5 files changed, 40 insertions(+), 22 deletions(-) diff --git a/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs index aa230609..4f94e248 100644 --- a/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs @@ -44,7 +44,7 @@ use pmoflac::{EncoderOptions, PcmFormat}; use tokio::io::{AsyncRead, ReadBuf}; use tokio::sync::{mpsc, watch, Mutex}; use tokio_util::sync::CancellationToken; -use tracing::debug; +use tracing::{debug, warn}; use crate::sinks::byte_stream_reader::{ByteStreamReader, PcmChunk}; use crate::sinks::chunk_to_pcm::chunk_to_pcm_bytes; @@ -228,7 +228,11 @@ impl NodeLogic for DirectOggFlacSinkLogic { duration_sec, }; if tx.send(pcm_chunk).await.is_err() { - debug!("DirectOggFlacSink: pcm_tx send failed (client disconnected), clearing pcm_tx"); + warn!( + ts = seg.timestamp_sec, + "DirectOggFlacSink: chunk dropped (client disconnected at {:.3}s), waiting for reconnect", + seg.timestamp_sec, + ); *self.pcm_tx.lock().await = None; } } @@ -348,7 +352,13 @@ async fn run_ogg_encoder( let ogg_page = Bytes::from(ogg.create_page(&frame, false, false, false)); if pipe_writer.write_all(&ogg_page).await.is_err() { - // Client déconnecté + // Client déconnecté — le pipe HTTP s'est rompu + warn!( + samples = encoded_samples, + "DirectOggFlacSink: OGG pipe broken after {} samples ({:.3}s), client disconnected", + encoded_samples, + encoded_samples as f64 / DIRECT_OGG_FLAC_SAMPLE_RATE as f64, + ); return Ok(()); } } diff --git a/pmoaudio/src/nodes/timer_buffer_node.rs b/pmoaudio/src/nodes/timer_buffer_node.rs index dae81935..756b9979 100644 --- a/pmoaudio/src/nodes/timer_buffer_node.rs +++ b/pmoaudio/src/nodes/timer_buffer_node.rs @@ -135,6 +135,14 @@ impl TimerBufferNodeLogic { self.buffer.len() ); + if self.buffer.is_empty() && self.buffered_time_sec < self.capacity_sec * 0.1 { + tracing::warn!( + "TimerBufferNode: buffer underrun at ts={:.3}s (capacity={:.1}s) — source too slow or stalled", + segment.timestamp_sec, + self.capacity_sec, + ); + } + send_to_children(std::any::type_name::(), output, segment).await?; } Ok(()) diff --git a/pmowebrenderer/src/pipeline.rs b/pmowebrenderer/src/pipeline.rs index 36528f89..4f8548d0 100644 --- a/pmowebrenderer/src/pipeline.rs +++ b/pmowebrenderer/src/pipeline.rs @@ -7,9 +7,12 @@ use std::sync::Arc; -use pmoaudio::{AudioSegment, PositionHandle, PositionTrackerNode, ResamplingNode, TimerBufferNode, ToI24Node}; +use pmoaudio::{AudioSegment, PositionTrackerNode, ResamplingNode, ToI24Node}; use pmometadata::{MemoryTrackMetadata, TrackMetadata}; -use pmoaudio_ext::sinks::{DirectOggFlacHandle, DirectOggFlacSink, DIRECT_OGG_FLAC_SAMPLE_RATE}; +use pmoaudio_ext::sinks::{ + OggFlacStreamHandle, StreamingOggFlacSink, + DIRECT_OGG_FLAC_BITS_PER_SAMPLE, DIRECT_OGG_FLAC_SAMPLE_RATE, +}; use pmoaudio_ext::UriSource; use pmoflac::EncoderOptions; use tokio::sync::mpsc; @@ -56,10 +59,10 @@ impl PipelineHandle { /// Pipeline audio complet pour une instance WebRenderer. /// /// Créé au `POST /register`. Le flux OGG-FLAC est accessible via `flac_handle` -/// (un seul client à la fois, reconnectable à chaque Play). +/// (multi-clients, chaque `subscribe()` retourne un nouveau flux depuis le point courant). pub struct InstancePipeline { - /// Handle vers le sink OGG-FLAC — clonable, reconnectable à chaque Play. - pub flac_handle: DirectOggFlacHandle, + /// Handle vers le sink OGG-FLAC — clonable, chaque subscribe() donne un flux live. + pub flac_handle: OggFlacStreamHandle, pub pipeline_handle: PipelineHandle, } @@ -79,21 +82,18 @@ impl InstancePipeline { // La backpressure remonte naturellement depuis le pipe duplex jusqu'à la source. use pmoaudio::pipeline::AudioPipelineNode; - let (sink, flac_handle) = DirectOggFlacSink::new(EncoderOptions::default()); + let (sink, flac_handle) = StreamingOggFlacSink::new( + EncoderOptions::default(), + DIRECT_OGG_FLAC_BITS_PER_SAMPLE, + ); - // Nœud de suivi de position : lit le timestamp des chunks sortant du buffer + // Nœud de suivi de position : lit le timestamp des chunks sortant vers le sink let (mut position_tracker, position_handle) = PositionTrackerNode::new(); position_tracker.register(sink.boxed()); - // Nœud de pacing : régule le débit pour éviter les rafales et pertes de segments - // 2s de buffer absorbe les irrégularités de la source réseau - let mut timer_buffer = TimerBufferNode::new(2.0); - timer_buffer.register(position_tracker.boxed()); - // Nœud de conversion de profondeur : tout type entier → I24 - // Placé avant le buffer pour réduire la mémoire utilisée let mut to_i24 = ToI24Node::new(); - to_i24.register(timer_buffer.boxed()); + to_i24.register(position_tracker.boxed()); // Nœud de rééchantillonnage : n'importe quel sample rate → 96 kHz let mut resampler = ResamplingNode::new(DIRECT_OGG_FLAC_SAMPLE_RATE); diff --git a/pmowebrenderer/src/registry.rs b/pmowebrenderer/src/registry.rs index 1a894e72..b49e2c06 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, reconnectable à chaque Play. - pub flac_handle: pmoaudio_ext::sinks::DirectOggFlacHandle, + /// Handle vers le sink OGG-FLAC — clonable, chaque subscribe() donne un flux live. + pub flac_handle: pmoaudio_ext::sinks::OggFlacStreamHandle, pub pipeline: PipelineHandle, pub created_at: SystemTime, } @@ -116,11 +116,11 @@ impl RendererRegistry { Ok((stream_url, udn)) } - /// Retourne le DirectOggFlacHandle pour l'endpoint /stream (clonable). + /// Retourne le OggFlacStreamHandle pour l'endpoint /stream (clonable). pub fn get_flac_handle( &self, instance_id: &str, - ) -> Option { + ) -> Option { self.instances .read() .get(instance_id) diff --git a/pmowebrenderer/src/stream.rs b/pmowebrenderer/src/stream.rs index b147c986..fa28a19a 100644 --- a/pmowebrenderer/src/stream.rs +++ b/pmowebrenderer/src/stream.rs @@ -44,7 +44,7 @@ pub async fn stream_handler( } }; - let stream = handle.connect().await; + let stream = handle.subscribe(); info!(instance_id = %instance_id, "FLAC stream started");