diff --git a/pmoparadise/examples/stream_block.rs b/pmoparadise/examples/stream_block.rs index 7355f259..458614fb 100644 --- a/pmoparadise/examples/stream_block.rs +++ b/pmoparadise/examples/stream_block.rs @@ -4,6 +4,9 @@ //! 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 @@ -38,7 +41,7 @@ use axum::{ use pmoaudio::{AudioPipelineNode, TimerNode}; use pmoaudio_ext::{StreamingFlacSink, StreamingOggFlacSink}; use pmoflac::EncoderOptions; -use pmoparadise::{RadioParadiseClient, RadioParadiseStreamSource}; +use pmoparadise::{RadioParadiseClient, RadioParadiseStreamSource, END_OF_BLOCKS_SIGNAL}; use pmoserver::{ServerBuilder, init_logging}; use std::env; use std::sync::Arc; @@ -206,7 +209,8 @@ async fn main() -> Result<(), Box> { let mut source_flac = RadioParadiseStreamSource::new(client.clone()); source_flac.push_block_id(block.event); - tracing::debug!("RadioParadiseStreamSource (FLAC) created with block {}", 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); // Calculate channel size to match max_lead_time // With 50ms chunks and 3.0s lead time: 3.0 / 0.05 = 60 chunks @@ -232,7 +236,8 @@ async fn main() -> Result<(), Box> { let mut source_ogg = RadioParadiseStreamSource::new(client); source_ogg.push_block_id(block.event); - tracing::debug!("RadioParadiseStreamSource (OGG) created with block {}", 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); diff --git a/pmoparadise/src/lib.rs b/pmoparadise/src/lib.rs index 960b3d94..8a901131 100644 --- a/pmoparadise/src/lib.rs +++ b/pmoparadise/src/lib.rs @@ -232,7 +232,7 @@ pub use models::{Block, DurationMs, EventId, NowPlaying, Song}; pub use source::RadioParadiseSource; #[cfg(feature = "pmoaudio")] -pub use radio_paradise_stream_source::RadioParadiseStreamSource; +pub use radio_paradise_stream_source::{RadioParadiseStreamSource, END_OF_BLOCKS_SIGNAL}; #[cfg(feature = "pmoserver")] pub use pmoserver_ext::{ diff --git a/pmoparadise/src/radio_paradise_stream_source.rs b/pmoparadise/src/radio_paradise_stream_source.rs index 7ce8a3d5..17bb3795 100644 --- a/pmoparadise/src/radio_paradise_stream_source.rs +++ b/pmoparadise/src/radio_paradise_stream_source.rs @@ -25,10 +25,10 @@ use tokio::io::AsyncReadExt; use tokio::sync::{mpsc, RwLock}; use tokio_util::{io::StreamReader, sync::CancellationToken}; -/// Timeout pour attendre un nouveau block ID -/// Pour une radio en temps réel, 3s est raisonnable -/// Pour des tests avec un seul bloc, on veut quelque chose de plus long -const BLOCK_ID_TIMEOUT_SECS: u64 = 3600; // 1 heure +/// Signal spécial pour indiquer qu'il n'y aura plus de blocs +/// Quand ce blockid est poussé dans la queue, le source termine proprement +/// après avoir fini de traiter le bloc en cours +pub const END_OF_BLOCKS_SIGNAL: EventId = EventId::MAX; /// Nombre de blocs récents à mémoriser pour éviter les re-téléchargements const RECENT_BLOCKS_CACHE_SIZE: usize = 10; @@ -512,34 +512,38 @@ impl NodeLogic for RadioParadiseStreamSourceLogic { let mut last_start_instant: Option = None; loop { - // Attendre un block ID (timeout court pour une radio) - tracing::debug!("Waiting for block_id from queue (timeout={}s)...", BLOCK_ID_TIMEOUT_SECS); - let event_id = match tokio::time::timeout( - Duration::from_secs(BLOCK_ID_TIMEOUT_SECS), - async { - while self.block_queue.is_empty() { - tracing::trace!("block_queue is empty, sleeping..."); - tokio::time::sleep(Duration::from_millis(100)).await; + // Attendre un block ID depuis la queue (pas de timeout - mode idle) + tracing::debug!("Waiting for block_id from queue (idle mode, no timeout)..."); + let event_id = loop { + // Vérifier d'abord le stop_token + if stop_token.is_cancelled() { + tracing::info!("Stop token cancelled while waiting for block_id"); + break None; + } - if stop_token.is_cancelled() { - tracing::debug!("stop_token cancelled while waiting for block_id"); - return None; - } - } - self.block_queue.pop_front() - } - ).await { - Ok(Some(id)) => { + // Essayer de pop un event_id + if let Some(id) = self.block_queue.pop_front() { tracing::debug!("Got event_id {} from queue", id); - id + + // Vérifier si c'est le signal de fin + if id == END_OF_BLOCKS_SIGNAL { + tracing::info!("Received END_OF_BLOCKS_SIGNAL, finishing after current block"); + break None; + } + + break Some(id); } - Ok(None) => { - tracing::debug!("Loop cancelled, breaking"); - break; - } // Cancelled - Err(_) => { - // Timeout - pas de nouveau bloc, on termine - tracing::warn!("Timeout waiting for block_id, breaking"); + + // Queue vide, attendre un peu et réessayer + tracing::trace!("block_queue is empty, sleeping 100ms..."); + tokio::time::sleep(Duration::from_millis(100)).await; + }; + + // Si on n'a pas d'event_id, on termine + let event_id = match event_id { + Some(id) => id, + None => { + tracing::info!("No more blocks to process, exiting loop"); break; } };