diff --git a/pmoaudio-ext/src/sinks/mod.rs b/pmoaudio-ext/src/sinks/mod.rs index 07b68315..72fc7e27 100755 --- a/pmoaudio-ext/src/sinks/mod.rs +++ b/pmoaudio-ext/src/sinks/mod.rs @@ -16,6 +16,9 @@ mod broadcast_pacing; #[cfg(feature = "http-stream")] mod flac_frame_utils; +#[cfg(feature = "http-stream")] +mod timed_broadcast; + #[cfg(feature = "http-stream")] mod streaming_flac_sink; diff --git a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs index 95cdb56f..3b2d6d9d 100644 --- a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs @@ -19,7 +19,7 @@ //! ↓ //! [Broadcaster Task] //! ↓ -//! broadcast::channel (FLAC bytes) +//! timed_broadcast::channel (FLAC bytes) //! ↓ //! Multiple clients via StreamHandle::subscribe() //! ├─ FLAC pure (for standard renderers) @@ -62,7 +62,11 @@ use std::sync::Arc; use std::task::{Context, Poll}; use std::time::Duration; -use super::{broadcast_pacing::BroadcastPacer, flac_frame_utils}; +use super::{ + broadcast_pacing::BroadcastPacer, + flac_frame_utils, + timed_broadcast::{self, TimedPacket, TryRecvError}, +}; use async_trait::async_trait; use bytes::Bytes; use pmoaudio::{ @@ -73,7 +77,7 @@ use pmoaudio::{ 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::sync::{mpsc, RwLock}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, trace, warn}; @@ -160,7 +164,7 @@ pub struct MetadataSnapshot { #[derive(Clone)] pub struct StreamHandle { /// Broadcast sender for FLAC bytes (pure mode) - flac_broadcast: broadcast::Sender, + flac_broadcast: timed_broadcast::Sender, /// Current track metadata (read-only for consumers) metadata: Arc>, @@ -189,6 +193,7 @@ impl StreamHandle { finished: false, handle: self.clone(), state: FlacStreamState::SendingHeader, + current_epoch: 0, } } @@ -220,6 +225,7 @@ impl StreamHandle { finished: false, handle: self.clone(), state: FlacStreamState::SendingHeader, + current_epoch: 0, } } @@ -247,11 +253,18 @@ enum FlacStreamState { /// Pure FLAC client stream (implements AsyncRead). pub struct FlacClientStream { - rx: broadcast::Receiver, + rx: timed_broadcast::Receiver, buffer: VecDeque, finished: bool, handle: StreamHandle, state: FlacStreamState, + current_epoch: u64, +} + +impl FlacClientStream { + pub fn current_epoch(&self) -> u64 { + self.current_epoch + } } impl AsyncRead for FlacClientStream { @@ -302,10 +315,11 @@ impl AsyncRead for FlacClientStream { // Try to receive more data match self.rx.try_recv() { - Ok(bytes) => { - self.buffer.extend(bytes.iter()); + Ok(packet) => { + self.current_epoch = packet.epoch; + self.buffer.extend(packet.payload.iter()); } - Err(broadcast::error::TryRecvError::Empty) => { + Err(TryRecvError::Empty) => { // No data available right now. // Schedule a wakeup after a small delay to avoid busy-loop polling. let waker = cx.waker().clone(); @@ -315,11 +329,11 @@ impl AsyncRead for FlacClientStream { }); return Poll::Pending; } - Err(broadcast::error::TryRecvError::Lagged(skipped)) => { + Err(TryRecvError::Lagged(skipped)) => { warn!("FLAC client lagged, skipped {} messages", skipped); // Continue to try receiving again } - Err(broadcast::error::TryRecvError::Closed) => { + Err(TryRecvError::Closed) => { self.finished = true; return Poll::Ready(Ok(())); } @@ -345,7 +359,7 @@ impl Drop for FlacClientStream { /// This stream injects ICY metadata blocks at regular intervals, /// allowing clients to display "Now Playing" information. pub struct IcyClientStream { - rx: broadcast::Receiver, + rx: timed_broadcast::Receiver, metadata: Arc>, metaint: usize, byte_count: usize, @@ -355,6 +369,13 @@ pub struct IcyClientStream { finished: bool, handle: StreamHandle, state: FlacStreamState, + current_epoch: u64, +} + +impl IcyClientStream { + pub fn current_epoch(&self) -> u64 { + self.current_epoch + } } impl IcyClientStream { @@ -491,22 +512,23 @@ impl AsyncRead for IcyClientStream { // Try to receive audio data match self.rx.try_recv() { - Ok(bytes) => { + Ok(packet) => { + self.current_epoch = packet.epoch; // 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); + let to_buffer = packet.payload.len().min(until_metadata); - self.buffer.extend(bytes[..to_buffer].iter()); + self.buffer.extend(packet.payload[..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() { + if to_buffer < packet.payload.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) => { + Err(TryRecvError::Empty) => { // No data available right now. // Schedule a wakeup after a small delay to avoid busy-loop polling. let waker = cx.waker().clone(); @@ -516,10 +538,10 @@ impl AsyncRead for IcyClientStream { }); return Poll::Pending; } - Err(broadcast::error::TryRecvError::Lagged(skipped)) => { + Err(TryRecvError::Lagged(skipped)) => { warn!("ICY client lagged, skipped {} messages", skipped); } - Err(broadcast::error::TryRecvError::Closed) => { + Err(TryRecvError::Closed) => { self.finished = true; return Poll::Ready(Ok(())); } @@ -552,7 +574,7 @@ struct StreamingFlacSinkLogic { pcm_tx: mpsc::Sender, pcm_rx: Option>, metadata: Arc>, - flac_broadcast: broadcast::Sender, + flac_broadcast: timed_broadcast::Sender, flac_header: Arc>>, encoder_state: Option, sample_rate: Option, @@ -749,6 +771,11 @@ impl NodeLogic for StreamingFlacSinkLogic { break; } + SyncMarker::TopZeroSync => { + self.flac_broadcast.mark_top_zero(); + trace!("TopZeroSync propagated to FLAC broadcast"); + } + _ => { trace!("Received other sync marker"); } @@ -781,7 +808,7 @@ impl NodeLogic for StreamingFlacSinkLogic { /// Ensures data is sent at FLAC frame boundaries to prevent sync errors in strict decoders like FFPlay. async fn broadcast_flac_stream( mut flac_stream: FlacEncodedStream, - broadcast_tx: broadcast::Sender, + broadcast_tx: timed_broadcast::Sender, header_cache: Arc>>, current_timestamp: Arc>, broadcast_max_lead_time: f64, @@ -813,7 +840,11 @@ async fn broadcast_flac_stream( // EOF - send any remaining data if !accumulator.is_empty() { let bytes = Bytes::from(std::mem::take(&mut accumulator)); - let _ = broadcast_tx.send(bytes); + let audio_ts = *current_timestamp.read().await; + if broadcast_tx.send(bytes.clone(), audio_ts).await.is_err() { + trace!("Broadcast closed before sending final FLAC data"); + break; + } } info!("FLAC encoder stream ended, total bytes: {}", total_bytes); break; @@ -931,15 +962,20 @@ async fn broadcast_flac_stream( } let num_receivers = broadcast_tx.receiver_count(); - if let Err(e) = broadcast_tx.send(bytes.clone()) { - // 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", - bytes.len(), - num_receivers - ); + match broadcast_tx.send(bytes.clone(), audio_timestamp).await { + Ok(_) => { + if num_receivers > 0 { + trace!( + "Broadcasted {} bytes to {} receivers", + bytes.len(), + num_receivers + ); + } + } + Err(_) => { + trace!("No active receivers for FLAC broadcast, terminating"); + return Ok(()); + } } } } @@ -1017,7 +1053,7 @@ impl StreamingFlacSink { ); // Broadcast channel for FLAC bytes - let (flac_broadcast, _) = broadcast::channel(broadcast_capacity); + let (flac_broadcast, _) = timed_broadcast::channel(broadcast_capacity); // FLAC header cache let flac_header = Arc::new(RwLock::new(None)); diff --git a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs index 89ca8743..a236a6c4 100644 --- a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs @@ -21,7 +21,7 @@ //! ↓ //! [OGG Wrapper Task] - wraps FLAC frames in OGG pages //! ↓ -//! broadcast::channel (OGG-FLAC bytes) +//! timed_broadcast::channel (OGG-FLAC bytes) //! ↓ //! Multiple HTTP clients //! ``` @@ -53,7 +53,11 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::task::{Context, Poll}; -use super::{broadcast_pacing::BroadcastPacer, flac_frame_utils}; +use super::{ + broadcast_pacing::BroadcastPacer, + flac_frame_utils, + timed_broadcast::{self, TimedPacket, TryRecvError}, +}; use async_trait::async_trait; use bytes::Bytes; use pmoaudio::{ @@ -64,7 +68,7 @@ use pmoaudio::{ 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::sync::{mpsc, RwLock}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, trace, warn}; @@ -97,7 +101,7 @@ pub use super::streaming_flac_sink::MetadataSnapshot; #[derive(Clone)] pub struct OggFlacStreamHandle { /// Broadcast sender for OGG-FLAC bytes - ogg_broadcast: broadcast::Sender, + ogg_broadcast: timed_broadcast::Sender, /// Current track metadata metadata: Arc>, @@ -126,6 +130,7 @@ impl OggFlacStreamHandle { finished: false, handle: self.clone(), state: OggFlacStreamState::SendingHeader, + current_epoch: 0, } } @@ -148,11 +153,19 @@ enum OggFlacStreamState { /// OGG-FLAC client stream (implements AsyncRead). pub struct OggFlacClientStream { - rx: broadcast::Receiver, + rx: timed_broadcast::Receiver, buffer: VecDeque, finished: bool, handle: OggFlacStreamHandle, state: OggFlacStreamState, + current_epoch: u64, +} + +impl OggFlacClientStream { + /// Dernier epoch observé (incrémenté à chaque TopZeroSync). + pub fn current_epoch(&self) -> u64 { + self.current_epoch + } } impl AsyncRead for OggFlacClientStream { @@ -203,10 +216,11 @@ impl AsyncRead for OggFlacClientStream { // Try to receive more data match self.rx.try_recv() { - Ok(bytes) => { - self.buffer.extend(bytes.iter()); + Ok(packet) => { + self.current_epoch = packet.epoch; + self.buffer.extend(packet.payload.iter()); } - Err(broadcast::error::TryRecvError::Empty) => { + Err(TryRecvError::Empty) => { // No data available right now. // Schedule a wakeup after a small delay to avoid busy-loop polling. let waker = cx.waker().clone(); @@ -216,10 +230,10 @@ impl AsyncRead for OggFlacClientStream { }); return Poll::Pending; } - Err(broadcast::error::TryRecvError::Lagged(skipped)) => { + Err(TryRecvError::Lagged(skipped)) => { warn!("OGG-FLAC client lagged, skipped {} messages", skipped); } - Err(broadcast::error::TryRecvError::Closed) => { + Err(TryRecvError::Closed) => { self.finished = true; return Poll::Ready(Ok(())); } @@ -252,7 +266,7 @@ struct StreamingOggFlacSinkLogic { pcm_tx: mpsc::Sender, pcm_rx: Option>, metadata: Arc>, - ogg_broadcast: broadcast::Sender, + ogg_broadcast: timed_broadcast::Sender, ogg_header: Arc>>, encoder_state: Option, sample_rate: Option, @@ -440,6 +454,11 @@ impl NodeLogic for StreamingOggFlacSinkLogic { break; } + SyncMarker::TopZeroSync => { + self.ogg_broadcast.mark_top_zero(); + trace!("TopZeroSync propagated to timed broadcast"); + } + _ => { trace!("Received other sync marker"); } @@ -521,7 +540,7 @@ impl StreamingOggFlacSink { broadcast_capacity, broadcast_max_lead_time ); - let (ogg_broadcast, _) = broadcast::channel(broadcast_capacity); + let (ogg_broadcast, _) = timed_broadcast::channel(broadcast_capacity); // OGG-FLAC header cache let ogg_header = Arc::new(RwLock::new(None)); @@ -751,7 +770,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result /// Ensures FLAC frames are only sent at frame boundaries to prevent sync errors in strict decoders like FFPlay. async fn broadcast_ogg_flac_stream( mut flac_stream: FlacEncodedStream, - broadcast_tx: broadcast::Sender, + broadcast_tx: timed_broadcast::Sender, header_cache: Arc>>, current_timestamp: Arc>, broadcast_max_lead_time: f64, @@ -811,9 +830,15 @@ async fn broadcast_ogg_flac_stream( ); // Broadcast header - let _ = broadcast_tx.send(bos_bytes); + if broadcast_tx.send(bos_bytes.clone(), 0.0).await.is_err() { + trace!("No receivers for BOS page, terminating broadcast"); + return Ok(()); + } total_ogg_bytes += comment_bytes.len() as u64; - let _ = broadcast_tx.send(comment_bytes); + if broadcast_tx.send(comment_bytes.clone(), 0.0).await.is_err() { + trace!("No receivers for comment page, terminating broadcast"); + return Ok(()); + } // Step 4: Read FLAC stream and create OGG packets // Use larger read buffer (16KB) to reduce syscalls and accumulator for frame boundary detection @@ -830,7 +855,11 @@ async fn broadcast_ogg_flac_stream( let eos_page = ogg_writer.create_page(&flac_accumulator, false, true, false); let eos_bytes = Bytes::from(eos_page); total_ogg_bytes += eos_bytes.len() as u64; - let _ = broadcast_tx.send(eos_bytes); + let eos_ts = *current_timestamp.read().await; + if broadcast_tx.send(eos_bytes.clone(), eos_ts).await.is_err() { + trace!("Broadcast closed before sending final EOS page"); + break; + } info!( "Sent final EOS page with {} bytes of data", flac_accumulator.len() @@ -840,7 +869,11 @@ async fn broadcast_ogg_flac_stream( 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); + let eos_ts = *current_timestamp.read().await; + if broadcast_tx.send(eos_bytes.clone(), eos_ts).await.is_err() { + trace!("Broadcast closed before sending empty EOS page"); + break; + } info!("Sent empty EOS page"); } @@ -979,12 +1012,13 @@ async fn broadcast_ogg_flac_stream( } // Envoyer au broadcast - match broadcast_tx.send(ogg_bytes.clone()) { + match broadcast_tx.send(ogg_bytes.clone(), audio_timestamp).await { Ok(n) => { trace!("Broadcasted OGG page with 1 FLAC frame ({} bytes), {} samples ({} bytes total with OGG overhead) to {} receivers", first_frame.len(), first_frame_samples, ogg_bytes.len(), n); } - Err(e) => { - trace!("No active receivers for OGG-FLAC broadcast: {}", e); + Err(_) => { + trace!("No active receivers for OGG-FLAC broadcast, terminating loop"); + return Ok(()); } } } diff --git a/pmoaudio-ext/src/sinks/timed_broadcast.rs b/pmoaudio-ext/src/sinks/timed_broadcast.rs new file mode 100644 index 00000000..205b6e52 --- /dev/null +++ b/pmoaudio-ext/src/sinks/timed_broadcast.rs @@ -0,0 +1,475 @@ +//! Broadcast channel avec TTL et propagation de TopZero. +//! Inspiré de `tokio::sync::broadcast` mais ajoute : +//! - Capacité bornée avec blocage des producteurs quand aucun slot n’est libre. +//! - Expiration automatique des messages (TTL) pour libérer les slots. +//! - Propagation d’un compteur `epoch` incrémenté sur chaque TopZeroSync. + +use std::{ + collections::VecDeque, + fmt, + sync::{ + atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, + Arc, Mutex, Weak, + }, + time::{Duration, Instant}, +}; + +use tokio::sync::Notify; +use tracing::warn; + +/// Paquet diffusé contenant la charge utile + méta timing. +#[derive(Clone)] +pub struct TimedPacket { + /// Charge utile diffusée aux clients. + pub payload: T, + /// Timestamp audio relatif (en secondes) pour pacing côté client. + pub audio_timestamp: f64, + /// Compteur incrémenté lorsqu'un TopZeroSync est reçu. + pub epoch: u64, +} + +impl fmt::Debug for TimedPacket { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TimedPacket") + .field("audio_timestamp", &self.audio_timestamp) + .field("epoch", &self.epoch) + .finish_non_exhaustive() + } +} + +/// Erreur remontée par `Receiver::try_recv`. +#[derive(Debug)] +pub enum TryRecvError { + Empty, + Lagged(u64), + Closed, +} + +/// Erreur remontée par `Receiver::recv`. +#[derive(Debug)] +pub enum RecvError { + Lagged(u64), + Closed, +} + +/// Erreur remontée par `Sender::send`. +#[derive(Debug)] +pub struct SendError(pub T); + +struct Entry { + seq: u64, + expires_at: Instant, + payload: T, + audio_timestamp: f64, + epoch: u64, +} + +struct State { + buffer: VecDeque>, + head_seq: u64, + next_seq: u64, + closed: bool, + epoch: u64, + epoch_start: Instant, + cursors: Vec>, +} + +impl State { + fn new(capacity: usize, epoch_start: Instant) -> Self { + Self { + buffer: VecDeque::with_capacity(capacity), + head_seq: 0, + next_seq: 0, + closed: false, + epoch: 0, + epoch_start, + cursors: Vec::new(), + } + } + + fn purge_expired(&mut self, now: Instant) -> bool { + let mut purged = 0u64; + while let Some(entry) = self.buffer.front() { + if entry.expires_at <= now { + self.buffer.pop_front(); + self.head_seq += 1; + purged += 1; + } else { + break; + } + } + if purged > 0 { + warn!( + "TimedBroadcast: purged {} expired packet(s) (head_seq={})", + purged, + self.head_seq + ); + return true; + } + false + } + + fn prune_consumed(&mut self) -> bool { + let mut min_next = self.next_seq; + let mut has_cursor = false; + self.cursors.retain(|weak| { + if let Some(cursor) = weak.upgrade() { + let pos = cursor.next_seq.load(Ordering::SeqCst); + if pos < min_next { + min_next = pos; + } + has_cursor = true; + true + } else { + false + } + }); + + if !has_cursor { + return false; + } + + let removable = min_next.saturating_sub(self.head_seq) as usize; + if removable == 0 { + return false; + } + + for _ in 0..removable { + if self.buffer.pop_front().is_some() { + self.head_seq += 1; + } + } + true + } +} + +struct Inner { + state: Mutex>, + data_notify: Notify, + space_notify: Notify, + capacity: usize, + sender_count: AtomicUsize, + receiver_count: AtomicUsize, + is_closed: AtomicBool, +} + +impl Inner { + fn new(capacity: usize) -> Self { + Self { + state: Mutex::new(State::new(capacity, Instant::now())), + data_notify: Notify::new(), + space_notify: Notify::new(), + capacity, + sender_count: AtomicUsize::new(1), + receiver_count: AtomicUsize::new(0), + is_closed: AtomicBool::new(false), + } + } + + fn close(&self) { + if !self + .is_closed + .swap(true, Ordering::SeqCst) + { + if let Ok(mut state) = self.state.lock() { + state.closed = true; + } + self.data_notify.notify_waiters(); + self.space_notify.notify_waiters(); + } + } +} + +/// Créé un channel broadcast temporisé. +pub fn channel(capacity: usize) -> (Sender, Receiver) { + assert!(capacity > 0, "capacity must be > 0"); + let inner = Arc::new(Inner::new(capacity)); + let next_seq = { + let state = inner + .state + .lock() + .expect("timed broadcast mutex poisoned"); + state.next_seq + }; + let sender = Sender { + inner: inner.clone(), + }; + let cursor = Arc::new(ReceiverCursor { + next_seq: AtomicU64::new(next_seq), + }); + { + let mut state = inner + .state + .lock() + .expect("timed broadcast mutex poisoned"); + state.cursors.push(Arc::downgrade(&cursor)); + } + inner.receiver_count.store(1, Ordering::SeqCst); + let receiver = Receiver { + inner, + next_seq, + cursor, + }; + (sender, receiver) +} + +/// Sender côté producteur. +pub struct Sender { + inner: Arc>, +} + +impl Clone for Sender { + fn clone(&self) -> Self { + self.inner.sender_count.fetch_add(1, Ordering::SeqCst); + Self { + inner: self.inner.clone(), + } + } +} + +impl Sender { + /// Diffuse un paquet. Bloque si la capacité est atteinte avec des paquets non périmés. + pub async fn send(&self, payload: T, audio_timestamp: f64) -> Result> + where + T: Clone, + { + let mut payload = Some(payload); + loop { + let mut wait_deadline = None; + { + let now = Instant::now(); + let mut state = self + .inner + .state + .lock() + .expect("timed broadcast mutex poisoned"); + + if state.closed { + return Err(SendError(payload.expect("payload already consumed"))); + } + + if state.purge_expired(now) { + self.inner.space_notify.notify_waiters(); + } + + if state.prune_consumed() { + self.inner.space_notify.notify_waiters(); + } + + if state.buffer.len() < self.inner.capacity { + let audio_offset = + Duration::from_secs_f64(audio_timestamp.max(0.0)); + let expires_at = state.epoch_start + audio_offset; + let entry = Entry { + seq: state.next_seq, + expires_at, + payload: payload + .take() + .expect("payload already consumed"), + audio_timestamp, + epoch: state.epoch, + }; + state.next_seq += 1; + state.buffer.push_back(entry); + let receivers = self.inner.receiver_count.load(Ordering::SeqCst); + drop(state); + self.inner.data_notify.notify_waiters(); + return Ok(receivers); + } + + wait_deadline = state.buffer.front().map(|entry| entry.expires_at); + } + + if let Some(deadline) = wait_deadline { + let deadline = tokio::time::Instant::from_std(deadline); + tokio::select! { + _ = self.inner.space_notify.notified() => {}, + _ = tokio::time::sleep_until(deadline) => {}, + } + } else { + self.inner.space_notify.notified().await; + } + } + } + + /// Crée un nouveau receiver abonné au flux. + pub fn subscribe(&self) -> Receiver { + let mut state = self + .inner + .state + .lock() + .expect("timed broadcast mutex poisoned"); + let next_seq = state.next_seq; + let cursor = Arc::new(ReceiverCursor { + next_seq: AtomicU64::new(next_seq), + }); + state.cursors.push(Arc::downgrade(&cursor)); + state.prune_consumed(); + drop(state); + + self.inner.receiver_count.fetch_add(1, Ordering::SeqCst); + + Receiver { + inner: self.inner.clone(), + next_seq, + cursor, + } + } + + /// Marque un TopZero : incrémente l'epoch pour les paquets suivants. + pub fn mark_top_zero(&self) { + let mut state = self + .inner + .state + .lock() + .expect("timed broadcast mutex poisoned"); + state.epoch = state.epoch.wrapping_add(1); + state.epoch_start = Instant::now(); + if !state.buffer.is_empty() { + state.head_seq = state.next_seq; + state.buffer.clear(); + self.inner.space_notify.notify_waiters(); + } + } + + /// Nombre actuel de receivers abonnés. + pub fn receiver_count(&self) -> usize { + self.inner.receiver_count.load(Ordering::SeqCst) + } + + /// Ferme explicitement le channel. + pub fn close(&self) { + self.inner.close(); + } +} + +impl Drop for Sender { + fn drop(&mut self) { + if self.inner.sender_count.fetch_sub(1, Ordering::SeqCst) == 1 { + self.inner.close(); + } + } +} + +/// Receiver côté consommateur. +pub struct Receiver { + inner: Arc>, + next_seq: u64, + cursor: Arc, +} + +struct ReceiverCursor { + next_seq: AtomicU64, +} + +impl Receiver +where + T: Clone, +{ + fn poll_entry(&mut self) -> Result, TryRecvError> { + let mut state = self + .inner + .state + .lock() + .expect("timed broadcast mutex poisoned"); + + if state.closed && state.buffer.is_empty() { + return Err(TryRecvError::Closed); + } + + let now = Instant::now(); + if state.purge_expired(now) { + self.inner.space_notify.notify_waiters(); + } + + if self.next_seq < state.head_seq { + let skipped = state.head_seq - self.next_seq; + self.next_seq = state.head_seq; + return Err(TryRecvError::Lagged(skipped)); + } + + let offset = (self.next_seq - state.head_seq) as usize; + if offset < state.buffer.len() { + let entry = state + .buffer + .get(offset) + .expect("invalid buffer offset"); + let packet = TimedPacket { + payload: entry.payload.clone(), + audio_timestamp: entry.audio_timestamp, + epoch: entry.epoch, + }; + self.next_seq += 1; + self.cursor + .next_seq + .store(self.next_seq, Ordering::SeqCst); + if state.prune_consumed() { + self.inner.space_notify.notify_waiters(); + } + return Ok(packet); + } + + if state.closed { + Err(TryRecvError::Closed) + } else { + Err(TryRecvError::Empty) + } + } + + /// Version synchrone utilisée dans `poll_read`. + pub fn try_recv(&mut self) -> Result, TryRecvError> { + self.poll_entry() + } + + /// Attends qu'un paquet soit disponible. + pub async fn recv(&mut self) -> Result, RecvError> { + loop { + match self.try_recv() { + Ok(packet) => return Ok(packet), + Err(TryRecvError::Empty) => { + self.inner.data_notify.notified().await; + } + Err(TryRecvError::Lagged(skipped)) => return Err(RecvError::Lagged(skipped)), + Err(TryRecvError::Closed) => return Err(RecvError::Closed), + } + } + } +} + +impl Clone for Receiver { + fn clone(&self) -> Self { + self.inner.receiver_count.fetch_add(1, Ordering::SeqCst); + let cursor = Arc::new(ReceiverCursor { + next_seq: AtomicU64::new(self.next_seq), + }); + { + let mut state = self + .inner + .state + .lock() + .expect("timed broadcast mutex poisoned"); + state.cursors.push(Arc::downgrade(&cursor)); + } + Self { + inner: self.inner.clone(), + next_seq: self.next_seq, + cursor, + } + } +} + +impl Drop for Receiver { + fn drop(&mut self) { + self.cursor + .next_seq + .store(self.next_seq, Ordering::SeqCst); + if let Ok(mut state) = self.inner.state.lock() { + if state.prune_consumed() { + self.inner.space_notify.notify_waiters(); + } + } + if self.inner.receiver_count.fetch_sub(1, Ordering::SeqCst) == 1 { + self.inner.space_notify.notify_waiters(); + } + } +} diff --git a/pmoparadise/examples/stream_block.rs b/pmoparadise/examples/stream_block.rs index 7ecc1887..3224bc6d 100644 --- a/pmoparadise/examples/stream_block.rs +++ b/pmoparadise/examples/stream_block.rs @@ -204,21 +204,19 @@ async fn main() -> Result<(), Box> { }; // ───────────────────────────────────────────────────────────────────────── - // Pipeline 1: FLAC streaming + // Unique pipeline feeding both FLAC and OGG sinks // ───────────────────────────────────────────────────────────────────────── - 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 + let mut source = RadioParadiseStreamSource::new(client); + source.push_block_id(block.event); + source.push_block_id(END_OF_BLOCKS_SIGNAL); // Signal: no more blocks after this one tracing::debug!( - "RadioParadiseStreamSource (FLAC) created with block {} + END signal", + "RadioParadiseStreamSource 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 buffer_sec = 10.0; + // Use SMALL channel size to make backpressure plus fan-out manageable. + let buffer_sec = 0.1; let max_lead_time = buffer_sec; let channel_size = 512; tracing::debug!( @@ -227,53 +225,31 @@ async fn main() -> Result<(), Box> { channel_size as f64 * 0.05 ); - let mut timer_flac = TimerBufferNode::with_channel_size(buffer_sec, channel_size); + let mut timer_node = TimerBufferNode::with_channel_size(buffer_sec, channel_size); tracing::debug!( - "TimerBufferNode (FLAC) created with {:.1}s buffer, {} chunk queue", + "TimerBufferNode created with {:.1}s buffer, {} chunk queue", buffer_sec, channel_size ); - // StreamingFlacSink doesn't take channel_size - it uses bits_per_sample (16, 24, or 32) + // Streaming sinks let (streaming_sink, stream_handle) = StreamingFlacSink::with_max_broadcast_lead(encoder_options.clone(), 16, max_lead_time); tracing::debug!("StreamingFlacSink created"); - timer_flac.register(Box::new(streaming_sink)); - source_flac.register(Box::new(timer_flac)); - tracing::info!( - "Pipeline 1 connected: RadioParadiseStreamSource → TimerBufferNode → 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 = TimerBufferNode::with_channel_size(buffer_sec, channel_size); - tracing::debug!( - "TimerBufferNode (OGG) created with {:.1}s buffer, {} chunk queue", - buffer_sec, - channel_size - ); - - // StreamingOggFlacSink doesn't take channel_size - it uses bits_per_sample (16, 24, or 32) let (ogg_sink, ogg_handle) = StreamingOggFlacSink::with_max_broadcast_lead(encoder_options, 16, max_lead_time); tracing::debug!("StreamingOggFlacSink created"); - timer_ogg.register(Box::new(ogg_sink)); - source_ogg.register(Box::new(timer_ogg)); + // timer_node.register(Box::new(streaming_sink)); + // timer_node.register(Box::new(ogg_sink)); + // source.register(Box::new(timer_node)); + + source.register(Box::new(streaming_sink)); + source.register(Box::new(ogg_sink)); tracing::info!( - "Pipeline 2 connected: RadioParadiseStreamSource → TimerBufferNode → StreamingOggFlacSink" + "Pipeline connected: StreamSource → TimerBufferNode → {{FLAC, OGG}} sinks" ); // ═══════════════════════════════════════════════════════════════════════════ @@ -332,27 +308,15 @@ async fn main() -> Result<(), Box> { // ═══════════════════════════════════════════════════════════════════════════ let stop_token = CancellationToken::new(); - let stop_token_flac = stop_token.clone(); - let stop_token_ogg = stop_token.clone(); + let pipeline_stop = 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; + // Start shared pipeline in background + let pipeline_handle = tokio::spawn(async move { + tracing::info!("[PIPELINE] Starting..."); + let result = Box::new(source).run(pipeline_stop).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), + Ok(()) => tracing::info!("[PIPELINE] Completed successfully"), + Err(e) => tracing::error!("[PIPELINE] Error: {}", e), } result }); @@ -366,17 +330,11 @@ async fn main() -> Result<(), Box> { 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), + // Wait for pipeline to finish + match pipeline_handle.await { + Ok(Ok(())) => tracing::info!("Pipeline completed successfully"), + Ok(Err(e)) => tracing::error!("Pipeline error: {}", e), + Err(e) => tracing::error!("Pipeline task error: {}", e), } tracing::info!("Shutdown complete");