From acf504aaec0e705f12517e61d45efcf72c612620 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 12 Nov 2025 00:14:51 +0000 Subject: [PATCH] Complete StreamingOggFlacSink implementation (FLAC passthrough) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implemented a fully functional OGG-FLAC streaming sink that: - Converts AudioChunk to PCM bytes (chunk_to_pcm_bytes) - Encodes to FLAC using pmoflac::encode_flac_stream - Broadcasts FLAC frames to multiple HTTP clients - Caches and resends header to late-joining clients - Tracks metadata from TrackBoundary markers - Uses ByteStreamReader for mpsc → AsyncRead conversion Current limitations (TODO): - OGG wrapping: Currently passes through pure FLAC (broadcast_ogg_flac_stream needs proper OGG page generation) - OGG chaining: TrackBoundary detection is in place but doesn't restart encoder with new metadata yet This provides a working base that compiles and should stream FLAC audio. OGG containerization and chaining will be added next. Architecture matches StreamingFlacSink pattern for consistency. --- .../src/sinks/streaming_ogg_flac_sink.rs | 296 +++++++++++++++++- 1 file changed, 290 insertions(+), 6 deletions(-) diff --git a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs index 90aa7fd3..09dd5b1b 100644 --- a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs @@ -60,9 +60,9 @@ use pmoaudio::{ AudioChunk, AudioError, AudioSegment, SyncMarker, TypeRequirement, TypedAudioNode, _AudioSegment, }; -use pmoflac::{EncoderOptions, PcmFormat}; +use pmoflac::{encode_flac_stream, EncoderOptions, FlacEncodedStream, PcmFormat}; use pmometadata::TrackMetadata; -use tokio::io::{AsyncRead, ReadBuf}; +use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf}; use tokio::sync::{broadcast, mpsc, RwLock}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, trace, warn}; @@ -212,17 +212,71 @@ impl Drop for OggFlacClientStream { } } +/// Internal state for encoder initialization. +struct EncoderState { + broadcaster_task: tokio::task::JoinHandle<()>, +} + /// Logic for the streaming OGG-FLAC sink. struct StreamingOggFlacSinkLogic { encoder_options: EncoderOptions, bits_per_sample: u8, + pcm_tx: mpsc::Sender>, + pcm_rx: Option>>, metadata: Arc>, ogg_broadcast: broadcast::Sender, ogg_header: Arc>>, + encoder_state: Option, sample_rate: Option, } impl StreamingOggFlacSinkLogic { + /// Initialize the FLAC encoder once we know the sample rate. + async fn initialize_encoder(&mut self, sample_rate: u32) -> Result<(), AudioError> { + if self.encoder_state.is_some() { + return Ok(()); // Already initialized + } + + info!("Initializing OGG-FLAC encoder with sample rate: {} Hz", sample_rate); + + // Take the PCM receiver (we only initialize once) + let pcm_rx = self.pcm_rx.take().ok_or_else(|| { + AudioError::ProcessingError("PCM receiver already consumed".into()) + })?; + + // Create ByteStreamReader for the encoder + let pcm_reader = ByteStreamReader::new(pcm_rx); + + // Create PCM format + let pcm_format = PcmFormat { + sample_rate, + channels: 2, + bits_per_sample: self.bits_per_sample, + }; + + // Start the FLAC encoder + let flac_stream = encode_flac_stream(pcm_reader, pcm_format, self.encoder_options.clone()) + .await + .map_err(|e| AudioError::ProcessingError(format!("Failed to start FLAC encoder: {}", e)))?; + + info!("OGG-FLAC encoder initialized successfully"); + + // Spawn OGG wrapper + broadcaster task + let ogg_broadcast = self.ogg_broadcast.clone(); + let ogg_header = self.ogg_header.clone(); + let broadcaster_task = tokio::spawn(async move { + if let Err(e) = broadcast_ogg_flac_stream(flac_stream, ogg_broadcast, ogg_header).await { + error!("OGG broadcaster task error: {}", e); + } + }); + + self.encoder_state = Some(EncoderState { broadcaster_task }); + + info!("OGG broadcaster task spawned"); + + Ok(()) + } + /// Update metadata from a TrackBoundary marker. async fn update_metadata( &mut self, @@ -294,21 +348,31 @@ impl NodeLogic for StreamingOggFlacSinkLogic { Some(seg) => { match &seg.segment { _AudioSegment::Chunk(chunk) => { - // Detect sample rate from first chunk + // Detect sample rate from first chunk and initialize encoder if self.sample_rate.is_none() { let sample_rate = chunk.sample_rate(); self.sample_rate = Some(sample_rate); info!("Detected sample rate: {} Hz", sample_rate); - // TODO: Initialize OGG-FLAC encoder + + // Initialize the FLAC encoder now + self.initialize_encoder(sample_rate).await?; } + // Convert chunk to PCM bytes + let pcm_bytes = chunk_to_pcm_bytes(&chunk, self.bits_per_sample)?; + trace!( - "Received chunk: {} samples @ {:.2}s", + "Sending PCM chunk: {} bytes, {} samples @ {:.2}s", + pcm_bytes.len(), chunk.len(), seg.timestamp_sec ); - // TODO: Convert chunk to PCM and send to encoder + // Send to FLAC encoder + if let Err(e) = self.pcm_tx.send(pcm_bytes).await { + warn!("Failed to send PCM data to encoder: {}", e); + break; + } } _AudioSegment::Sync(marker) => { @@ -379,6 +443,9 @@ impl StreamingOggFlacSink { 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())); @@ -403,9 +470,12 @@ impl StreamingOggFlacSink { let logic = StreamingOggFlacSinkLogic { encoder_options, bits_per_sample, + pcm_tx, + pcm_rx: Some(pcm_rx), metadata, ogg_broadcast, ogg_header, + encoder_state: None, sample_rate: None, }; @@ -445,3 +515,217 @@ impl TypedAudioNode for StreamingOggFlacSink { None } } + +/// AsyncRead adapter for mpsc::Receiver>. +struct ByteStreamReader { + rx: mpsc::Receiver>, + buffer: VecDeque, + finished: bool, +} + +impl ByteStreamReader { + fn new(rx: mpsc::Receiver>) -> Self { + Self { + rx, + buffer: VecDeque::new(), + finished: false, + } + } +} + +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(bytes)) => { + if bytes.is_empty() { + continue; + } + self.buffer.extend(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. +/// +/// For now, this is a simplified version that just passes through FLAC bytes without OGG wrapping. +/// TODO: Implement proper OGG page generation with BOS/EOS flags and Vorbis Comments. +async fn broadcast_ogg_flac_stream( + mut flac_stream: FlacEncodedStream, + broadcast_tx: broadcast::Sender, + header_cache: Arc>>, +) -> Result<(), AudioError> { + info!("OGG-FLAC broadcaster task started (FLAC passthrough mode - OGG wrapping TODO)"); + + let mut buffer = vec![0u8; 8192]; // 8KB buffer for reading + let mut total_bytes = 0u64; + let mut header_captured = false; + + loop { + match flac_stream.read(&mut buffer).await { + Ok(0) => { + // EOF + info!("OGG-FLAC encoder stream ended, total bytes: {}", total_bytes); + break; + } + Ok(n) => { + total_bytes += n as u64; + trace!("Read {} bytes from FLAC encoder (total: {})", n, total_bytes); + + // Broadcast to all clients (TODO: wrap in OGG pages) + let bytes = Bytes::copy_from_slice(&buffer[..n]); + + // Capture first chunk as header if it contains "fLaC" + if !header_captured && bytes.len() >= 4 && &bytes[0..4] == b"fLaC" { + *header_cache.write().await = Some(bytes.clone()); + header_captured = true; + info!("FLAC header captured ({} bytes) - will be wrapped in OGG later", bytes.len()); + } + + if let Err(e) = broadcast_tx.send(bytes) { + // No receivers, but that's okay - clients may not be connected yet + trace!("No active receivers for OGG-FLAC broadcast: {}", e); + } + } + Err(e) => { + error!("Error reading from FLAC encoder: {}", e); + return Err(AudioError::ProcessingError(format!( + "FLAC encoder read error: {}", + e + ))); + } + } + } + + // Wait for the encoder to finish cleanly + if let Err(e) = flac_stream.wait().await { + error!("FLAC encoder error during cleanup: {}", e); + return Err(AudioError::ProcessingError(format!( + "FLAC encoder error: {}", + e + ))); + } + + info!("OGG-FLAC broadcaster task completed successfully"); + Ok(()) +}