From 12389dd7c1750723037efd0e4cd8710a9876bfd5 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Thu, 20 Nov 2025 07:29:31 +0100 Subject: [PATCH] debut de refactoring des stream sinks --- pmoaudio-ext/src/sinks/byte_stream_reader.rs | 90 ++++ pmoaudio-ext/src/sinks/chunk_to_pcm.rs | 94 ++++ pmoaudio-ext/src/sinks/flac_frame_utils.rs | 102 ++++ pmoaudio-ext/src/sinks/mod.rs | 3 + pmoaudio-ext/src/sinks/streaming_flac_sink.rs | 484 ++++++------------ .../src/sinks/streaming_ogg_flac_sink.rs | 386 ++------------ pmoaudio-ext/src/sinks/timed_broadcast.rs | 22 + 7 files changed, 514 insertions(+), 667 deletions(-) create mode 100644 pmoaudio-ext/src/sinks/byte_stream_reader.rs create mode 100644 pmoaudio-ext/src/sinks/chunk_to_pcm.rs diff --git a/pmoaudio-ext/src/sinks/byte_stream_reader.rs b/pmoaudio-ext/src/sinks/byte_stream_reader.rs new file mode 100644 index 00000000..9de6f3cd --- /dev/null +++ b/pmoaudio-ext/src/sinks/byte_stream_reader.rs @@ -0,0 +1,90 @@ +use std::io; +use std::{collections::VecDeque, pin::Pin, sync::Arc, task::{Context, Poll}}; + +use tokio::{io::{AsyncRead, ReadBuf}, sync::{RwLock, mpsc}}; + +/// PCM chunk with audio data and timestamp for precise pacing. +#[derive(Debug)] +pub struct PcmChunk { + /// Raw PCM audio bytes + pub bytes: Vec, + /// Timestamp in seconds (from AudioSegment) + pub timestamp_sec: f64, + /// Duration in seconds of this PCM chunk (samples / sample_rate) + pub duration_sec: f64, +} + +/// AsyncRead adapter for mpsc::Receiver. +/// Extracts bytes from PcmChunk and provides them to the FLAC encoder. +pub struct ByteStreamReader { + rx: mpsc::Receiver, + buffer: VecDeque, + finished: bool, + /// Shared timestamp for broadcaster pacing + current_timestamp: Arc>, + /// Shared duration for broadcaster pacing + current_duration: Arc>, +} + +impl ByteStreamReader { + pub fn new( + rx: mpsc::Receiver, + current_timestamp: Arc>, + current_duration: Arc>, + ) -> Self { + Self { + rx, + buffer: VecDeque::new(), + finished: false, + current_timestamp, + current_duration, + } + } +} + +impl AsyncRead for ByteStreamReader { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + loop { + if !self.buffer.is_empty() { + let to_copy = self.buffer.len().min(buf.remaining()); + if to_copy == 0 { + return Poll::Ready(Ok(())); + } + + let slice = self.buffer.make_contiguous(); + buf.put_slice(&slice[..to_copy]); + self.buffer.drain(..to_copy); + return Poll::Ready(Ok(())); + } + + if self.finished { + return Poll::Ready(Ok(())); + } + + match Pin::new(&mut self.rx).poll_recv(cx) { + Poll::Ready(Some(chunk)) => { + if chunk.bytes.is_empty() { + continue; + } + // Update shared timestamp and duration for broadcaster pacing + if let Ok(mut ts) = self.current_timestamp.try_write() { + *ts = chunk.timestamp_sec; + } + if let Ok(mut dur) = self.current_duration.try_write() { + *dur = chunk.duration_sec; + } + self.buffer.extend(chunk.bytes); + } + Poll::Ready(None) => { + self.finished = true; + return Poll::Ready(Ok(())); + } + Poll::Pending => return Poll::Pending, + } + } + } +} diff --git a/pmoaudio-ext/src/sinks/chunk_to_pcm.rs b/pmoaudio-ext/src/sinks/chunk_to_pcm.rs new file mode 100644 index 00000000..d9652ba8 --- /dev/null +++ b/pmoaudio-ext/src/sinks/chunk_to_pcm.rs @@ -0,0 +1,94 @@ +use pmoaudio::{AudioChunk, AudioError}; + +/// Convert an AudioChunk to PCM bytes with specified bit depth. +pub(crate) fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result, AudioError> { + match chunk { + AudioChunk::F32(_) | AudioChunk::F64(_) => { + return Err(AudioError::ProcessingError( + "StreamingFlacSink only supports integer audio chunks".into(), + )); + } + _ => {} + } + + let len = chunk.len(); + let bytes_per_frame = (bits_per_sample / 8) as usize * 2; + let mut bytes = Vec::with_capacity(len * bytes_per_frame); + + match (chunk, bits_per_sample) { + (AudioChunk::I16(data), 16) => { + for frame in data.get_frames() { + bytes.extend_from_slice(&frame[0].to_le_bytes()); + bytes.extend_from_slice(&frame[1].to_le_bytes()); + } + } + (AudioChunk::I16(data), 24) => { + for frame in data.get_frames() { + let left = (frame[0] as i32) << 8; + let right = (frame[1] as i32) << 8; + bytes.extend_from_slice(&left.to_le_bytes()[..3]); + bytes.extend_from_slice(&right.to_le_bytes()[..3]); + } + } + (AudioChunk::I16(data), 32) => { + for frame in data.get_frames() { + let left = (frame[0] as i32) << 16; + let right = (frame[1] as i32) << 16; + bytes.extend_from_slice(&left.to_le_bytes()); + bytes.extend_from_slice(&right.to_le_bytes()); + } + } + (AudioChunk::I24(data), 16) => { + for frame in data.get_frames() { + let left = (frame[0].as_i32() >> 8) as i16; + let right = (frame[1].as_i32() >> 8) as i16; + bytes.extend_from_slice(&left.to_le_bytes()); + bytes.extend_from_slice(&right.to_le_bytes()); + } + } + (AudioChunk::I24(data), 24) => { + for frame in data.get_frames() { + bytes.extend_from_slice(&frame[0].as_i32().to_le_bytes()[..3]); + bytes.extend_from_slice(&frame[1].as_i32().to_le_bytes()[..3]); + } + } + (AudioChunk::I24(data), 32) => { + for frame in data.get_frames() { + let left = frame[0].as_i32() << 8; + let right = frame[1].as_i32() << 8; + bytes.extend_from_slice(&left.to_le_bytes()); + bytes.extend_from_slice(&right.to_le_bytes()); + } + } + (AudioChunk::I32(data), 16) => { + for frame in data.get_frames() { + let left = (frame[0] >> 16) as i16; + let right = (frame[1] >> 16) as i16; + bytes.extend_from_slice(&left.to_le_bytes()); + bytes.extend_from_slice(&right.to_le_bytes()); + } + } + (AudioChunk::I32(data), 24) => { + for frame in data.get_frames() { + let left = frame[0] >> 8; + let right = frame[1] >> 8; + bytes.extend_from_slice(&left.to_le_bytes()[..3]); + bytes.extend_from_slice(&right.to_le_bytes()[..3]); + } + } + (AudioChunk::I32(data), 32) => { + for frame in data.get_frames() { + bytes.extend_from_slice(&frame[0].to_le_bytes()); + bytes.extend_from_slice(&frame[1].to_le_bytes()); + } + } + _ => { + return Err(AudioError::ProcessingError(format!( + "Unsupported bits_per_sample: {}", + bits_per_sample + ))); + } + } + + Ok(bytes) +} diff --git a/pmoaudio-ext/src/sinks/flac_frame_utils.rs b/pmoaudio-ext/src/sinks/flac_frame_utils.rs index 2b9a5bb1..2157ddfb 100644 --- a/pmoaudio-ext/src/sinks/flac_frame_utils.rs +++ b/pmoaudio-ext/src/sinks/flac_frame_utils.rs @@ -7,6 +7,10 @@ //! Frame header validation includes CRC-8 verification as per FLAC specification //! to eliminate false positives that would cause decoder errors. +use pmoaudio::AudioError; +use pmoflac::FlacEncodedStream; +use tokio::io::AsyncReadExt; + /// Validate and parse FLAC block size from frame header /// /// Returns the number of samples in the frame if the header is valid, or None if: @@ -365,6 +369,104 @@ pub(crate) fn find_complete_frames_with_samples(data: &[u8]) -> (usize, u64) { } } + +/// Extract sample rate from STREAMINFO block in FLAC header +pub(crate) fn extract_sample_rate_from_streaminfo(flac_header: &[u8]) -> Result { + // Verify we have at least "fLaC" magic + STREAMINFO block header + if flac_header.len() < 8 { + return Err(AudioError::ProcessingError("FLAC header too short".into())); + } + + if &flac_header[0..4] != b"fLaC" { + return Err(AudioError::ProcessingError("Invalid FLAC magic".into())); + } + + // First metadata block should be STREAMINFO (type 0) + let block_type = flac_header[4] & 0x7F; + if block_type != 0 { + return Err(AudioError::ProcessingError( + "First block is not STREAMINFO".into(), + )); + } + + // STREAMINFO data starts at offset 8 (after magic + block header) + // Sample rate is at offset 10-12 of STREAMINFO data (bytes 18-20 of header) + if flac_header.len() < 21 { + return Err(AudioError::ProcessingError( + "STREAMINFO block truncated".into(), + )); + } + + // Sample rate: 20 bits starting at byte 10 of STREAMINFO + // Format: [byte10: SSSSSSSS] [byte11: SSSSSSSS] [byte12: SSSSCCCC] + // S = sample rate bits, C = channels bits + let byte10 = flac_header[18] as u32; + let byte11 = flac_header[19] as u32; + let byte12 = flac_header[20] as u32; + + // Extract 20 bits for sample rate (top 20 bits of 3 bytes) + let sample_rate = (byte10 << 12) | (byte11 << 4) | (byte12 >> 4); + + if sample_rate == 0 { + return Err(AudioError::ProcessingError( + "Invalid sample rate (0)".into(), + )); + } + + Ok(sample_rate) +} + +/// Read FLAC header (fLaC + all metadata blocks until first frame) +pub(crate) async fn read_flac_header(stream: &mut FlacEncodedStream) -> Result, AudioError> { + let mut header = Vec::new(); + let mut buffer = [0u8; 4]; + + // Read "fLaC" magic + stream + .read_exact(&mut buffer) + .await + .map_err(|e| AudioError::ProcessingError(format!("Failed to read FLAC magic: {}", e)))?; + + if &buffer != b"fLaC" { + return Err(AudioError::ProcessingError( + "Invalid FLAC stream: missing fLaC magic".into(), + )); + } + + header.extend_from_slice(&buffer); + + // Read metadata blocks + loop { + // Read metadata block header (1 byte type + 3 bytes length) + let mut block_header = [0u8; 4]; + stream.read_exact(&mut block_header).await.map_err(|e| { + AudioError::ProcessingError(format!("Failed to read metadata block header: {}", e)) + })?; + + let is_last = (block_header[0] & 0x80) != 0; + let block_length = + u32::from_be_bytes([0, block_header[1], block_header[2], block_header[3]]) as usize; + + header.extend_from_slice(&block_header); + + // Read metadata block data + let mut block_data = vec![0u8; block_length]; + stream.read_exact(&mut block_data).await.map_err(|e| { + AudioError::ProcessingError(format!("Failed to read metadata block data: {}", e)) + })?; + + header.extend_from_slice(&block_data); + + if is_last { + break; + } + } + + Ok(header) +} + + + #[cfg(test)] mod tests { use super::*; diff --git a/pmoaudio-ext/src/sinks/mod.rs b/pmoaudio-ext/src/sinks/mod.rs index 72fc7e27..1fab069f 100755 --- a/pmoaudio-ext/src/sinks/mod.rs +++ b/pmoaudio-ext/src/sinks/mod.rs @@ -4,6 +4,9 @@ //! et ne peuvent pas être placés directement dans pmoaudio sans créer //! de dépendances cycliques. +pub mod byte_stream_reader; +pub mod chunk_to_pcm; + #[cfg(feature = "cache-sink")] mod flac_cache_sink; diff --git a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs index 565df4f6..ffb38bd9 100644 --- a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs @@ -71,7 +71,7 @@ use async_trait::async_trait; use bytes::Bytes; use pmoaudio::{ pipeline::{AudioPipelineNode, Node, NodeLogic, PipelineHandle, StopReason}, - AudioChunk, AudioError, AudioSegment, SyncMarker, TypeRequirement, TypedAudioNode, + AudioError, AudioSegment, SyncMarker, TypeRequirement, TypedAudioNode, _AudioSegment, }; use pmoflac::{encode_flac_stream, EncoderOptions, FlacEncodedStream, PcmFormat}; @@ -81,44 +81,16 @@ use tokio::sync::{mpsc, RwLock}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, trace, warn}; +use crate::byte_stream_reader::{PcmChunk,ByteStreamReader}; +use crate::chunk_to_pcm::chunk_to_pcm_bytes; +use crate::sinks::timed_broadcast::{DEFAULT_BROADCAST_MAX_LEAD_TIME, calculate_broadcast_capacity}; + /// Default ICY metadata interval (bytes of audio between metadata blocks). /// Standard value used by most streaming servers. const DEFAULT_ICY_METAINT: usize = 16000; -/// Default maximum lead time for HTTP broadcast pacing (in seconds). -/// The broadcaster will sleep if it's ahead of real-time by more than this amount. -const DEFAULT_BROADCAST_MAX_LEAD_TIME: f64 = 0.5; -/// Calculate broadcast channel capacity based on max_lead_time. -/// -/// Estimates the number of items needed to buffer max_lead_time seconds of audio. -/// Assumes ~20 items per second (50ms per chunk). -/// -/// # Arguments -/// -/// * `max_lead_time` - Maximum lead time in seconds -/// -/// # Returns -/// -/// Broadcast channel capacity (minimum 100 items) -fn calculate_broadcast_capacity(max_lead_time: f64) -> usize { - // Estimation: ~20 items/second (chunks de 50ms en moyenne) - // Pour 10s: 200 items - let estimated_items_per_second = 20.0; - let capacity = (max_lead_time * estimated_items_per_second) as usize; - capacity.max(100) // Minimum 100 items -} -/// PCM chunk with audio data and timestamp for precise pacing. -#[derive(Debug)] -struct PcmChunk { - /// Raw PCM audio bytes - bytes: Vec, - /// Timestamp in seconds (from AudioSegment) - timestamp_sec: f64, - /// Duration in seconds of this PCM chunk (samples / sample_rate) - duration_sec: f64, -} /// Snapshot of track metadata at a point in time. /// @@ -166,7 +138,7 @@ pub struct MetadataSnapshot { #[derive(Clone)] pub struct StreamHandle { /// Broadcast sender for FLAC bytes (pure mode) - flac_broadcast: timed_broadcast::Sender, + broadcast: timed_broadcast::Sender, /// Current track metadata (read-only for consumers) metadata: Arc>, @@ -178,7 +150,7 @@ pub struct StreamHandle { stop_token: CancellationToken, /// Cached FLAC header (sent to new subscribers first) - flac_header: Arc>>, + header: Arc>>, auto_stop: Arc, } @@ -192,7 +164,7 @@ impl StreamHandle { debug!("New FLAC client subscribed (total: {})", count + 1); FlacClientStream { - rx: self.flac_broadcast.subscribe(), + rx: self.broadcast.subscribe(), buffer: VecDeque::new(), finished: false, handle: self.clone(), @@ -219,7 +191,7 @@ impl StreamHandle { ); IcyClientStream { - rx: self.flac_broadcast.subscribe(), + rx: self.broadcast.subscribe(), metadata: self.metadata.clone(), metaint, byte_count: 0, @@ -291,7 +263,7 @@ impl AsyncRead for FlacClientStream { loop { // If in header state, send the header first if matches!(self.state, FlacStreamState::SendingHeader) { - let header_opt = if let Ok(guard) = self.handle.flac_header.try_read() { + let header_opt = if let Ok(guard) = self.handle.header.try_read() { guard.clone() } else { None @@ -470,7 +442,7 @@ impl AsyncRead for IcyClientStream { loop { // If in header state, send the header first if matches!(self.state, FlacStreamState::SendingHeader) { - let header_opt = if let Ok(guard) = self.handle.flac_header.try_read() { + let header_opt = if let Ok(guard) = self.handle.header.try_read() { guard.clone() } else { None @@ -606,8 +578,8 @@ struct StreamingFlacSinkLogic { pcm_tx: Option>, pcm_rx: Option>, metadata: Arc>, - flac_broadcast: timed_broadcast::Sender, - flac_header: Arc>>, + broadcast: timed_broadcast::Sender, + header: Arc>>, encoder_state: Option, sample_rate: Option, broadcast_max_lead_time: f64, @@ -662,14 +634,14 @@ impl StreamingFlacSinkLogic { debug!("FLAC encoder initialized successfully"); // Spawn broadcaster task with timestamp and duration for pacing - let flac_broadcast = self.flac_broadcast.clone(); - let flac_header = self.flac_header.clone(); + let broadcast = self.broadcast.clone(); + let header = self.header.clone(); let max_lead = self.broadcast_max_lead_time; let broadcaster_task = tokio::spawn(async move { if let Err(e) = broadcast_flac_stream( flac_stream, - flac_broadcast, - flac_header, + broadcast, + header, current_timestamp, current_duration, max_lead, @@ -931,6 +903,134 @@ impl NodeLogic for StreamingFlacSinkLogic { } } + +/// Streaming FLAC sink for multi-client HTTP streaming. +pub struct StreamingFlacSink { + inner: Node, +} + +impl StreamingFlacSink { + /// Create a new streaming FLAC sink. + /// + /// # Arguments + /// + /// * `encoder_options` - FLAC encoder configuration + /// * `bits_per_sample` - Target bit depth (16, 24, or 32) + /// + /// # Returns + /// + /// A tuple of `(sink, handle)` where: + /// - `sink` is added to the audio pipeline + /// - `handle` is used by HTTP handlers to serve streams + pub fn new( + encoder_options: EncoderOptions, + bits_per_sample: u8, + ) -> (Self, StreamHandle) { + Self::with_max_broadcast_lead( + encoder_options, + bits_per_sample, + DEFAULT_BROADCAST_MAX_LEAD_TIME, + ) + } + + /// Create a sink with a custom broadcast pacing limit. + pub fn with_max_broadcast_lead( + encoder_options: EncoderOptions, + bits_per_sample: u8, + broadcast_max_lead_time: f64, + ) -> (Self, StreamHandle) { + // Validate bit depth + if ![16, 24, 32].contains(&bits_per_sample) { + panic!("bits_per_sample must be 16, 24, or 32"); + } + + // Create PCM channel (bounded for backpressure) + let (pcm_tx, pcm_rx) = mpsc::channel::(16); + + // Shared metadata + let metadata = Arc::new(RwLock::new(MetadataSnapshot::default())); + + // Capacity calculated from max_lead_time to ensure enough buffering + let broadcast_capacity = calculate_broadcast_capacity(broadcast_max_lead_time); + debug!( + "Streaming Sink: using broadcast capacity of {} items (max_lead_time={:.1}s)", + broadcast_capacity, + broadcast_max_lead_time + ); + + // Broadcast channel for FLAC bytes + let (broadcast, _) = timed_broadcast::channel(broadcast_capacity); + + // FLAC header cache + let header = Arc::new(RwLock::new(None)); + + // Stop token and client counter + let stop_token = CancellationToken::new(); + let active_clients = Arc::new(AtomicUsize::new(0)); + + let handle = StreamHandle { + broadcast: broadcast.clone(), + metadata: metadata.clone(), + active_clients, + stop_token: stop_token.clone(), + header: header.clone(), + auto_stop: Arc::new(AtomicBool::new(true)), + }; + + let logic = StreamingFlacSinkLogic { + encoder_options, + bits_per_sample, + pcm_tx: Some(pcm_tx), + pcm_rx: Some(pcm_rx), + metadata, + broadcast, + header, + encoder_state: None, + sample_rate: None, + broadcast_max_lead_time: broadcast_max_lead_time.max(0.0), + first_chunk_timestamp_checked: false, + timestamp_offset_sec: 0.0, + current_timestamp: Arc::new(RwLock::new(0.0)), + }; + + let sink = Self { + inner: Node::new_with_input(logic, 16), + }; + + (sink, handle) + } +} + +#[async_trait] +impl AudioPipelineNode for StreamingFlacSink { + fn get_tx(&self) -> Option>> { + self.inner.get_tx() + } + + fn register(&mut self, _child: Box) { + panic!("StreamingFlacSink is a terminal sink and cannot have children"); + } + + async fn run(self: Box, stop_token: CancellationToken) -> Result<(), AudioError> { + Box::new(self.inner).run(stop_token).await + } + + fn start(self: Box) -> PipelineHandle { + Box::new(self.inner).start() + } +} + +impl TypedAudioNode for StreamingFlacSink { + fn input_type(&self) -> Option { + Some(TypeRequirement::any_integer()) + } + + fn output_type(&self) -> Option { + None + } +} + + /// Broadcaster task: reads FLAC bytes from encoder and broadcasts to all clients. /// Implements precise real-time pacing based on audio timestamps. /// Ensures data is sent at FLAC frame boundaries to prevent sync errors in strict decoders like FFPlay. @@ -963,9 +1063,12 @@ async fn broadcast_flac_stream( let mut broadcast_count = 0u64; let mut total_read_time = 0.0f64; let mut read_count = 0u64; - let mut encoded_samples = 0u64; let sample_rate_f64 = sample_rate as f64; + // Sample counter for calculating accurate timestamps (reset on new headers) + let mut encoded_samples = 0u64; + + loop { let read_start = std::time::Instant::now(); match flac_stream.read(&mut read_buffer).await { @@ -1176,292 +1279,3 @@ async fn broadcast_flac_stream( Ok(()) } -/// Streaming FLAC sink for multi-client HTTP streaming. -pub struct StreamingFlacSink { - inner: Node, -} - -impl StreamingFlacSink { - /// Create a new streaming FLAC sink. - /// - /// # Arguments - /// - /// * `encoder_options` - FLAC encoder configuration - /// * `bits_per_sample` - Target bit depth (16, 24, or 32) - /// - /// # Returns - /// - /// A tuple of `(sink, handle)` where: - /// - `sink` is added to the audio pipeline - /// - `handle` is used by HTTP handlers to serve streams - pub fn new(encoder_options: EncoderOptions, bits_per_sample: u8) -> (Self, StreamHandle) { - Self::with_max_broadcast_lead( - encoder_options, - bits_per_sample, - DEFAULT_BROADCAST_MAX_LEAD_TIME, - ) - } - - /// Create a sink with a custom broadcast pacing limit. - pub fn with_max_broadcast_lead( - encoder_options: EncoderOptions, - bits_per_sample: u8, - broadcast_max_lead_time: f64, - ) -> (Self, StreamHandle) { - // Validate bit depth - if ![16, 24, 32].contains(&bits_per_sample) { - panic!("bits_per_sample must be 16, 24, or 32"); - } - - // Create PCM channel (bounded for backpressure) - let (pcm_tx, pcm_rx) = mpsc::channel::(16); - - // Shared metadata - let metadata = Arc::new(RwLock::new(MetadataSnapshot::default())); - - // Calculate broadcast capacity based on max_lead_time - let broadcast_capacity = calculate_broadcast_capacity(broadcast_max_lead_time); - debug!( - "StreamingFlacSink: using broadcast capacity of {} items (max_lead_time={:.1}s)", - broadcast_capacity, broadcast_max_lead_time - ); - - // Broadcast channel for FLAC bytes - let (flac_broadcast, _) = timed_broadcast::channel(broadcast_capacity); - - // FLAC header cache - let flac_header = Arc::new(RwLock::new(None)); - - // Stop token and client counter - let stop_token = CancellationToken::new(); - let active_clients = Arc::new(AtomicUsize::new(0)); - - let handle = StreamHandle { - flac_broadcast: flac_broadcast.clone(), - metadata: metadata.clone(), - active_clients, - stop_token: stop_token.clone(), - flac_header: flac_header.clone(), - auto_stop: Arc::new(AtomicBool::new(true)), - }; - - let logic = StreamingFlacSinkLogic { - encoder_options, - bits_per_sample, - pcm_tx: Some(pcm_tx), - pcm_rx: Some(pcm_rx), - metadata, - flac_broadcast, - flac_header, - encoder_state: None, - sample_rate: None, - broadcast_max_lead_time: broadcast_max_lead_time.max(0.0), - first_chunk_timestamp_checked: false, - timestamp_offset_sec: 0.0, - current_timestamp: Arc::new(RwLock::new(0.0)), - }; - - let sink = Self { - inner: Node::new_with_input(logic, 16), - }; - - (sink, handle) - } -} - -#[async_trait] -impl AudioPipelineNode for StreamingFlacSink { - fn get_tx(&self) -> Option>> { - self.inner.get_tx() - } - - fn register(&mut self, _child: Box) { - panic!("StreamingFlacSink is a terminal sink and cannot have children"); - } - - async fn run(self: Box, stop_token: CancellationToken) -> Result<(), AudioError> { - Box::new(self.inner).run(stop_token).await - } - - fn start(self: Box) -> PipelineHandle { - Box::new(self.inner).start() - } -} - -impl TypedAudioNode for StreamingFlacSink { - fn input_type(&self) -> Option { - Some(TypeRequirement::any_integer()) - } - - fn output_type(&self) -> Option { - None - } -} - -/// Convert an AudioChunk to PCM bytes with specified bit depth. -fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result, AudioError> { - match chunk { - AudioChunk::F32(_) | AudioChunk::F64(_) => { - return Err(AudioError::ProcessingError( - "StreamingFlacSink only supports integer audio chunks".into(), - )); - } - _ => {} - } - - let len = chunk.len(); - let bytes_per_frame = (bits_per_sample / 8) as usize * 2; - let mut bytes = Vec::with_capacity(len * bytes_per_frame); - - match (chunk, bits_per_sample) { - (AudioChunk::I16(data), 16) => { - for frame in data.get_frames() { - bytes.extend_from_slice(&frame[0].to_le_bytes()); - bytes.extend_from_slice(&frame[1].to_le_bytes()); - } - } - (AudioChunk::I16(data), 24) => { - for frame in data.get_frames() { - let left = (frame[0] as i32) << 8; - let right = (frame[1] as i32) << 8; - bytes.extend_from_slice(&left.to_le_bytes()[..3]); - bytes.extend_from_slice(&right.to_le_bytes()[..3]); - } - } - (AudioChunk::I16(data), 32) => { - for frame in data.get_frames() { - let left = (frame[0] as i32) << 16; - let right = (frame[1] as i32) << 16; - bytes.extend_from_slice(&left.to_le_bytes()); - bytes.extend_from_slice(&right.to_le_bytes()); - } - } - (AudioChunk::I24(data), 16) => { - for frame in data.get_frames() { - let left = (frame[0].as_i32() >> 8) as i16; - let right = (frame[1].as_i32() >> 8) as i16; - bytes.extend_from_slice(&left.to_le_bytes()); - bytes.extend_from_slice(&right.to_le_bytes()); - } - } - (AudioChunk::I24(data), 24) => { - for frame in data.get_frames() { - bytes.extend_from_slice(&frame[0].as_i32().to_le_bytes()[..3]); - bytes.extend_from_slice(&frame[1].as_i32().to_le_bytes()[..3]); - } - } - (AudioChunk::I24(data), 32) => { - for frame in data.get_frames() { - let left = frame[0].as_i32() << 8; - let right = frame[1].as_i32() << 8; - bytes.extend_from_slice(&left.to_le_bytes()); - bytes.extend_from_slice(&right.to_le_bytes()); - } - } - (AudioChunk::I32(data), 16) => { - for frame in data.get_frames() { - let left = (frame[0] >> 16) as i16; - let right = (frame[1] >> 16) as i16; - bytes.extend_from_slice(&left.to_le_bytes()); - bytes.extend_from_slice(&right.to_le_bytes()); - } - } - (AudioChunk::I32(data), 24) => { - for frame in data.get_frames() { - let left = frame[0] >> 8; - let right = frame[1] >> 8; - bytes.extend_from_slice(&left.to_le_bytes()[..3]); - bytes.extend_from_slice(&right.to_le_bytes()[..3]); - } - } - (AudioChunk::I32(data), 32) => { - for frame in data.get_frames() { - bytes.extend_from_slice(&frame[0].to_le_bytes()); - bytes.extend_from_slice(&frame[1].to_le_bytes()); - } - } - _ => { - return Err(AudioError::ProcessingError(format!( - "Unsupported bits_per_sample: {}", - bits_per_sample - ))); - } - } - - Ok(bytes) -} - -/// AsyncRead adapter for mpsc::Receiver. -/// Extracts bytes from PcmChunk and provides them to the FLAC encoder. -struct ByteStreamReader { - rx: mpsc::Receiver, - buffer: VecDeque, - finished: bool, - /// Shared timestamp for broadcaster pacing - current_timestamp: Arc>, - /// Shared duration for broadcaster pacing - current_duration: Arc>, -} - -impl ByteStreamReader { - fn new( - rx: mpsc::Receiver, - current_timestamp: Arc>, - current_duration: Arc>, - ) -> Self { - Self { - rx, - buffer: VecDeque::new(), - finished: false, - current_timestamp, - current_duration, - } - } -} - -impl AsyncRead for ByteStreamReader { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut ReadBuf<'_>, - ) -> Poll> { - loop { - if !self.buffer.is_empty() { - let to_copy = self.buffer.len().min(buf.remaining()); - if to_copy == 0 { - return Poll::Ready(Ok(())); - } - - let slice = self.buffer.make_contiguous(); - buf.put_slice(&slice[..to_copy]); - self.buffer.drain(..to_copy); - return Poll::Ready(Ok(())); - } - - if self.finished { - return Poll::Ready(Ok(())); - } - - match Pin::new(&mut self.rx).poll_recv(cx) { - Poll::Ready(Some(chunk)) => { - if chunk.bytes.is_empty() { - continue; - } - // Update shared timestamp and duration for broadcaster pacing - if let Ok(mut ts) = self.current_timestamp.try_write() { - *ts = chunk.timestamp_sec; - } - if let Ok(mut dur) = self.current_duration.try_write() { - *dur = chunk.duration_sec; - } - self.buffer.extend(chunk.bytes); - } - Poll::Ready(None) => { - self.finished = true; - return Poll::Ready(Ok(())); - } - Poll::Pending => return Poll::Pending, - } - } - } -} diff --git a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs index dd6a9257..ddec2584 100644 --- a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs @@ -62,7 +62,7 @@ use async_trait::async_trait; use bytes::Bytes; use pmoaudio::{ pipeline::{AudioPipelineNode, Node, NodeLogic, PipelineHandle, StopReason}, - AudioChunk, AudioError, AudioSegment, SyncMarker, TypeRequirement, TypedAudioNode, + AudioError, AudioSegment, SyncMarker, TypeRequirement, TypedAudioNode, _AudioSegment, }; use pmoflac::{encode_flac_stream, EncoderOptions, FlacEncodedStream, PcmFormat}; @@ -72,29 +72,10 @@ use tokio::sync::{mpsc, RwLock}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, trace, warn}; -/// Default maximum lead time for HTTP broadcast pacing (in seconds). -const DEFAULT_BROADCAST_MAX_LEAD_TIME: f64 = 0.0; - -/// Calculate broadcast channel capacity based on max lead time. -/// -/// Estimate: ~20 OGG pages per second (assuming 50ms chunks). -/// Minimum capacity: 100 items for buffering even with 0 lead time. -fn calculate_broadcast_capacity(max_lead_time: f64) -> usize { - let estimated_items_per_second = 20.0; - let capacity = (max_lead_time * estimated_items_per_second) as usize; - capacity.max(100) // Minimum 100 items -} - -/// PCM chunk with audio data and timestamp for precise pacing. -#[derive(Debug)] -struct PcmChunk { - /// Raw PCM audio bytes - bytes: Vec, - /// Timestamp in seconds (from AudioSegment) - timestamp_sec: f64, - /// Duration in seconds of this PCM chunk (samples / sample_rate) - duration_sec: f64, -} +use crate::byte_stream_reader::{PcmChunk,ByteStreamReader}; +use crate::chunk_to_pcm::chunk_to_pcm_bytes; +use crate::sinks::flac_frame_utils::{extract_sample_rate_from_streaminfo, read_flac_header}; +use crate::sinks::timed_broadcast::{DEFAULT_BROADCAST_MAX_LEAD_TIME, calculate_broadcast_capacity}; /// Snapshot of track metadata (reuse from streaming_flac_sink) pub use super::streaming_flac_sink::MetadataSnapshot; @@ -103,9 +84,10 @@ pub use super::streaming_flac_sink::MetadataSnapshot; #[derive(Clone)] pub struct OggFlacStreamHandle { /// Broadcast sender for OGG-FLAC bytes - ogg_broadcast: timed_broadcast::Sender, + + broadcast: timed_broadcast::Sender, - /// Current track metadata + /// Current track metadata (read-only for consumers) metadata: Arc>, /// Active client counter @@ -115,7 +97,7 @@ pub struct OggFlacStreamHandle { stop_token: CancellationToken, /// Cached OGG-FLAC header (sent to new subscribers first) - ogg_header: Arc>>, + header: Arc>>, auto_stop: Arc, } @@ -129,7 +111,7 @@ impl OggFlacStreamHandle { debug!("New OGG-FLAC client subscribed (total: {})", count + 1); OggFlacClientStream { - rx: self.ogg_broadcast.subscribe(), + rx: self.broadcast.subscribe(), buffer: VecDeque::new(), finished: false, handle: self.clone(), @@ -185,7 +167,7 @@ impl AsyncRead for OggFlacClientStream { loop { // If in header state, send the header first if matches!(self.state, OggFlacStreamState::SendingHeader) { - let header_opt = if let Ok(guard) = self.handle.ogg_header.try_read() { + let header_opt = if let Ok(guard) = self.handle.header.try_read() { guard.clone() } else { None @@ -278,8 +260,8 @@ struct StreamingOggFlacSinkLogic { pcm_tx: Option>, pcm_rx: Option>, metadata: Arc>, - ogg_broadcast: timed_broadcast::Sender, - ogg_header: Arc>>, + broadcast: timed_broadcast::Sender, + header: Arc>>, encoder_state: Option, sample_rate: Option, broadcast_max_lead_time: f64, @@ -333,14 +315,14 @@ impl StreamingOggFlacSinkLogic { debug!("OGG-FLAC encoder initialized successfully"); // Spawn OGG wrapper + broadcaster task with timestamp and duration for pacing - let ogg_broadcast = self.ogg_broadcast.clone(); - let ogg_header = self.ogg_header.clone(); + let broadcast = self.broadcast.clone(); + let header = self.header.clone(); let max_lead = self.broadcast_max_lead_time; let broadcaster_task = tokio::spawn(async move { if let Err(e) = broadcast_ogg_flac_stream( flac_stream, - ogg_broadcast, - ogg_header, + broadcast, + header, current_timestamp, current_duration, max_lead, @@ -390,11 +372,12 @@ impl StreamingOggFlacSinkLogic { snapshot.version += 1; debug!( - "OGG-FLAC metadata updated: v{} @ {:.2}s - {} - {}", + "Metadata updated: v{} @ {:.2}s - {} - {} (cover_pk: {:?})", snapshot.version, timestamp_sec, snapshot.artist.as_deref().unwrap_or("?"), - snapshot.title.as_deref().unwrap_or("?") + snapshot.title.as_deref().unwrap_or("?"), + snapshot.cover_pk ); Ok(()) @@ -623,29 +606,28 @@ impl StreamingOggFlacSink { // Shared metadata let metadata = Arc::new(RwLock::new(MetadataSnapshot::default())); - // Broadcast channel for OGG-FLAC bytes // Capacity calculated from max_lead_time to ensure enough buffering let broadcast_capacity = calculate_broadcast_capacity(broadcast_max_lead_time); - tracing::trace!( - "OGG-FLAC broadcast capacity: {} items (for {:.1}s max lead time)", + debug!( + "Streaming Sink: using broadcast capacity of {} items (max_lead_time={:.1}s)", broadcast_capacity, broadcast_max_lead_time ); - let (ogg_broadcast, _) = timed_broadcast::channel(broadcast_capacity); + let (broadcast, _) = timed_broadcast::channel(broadcast_capacity); // OGG-FLAC header cache - let ogg_header = Arc::new(RwLock::new(None)); + let header = Arc::new(RwLock::new(None)); // Stop token and client counter let stop_token = CancellationToken::new(); let active_clients = Arc::new(AtomicUsize::new(0)); let handle = OggFlacStreamHandle { - ogg_broadcast: ogg_broadcast.clone(), + broadcast: broadcast.clone(), metadata: metadata.clone(), active_clients, stop_token: stop_token.clone(), - ogg_header: ogg_header.clone(), + header: header.clone(), auto_stop: Arc::new(AtomicBool::new(true)), }; @@ -655,8 +637,8 @@ impl StreamingOggFlacSink { pcm_tx: Some(pcm_tx), pcm_rx: Some(pcm_rx), metadata, - ogg_broadcast, - ogg_header, + broadcast, + header, encoder_state: None, sample_rate: None, broadcast_max_lead_time: broadcast_max_lead_time.max(0.0), @@ -701,173 +683,7 @@ impl TypedAudioNode for StreamingOggFlacSink { } } -/// AsyncRead adapter for mpsc::Receiver. -/// Extracts bytes from PcmChunk and provides them to the FLAC encoder. -struct ByteStreamReader { - rx: mpsc::Receiver, - buffer: VecDeque, - finished: bool, - /// Shared timestamp for broadcaster pacing - current_timestamp: Arc>, - /// Shared duration for broadcaster pacing - current_duration: Arc>, -} -impl ByteStreamReader { - fn new( - rx: mpsc::Receiver, - current_timestamp: Arc>, - current_duration: Arc>, - ) -> Self { - Self { - rx, - buffer: VecDeque::new(), - finished: false, - current_timestamp, - current_duration, - } - } -} - -impl AsyncRead for ByteStreamReader { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut ReadBuf<'_>, - ) -> Poll> { - loop { - if !self.buffer.is_empty() { - let to_copy = self.buffer.len().min(buf.remaining()); - if to_copy == 0 { - return Poll::Ready(Ok(())); - } - - let slice = self.buffer.make_contiguous(); - buf.put_slice(&slice[..to_copy]); - self.buffer.drain(..to_copy); - return Poll::Ready(Ok(())); - } - - if self.finished { - return Poll::Ready(Ok(())); - } - - match Pin::new(&mut self.rx).poll_recv(cx) { - Poll::Ready(Some(chunk)) => { - if chunk.bytes.is_empty() { - continue; - } - // Update shared timestamp and duration for broadcaster pacing - if let Ok(mut ts) = self.current_timestamp.try_write() { - *ts = chunk.timestamp_sec; - } - if let Ok(mut dur) = self.current_duration.try_write() { - *dur = chunk.duration_sec; - } - self.buffer.extend(chunk.bytes); - } - Poll::Ready(None) => { - self.finished = true; - return Poll::Ready(Ok(())); - } - Poll::Pending => return Poll::Pending, - } - } - } -} - -/// Convert an AudioChunk to PCM bytes with specified bit depth. -fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result, AudioError> { - match chunk { - AudioChunk::F32(_) | AudioChunk::F64(_) => { - return Err(AudioError::ProcessingError( - "StreamingOggFlacSink only supports integer audio chunks".into(), - )); - } - _ => {} - } - - let len = chunk.len(); - let bytes_per_frame = (bits_per_sample / 8) as usize * 2; - let mut bytes = Vec::with_capacity(len * bytes_per_frame); - - match (chunk, bits_per_sample) { - (AudioChunk::I16(data), 16) => { - for frame in data.get_frames() { - bytes.extend_from_slice(&frame[0].to_le_bytes()); - bytes.extend_from_slice(&frame[1].to_le_bytes()); - } - } - (AudioChunk::I16(data), 24) => { - for frame in data.get_frames() { - let left = (frame[0] as i32) << 8; - let right = (frame[1] as i32) << 8; - bytes.extend_from_slice(&left.to_le_bytes()[..3]); - bytes.extend_from_slice(&right.to_le_bytes()[..3]); - } - } - (AudioChunk::I16(data), 32) => { - for frame in data.get_frames() { - let left = (frame[0] as i32) << 16; - let right = (frame[1] as i32) << 16; - bytes.extend_from_slice(&left.to_le_bytes()); - bytes.extend_from_slice(&right.to_le_bytes()); - } - } - (AudioChunk::I24(data), 16) => { - for frame in data.get_frames() { - let left = (frame[0].as_i32() >> 8) as i16; - let right = (frame[1].as_i32() >> 8) as i16; - bytes.extend_from_slice(&left.to_le_bytes()); - bytes.extend_from_slice(&right.to_le_bytes()); - } - } - (AudioChunk::I24(data), 24) => { - for frame in data.get_frames() { - bytes.extend_from_slice(&frame[0].as_i32().to_le_bytes()[..3]); - bytes.extend_from_slice(&frame[1].as_i32().to_le_bytes()[..3]); - } - } - (AudioChunk::I24(data), 32) => { - for frame in data.get_frames() { - let left = frame[0].as_i32() << 8; - let right = frame[1].as_i32() << 8; - bytes.extend_from_slice(&left.to_le_bytes()); - bytes.extend_from_slice(&right.to_le_bytes()); - } - } - (AudioChunk::I32(data), 16) => { - for frame in data.get_frames() { - let left = (frame[0] >> 16) as i16; - let right = (frame[1] >> 16) as i16; - bytes.extend_from_slice(&left.to_le_bytes()); - bytes.extend_from_slice(&right.to_le_bytes()); - } - } - (AudioChunk::I32(data), 24) => { - for frame in data.get_frames() { - let left = frame[0] >> 8; - let right = frame[1] >> 8; - bytes.extend_from_slice(&left.to_le_bytes()[..3]); - bytes.extend_from_slice(&right.to_le_bytes()[..3]); - } - } - (AudioChunk::I32(data), 32) => { - for frame in data.get_frames() { - bytes.extend_from_slice(&frame[0].to_le_bytes()); - bytes.extend_from_slice(&frame[1].to_le_bytes()); - } - } - _ => { - return Err(AudioError::ProcessingError(format!( - "Unsupported bits_per_sample: {}", - bits_per_sample - ))); - } - } - - Ok(bytes) -} /// OGG wrapper + broadcaster task: reads FLAC bytes from encoder, wraps in OGG pages, and broadcasts. /// Implements precise real-time pacing based on audio timestamps. @@ -882,14 +698,14 @@ async fn broadcast_ogg_flac_stream( timestamp_offset_sec: f64, ) -> Result<(), AudioError> { trace!( - "OGG-FLAC broadcaster task started with FLAC frame boundary detection (max_lead={:.3}s)", + "Broadcaster task started with FLAC frame boundary detection (max_lead={:.3}s)", broadcast_max_lead_time ); let stream_serial = rand::random::(); let mut ogg_writer = OggPageWriter::new(stream_serial); - let mut total_ogg_bytes = 0u64; + let mut total_bytes = 0u64; let mut header_captured = false; let mut pacer = BroadcastPacer::new(broadcast_max_lead_time, "OGG"); let mut last_granule_update_time = 0.0f64; @@ -951,7 +767,7 @@ async fn broadcast_ogg_flac_stream( return Ok(()); } } - total_ogg_bytes += comment_bytes.len() as u64; + total_bytes += comment_bytes.len() as u64; match broadcast_tx.send(comment_bytes.clone(), 0.0, 0.0).await { Ok(_) => {} Err(SendError::Expired(_)) => { @@ -968,17 +784,17 @@ async fn broadcast_ogg_flac_stream( // Use larger read buffer (16KB) to reduce syscalls and accumulator for frame boundary detection // The accumulator is necessary to ensure we only send complete FLAC frames let mut read_buffer = vec![0u8; 16384]; - let mut flac_accumulator = Vec::with_capacity(32768); + let mut accumulator = Vec::with_capacity(32768); loop { let read_start = std::time::Instant::now(); match flac_stream.read(&mut read_buffer).await { Ok(0) => { // EOF - create final page with EOS flag and any remaining data - if !flac_accumulator.is_empty() { - let eos_page = ogg_writer.create_page(&flac_accumulator, false, true, false); + if !accumulator.is_empty() { + let eos_page = ogg_writer.create_page(&accumulator, false, true, false); let eos_bytes = Bytes::from(eos_page); - total_ogg_bytes += eos_bytes.len() as u64; + total_bytes += eos_bytes.len() as u64; let eos_ts = *current_timestamp.read().await; let eos_dur = *current_duration.read().await; match broadcast_tx.send(eos_bytes.clone(), eos_ts, eos_dur).await { @@ -992,13 +808,13 @@ async fn broadcast_ogg_flac_stream( } trace!( "Sent final EOS page with {} bytes of data", - flac_accumulator.len() + accumulator.len() ); } else { // Send empty EOS page (metadata page, duration=0.0) let eos_page = ogg_writer.create_page(&[], false, true, false); let eos_bytes = Bytes::from(eos_page); - total_ogg_bytes += eos_bytes.len() as u64; + total_bytes += eos_bytes.len() as u64; let eos_ts = *current_timestamp.read().await; match broadcast_tx.send(eos_bytes.clone(), eos_ts, 0.0).await { Ok(_) => {} @@ -1014,7 +830,7 @@ async fn broadcast_ogg_flac_stream( trace!( "OGG-FLAC stream ended, total OGG bytes: {}", - total_ogg_bytes + total_bytes ); break; } @@ -1034,11 +850,11 @@ async fn broadcast_ogg_flac_stream( } // Append to accumulator - flac_accumulator.extend_from_slice(&read_buffer[..n]); + accumulator.extend_from_slice(&read_buffer[..n]); trace!( "OGG: accumulator now {} bytes after reading {} bytes", - flac_accumulator.len(), + accumulator.len(), n ); @@ -1046,22 +862,22 @@ async fn broadcast_ogg_flac_stream( // OGG-FLAC spec requires: "Each audio data packet contains one complete FLAC frame" loop { // Find all complete frames in the accumulator - if flac_accumulator.len() < 4 { + if accumulator.len() < 4 { break; // Need at least 4 bytes for sync code check } // Find all sync positions with their sample counts // Use CRC-8 validation to eliminate false positives let mut sync_data = Vec::new(); - for i in 0..flac_accumulator.len() - 1 { - let byte1 = flac_accumulator[i]; - let byte2 = flac_accumulator[i + 1]; + for i in 0..accumulator.len() - 1 { + let byte1 = accumulator[i]; + let byte2 = accumulator[i + 1]; if byte1 == 0xFF && byte2 >= 0xF8 && byte2 <= 0xFE { // Validate frame header with CRC-8 to avoid false positives - if flac_frame_utils::validate_frame_header_crc(&flac_accumulator, i) { + if flac_frame_utils::validate_frame_header_crc(&accumulator, i) { if let Some(samples) = - flac_frame_utils::parse_flac_block_size(&flac_accumulator, i) + flac_frame_utils::parse_flac_block_size(&accumulator, i) { sync_data.push((i, samples)); } @@ -1085,13 +901,13 @@ async fn broadcast_ogg_flac_stream( "OGG-FLAC: Skipping {} bytes of garbage data before first frame", first_frame_start ); - flac_accumulator.drain(0..first_frame_start); + accumulator.drain(0..first_frame_start); continue; } // Extract just the first frame let first_frame: Vec = - flac_accumulator.drain(0..second_frame_start).collect(); + accumulator.drain(0..second_frame_start).collect(); // ╔═══════════════════════════════════════════════════════════════╗ // ║ BACKPRESSURE INTELLIGENTE BASÉE SUR LE TIMING ║ @@ -1131,8 +947,8 @@ async fn broadcast_ogg_flac_stream( // Wrap this single FLAC frame in ONE OGG page (per OGG-FLAC spec) let ogg_page = ogg_writer.create_page(&first_frame, false, false, false); - let ogg_bytes = Bytes::from(ogg_page); - total_ogg_bytes += ogg_bytes.len() as u64; + let bytes = Bytes::from(ogg_page); + total_bytes += bytes.len() as u64; // Measure broadcast interval for burst detection let broadcast_interval = last_broadcast_time.elapsed().as_secs_f64(); @@ -1157,17 +973,17 @@ async fn broadcast_ogg_flac_stream( "OGG: {} broadcasts sent, avg_interval={:.3}s, accumulator={} bytes", broadcast_count, last_broadcast_time.elapsed().as_secs_f64() / broadcast_count as f64, - flac_accumulator.len() + accumulator.len() ); } // Envoyer au broadcast match broadcast_tx - .send(ogg_bytes.clone(), audio_timestamp, segment_duration) + .send(bytes.clone(), audio_timestamp, segment_duration) .await { Ok(n) => { - trace!("Broadcasted OGG page with 1 FLAC frame ({} bytes), {} samples ({} bytes total with OGG overhead) to {} receivers (ts={:.3}s, dur={:.3}s)", first_frame.len(), first_frame_samples, ogg_bytes.len(), n, audio_timestamp, segment_duration); + trace!("Broadcasted OGG page with 1 FLAC frame ({} bytes), {} samples ({} bytes total with OGG overhead) to {} receivers (ts={:.3}s, dur={:.3}s)", first_frame.len(), first_frame_samples, bytes.len(), n, audio_timestamp, segment_duration); } Err(SendError::Expired(_)) => { trace!( @@ -1203,104 +1019,10 @@ async fn broadcast_ogg_flac_stream( ))); } - trace!("OGG-FLAC broadcaster task completed successfully"); + trace!("Broadcaster task completed successfully"); Ok(()) } -/// Extract sample rate from STREAMINFO block in FLAC header -fn extract_sample_rate_from_streaminfo(flac_header: &[u8]) -> Result { - // Verify we have at least "fLaC" magic + STREAMINFO block header - if flac_header.len() < 8 { - return Err(AudioError::ProcessingError("FLAC header too short".into())); - } - - if &flac_header[0..4] != b"fLaC" { - return Err(AudioError::ProcessingError("Invalid FLAC magic".into())); - } - - // First metadata block should be STREAMINFO (type 0) - let block_type = flac_header[4] & 0x7F; - if block_type != 0 { - return Err(AudioError::ProcessingError( - "First block is not STREAMINFO".into(), - )); - } - - // STREAMINFO data starts at offset 8 (after magic + block header) - // Sample rate is at offset 10-12 of STREAMINFO data (bytes 18-20 of header) - if flac_header.len() < 21 { - return Err(AudioError::ProcessingError( - "STREAMINFO block truncated".into(), - )); - } - - // Sample rate: 20 bits starting at byte 10 of STREAMINFO - // Format: [byte10: SSSSSSSS] [byte11: SSSSSSSS] [byte12: SSSSCCCC] - // S = sample rate bits, C = channels bits - let byte10 = flac_header[18] as u32; - let byte11 = flac_header[19] as u32; - let byte12 = flac_header[20] as u32; - - // Extract 20 bits for sample rate (top 20 bits of 3 bytes) - let sample_rate = (byte10 << 12) | (byte11 << 4) | (byte12 >> 4); - - if sample_rate == 0 { - return Err(AudioError::ProcessingError( - "Invalid sample rate (0)".into(), - )); - } - - Ok(sample_rate) -} - -/// Read FLAC header (fLaC + all metadata blocks until first frame) -async fn read_flac_header(stream: &mut FlacEncodedStream) -> Result, AudioError> { - let mut header = Vec::new(); - let mut buffer = [0u8; 4]; - - // Read "fLaC" magic - stream - .read_exact(&mut buffer) - .await - .map_err(|e| AudioError::ProcessingError(format!("Failed to read FLAC magic: {}", e)))?; - - if &buffer != b"fLaC" { - return Err(AudioError::ProcessingError( - "Invalid FLAC stream: missing fLaC magic".into(), - )); - } - - header.extend_from_slice(&buffer); - - // Read metadata blocks - loop { - // Read metadata block header (1 byte type + 3 bytes length) - let mut block_header = [0u8; 4]; - stream.read_exact(&mut block_header).await.map_err(|e| { - AudioError::ProcessingError(format!("Failed to read metadata block header: {}", e)) - })?; - - let is_last = (block_header[0] & 0x80) != 0; - let block_length = - u32::from_be_bytes([0, block_header[1], block_header[2], block_header[3]]) as usize; - - header.extend_from_slice(&block_header); - - // Read metadata block data - let mut block_data = vec![0u8; block_length]; - stream.read_exact(&mut block_data).await.map_err(|e| { - AudioError::ProcessingError(format!("Failed to read metadata block data: {}", e)) - })?; - - header.extend_from_slice(&block_data); - - if is_last { - break; - } - } - - Ok(header) -} /// Create OGG-FLAC identification packet (first packet in BOS page) /// Format: https://xiph.org/flac/ogg_mapping.html diff --git a/pmoaudio-ext/src/sinks/timed_broadcast.rs b/pmoaudio-ext/src/sinks/timed_broadcast.rs index ed09ffa5..e4930325 100644 --- a/pmoaudio-ext/src/sinks/timed_broadcast.rs +++ b/pmoaudio-ext/src/sinks/timed_broadcast.rs @@ -20,6 +20,8 @@ use tracing::{info, trace, warn}; /// Tolérance pour détecter un timestamp à zéro (TopZero). const TOP_ZERO_EPSILON: f64 = 1e-9; +pub const DEFAULT_BROADCAST_MAX_LEAD_TIME: f64 = 0.5; + /// Paquet diffusé contenant la charge utile + méta timing. #[derive(Clone)] pub struct TimedPacket { @@ -543,3 +545,23 @@ impl Drop for Receiver { } } } + +/// Calculate broadcast channel capacity based on max_lead_time. +/// +/// Estimates the number of items needed to buffer max_lead_time seconds of audio. +/// Assumes ~20 items per second (50ms per chunk). +/// +/// # Arguments +/// +/// * `max_lead_time` - Maximum lead time in seconds +/// +/// # Returns +/// +/// Broadcast channel capacity (minimum 100 items) +pub(crate) fn calculate_broadcast_capacity(max_lead_time: f64) -> usize { + // Estimation: ~20 items/second (chunks de 50ms en moyenne) + // Pour 10s: 200 items + let estimated_items_per_second = 20.0; + let capacity = (max_lead_time * estimated_items_per_second) as usize; + capacity.max(100) // Minimum 100 items +}