diff --git a/Cargo.lock b/Cargo.lock index b4fc4f12..2e8bcf48 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2867,6 +2867,7 @@ name = "pmoaudio-ext" version = "0.1.0" dependencies = [ "async-trait", + "bytes", "pmoaudio", "pmoaudiocache", "pmocache", @@ -2874,6 +2875,7 @@ dependencies = [ "pmoflac", "pmometadata", "pmoplaylist", + "serde", "tokio", "tokio-util", "tracing", diff --git a/pmoaudio-ext/Cargo.toml b/pmoaudio-ext/Cargo.toml index 44212103..2bc3f364 100644 --- a/pmoaudio-ext/Cargo.toml +++ b/pmoaudio-ext/Cargo.toml @@ -16,6 +16,11 @@ pmometadata = { path = "../pmometadata", optional = true } # Optional dependencies for playlist integration pmoplaylist = { path = "../pmoplaylist", optional = true } pmocache = { path = "../pmocache", optional = true } + +# Optional dependencies for streaming feature +bytes = { version = "1.5", optional = true } +serde = { version = "1.0", features = ["derive"], optional = true } + # Async runtime tokio = { version = "1.0", features = ["full"] } tokio-util = { version = "0.7" } @@ -28,4 +33,5 @@ tracing = "0.1" default = [] cache-sink = ["dep:pmoaudiocache", "dep:pmoflac", "dep:pmometadata"] playlist = ["cache-sink", "dep:pmoplaylist", "dep:pmocache"] -all = ["cache-sink", "playlist"] +streaming = ["dep:pmoflac", "dep:pmometadata", "dep:bytes", "dep:serde"] +all = ["cache-sink", "playlist", "streaming"] diff --git a/pmoaudio-ext/src/sinks/mod.rs b/pmoaudio-ext/src/sinks/mod.rs index 9cb71261..189d1e11 100755 --- a/pmoaudio-ext/src/sinks/mod.rs +++ b/pmoaudio-ext/src/sinks/mod.rs @@ -9,3 +9,15 @@ mod flac_cache_sink; #[cfg(feature = "cache-sink")] pub use flac_cache_sink::{FlacCacheSink, FlacCacheSinkStats, TrackStats}; + +#[cfg(feature = "streaming")] +mod streaming_flac_sink; + +#[cfg(feature = "streaming")] +mod streaming_ogg_flac_sink; + +#[cfg(feature = "streaming")] +pub use streaming_flac_sink::{StreamingFlacSink, StreamHandle, MetadataSnapshot}; + +#[cfg(feature = "streaming")] +pub use streaming_ogg_flac_sink::{StreamingOggFlacSink, OggFlacStreamHandle}; diff --git a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs new file mode 100644 index 00000000..ce0ececa --- /dev/null +++ b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs @@ -0,0 +1,1071 @@ +//! Streaming FLAC sink for multi-track radio-style streaming over HTTP. +//! +//! This sink encodes incoming audio segments into a continuous FLAC stream, +//! broadcasts it to multiple concurrent clients (UPnP renderers, web players, etc.), +//! and supports ICY metadata for "Now Playing" updates. +//! +//! # Architecture +//! +//! ```text +//! AudioSegment Pipeline +//! ↓ +//! StreamingFlacSink +//! ↓ +//! [Convert AudioChunk → PCM bytes] +//! ↓ +//! ByteStreamReader (AsyncRead) +//! ↓ +//! pmoflac::encode_flac_stream() +//! ↓ +//! [Broadcaster Task] +//! ↓ +//! broadcast::channel (FLAC bytes) +//! ↓ +//! Multiple clients via StreamHandle::subscribe() +//! ├─ FLAC pure (for standard renderers) +//! └─ ICY-wrapped FLAC (for metadata-aware clients) +//! ``` +//! +//! # Usage Example +//! +//! ```no_run +//! use pmoaudio_ext::sinks::StreamingFlacSink; +//! use pmoflac::EncoderOptions; +//! +//! // Create the sink and get the handle for HTTP serving +//! let (sink, handle) = StreamingFlacSink::new( +//! EncoderOptions::default(), +//! 16, // bits per sample +//! ); +//! +//! // Add to audio pipeline +//! source.register(Box::new(sink)); +//! +//! // In your HTTP handler (e.g., pmoparadise): +//! if headers.get("Icy-MetaData") == Some("1") { +//! // ICY mode with metadata updates +//! let stream = handle.subscribe_icy(); +//! response.header("icy-metaint", "16000"); +//! Body::from_stream(ReaderStream::new(stream)) +//! } else { +//! // Pure FLAC mode +//! let stream = handle.subscribe_flac(); +//! Body::from_stream(ReaderStream::new(stream)) +//! } +//! ``` + +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 std::time::Duration; + +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::{encode_flac_stream, EncoderOptions, FlacEncodedStream, PcmFormat}; +use pmometadata::TrackMetadata; +use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf}; +use tokio::sync::{broadcast, mpsc, RwLock}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, error, info, trace, warn}; + +/// Default ICY metadata interval (bytes of audio between metadata blocks). +/// Standard value used by most streaming servers. +const DEFAULT_ICY_METAINT: usize = 16000; + +/// Broadcast channel capacity for FLAC bytes. +/// Set to 128 to provide ~10 seconds of buffer for network jitter. +/// With TimerNode pacing the stream to real-time, this is sufficient +/// while keeping metadata synchronized (larger buffers cause metadata drift). +const BROADCAST_CAPACITY: usize = 128; + +/// 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. +/// This is much smaller than the pipeline TimerNode's 3.0s to provide tighter control. +const BROADCAST_MAX_LEAD_TIME: f64 = 0.5; + +/// 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, +} + +/// Snapshot of track metadata at a point in time. +/// +/// This structure is shared between the sink and clients to provide +/// real-time metadata updates as tracks change in a continuous stream. +#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] +pub struct MetadataSnapshot { + /// Track title + pub title: Option, + /// Artist name + pub artist: Option, + /// Album name + pub album: Option, + /// Track duration + #[serde(skip_serializing_if = "Option::is_none")] + pub duration: Option, + /// Cover image URL (external/original) + #[serde(skip_serializing_if = "Option::is_none")] + pub cover_url: Option, + /// Cover primary key in local cache (for constructing server URL) + #[serde(skip_serializing_if = "Option::is_none")] + pub cover_pk: Option, + /// Track number + #[serde(skip_serializing_if = "Option::is_none")] + pub track_number: Option, + /// Album artist + #[serde(skip_serializing_if = "Option::is_none")] + pub album_artist: Option, + /// Genre + #[serde(skip_serializing_if = "Option::is_none")] + pub genre: Option, + /// Year + #[serde(skip_serializing_if = "Option::is_none")] + pub year: Option, + /// Audio timestamp where this metadata became active (seconds) + pub audio_timestamp_sec: f64, + /// Version counter incremented on each update (for client-side change detection) + pub version: u64, +} + +/// Handle for accessing the FLAC stream and metadata from HTTP handlers. +/// +/// This handle is designed to be cloned and used by multiple HTTP clients +/// simultaneously. Each client gets its own independent stream by subscribing. +#[derive(Clone)] +pub struct StreamHandle { + /// Broadcast sender for FLAC bytes (pure mode) + flac_broadcast: broadcast::Sender, + + /// Current track metadata (read-only for consumers) + metadata: Arc>, + + /// Active client counter + active_clients: Arc, + + /// Stop token to signal pipeline shutdown + stop_token: CancellationToken, + + /// Cached FLAC header (sent to new subscribers first) + flac_header: Arc>>, +} + +impl StreamHandle { + /// Subscribe to the FLAC stream in pure mode (no ICY metadata). + /// + /// Returns an `AsyncRead` stream suitable for use with `tokio_util::io::ReaderStream`. + pub fn subscribe_flac(&self) -> FlacClientStream { + let count = self.active_clients.fetch_add(1, Ordering::SeqCst); + debug!("New FLAC client subscribed (total: {})", count + 1); + + FlacClientStream { + rx: self.flac_broadcast.subscribe(), + buffer: VecDeque::new(), + finished: false, + handle: self.clone(), + state: FlacStreamState::SendingHeader, + } + } + + /// Subscribe to the FLAC stream with ICY metadata injection. + /// + /// Returns an `AsyncRead` stream that injects ICY metadata blocks + /// at regular intervals (default: every 16000 bytes). + pub fn subscribe_icy(&self) -> IcyClientStream { + self.subscribe_icy_with_interval(DEFAULT_ICY_METAINT) + } + + /// Subscribe to the FLAC stream with custom ICY metadata interval. + pub fn subscribe_icy_with_interval(&self, metaint: usize) -> IcyClientStream { + let count = self.active_clients.fetch_add(1, Ordering::SeqCst); + debug!("New ICY client subscribed (total: {}, metaint: {})", count + 1, metaint); + + IcyClientStream { + rx: self.flac_broadcast.subscribe(), + metadata: self.metadata.clone(), + metaint, + byte_count: 0, + buffer: VecDeque::new(), + current_metadata_version: 0, + cached_icy_metadata: Bytes::new(), + finished: false, + handle: self.clone(), + state: FlacStreamState::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) + } + + /// Check if the stream should be stopped (no more clients). + pub fn should_stop(&self) -> bool { + self.active_clients.load(Ordering::SeqCst) == 0 + } +} + +/// State for FLAC stream subscription. +enum FlacStreamState { + SendingHeader, + Streaming, +} + +/// Pure FLAC client stream (implements AsyncRead). +pub struct FlacClientStream { + rx: broadcast::Receiver, + buffer: VecDeque, + finished: bool, + handle: StreamHandle, + state: FlacStreamState, +} + +impl AsyncRead for FlacClientStream { + 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, FlacStreamState::SendingHeader) { + let header_opt = if let Ok(guard) = self.handle.flac_header.try_read() { + guard.clone() + } else { + None + }; + + if let Some(header) = header_opt { + self.buffer.extend(header.iter()); + info!("Sending cached FLAC header to new client ({} bytes)", header.len()); + self.state = FlacStreamState::Streaming; + continue; // Now copy header to output buffer + } else { + // Header not yet captured or can't acquire lock, skip to streaming + self.state = FlacStreamState::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) => { + // No data available right now. + // Schedule a wakeup after a small delay to avoid busy-loop polling. + let waker = cx.waker().clone(); + tokio::spawn(async move { + tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; + waker.wake(); + }); + return Poll::Pending; + } + Err(broadcast::error::TryRecvError::Lagged(skipped)) => { + warn!("FLAC client lagged, skipped {} messages", skipped); + // Continue to try receiving again + } + Err(broadcast::error::TryRecvError::Closed) => { + self.finished = true; + return Poll::Ready(Ok(())); + } + } + } + } +} + +impl Drop for FlacClientStream { + fn drop(&mut self) { + let count = self.handle.active_clients.fetch_sub(1, Ordering::SeqCst); + debug!("FLAC client disconnected (remaining: {})", count - 1); + + if count == 1 { + info!("Last client disconnected, signaling pipeline stop"); + self.handle.stop_token.cancel(); + } + } +} + +/// ICY-wrapped FLAC client stream (implements AsyncRead). +/// +/// This stream injects ICY metadata blocks at regular intervals, +/// allowing clients to display "Now Playing" information. +pub struct IcyClientStream { + rx: broadcast::Receiver, + metadata: Arc>, + metaint: usize, + byte_count: usize, + buffer: VecDeque, + current_metadata_version: u64, + cached_icy_metadata: Bytes, + finished: bool, + handle: StreamHandle, + state: FlacStreamState, +} + +impl IcyClientStream { + /// Format metadata as ICY metadata block. + /// + /// ICY format: StreamTitle='Artist - Title';StreamUrl='url'; + /// Padded to multiple of 16 bytes, prefixed with length byte. + /// + /// If cover_pk is available, constructs a URL for the cover image: + /// - If pmoserver is initialized: http://server/covers/image/{pk}/256 + /// - Otherwise: relative URL /covers/image/{pk}/256 + fn format_icy_metadata(meta: &MetadataSnapshot) -> Bytes { + let title = meta.title.as_deref().unwrap_or("Unknown"); + let artist = meta.artist.as_deref().unwrap_or("Unknown Artist"); + + // Build ICY metadata string with cover URL if available + let mut metadata_str = format!("StreamTitle='{} - {}';", artist, title); + + // Add cover URL if we have a cover_pk + if let Some(pk) = &meta.cover_pk { + // Use relative URL /covers/image/{pk}/256 + // This works when streaming from the same server that serves covers + // VLC and other players will resolve relative URLs correctly + metadata_str.push_str(&format!("StreamUrl='/covers/image/{}/256';", pk)); + } else if let Some(url) = &meta.cover_url { + // Fallback to external cover URL if no local pk + metadata_str.push_str(&format!("StreamUrl='{}';", url)); + } + + // ICY metadata is padded to multiple of 16 bytes + let metadata_bytes = metadata_str.as_bytes(); + let length = metadata_bytes.len(); + let padded_length = ((length + 15) / 16) * 16; + let length_byte = (padded_length / 16) as u8; + + let mut result = Vec::with_capacity(1 + padded_length); + result.push(length_byte); + result.extend_from_slice(metadata_bytes); + result.resize(1 + padded_length, 0); // Pad with zeros + + Bytes::from(result) + } + + /// Get metadata block if it needs to be inserted. + async fn get_metadata_if_changed(&mut self) -> Option { + let meta = self.metadata.read().await; + if meta.version > self.current_metadata_version { + self.current_metadata_version = meta.version; + let icy_meta = Self::format_icy_metadata(&meta); + self.cached_icy_metadata = icy_meta.clone(); + Some(icy_meta) + } else if self.byte_count == 0 { + // Always send metadata at the start + Some(self.cached_icy_metadata.clone()) + } else { + // No change, send empty metadata block + Some(Bytes::from(vec![0u8])) + } + } +} + +impl AsyncRead for IcyClientStream { + 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, FlacStreamState::SendingHeader) { + let header_opt = if let Ok(guard) = self.handle.flac_header.try_read() { + guard.clone() + } else { + None + }; + + if let Some(header) = header_opt { + self.buffer.extend(header.iter()); + info!("Sending cached FLAC header to new ICY client ({} bytes)", header.len()); + self.state = FlacStreamState::Streaming; + continue; // Now copy header to output buffer + } else { + // Header not yet captured or can't acquire lock, skip to streaming + self.state = FlacStreamState::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(())); + } + + // Check if we need to insert metadata + if self.byte_count % self.metaint == 0 && self.byte_count > 0 { + // Time to insert ICY metadata + // Use try_read to avoid blocking in poll context + let update = { + if let Ok(meta) = self.metadata.try_read() { + if meta.version > self.current_metadata_version { + Some((meta.version, Self::format_icy_metadata(&meta))) + } else { + None + } + } else { + None + } + }; + + if let Some((new_version, new_metadata)) = update { + self.current_metadata_version = new_version; + self.cached_icy_metadata = new_metadata; + } + + let icy_data = self.cached_icy_metadata.clone(); + self.buffer.extend(icy_data.iter()); + self.byte_count = 0; // Reset counter after metadata + continue; + } + + // Try to receive audio data + match self.rx.try_recv() { + Ok(bytes) => { + // Calculate how many bytes until next metadata block + let until_metadata = self.metaint - (self.byte_count % self.metaint); + let to_buffer = bytes.len().min(until_metadata); + + self.buffer.extend(bytes[..to_buffer].iter()); + self.byte_count += to_buffer; + + // If we have more data, we'll process it in the next iteration + if to_buffer < bytes.len() { + // Save remaining for next iteration + // For now, we'll just drop it and get it again + // TODO: Improve this + } + } + Err(broadcast::error::TryRecvError::Empty) => { + // No data available right now. + // Schedule a wakeup after a small delay to avoid busy-loop polling. + let waker = cx.waker().clone(); + tokio::spawn(async move { + tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; + waker.wake(); + }); + return Poll::Pending; + } + Err(broadcast::error::TryRecvError::Lagged(skipped)) => { + warn!("ICY client lagged, skipped {} messages", skipped); + } + Err(broadcast::error::TryRecvError::Closed) => { + self.finished = true; + return Poll::Ready(Ok(())); + } + } + } + } +} + +impl Drop for IcyClientStream { + fn drop(&mut self) { + let count = self.handle.active_clients.fetch_sub(1, Ordering::SeqCst); + debug!("ICY client disconnected (remaining: {})", count - 1); + + if count == 1 { + info!("Last client disconnected, signaling pipeline stop"); + self.handle.stop_token.cancel(); + } + } +} + +/// Internal state for encoder initialization. +struct EncoderState { + broadcaster_task: tokio::task::JoinHandle<()>, +} + +/// Logic for the streaming FLAC sink. +struct StreamingFlacSinkLogic { + encoder_options: EncoderOptions, + bits_per_sample: u8, + pcm_tx: mpsc::Sender, + pcm_rx: Option>, + metadata: Arc>, + flac_broadcast: broadcast::Sender, + flac_header: Arc>>, + encoder_state: Option, + sample_rate: Option, +} + +impl StreamingFlacSinkLogic { + /// 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 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 shared timestamp for pacing + let current_timestamp = Arc::new(RwLock::new(0.0f64)); + + // Create ByteStreamReader for the encoder + let pcm_reader = ByteStreamReader::new(pcm_rx, current_timestamp.clone()); + + // 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!("FLAC encoder initialized successfully"); + + // Spawn broadcaster task with timestamp for pacing + let flac_broadcast = self.flac_broadcast.clone(); + let flac_header = self.flac_header.clone(); + let broadcaster_task = tokio::spawn(async move { + if let Err(e) = broadcast_flac_stream(flac_stream, flac_broadcast, flac_header, current_timestamp).await { + error!("Broadcaster task error: {}", e); + } + }); + + self.encoder_state = Some(EncoderState { broadcaster_task }); + + info!("Broadcaster task spawned"); + + Ok(()) + } + + /// 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!( + "Metadata updated: v{} @ {:.2}s - {} - {} (cover_pk: {:?})", + snapshot.version, + timestamp_sec, + snapshot.artist.as_deref().unwrap_or("?"), + snapshot.title.as_deref().unwrap_or("?"), + snapshot.cover_pk + ); + + Ok(()) + } +} + +#[async_trait] +impl NodeLogic for StreamingFlacSinkLogic { + async fn process( + &mut self, + input: Option>>, + _output: Vec>>, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + let mut input = input.ok_or_else(|| { + AudioError::ProcessingError("StreamingFlacSink requires an input".into()) + })?; + + info!("StreamingFlacSink started"); + + // We'll initialize the encoder lazily when we get the first chunk + // For now, just process segments + + loop { + tokio::select! { + _ = stop_token.cancelled() => { + info!("StreamingFlacSink stopped by cancellation"); + break; + } + + segment = input.recv() => { + match segment { + Some(seg) => { + match &seg.segment { + _AudioSegment::Chunk(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); + + // 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!( + "Sending PCM chunk: {} bytes, {} samples @ {:.2}s", + pcm_bytes.len(), + chunk.len(), + seg.timestamp_sec + ); + + // Send to FLAC encoder with timestamp + let pcm_chunk = PcmChunk { + bytes: pcm_bytes, + timestamp_sec: seg.timestamp_sec, + }; + if let Err(e) = self.pcm_tx.send(pcm_chunk).await { + warn!("Failed to send PCM data to encoder: {}", e); + break; + } + } + + _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); + } + } + + SyncMarker::EndOfStream => { + info!("End of stream marker received"); + break; + } + + _ => { + trace!("Received other sync marker"); + } + } + } + } + } + + None => { + info!("Input channel closed"); + break; + } + } + } + } + } + + info!("StreamingFlacSink processing complete"); + Ok(()) + } + + async fn cleanup(&mut self, reason: StopReason) -> Result<(), AudioError> { + info!("StreamingFlacSink cleanup: {:?}", reason); + Ok(()) + } +} + +/// Broadcaster task: reads FLAC bytes from encoder and broadcasts to all clients. +/// Implements precise real-time pacing based on audio timestamps. +async fn broadcast_flac_stream( + mut flac_stream: FlacEncodedStream, + broadcast_tx: broadcast::Sender, + header_cache: Arc>>, + current_timestamp: Arc>, +) -> Result<(), AudioError> { + info!("Broadcaster task started with precise timestamp-based pacing"); + + // Reduced buffer size from 8KB to 512 bytes for smoother streaming + // This prevents burst transmission that causes buffer cycling in FFPlay + let mut buffer = vec![0u8; 512]; + let mut total_bytes = 0u64; + let mut header_captured = false; + let start_time = std::time::Instant::now(); + + loop { + match flac_stream.read(&mut buffer).await { + Ok(0) => { + // EOF + info!("FLAC encoder stream ended, total bytes: {}", total_bytes); + break; + } + Ok(n) => { + total_bytes += n as u64; + if total_bytes % 100000 == 0 || total_bytes < 10000 { + trace!("Read {} bytes from FLAC encoder (total: {})", n, total_bytes); + } + + // Precise pacing based on audio timestamp + let audio_timestamp = *current_timestamp.read().await; + let elapsed = start_time.elapsed().as_secs_f64(); + let lead_time = audio_timestamp - elapsed; + + if lead_time > BROADCAST_MAX_LEAD_TIME { + let sleep_duration = lead_time - BROADCAST_MAX_LEAD_TIME; + debug!( + "Broadcaster pacing: sleeping {:.3}s (audio_ts={:.3}s, elapsed={:.3}s, lead={:.3}s)", + sleep_duration, audio_timestamp, elapsed, lead_time + ); + tokio::time::sleep(tokio::time::Duration::from_secs_f64(sleep_duration)).await; + } + + // Broadcast to all clients + 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)", bytes.len()); + } + + let num_receivers = broadcast_tx.receiver_count(); + 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 FLAC broadcast: {}", e); + } else if num_receivers > 0 { + trace!("Broadcasted {} bytes to {} receivers", n, num_receivers); + } + } + 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!("Broadcaster task completed successfully"); + 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) { + // 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())); + + // Broadcast channel for FLAC bytes + let (flac_broadcast, _) = 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(), + }; + + let logic = StreamingFlacSinkLogic { + encoder_options, + bits_per_sample, + pcm_tx, + pcm_rx: Some(pcm_rx), + metadata, + flac_broadcast, + flac_header, + encoder_state: None, + sample_rate: None, + }; + + 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>, +} + +impl ByteStreamReader { + fn new(rx: mpsc::Receiver, current_timestamp: Arc>) -> Self { + Self { + rx, + buffer: VecDeque::new(), + finished: false, + current_timestamp, + } + } +} + +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 for broadcaster pacing + if let Ok(mut ts) = self.current_timestamp.try_write() { + *ts = chunk.timestamp_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 new file mode 100644 index 00000000..20053c40 --- /dev/null +++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs @@ -0,0 +1,1067 @@ +//! 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::{encode_flac_stream, EncoderOptions, FlacEncodedStream, PcmFormat}; +use pmometadata::TrackMetadata; +use tokio::io::{AsyncRead, AsyncReadExt, 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; + +/// 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 BROADCAST_MAX_LEAD_TIME: f64 = 0.5; + +/// 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, +} + +/// 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) => { + // No data available right now. + // Schedule a wakeup after a small delay to avoid busy-loop polling. + let waker = cx.waker().clone(); + tokio::spawn(async move { + tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; + waker.wake(); + }); + 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(); + } + } +} + +/// 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 shared timestamp for pacing + let current_timestamp = Arc::new(RwLock::new(0.0f64)); + + // Create ByteStreamReader for the encoder + let pcm_reader = ByteStreamReader::new(pcm_rx, current_timestamp.clone()); + + // 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 with timestamp for pacing + 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, current_timestamp).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, + 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 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); + + // 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!( + "Sending PCM chunk: {} bytes, {} samples @ {:.2}s", + pcm_bytes.len(), + chunk.len(), + seg.timestamp_sec + ); + + // Send to FLAC encoder with timestamp + let pcm_chunk = PcmChunk { + bytes: pcm_bytes, + timestamp_sec: seg.timestamp_sec, + }; + if let Err(e) = self.pcm_tx.send(pcm_chunk).await { + warn!("Failed to send PCM data to encoder: {}", e); + break; + } + } + + _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"); + } + + // Create PCM channel (bounded for backpressure) + let (pcm_tx, pcm_rx) = mpsc::channel::(16); + + // 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, + pcm_tx, + pcm_rx: Some(pcm_rx), + metadata, + ogg_broadcast, + ogg_header, + encoder_state: None, + 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 + } +} + +/// 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>, +} + +impl ByteStreamReader { + fn new(rx: mpsc::Receiver, current_timestamp: Arc>) -> Self { + Self { + rx, + buffer: VecDeque::new(), + finished: false, + current_timestamp, + } + } +} + +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 for broadcaster pacing + if let Ok(mut ts) = self.current_timestamp.try_write() { + *ts = chunk.timestamp_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. +async fn broadcast_ogg_flac_stream( + mut flac_stream: FlacEncodedStream, + broadcast_tx: broadcast::Sender, + header_cache: Arc>>, + current_timestamp: Arc>, +) -> Result<(), AudioError> { + info!("OGG-FLAC broadcaster task started with precise timestamp-based pacing"); + + let stream_serial = rand::random::(); + let mut ogg_writer = OggPageWriter::new(stream_serial); + + let mut total_ogg_bytes = 0u64; + let mut header_captured = false; + let start_time = std::time::Instant::now(); + + // Step 1: Read FLAC header (fLaC + metadata blocks) + let flac_header = read_flac_header(&mut flac_stream).await?; + info!("Read FLAC header: {} bytes", flac_header.len()); + + // Step 2: Create OGG-FLAC identification packet (BOS) + // Format according to https://xiph.org/flac/ogg_mapping.html + let ogg_flac_id = create_ogg_flac_identification(&flac_header)?; + info!("Created OGG-FLAC identification packet: {} bytes", ogg_flac_id.len()); + + let bos_page = ogg_writer.create_page(&ogg_flac_id, true, false, false); + let bos_bytes = Bytes::from(bos_page); + + // Step 3: Create Vorbis Comment page (empty for now, metadata comes from /metadata endpoint) + let vorbis_comment = create_empty_vorbis_comment(); + let comment_page = ogg_writer.create_page(&vorbis_comment, false, false, false); + let comment_bytes = Bytes::from(comment_page); + + // Cache the header (BOS + Comment pages) + let mut cached_header = Vec::new(); + cached_header.extend_from_slice(&bos_bytes); + cached_header.extend_from_slice(&comment_bytes); + *header_cache.write().await = Some(Bytes::from(cached_header)); + header_captured = true; + info!("OGG-FLAC header cached ({} bytes: BOS + Vorbis Comment)", bos_bytes.len() + comment_bytes.len()); + + // Broadcast header + let _ = broadcast_tx.send(bos_bytes); + total_ogg_bytes += comment_bytes.len() as u64; + let _ = broadcast_tx.send(comment_bytes); + + // Step 4: Read FLAC stream and create OGG packets + // According to OGG FLAC spec, we put the complete FLAC stream in a single logical bitstream, + // but split it into reasonable page sizes for streaming + + let mut flac_data = Vec::new(); + let mut read_buffer = vec![0u8; 8192]; + + loop { + match flac_stream.read(&mut read_buffer).await { + Ok(0) => { + // EOF - create final page with EOS flag and any remaining data + if !flac_data.is_empty() { + let eos_page = ogg_writer.create_page(&flac_data, false, true, false); + let eos_bytes = Bytes::from(eos_page); + total_ogg_bytes += eos_bytes.len() as u64; + let _ = broadcast_tx.send(eos_bytes); + info!("Sent final EOS page with {} bytes of data", flac_data.len()); + } else { + // Send empty EOS page + 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; + let _ = broadcast_tx.send(eos_bytes); + info!("Sent empty EOS page"); + } + + info!("OGG-FLAC stream ended, total OGG bytes: {}", total_ogg_bytes); + break; + } + Ok(n) => { + // Precise pacing based on audio timestamp + let audio_timestamp = *current_timestamp.read().await; + let elapsed = start_time.elapsed().as_secs_f64(); + let lead_time = audio_timestamp - elapsed; + + if lead_time > BROADCAST_MAX_LEAD_TIME { + let sleep_duration = lead_time - BROADCAST_MAX_LEAD_TIME; + debug!( + "OGG broadcaster pacing: sleeping {:.3}s (audio_ts={:.3}s, elapsed={:.3}s, lead={:.3}s)", + sleep_duration, audio_timestamp, elapsed, lead_time + ); + tokio::time::sleep(tokio::time::Duration::from_secs_f64(sleep_duration)).await; + } + + // Accumulate FLAC data + flac_data.extend_from_slice(&read_buffer[..n]); + + // Create pages when we have a reasonable amount of data (8KB chunks) + // This respects FLAC frame boundaries better than arbitrary 4KB splits + while flac_data.len() >= 8192 { + let chunk = flac_data.drain(..8192).collect::>(); + let ogg_page = ogg_writer.create_page(&chunk, false, false, false); + let ogg_bytes = Bytes::from(ogg_page); + total_ogg_bytes += ogg_bytes.len() as u64; + + if let Err(e) = broadcast_tx.send(ogg_bytes) { + 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(()) +} + +/// 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 +fn create_ogg_flac_identification(flac_header: &[u8]) -> Result, AudioError> { + // Verify we have at least "fLaC" magic + if flac_header.len() < 4 || &flac_header[0..4] != b"fLaC" { + return Err(AudioError::ProcessingError("Invalid FLAC header".into())); + } + + // Extract STREAMINFO block (first metadata block) + // Format: 1 byte type+flags, 3 bytes length, N bytes data + if flac_header.len() < 8 { + return Err(AudioError::ProcessingError("FLAC header too short".into())); + } + + let first_block_type = flac_header[4] & 0x7F; // Remove last-metadata-block flag + if first_block_type != 0 { + return Err(AudioError::ProcessingError("First FLAC metadata block is not STREAMINFO".into())); + } + + // Extract block length (3 bytes big-endian after type byte) + let block_length = u32::from_be_bytes([0, flac_header[5], flac_header[6], flac_header[7]]) as usize; + + info!("STREAMINFO block_length = {} bytes", block_length); + + // STREAMINFO should be exactly 34 bytes of data + if block_length != 34 { + warn!("STREAMINFO block length is {} (expected 34)", block_length); + } + + // Total STREAMINFO block size = 1 (type) + 3 (length) + block_length + let streaminfo_size = 4 + block_length; + + if flac_header.len() < 4 + streaminfo_size { + return Err(AudioError::ProcessingError("FLAC header truncated".into())); + } + + // Extract just the STREAMINFO block (type + length + data) + let streaminfo = &flac_header[4..4 + streaminfo_size]; + + info!("Extracted STREAMINFO: {} bytes (type+length+data)", streaminfo.len()); + + let mut packet = Vec::new(); + + // OGG-FLAC identification header + packet.push(0x7F); // Byte 0: 0x7F + packet.extend_from_slice(b"FLAC"); // Bytes 1-4: "FLAC" + packet.push(0x01); // Byte 5: Major version + packet.push(0x00); // Byte 6: Minor version + packet.extend_from_slice(&1u16.to_be_bytes()); // Bytes 7-8: 1 header packet (Vorbis Comment) + packet.extend_from_slice(b"fLaC"); // Bytes 9-12: Native FLAC signature + packet.extend_from_slice(streaminfo); // Bytes 13+: STREAMINFO block only + + Ok(packet) +} + +/// Create empty Vorbis Comment block +fn create_empty_vorbis_comment() -> Vec { + let mut data = Vec::new(); + + // Vendor string + let vendor = "pmoaudio OGG-FLAC streamer"; + let vendor_bytes = vendor.as_bytes(); + data.extend_from_slice(&(vendor_bytes.len() as u32).to_le_bytes()); + data.extend_from_slice(vendor_bytes); + + // Number of comments (0 for now - metadata via /metadata endpoint) + data.extend_from_slice(&0u32.to_le_bytes()); + + data +} + +/// OGG page writer (same as in pmoflac::ogg_flac_encoder) +struct OggPageWriter { + stream_serial: u32, + page_sequence: u32, + granule_position: u64, +} + +impl OggPageWriter { + fn new(stream_serial: u32) -> Self { + Self { + stream_serial, + page_sequence: 0, + granule_position: 0, + } + } + + fn create_page(&mut self, packet_data: &[u8], is_bos: bool, is_eos: bool, is_continuation: bool) -> Vec { + use std::io::Write; + + let mut segments = Vec::new(); + let mut remaining = packet_data.len(); + + // Segment the packet into 255-byte chunks + while remaining > 0 { + let segment_size = remaining.min(255); + segments.push(segment_size as u8); + remaining -= segment_size; + } + + // If packet ends exactly on a 255-byte boundary, add empty segment + if !packet_data.is_empty() && packet_data.len() % 255 == 0 && !is_continuation { + segments.push(0); + } + + let segment_count = segments.len(); + let header_size = 27 + segment_count; + let total_size = header_size + packet_data.len(); + + let mut page = Vec::with_capacity(total_size); + + // OGG page header + page.write_all(b"OggS").unwrap(); + page.write_all(&[0]).unwrap(); // Version + + // Header type + let mut header_type = 0u8; + if is_continuation { + header_type |= 0x01; + } + if is_bos { + header_type |= 0x02; + } + if is_eos { + header_type |= 0x04; + } + page.write_all(&[header_type]).unwrap(); + + // Granule position + page.write_all(&self.granule_position.to_le_bytes()).unwrap(); + + // Stream serial number + page.write_all(&self.stream_serial.to_le_bytes()).unwrap(); + + // Page sequence number + page.write_all(&self.page_sequence.to_le_bytes()).unwrap(); + self.page_sequence += 1; + + // CRC checksum (zero for now, calculated later) + let crc_offset = page.len(); + page.write_all(&[0, 0, 0, 0]).unwrap(); + + // Number of segments + page.write_all(&[segment_count as u8]).unwrap(); + + // Segment table + page.write_all(&segments).unwrap(); + + // Packet data + page.write_all(packet_data).unwrap(); + + // Calculate and insert CRC32 + let crc = calculate_ogg_crc(&page); + page[crc_offset..crc_offset + 4].copy_from_slice(&crc.to_le_bytes()); + + page + } +} + +/// Calculate OGG CRC32 checksum +fn calculate_ogg_crc(data: &[u8]) -> u32 { + const CRC_TABLE: [u32; 256] = generate_crc_table(); + + let mut crc: u32 = 0; + for &byte in data { + crc = (crc << 8) ^ CRC_TABLE[((crc >> 24) ^ (byte as u32)) as usize]; + } + crc +} + +/// Generate CRC lookup table at compile time +const fn generate_crc_table() -> [u32; 256] { + let mut table = [0u32; 256]; + let mut i = 0; + while i < 256 { + let mut r = i << 24; + let mut j = 0; + while j < 8 { + if (r & 0x80000000) != 0 { + r = (r << 1) ^ 0x04c11db7; + } else { + r <<= 1; + } + j += 1; + } + table[i as usize] = r; + i += 1; + } + table +} diff --git a/pmoparadise/Cargo.toml b/pmoparadise/Cargo.toml index baf25782..f662a1b1 100644 --- a/pmoparadise/Cargo.toml +++ b/pmoparadise/Cargo.toml @@ -51,8 +51,8 @@ symphonia = { version = "0.5", features = ["all"] } # Audio decoding - claxon for FLAC streaming claxon = "0.4" -# pmoaudio-ext with playlist support (optional for examples) -pmoaudio-ext = { path = "../pmoaudio-ext", optional = true, features = ["playlist"] } +# pmoaudio-ext with playlist and streaming support (optional for examples) +pmoaudio-ext = { path = "../pmoaudio-ext", optional = true, features = ["playlist", "streaming"] } # Common music source traits pmosource = { path = "../pmosource" } diff --git a/pmoparadise/examples/stream_block.rs b/pmoparadise/examples/stream_block.rs new file mode 100644 index 00000000..5447ff80 --- /dev/null +++ b/pmoparadise/examples/stream_block.rs @@ -0,0 +1,350 @@ +//! Streams a Radio Paradise block via HTTP using pmoserver +//! +//! This example demonstrates streaming a single Radio Paradise block +//! using the StreamingFlacSink over HTTP via pmoserver. Perfect for +//! testing with VLC or other media players that support HTTP streaming. +//! +//! The example streams ONE block then terminates cleanly using END_OF_BLOCKS_SIGNAL. +//! For continuous streaming, push multiple block_ids without the END signal. +//! +//! Architecture: +//! ```text +//! RadioParadiseStreamSource → TimerNode → StreamingFlacSink +//! ↓ +//! StreamHandle +//! ↓ +//! pmoserver (Axum) +//! ↓ +//! VLC / Media Player Client +//! ``` +//! +//! Usage: +//! cargo run --example stream_block --features full -- +//! +//! Example: +//! cargo run --example stream_block --features full -- 0 # Main Mix +//! +//! Then open in VLC: +//! vlc http://localhost:8080/test/stream (pure FLAC) +//! vlc http://localhost:8080/test/stream-ogg (OGG-FLAC streaming container) +//! vlc http://localhost:8080/test/stream-icy (FLAC + ICY metadata) +//! +//! To check current metadata: +//! curl http://localhost:8080/test/metadata + +use axum::{ + body::Body, + extract::State, + http::{HeaderMap, StatusCode}, + response::{IntoResponse, Response}, +}; +use pmoaudio::{AudioPipelineNode, TimerNode}; +use pmoaudio_ext::{StreamingFlacSink, StreamingOggFlacSink}; +use pmoflac::EncoderOptions; +use pmoparadise::{RadioParadiseClient, RadioParadiseStreamSource, END_OF_BLOCKS_SIGNAL}; +use pmoserver::{ServerBuilder, init_logging}; +use std::env; +use std::sync::Arc; +use tokio_util::io::ReaderStream; +use tokio_util::sync::CancellationToken; + +/// Shared application state +struct AppState { + stream_handle: pmoaudio_ext::StreamHandle, + ogg_handle: pmoaudio_ext::OggFlacStreamHandle, +} + +/// Main HTTP handler for streaming (pure FLAC, no ICY metadata) +async fn stream_handler( + State(state): State>, + _headers: HeaderMap, +) -> Result { + tracing::info!("New client connected (pure FLAC mode)"); + + // Pure FLAC stream without ICY metadata + let flac_stream = state.stream_handle.subscribe_flac(); + + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/flac") + .header("Cache-Control", "no-cache, no-store") + .body(Body::from_stream(ReaderStream::new(flac_stream))) + .unwrap()) +} + +/// ICY streaming handler (FLAC with embedded metadata) +async fn stream_icy_handler( + State(state): State>, + _headers: HeaderMap, +) -> Result { + tracing::info!("New client connected (ICY mode)"); + + // FLAC stream with ICY metadata + let icy_stream = state.stream_handle.subscribe_icy(); + + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/flac") + .header("icy-metaint", "16000") + .header("icy-name", "Radio Paradise Stream Test") + .header("icy-genre", "Eclectic") + .header("icy-pub", "1") + .header("Cache-Control", "no-cache, no-store") + .body(Body::from_stream(ReaderStream::new(icy_stream))) + .unwrap()) +} + +/// OGG-FLAC streaming handler +async fn stream_ogg_handler( + State(state): State>, + _headers: HeaderMap, +) -> Result { + tracing::info!("New client connected (OGG-FLAC mode)"); + + // OGG-FLAC stream + let ogg_stream = state.ogg_handle.subscribe(); + + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/ogg") + .header("Cache-Control", "no-cache, no-store") + .body(Body::from_stream(ReaderStream::new(ogg_stream))) + .unwrap()) +} + +/// Metadata endpoint (JSON) +async fn metadata_handler(State(state): State>) -> impl IntoResponse { + let metadata = state.stream_handle.get_metadata().await; + axum::Json(metadata) +} + +/// Health check endpoint +async fn health_handler() -> &'static str { + "OK" +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + // Initialize logging via pmoserver + let _log_state = init_logging(); + + tracing::info!("=== Radio Paradise HTTP Streaming Test ==="); + + // Parse arguments + let args: Vec = env::args().collect(); + if args.len() < 2 { + eprintln!("Usage: {} ", args[0]); + eprintln!(); + eprintln!("Streams a Radio Paradise block via HTTP for testing."); + eprintln!(); + eprintln!("Channel IDs:"); + eprintln!(" 0 - Main Mix (eclectic, diverse mix)"); + eprintln!(" 1 - Mellow Mix (smooth, chilled music)"); + eprintln!(" 2 - Rock Mix (classic & modern rock)"); + eprintln!(" 3 - World/Etc Mix (global sounds)"); + eprintln!(); + eprintln!("After starting, open in VLC:"); + eprintln!(" vlc http://localhost:8080/test/stream (pure FLAC)"); + eprintln!(" vlc http://localhost:8080/test/stream-ogg (OGG-FLAC container)"); + eprintln!(" vlc http://localhost:8080/test/stream-icy (FLAC + ICY metadata)"); + std::process::exit(1); + } + + let channel_id: u8 = match args[1].parse() { + Ok(id) if id <= 3 => id, + _ => { + eprintln!("Error: channel_id must be a number between 0 and 3"); + std::process::exit(1); + } + }; + + tracing::info!("Channel ID: {}", channel_id); + + // ═══════════════════════════════════════════════════════════════════════════ + // Fetch block metadata + // ═══════════════════════════════════════════════════════════════════════════ + + tracing::info!("Fetching current block metadata..."); + let client = RadioParadiseClient::builder() + .channel(channel_id) + .build() + .await?; + + let block = client.get_block(None).await?; + + tracing::info!("Block Information:"); + tracing::info!(" Event ID: {}", block.event); + tracing::info!(" Songs: {}", block.song_count()); + tracing::info!(" Duration: {:.1} minutes", block.length as f64 / 60000.0); + tracing::info!(""); + + tracing::info!("Tracklist:"); + for (index, song) in block.songs_ordered() { + tracing::info!( + " {:2}. {} - {} ({})", + index + 1, + song.artist, + song.title, + song.album.as_deref().unwrap_or("Unknown Album") + ); + } + tracing::info!(""); + + // ═══════════════════════════════════════════════════════════════════════════ + // Create streaming pipelines (FLAC and OGG-FLAC) + // ═══════════════════════════════════════════════════════════════════════════ + + tracing::info!("Creating streaming pipelines..."); + + // Encoder options (shared) + let encoder_options = EncoderOptions { + compression_level: 5, + verify: false, + ..Default::default() + }; + + // ───────────────────────────────────────────────────────────────────────── + // Pipeline 1: FLAC streaming + // ───────────────────────────────────────────────────────────────────────── + + let mut source_flac = RadioParadiseStreamSource::new(client.clone()); + source_flac.push_block_id(block.event); + source_flac.push_block_id(END_OF_BLOCKS_SIGNAL); // Signal: no more blocks after this one + tracing::debug!("RadioParadiseStreamSource (FLAC) created with block {} + END signal", block.event); + + // Use SMALL channel size to make backpressure more reactive + // Instead of trying to buffer 3s of audio (60 chunks), use a much smaller buffer + // This forces tighter backpressure control + let max_lead_time = 3.0; + let channel_size = 8; // Small buffer for reactive backpressure + tracing::debug!("Using channel size: {} chunks ({:.1}s buffer at 50ms/chunk)", channel_size, channel_size as f64 * 0.05); + + let mut timer_flac = TimerNode::with_channel_size(max_lead_time, channel_size); + tracing::debug!("TimerNode (FLAC) created with {:.1}s max lead time, {} chunk buffer", max_lead_time, channel_size); + + // StreamingFlacSink doesn't take channel_size - it uses bits_per_sample (16, 24, or 32) + let (streaming_sink, stream_handle) = StreamingFlacSink::new(encoder_options.clone(), 16); + tracing::debug!("StreamingFlacSink created"); + + timer_flac.register(Box::new(streaming_sink)); + source_flac.register(Box::new(timer_flac)); + tracing::info!("Pipeline 1 connected: RadioParadiseStreamSource → TimerNode → StreamingFlacSink"); + + // ───────────────────────────────────────────────────────────────────────── + // Pipeline 2: OGG-FLAC streaming + // ───────────────────────────────────────────────────────────────────────── + + let mut source_ogg = RadioParadiseStreamSource::new(client); + source_ogg.push_block_id(block.event); + source_ogg.push_block_id(END_OF_BLOCKS_SIGNAL); // Signal: no more blocks after this one + tracing::debug!("RadioParadiseStreamSource (OGG) created with block {} + END signal", block.event); + + let mut timer_ogg = TimerNode::with_channel_size(max_lead_time, channel_size); + tracing::debug!("TimerNode (OGG) created with {:.1}s max lead time, {} chunk buffer", max_lead_time, channel_size); + + // StreamingOggFlacSink doesn't take channel_size - it uses bits_per_sample (16, 24, or 32) + let (ogg_sink, ogg_handle) = StreamingOggFlacSink::new(encoder_options, 16); + tracing::debug!("StreamingOggFlacSink created"); + + timer_ogg.register(Box::new(ogg_sink)); + source_ogg.register(Box::new(timer_ogg)); + tracing::info!("Pipeline 2 connected: RadioParadiseStreamSource → TimerNode → StreamingOggFlacSink"); + + // ═══════════════════════════════════════════════════════════════════════════ + // Setup pmoserver with streaming routes + // ═══════════════════════════════════════════════════════════════════════════ + + tracing::info!("Setting up pmoserver..."); + + let mut server = ServerBuilder::new("RadioParadiseStreamTest", "http://localhost", 8080) + .build(); + + let app_state = Arc::new(AppState { + stream_handle, + ogg_handle, + }); + + // Add streaming routes + server.add_handler_with_state("/test/stream", stream_handler, app_state.clone()).await; + server.add_handler_with_state("/test/stream-icy", stream_icy_handler, app_state.clone()).await; + server.add_handler_with_state("/test/stream-ogg", stream_ogg_handler, app_state.clone()).await; + + // Add metadata route + server.add_handler_with_state("/test/metadata", metadata_handler, app_state.clone()).await; + + // Add health check + server.add_handler("/test/health", health_handler).await; + + tracing::info!(""); + tracing::info!("========================================"); + tracing::info!("Ready to stream!"); + tracing::info!(""); + tracing::info!("Pure FLAC stream (for VLC, standard players):"); + tracing::info!(" vlc http://localhost:8080/test/stream"); + tracing::info!(""); + tracing::info!("OGG-FLAC stream (streaming container with metadata support):"); + tracing::info!(" vlc http://localhost:8080/test/stream-ogg"); + tracing::info!(""); + tracing::info!("FLAC + ICY metadata stream (for ICY-aware clients):"); + tracing::info!(" http://localhost:8080/test/stream-icy"); + tracing::info!(""); + tracing::info!("Metadata endpoint (JSON):"); + tracing::info!(" curl http://localhost:8080/test/metadata"); + tracing::info!("========================================"); + tracing::info!(""); + + // ═══════════════════════════════════════════════════════════════════════════ + // Start pipelines and server + // ═══════════════════════════════════════════════════════════════════════════ + + let stop_token = CancellationToken::new(); + let stop_token_flac = stop_token.clone(); + let stop_token_ogg = stop_token.clone(); + + // Start FLAC pipeline in background + let pipeline_flac_handle = tokio::spawn(async move { + tracing::info!("[PIPELINE-FLAC] Starting..."); + let result = Box::new(source_flac).run(stop_token_flac).await; + match &result { + Ok(()) => tracing::info!("[PIPELINE-FLAC] Completed successfully"), + Err(e) => tracing::error!("[PIPELINE-FLAC] Error: {}", e), + } + result + }); + + // Start OGG-FLAC pipeline in background + let pipeline_ogg_handle = tokio::spawn(async move { + tracing::info!("[PIPELINE-OGG] Starting..."); + let result = Box::new(source_ogg).run(stop_token_ogg).await; + match &result { + Ok(()) => tracing::info!("[PIPELINE-OGG] Completed successfully"), + Err(e) => tracing::error!("[PIPELINE-OGG] Error: {}", e), + } + result + }); + + // Start pmoserver (blocks until Ctrl+C) + tracing::info!("[SERVER] Starting pmoserver..."); + server.start().await; + server.wait().await; + + // Server stopped, cancel pipelines + tracing::info!("Server stopped, canceling pipelines..."); + stop_token.cancel(); + + // Wait for both pipelines to finish + match pipeline_flac_handle.await { + Ok(Ok(())) => tracing::info!("FLAC pipeline completed successfully"), + Ok(Err(e)) => tracing::error!("FLAC pipeline error: {}", e), + Err(e) => tracing::error!("FLAC pipeline task error: {}", e), + } + + match pipeline_ogg_handle.await { + Ok(Ok(())) => tracing::info!("OGG-FLAC pipeline completed successfully"), + Ok(Err(e)) => tracing::error!("OGG-FLAC pipeline error: {}", e), + Err(e) => tracing::error!("OGG-FLAC pipeline task error: {}", e), + } + + tracing::info!("Shutdown complete"); + Ok(()) +}