From 8518740f3e4e6237931b1a23d303340f8b87bd58 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 12 Nov 2025 00:08:17 +0000 Subject: [PATCH] Add StreamingOggFlacSink foundation (WIP - cannot compile) Created the basic structure for OGG-FLAC streaming sink with: - OggFlacStreamHandle for HTTP client subscriptions - OggFlacClientStream implementing AsyncRead - StreamingOggFlacSinkLogic with metadata tracking - TrackBoundary detection for OGG chaining (TODO) Structure follows StreamingFlacSink pattern but designed for: 1. OGG container wrapping around FLAC frames 2. OGG chaining on TrackBoundary markers 3. Vorbis Comments metadata updates per track LIMITATION: Cannot compile/test due to missing system dependencies (alsa-sys, libsoxr-sys). The code structure is complete but encoding logic needs to be implemented and tested on a machine with proper deps. Next steps: - Implement chunk_to_pcm_bytes conversion - Implement OGG page wrapper task - Implement OGG chaining logic - Test on system with alsa-dev installed --- pmoaudio-ext/src/sinks/mod.rs | 6 + .../src/sinks/streaming_ogg_flac_sink.rs | 447 ++++++++++++++++++ 2 files changed, 453 insertions(+) create mode 100644 pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs diff --git a/pmoaudio-ext/src/sinks/mod.rs b/pmoaudio-ext/src/sinks/mod.rs index 6e014758..55a1b1b8 100755 --- a/pmoaudio-ext/src/sinks/mod.rs +++ b/pmoaudio-ext/src/sinks/mod.rs @@ -15,3 +15,9 @@ mod streaming_flac_sink; #[cfg(feature = "http-stream")] pub use streaming_flac_sink::{StreamingFlacSink, StreamHandle, MetadataSnapshot, FlacClientStream, IcyClientStream}; + +#[cfg(feature = "http-stream")] +mod streaming_ogg_flac_sink; + +#[cfg(feature = "http-stream")] +pub use streaming_ogg_flac_sink::{StreamingOggFlacSink, OggFlacStreamHandle, OggFlacClientStream}; diff --git a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs new file mode 100644 index 00000000..90aa7fd3 --- /dev/null +++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs @@ -0,0 +1,447 @@ +//! Streaming OGG-FLAC sink for multi-track radio-style streaming over HTTP. +//! +//! This sink encodes incoming audio segments into OGG-FLAC format with proper +//! OGG chaining for track boundaries. Unlike pure FLAC, OGG-FLAC supports +//! embedded metadata via Vorbis Comments that update with each track. +//! +//! # Architecture +//! +//! ```text +//! AudioSegment Pipeline +//! ↓ +//! StreamingOggFlacSink +//! ↓ +//! [TrackBoundary detection] +//! ↓ +//! [Convert AudioChunk → PCM bytes] +//! ↓ +//! ByteStreamReader (AsyncRead) +//! ↓ +//! pmoflac::encode_flac_stream() +//! ↓ +//! [OGG Wrapper Task] - wraps FLAC frames in OGG pages +//! ↓ +//! broadcast::channel (OGG-FLAC bytes) +//! ↓ +//! Multiple HTTP clients +//! ``` +//! +//! # OGG Chaining +//! +//! When a `TrackBoundary` marker is received: +//! 1. Flush current FLAC encoder +//! 2. Write OGG page with EOS flag (End of Stream) +//! 3. Extract metadata from TrackBoundary +//! 4. Start new logical bitstream with BOS flag (Beginning of Stream) +//! 5. Write new OGG-FLAC headers with updated Vorbis Comments +//! 6. Continue encoding +//! +//! This allows seamless track changes with metadata updates. +//! +//! # 100% Streaming Guarantee +//! +//! - No track buffering: AudioChunks are converted to PCM immediately +//! - FLAC encoder produces frames as soon as it has enough samples +//! - OGG wrapper reads FLAC frames and creates pages on-the-fly +//! - Pages are broadcast immediately to connected clients +//! - TrackBoundary only triggers encoder flush (no data accumulation) + +use std::collections::VecDeque; +use std::io; +use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::task::{Context, Poll}; + +use async_trait::async_trait; +use bytes::Bytes; +use pmoaudio::{ + pipeline::{AudioPipelineNode, Node, NodeLogic, PipelineHandle, StopReason}, + AudioChunk, AudioError, AudioSegment, SyncMarker, TypeRequirement, TypedAudioNode, + _AudioSegment, +}; +use pmoflac::{EncoderOptions, PcmFormat}; +use pmometadata::TrackMetadata; +use tokio::io::{AsyncRead, ReadBuf}; +use tokio::sync::{broadcast, mpsc, RwLock}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, error, info, trace, warn}; + +/// Broadcast channel capacity for OGG-FLAC bytes. +/// Same as StreamingFlacSink for consistency. +const BROADCAST_CAPACITY: usize = 128; + +/// Snapshot of track metadata (reuse from streaming_flac_sink) +pub use super::streaming_flac_sink::MetadataSnapshot; + +/// Handle for accessing the OGG-FLAC stream and metadata from HTTP handlers. +#[derive(Clone)] +pub struct OggFlacStreamHandle { + /// Broadcast sender for OGG-FLAC bytes + ogg_broadcast: broadcast::Sender, + + /// Current track metadata + metadata: Arc>, + + /// Active client counter + active_clients: Arc, + + /// Stop token to signal pipeline shutdown + stop_token: CancellationToken, + + /// Cached OGG-FLAC header (sent to new subscribers first) + ogg_header: Arc>>, +} + +impl OggFlacStreamHandle { + /// Subscribe to the OGG-FLAC stream. + /// + /// Returns an `AsyncRead` stream suitable for use with `tokio_util::io::ReaderStream`. + pub fn subscribe(&self) -> OggFlacClientStream { + let count = self.active_clients.fetch_add(1, Ordering::SeqCst); + debug!("New OGG-FLAC client subscribed (total: {})", count + 1); + + OggFlacClientStream { + rx: self.ogg_broadcast.subscribe(), + buffer: VecDeque::new(), + finished: false, + handle: self.clone(), + state: OggFlacStreamState::SendingHeader, + } + } + + /// Get the current metadata snapshot. + pub async fn get_metadata(&self) -> MetadataSnapshot { + self.metadata.read().await.clone() + } + + /// Get the number of active clients. + pub fn active_client_count(&self) -> usize { + self.active_clients.load(Ordering::SeqCst) + } +} + +/// State for OGG-FLAC stream subscription. +enum OggFlacStreamState { + SendingHeader, + Streaming, +} + +/// OGG-FLAC client stream (implements AsyncRead). +pub struct OggFlacClientStream { + rx: broadcast::Receiver, + buffer: VecDeque, + finished: bool, + handle: OggFlacStreamHandle, + state: OggFlacStreamState, +} + +impl AsyncRead for OggFlacClientStream { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + 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() { + guard.clone() + } else { + None + }; + + if let Some(header) = header_opt { + self.buffer.extend(header.iter()); + info!("Sending cached OGG-FLAC header to new client ({} bytes)", header.len()); + self.state = OggFlacStreamState::Streaming; + continue; // Now copy header to output buffer + } else { + // Header not yet captured, skip to streaming + self.state = OggFlacStreamState::Streaming; + } + } + + // If we have buffered data, copy it + 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(())); + } + + // Try to receive more data + match self.rx.try_recv() { + Ok(bytes) => { + self.buffer.extend(bytes.iter()); + } + Err(broadcast::error::TryRecvError::Empty) => { + cx.waker().wake_by_ref(); + return Poll::Pending; + } + Err(broadcast::error::TryRecvError::Lagged(skipped)) => { + warn!("OGG-FLAC client lagged, skipped {} messages", skipped); + } + Err(broadcast::error::TryRecvError::Closed) => { + self.finished = true; + return Poll::Ready(Ok(())); + } + } + } + } +} + +impl Drop for OggFlacClientStream { + fn drop(&mut self) { + let count = self.handle.active_clients.fetch_sub(1, Ordering::SeqCst); + debug!("OGG-FLAC client disconnected (remaining: {})", count - 1); + + if count == 1 { + info!("Last OGG-FLAC client disconnected, signaling pipeline stop"); + self.handle.stop_token.cancel(); + } + } +} + +/// Logic for the streaming OGG-FLAC sink. +struct StreamingOggFlacSinkLogic { + encoder_options: EncoderOptions, + bits_per_sample: u8, + metadata: Arc>, + ogg_broadcast: broadcast::Sender, + ogg_header: Arc>>, + sample_rate: Option, +} + +impl StreamingOggFlacSinkLogic { + /// Update metadata from a TrackBoundary marker. + async fn update_metadata( + &mut self, + metadata_lock: &Arc>, + timestamp_sec: f64, + ) -> Result<(), AudioError> { + let metadata = metadata_lock.read().await; + + let mut snapshot = self.metadata.write().await; + + // Extract all metadata fields + snapshot.title = metadata.get_title().await.ok().flatten(); + snapshot.artist = metadata.get_artist().await.ok().flatten(); + snapshot.album = metadata.get_album().await.ok().flatten(); + snapshot.duration = metadata.get_duration().await.ok().flatten(); + snapshot.cover_url = metadata.get_cover_url().await.ok().flatten(); + snapshot.cover_pk = metadata.get_cover_pk().await.ok().flatten(); + snapshot.year = metadata.get_year().await.ok().flatten(); + + // Extract extra fields + if let Ok(Some(extra)) = metadata.get_extra().await { + snapshot.genre = extra.get("genre").cloned(); + snapshot.track_number = extra + .get("track_number") + .and_then(|s| s.parse::().ok()); + } + + snapshot.audio_timestamp_sec = timestamp_sec; + snapshot.version += 1; + + debug!( + "OGG-FLAC metadata updated: v{} @ {:.2}s - {} - {}", + snapshot.version, + timestamp_sec, + snapshot.artist.as_deref().unwrap_or("?"), + snapshot.title.as_deref().unwrap_or("?") + ); + + Ok(()) + } +} + +#[async_trait] +impl NodeLogic for StreamingOggFlacSinkLogic { + async fn process( + &mut self, + input: Option>>, + _output: Vec>>, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + let mut input = input.ok_or_else(|| { + AudioError::ProcessingError("StreamingOggFlacSink requires an input".into()) + })?; + + info!("StreamingOggFlacSink started"); + + // TODO: Implement OGG-FLAC encoding logic + // For now, just process segments without encoding + + loop { + tokio::select! { + _ = stop_token.cancelled() => { + info!("StreamingOggFlacSink stopped by cancellation"); + break; + } + + segment = input.recv() => { + match segment { + Some(seg) => { + match &seg.segment { + _AudioSegment::Chunk(chunk) => { + // Detect sample rate from first chunk + 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 + } + + trace!( + "Received chunk: {} samples @ {:.2}s", + chunk.len(), + seg.timestamp_sec + ); + + // TODO: Convert chunk to PCM and send to encoder + } + + _AudioSegment::Sync(marker) => { + match marker.as_ref() { + SyncMarker::TrackBoundary { metadata } => { + if let Err(e) = self.update_metadata(metadata, seg.timestamp_sec).await { + error!("Failed to update metadata: {}", e); + } + // TODO: Implement OGG chaining (EOS → new BOS) + } + + SyncMarker::EndOfStream => { + info!("End of stream marker received"); + break; + } + + _ => { + trace!("Received other sync marker"); + } + } + } + } + } + + None => { + info!("Input channel closed"); + break; + } + } + } + } + } + + info!("StreamingOggFlacSink processing complete"); + Ok(()) + } + + async fn cleanup(&mut self, reason: StopReason) -> Result<(), AudioError> { + info!("StreamingOggFlacSink cleanup: {:?}", reason); + Ok(()) + } +} + +/// Streaming OGG-FLAC sink for multi-client HTTP streaming with track metadata. +pub struct StreamingOggFlacSink { + inner: Node, +} + +impl StreamingOggFlacSink { + /// Create a new streaming OGG-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, OggFlacStreamHandle) { + // Validate bit depth + if ![16, 24, 32].contains(&bits_per_sample) { + panic!("bits_per_sample must be 16, 24, or 32"); + } + + // Shared metadata + let metadata = Arc::new(RwLock::new(MetadataSnapshot::default())); + + // Broadcast channel for OGG-FLAC bytes + let (ogg_broadcast, _) = broadcast::channel(BROADCAST_CAPACITY); + + // OGG-FLAC header cache + let ogg_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(), + metadata: metadata.clone(), + active_clients, + stop_token: stop_token.clone(), + ogg_header: ogg_header.clone(), + }; + + let logic = StreamingOggFlacSinkLogic { + encoder_options, + bits_per_sample, + metadata, + ogg_broadcast, + ogg_header, + sample_rate: None, + }; + + let sink = Self { + inner: Node::new_with_input(logic, 16), + }; + + (sink, handle) + } +} + +#[async_trait] +impl AudioPipelineNode for StreamingOggFlacSink { + fn get_tx(&self) -> Option>> { + self.inner.get_tx() + } + + fn register(&mut self, _child: Box) { + panic!("StreamingOggFlacSink 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 StreamingOggFlacSink { + fn input_type(&self) -> Option { + Some(TypeRequirement::any_integer()) + } + + fn output_type(&self) -> Option { + None + } +}