diff --git a/pmoaudio-ext/src/lib.rs b/pmoaudio-ext/src/lib.rs index e9003a3e..7b9727a9 100755 --- a/pmoaudio-ext/src/lib.rs +++ b/pmoaudio-ext/src/lib.rs @@ -22,14 +22,14 @@ //! Aucune des crates ci-dessus ne dépend de `pmoaudio-ext`, évitant ainsi //! tout cycle de dépendances. -#[cfg(feature = "cache-sink")] +#[cfg(any(feature = "cache-sink", feature = "http-stream"))] pub mod sinks; #[cfg(feature = "playlist")] pub mod sources; // Re-exports pour faciliter l'utilisation -#[cfg(feature = "cache-sink")] +#[cfg(any(feature = "cache-sink", feature = "http-stream"))] pub use sinks::*; #[cfg(feature = "playlist")] diff --git a/pmoaudio-ext/src/sinks/broadcast_pacing.rs b/pmoaudio-ext/src/sinks/broadcast_pacing.rs index f0f409fd..685f15f7 100644 --- a/pmoaudio-ext/src/sinks/broadcast_pacing.rs +++ b/pmoaudio-ext/src/sinks/broadcast_pacing.rs @@ -76,10 +76,7 @@ impl BroadcastPacer { if lead_time < 0.0 { warn!( "{}: Dropping late frame: audio_ts={:.3}s, elapsed={:.3}s, lag={:.3}s", - self.label, - audio_timestamp, - elapsed, - -lead_time + self.label, audio_timestamp, elapsed, -lead_time ); return Err(SkipFrame); } diff --git a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs index 3b2d6d9d..aa81021c 100644 --- a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs @@ -57,7 +57,7 @@ use std::collections::VecDeque; use std::io; use std::pin::Pin; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::task::{Context, Poll}; use std::time::Duration; @@ -177,6 +177,8 @@ pub struct StreamHandle { /// Cached FLAC header (sent to new subscribers first) flac_header: Arc>>, + + auto_stop: Arc, } impl StreamHandle { @@ -243,6 +245,11 @@ impl StreamHandle { pub fn should_stop(&self) -> bool { self.active_clients.load(Ordering::SeqCst) == 0 } + + /// Enable or disable automatic pipeline shutdown when the last client disconnects. + pub fn set_auto_stop(&self, enabled: bool) { + self.auto_stop.store(enabled, Ordering::SeqCst); + } } /// State for FLAC stream subscription. @@ -348,8 +355,12 @@ impl Drop for FlacClientStream { debug!("FLAC client disconnected (remaining: {})", count - 1); if count == 1 { - info!("Last client disconnected, signaling pipeline stop"); - self.handle.stop_token.cancel(); + if self.handle.auto_stop.load(Ordering::SeqCst) { + info!("Last client disconnected, signaling pipeline stop"); + self.handle.stop_token.cancel(); + } else { + info!("Last client disconnected, keeping pipeline alive"); + } } } } @@ -556,8 +567,12 @@ impl Drop for IcyClientStream { debug!("ICY client disconnected (remaining: {})", count - 1); if count == 1 { - info!("Last client disconnected, signaling pipeline stop"); - self.handle.stop_token.cancel(); + if self.handle.auto_stop.load(Ordering::SeqCst) { + info!("Last client disconnected, signaling pipeline stop"); + self.handle.stop_token.cancel(); + } else { + info!("Last client disconnected, keeping pipeline alive"); + } } } } @@ -1068,6 +1083,7 @@ impl StreamingFlacSink { active_clients, stop_token: stop_token.clone(), flac_header: flac_header.clone(), + auto_stop: Arc::new(AtomicBool::new(true)), }; let logic = StreamingFlacSinkLogic { diff --git a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs index a236a6c4..106d6939 100644 --- a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs @@ -49,7 +49,7 @@ use std::collections::VecDeque; use std::io; use std::pin::Pin; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::task::{Context, Poll}; @@ -114,6 +114,8 @@ pub struct OggFlacStreamHandle { /// Cached OGG-FLAC header (sent to new subscribers first) ogg_header: Arc>>, + + auto_stop: Arc, } impl OggFlacStreamHandle { @@ -143,6 +145,10 @@ impl OggFlacStreamHandle { pub fn active_client_count(&self) -> usize { self.active_clients.load(Ordering::SeqCst) } + + pub fn set_auto_stop(&self, enabled: bool) { + self.auto_stop.store(enabled, Ordering::SeqCst); + } } /// State for OGG-FLAC stream subscription. @@ -248,8 +254,12 @@ impl Drop for OggFlacClientStream { 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(); + if self.handle.auto_stop.load(Ordering::SeqCst) { + info!("Last OGG-FLAC client disconnected, signaling pipeline stop"); + self.handle.stop_token.cancel(); + } else { + info!("Last OGG-FLAC client disconnected, keeping pipeline alive"); + } } } } @@ -555,6 +565,7 @@ impl StreamingOggFlacSink { active_clients, stop_token: stop_token.clone(), ogg_header: ogg_header.clone(), + auto_stop: Arc::new(AtomicBool::new(true)), }; let logic = StreamingOggFlacSinkLogic { diff --git a/pmoaudio-ext/src/sinks/timed_broadcast.rs b/pmoaudio-ext/src/sinks/timed_broadcast.rs index 205b6e52..e73b5f8d 100644 --- a/pmoaudio-ext/src/sinks/timed_broadcast.rs +++ b/pmoaudio-ext/src/sinks/timed_broadcast.rs @@ -99,7 +99,7 @@ impl State { } } if purged > 0 { - warn!( + tracing::debug!( "TimedBroadcast: purged {} expired packet(s) (head_seq={})", purged, self.head_seq @@ -167,10 +167,7 @@ impl Inner { } fn close(&self) { - if !self - .is_closed - .swap(true, Ordering::SeqCst) - { + if !self.is_closed.swap(true, Ordering::SeqCst) { if let Ok(mut state) = self.state.lock() { state.closed = true; } @@ -185,10 +182,7 @@ 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"); + let state = inner.state.lock().expect("timed broadcast mutex poisoned"); state.next_seq }; let sender = Sender { @@ -198,10 +192,7 @@ pub fn channel(capacity: usize) -> (Sender, Receiver) { next_seq: AtomicU64::new(next_seq), }); { - let mut state = inner - .state - .lock() - .expect("timed broadcast mutex poisoned"); + let mut state = inner.state.lock().expect("timed broadcast mutex poisoned"); state.cursors.push(Arc::downgrade(&cursor)); } inner.receiver_count.store(1, Ordering::SeqCst); @@ -257,15 +248,12 @@ impl Sender { } if state.buffer.len() < self.inner.capacity { - let audio_offset = - Duration::from_secs_f64(audio_timestamp.max(0.0)); + 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"), + payload: payload.take().expect("payload already consumed"), audio_timestamp, epoch: state.epoch, }; @@ -390,19 +378,14 @@ where 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 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); + self.cursor.next_seq.store(self.next_seq, Ordering::SeqCst); if state.prune_consumed() { self.inner.space_notify.notify_waiters(); } @@ -460,9 +443,7 @@ impl Clone for Receiver { impl Drop for Receiver { fn drop(&mut self) { - self.cursor - .next_seq - .store(self.next_seq, Ordering::SeqCst); + 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(); diff --git a/pmoaudio/src/nodes/timer_node.rs b/pmoaudio/src/nodes/timer_node.rs index eb10f999..67038319 100644 --- a/pmoaudio/src/nodes/timer_node.rs +++ b/pmoaudio/src/nodes/timer_node.rs @@ -340,4 +340,3 @@ impl TypedAudioNode for TimerNode { Some(TypeRequirement::any()) } } - diff --git a/pmoparadise/examples/serve_channels.rs b/pmoparadise/examples/serve_channels.rs new file mode 100644 index 00000000..19c4aef9 --- /dev/null +++ b/pmoparadise/examples/serve_channels.rs @@ -0,0 +1,157 @@ +//! Minimal HTTP server exposing all four Radio Paradise channels. +//! +//! Routes: +//! - `/radioparadise/stream//flac` +//! - `/radioparadise/stream//ogg` +//! - `/radioparadise/stream//icy` +//! - `/radioparadise/metadata/` + +use std::sync::Arc; + +use axum::{ + body::Body, + extract::State, + http::StatusCode, + response::{IntoResponse, Response}, + Json, +}; +use pmoparadise::{channels::ALL_CHANNELS, stream_channel::ParadiseChannelManager}; +use pmoserver::{init_logging, ServerBuilder}; +use tokio_util::io::ReaderStream; +use tracing::info; + +#[derive(Clone)] +struct AppState { + manager: Arc, +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let _ = init_logging(); + + info!("Initializing Radio Paradise channels..."); + let manager = Arc::new(ParadiseChannelManager::with_defaults().await?); + let app_state = Arc::new(AppState { + manager: manager.clone(), + }); + + let mut server = ServerBuilder::new("RadioParadiseChannels", "http://localhost", 8080).build(); + + for descriptor in ALL_CHANNELS.iter() { + let slug = descriptor.slug; + let flac_path = format!("/radioparadise/stream/{}/flac", slug); + let ogg_path = format!("/radioparadise/stream/{}/ogg", slug); + let icy_path = format!("/radioparadise/stream/{}/icy", slug); + let meta_path = format!("/radioparadise/metadata/{}", slug); + let channel_id = descriptor.id; + + server + .add_handler_with_state( + &flac_path, + move |State(state): State>| { + let manager = state.manager.clone(); + async move { stream_flac(manager, channel_id).await } + }, + app_state.clone(), + ) + .await; + + server + .add_handler_with_state( + &ogg_path, + move |State(state): State>| { + let manager = state.manager.clone(); + async move { stream_ogg(manager, channel_id).await } + }, + app_state.clone(), + ) + .await; + + server + .add_handler_with_state( + &icy_path, + move |State(state): State>| { + let manager = state.manager.clone(); + async move { stream_icy(manager, channel_id).await } + }, + app_state.clone(), + ) + .await; + + server + .add_handler_with_state( + &meta_path, + move |State(state): State>| { + let manager = state.manager.clone(); + async move { get_metadata(manager, channel_id).await } + }, + app_state.clone(), + ) + .await; + } + + info!("========================================"); + info!("Radio Paradise streaming server running on http://localhost:8080"); + info!("Available channels:"); + for descriptor in ALL_CHANNELS.iter() { + info!( + " {}: /radioparadise/stream/{}/flac (also /ogg, /icy, metadata)", + descriptor.display_name, descriptor.slug + ); + } + info!("Press Ctrl+C to stop."); + info!("========================================"); + + server.start().await; + server.wait().await; + Ok(()) +} + +async fn stream_flac( + manager: Arc, + channel_id: u8, +) -> Result { + let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?; + let stream = channel.subscribe_flac(); + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/flac") + .body(Body::from_stream(ReaderStream::new(stream))) + .unwrap()) +} + +async fn stream_ogg( + manager: Arc, + channel_id: u8, +) -> Result { + let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?; + let stream = channel.subscribe_ogg(); + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/ogg") + .body(Body::from_stream(ReaderStream::new(stream))) + .unwrap()) +} + +async fn stream_icy( + manager: Arc, + channel_id: u8, +) -> Result { + let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?; + let stream = channel.subscribe_icy(); + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/flac") + .header("icy-metaint", "16000") + .body(Body::from_stream(ReaderStream::new(stream))) + .unwrap()) +} + +async fn get_metadata( + manager: Arc, + channel_id: u8, +) -> Result { + let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?; + let metadata = channel.metadata().await; + Ok(Json(metadata)) +} diff --git a/pmoparadise/examples/stream_block.rs b/pmoparadise/examples/stream_block.rs index 3224bc6d..d7849d63 100644 --- a/pmoparadise/examples/stream_block.rs +++ b/pmoparadise/examples/stream_block.rs @@ -248,9 +248,7 @@ async fn main() -> Result<(), Box> { source.register(Box::new(streaming_sink)); source.register(Box::new(ogg_sink)); - tracing::info!( - "Pipeline connected: StreamSource → TimerBufferNode → {{FLAC, OGG}} sinks" - ); + tracing::info!("Pipeline connected: StreamSource → TimerBufferNode → {{FLAC, OGG}} sinks"); // ═══════════════════════════════════════════════════════════════════════════ // Setup pmoserver with streaming routes @@ -267,19 +265,36 @@ async fn main() -> Result<(), Box> { }); // Add streaming routes + let base = "/radioparadise/test"; server - .add_handler_with_state("/test/stream", stream_handler, app_state.clone()) + .add_handler_with_state( + &format!("{}/stream", base), + stream_handler, + app_state.clone(), + ) .await; server - .add_handler_with_state("/test/stream-icy", stream_icy_handler, app_state.clone()) + .add_handler_with_state( + &format!("{}/stream-icy", base), + stream_icy_handler, + app_state.clone(), + ) .await; server - .add_handler_with_state("/test/stream-ogg", stream_ogg_handler, app_state.clone()) + .add_handler_with_state( + &format!("{}/stream-ogg", base), + stream_ogg_handler, + app_state.clone(), + ) .await; // Add metadata route server - .add_handler_with_state("/test/metadata", metadata_handler, app_state.clone()) + .add_handler_with_state( + &format!("{}/metadata", base), + metadata_handler, + app_state.clone(), + ) .await; // Add health check @@ -290,16 +305,16 @@ async fn main() -> Result<(), Box> { 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!(" vlc http://localhost:8080{}/stream", base); tracing::info!(""); tracing::info!("OGG-FLAC stream (streaming container with metadata support):"); - tracing::info!(" vlc http://localhost:8080/test/stream-ogg"); + tracing::info!(" vlc http://localhost:8080{}/stream-ogg", base); tracing::info!(""); tracing::info!("FLAC + ICY metadata stream (for ICY-aware clients):"); - tracing::info!(" http://localhost:8080/test/stream-icy"); + tracing::info!(" http://localhost:8080{}/stream-icy", base); tracing::info!(""); tracing::info!("Metadata endpoint (JSON):"); - tracing::info!(" curl http://localhost:8080/test/metadata"); + tracing::info!(" curl http://localhost:8080{}/metadata", base); tracing::info!("========================================"); tracing::info!(""); diff --git a/pmoparadise/src/lib.rs b/pmoparadise/src/lib.rs index 376d97d9..b0b7d0b5 100644 --- a/pmoparadise/src/lib.rs +++ b/pmoparadise/src/lib.rs @@ -228,6 +228,9 @@ pub mod config_ext; #[cfg(feature = "pmoaudio")] pub mod radio_paradise_stream_source; +#[cfg(feature = "pmoaudio")] +pub mod stream_channel; + // Re-exports for convenience pub use client::{ClientBuilder, RadioParadiseClient}; pub use error::{Error, Result}; @@ -237,6 +240,11 @@ pub use source::RadioParadiseSource; #[cfg(feature = "pmoaudio")] pub use radio_paradise_stream_source::{RadioParadiseStreamSource, END_OF_BLOCKS_SIGNAL}; +#[cfg(feature = "pmoaudio")] +pub use stream_channel::{ + ParadiseChannelManager, ParadiseStreamChannel, ParadiseStreamChannelConfig, +}; + #[cfg(feature = "pmoserver")] pub use pmoserver_ext::{ create_api_router, RadioParadiseApiDoc, RadioParadiseExt, RadioParadiseState, diff --git a/pmoparadise/src/pmoserver_ext.rs b/pmoparadise/src/pmoserver_ext.rs index 5c19d5ad..0f4867da 100644 --- a/pmoparadise/src/pmoserver_ext.rs +++ b/pmoparadise/src/pmoserver_ext.rs @@ -341,12 +341,14 @@ pub struct RadioParadiseApiDoc; /// Crée le router pour l'API Radio Paradise pub fn create_api_router(state: RadioParadiseState) -> Router { - Router::new() + let api = Router::new() .route("/now-playing", get(get_now_playing)) .route("/block/current", get(get_current_block)) .route("/block/{event_id}", get(get_block_by_id)) .route("/channels", get(get_channels)) - .with_state(state) + .with_state(state); + + Router::new().nest("/radioparadise", api) } /// Trait d'extension pour pmoserver::Server diff --git a/pmoparadise/src/radio_paradise_stream_source.rs b/pmoparadise/src/radio_paradise_stream_source.rs index c5bf5f7e..57afbc92 100644 --- a/pmoparadise/src/radio_paradise_stream_source.rs +++ b/pmoparadise/src/radio_paradise_stream_source.rs @@ -19,11 +19,11 @@ use pmoflac::decode_audio_stream; use pmometadata::{MemoryTrackMetadata, TrackMetadata}; use std::{ collections::VecDeque, - sync::Arc, + sync::{Arc, Mutex}, time::{Duration, Instant}, }; use tokio::io::AsyncReadExt; -use tokio::sync::{mpsc, RwLock}; +use tokio::sync::{mpsc, Notify, RwLock}; use tokio_util::{io::StreamReader, sync::CancellationToken}; /// Signal spécial pour indiquer qu'il n'y aura plus de blocs @@ -34,6 +34,58 @@ 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; +/// Handle pour alimenter la queue de blocs pendant que la source tourne. +#[derive(Clone, Default)] +pub struct BlockQueueHandle { + queue: Arc>>, + notify: Arc, +} + +impl BlockQueueHandle { + fn new() -> Self { + Self { + queue: Arc::new(Mutex::new(VecDeque::new())), + notify: Arc::new(Notify::new()), + } + } + + /// Enfile un block pour traitement. + pub fn enqueue(&self, event_id: EventId) { + { + let mut queue = self.queue.lock().expect("block queue poisoned"); + queue.push_back(event_id); + } + self.notify.notify_one(); + } + + /// Retire le prochain block s'il existe. + fn pop(&self) -> Option { + let mut queue = self.queue.lock().expect("block queue poisoned"); + queue.pop_front() + } + + /// Nombre d'éléments en attente. + pub fn len(&self) -> usize { + let queue = self.queue.lock().expect("block queue poisoned"); + queue.len() + } + + fn snapshot(&self) -> Vec { + let queue = self.queue.lock().expect("block queue poisoned"); + queue.iter().copied().collect() + } + + fn front(&self) -> Option { + let queue = self.queue.lock().expect("block queue poisoned"); + queue.front().copied() + } + + fn back(&self) -> Option { + let queue = self.queue.lock().expect("block queue poisoned"); + queue.back().copied() + } +} + // ═══════════════════════════════════════════════════════════════════════════ // RadioParadiseStreamSourceLogic - Logique métier pure // ═══════════════════════════════════════════════════════════════════════════ @@ -43,12 +95,21 @@ pub struct RadioParadiseStreamSourceLogic { client: RadioParadiseClient, chunk_frames: usize, recent_blocks: VecDeque, - block_queue: VecDeque, + block_queue: BlockQueueHandle, stats: Arc, } impl RadioParadiseStreamSourceLogic { pub fn new(client: RadioParadiseClient, chunk_duration_ms: u32) -> Self { + let handle = BlockQueueHandle::new(); + Self::with_queue(client, chunk_duration_ms, handle) + } + + fn with_queue( + client: RadioParadiseClient, + chunk_duration_ms: u32, + block_queue: BlockQueueHandle, + ) -> Self { // Calculer chunk_frames pour la durée cible (on suppose 44.1kHz) let chunk_frames = ((chunk_duration_ms as f64 / 1000.0) * 44100.0) as usize; @@ -56,14 +117,14 @@ impl RadioParadiseStreamSourceLogic { client, chunk_frames, recent_blocks: VecDeque::with_capacity(RECENT_BLOCKS_CACHE_SIZE), - block_queue: VecDeque::new(), + block_queue, stats: NodeStats::new("RadioParadiseStreamSource"), } } /// Ajoute un block ID à la file d'attente - pub fn push_block_id(&mut self, event_id: EventId) { - self.block_queue.push_back(event_id); + pub fn push_block_id(&self, event_id: EventId) { + self.block_queue.enqueue(event_id); } /// Vérifie si un bloc a été téléchargé récemment @@ -556,7 +617,7 @@ impl NodeLogic for RadioParadiseStreamSourceLogic { "RadioParadiseStreamSource::process() started, block_queue has {} items", self.block_queue.len() ); - for (i, event_id) in self.block_queue.iter().enumerate() { + for (i, event_id) in self.block_queue.snapshot().iter().enumerate() { tracing::debug!(" block_queue[{}] = {}", i, event_id); } @@ -575,7 +636,7 @@ impl NodeLogic for RadioParadiseStreamSourceLogic { } // Essayer de pop un event_id - if let Some(id) = self.block_queue.pop_front() { + if let Some(id) = self.block_queue.pop() { tracing::debug!("Got event_id {} from queue", id); // Vérifier si c'est le signal de fin @@ -589,9 +650,12 @@ impl NodeLogic for RadioParadiseStreamSourceLogic { break Some(id); } - // Queue vide, attendre un peu et réessayer - tracing::trace!("block_queue is empty, sleeping 100ms..."); - tokio::time::sleep(Duration::from_millis(100)).await; + tracing::trace!("block_queue is empty, waiting for new events..."); + tokio::select! { + _ = stop_token.cancelled() => break None, + _ = self.block_queue.notify.notified() => {}, + _ = tokio::time::sleep(Duration::from_millis(100)) => {} + }; }; // Si on n'a pas d'event_id, on termine @@ -679,6 +743,7 @@ impl NodeLogic for RadioParadiseStreamSourceLogic { pub struct RadioParadiseStreamSource { inner: Node, + block_handle: BlockQueueHandle, } impl RadioParadiseStreamSource { @@ -689,15 +754,23 @@ impl RadioParadiseStreamSource { /// Crée une nouvelle source avec durée de chunk personnalisée pub fn with_chunk_duration(client: RadioParadiseClient, chunk_duration_ms: u32) -> Self { - let logic = RadioParadiseStreamSourceLogic::new(client, chunk_duration_ms); + let handle = BlockQueueHandle::new(); + let logic = + RadioParadiseStreamSourceLogic::with_queue(client, chunk_duration_ms, handle.clone()); Self { inner: Node::new_source(logic), + block_handle: handle, } } /// Ajoute un block ID à la file d'attente de téléchargement - pub fn push_block_id(&mut self, event_id: EventId) { - self.inner.logic_mut().push_block_id(event_id); + pub fn push_block_id(&self, event_id: EventId) { + self.block_handle.enqueue(event_id); + } + + /// Retourne un handle permettant d'enfiler des blocks dynamiquement. + pub fn block_handle(&self) -> BlockQueueHandle { + self.block_handle.clone() } } @@ -907,7 +980,7 @@ mod tests { logic.push_block_id(300); assert_eq!(logic.block_queue.len(), 3); - assert_eq!(logic.block_queue.front(), Some(&100)); - assert_eq!(logic.block_queue.back(), Some(&300)); + assert_eq!(logic.block_queue.front(), Some(100)); + assert_eq!(logic.block_queue.back(), Some(300)); } } diff --git a/pmoparadise/src/stream_channel.rs b/pmoparadise/src/stream_channel.rs new file mode 100644 index 00000000..e554cb71 --- /dev/null +++ b/pmoparadise/src/stream_channel.rs @@ -0,0 +1,381 @@ +use std::{ + collections::HashMap, + pin::Pin, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, + task::{Context, Poll}, + time::Duration, +}; + +use crate::{ + channels::{ChannelDescriptor, ParadiseChannelKind, ALL_CHANNELS}, + client::RadioParadiseClient, + radio_paradise_stream_source::RadioParadiseStreamSource, +}; +use anyhow::Result; +use pmoaudio::AudioPipelineNode; +use pmoaudio_ext::{ + FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, OggFlacStreamHandle, + StreamHandle, StreamingFlacSink, StreamingOggFlacSink, +}; +use pmoflac::EncoderOptions; +use tokio::io::{AsyncRead, ReadBuf}; +use tokio::sync::Notify; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; +use tracing::{error, info, warn}; + +/// Configuration pour un canal Radio Paradise. +#[derive(Clone, Debug)] +pub struct ParadiseStreamChannelConfig { + /// Durée maximale (en secondes) d'avance acceptée par le broadcast. + pub max_lead_seconds: f64, +} + +impl Default for ParadiseStreamChannelConfig { + fn default() -> Self { + Self { + max_lead_seconds: 1.0, + } + } +} + +#[cfg(feature = "pmoconfig")] +impl ParadiseStreamChannelConfig { + pub fn from_config(cfg: &pmoconfig::Config, channel: ParadiseChannelKind) -> Self { + use serde_yaml::Value; + let path = [ + "sources", + "radio_paradise", + "channels", + channel.slug(), + "max_lead_seconds", + ]; + match cfg.get_value(&path) { + Ok(Value::Number(num)) => { + if let Some(v) = num.as_f64() { + Self { + max_lead_seconds: v.max(0.1), + } + } else { + let default = Self::default(); + let _ = + cfg.set_value(&path, Value::String(default.max_lead_seconds.to_string())); + default + } + } + Ok(Value::String(s)) => { + if let Ok(v) = s.parse::() { + Self { + max_lead_seconds: v.max(0.1), + } + } else { + let default = Self::default(); + let _ = + cfg.set_value(&path, Value::String(default.max_lead_seconds.to_string())); + default + } + } + _ => { + let default = Self::default(); + let _ = cfg.set_value(&path, Value::String(default.max_lead_seconds.to_string())); + default + } + } + } +} + +/// Stream complet (FLAC pur + OGG-FLAC) pour un canal Radio Paradise. +pub struct ParadiseStreamChannel { + descriptor: ChannelDescriptor, + state: Arc, + pipeline_handle: JoinHandle<()>, + feeder_handle: JoinHandle<()>, +} + +impl ParadiseStreamChannel { + /// Crée un canal avec client déjà configuré. + pub fn with_client( + descriptor: ChannelDescriptor, + client: RadioParadiseClient, + config: ParadiseStreamChannelConfig, + ) -> Self { + let mut source = RadioParadiseStreamSource::new(client.clone()); + let block_handle = source.block_handle(); + + let (flac_sink, stream_handle) = StreamingFlacSink::with_max_broadcast_lead( + EncoderOptions::default(), + 16, + config.max_lead_seconds, + ); + let (ogg_sink, ogg_handle) = StreamingOggFlacSink::with_max_broadcast_lead( + EncoderOptions::default(), + 16, + config.max_lead_seconds, + ); + + source.register(Box::new(flac_sink)); + source.register(Box::new(ogg_sink)); + stream_handle.set_auto_stop(false); + ogg_handle.set_auto_stop(false); + + let stop_token = CancellationToken::new(); + let pipeline_stop = stop_token.clone(); + let pipeline_handle = tokio::spawn(async move { + info!( + "RadioParadise stream pipeline started for channel {}", + descriptor.display_name + ); + if let Err(e) = Box::new(source).run(pipeline_stop).await { + error!( + "Pipeline error for channel {}: {}", + descriptor.display_name, e + ); + } + }); + + let state = Arc::new(ChannelState { + descriptor, + config, + client, + block_handle, + stream_handle, + ogg_handle, + active_clients: AtomicUsize::new(0), + activity_notify: Notify::new(), + stop_token, + }); + + let feeder_state = state.clone(); + let feeder_handle = tokio::spawn(async move { + feeder_state.run_scheduler().await; + }); + + Self { + descriptor, + state, + pipeline_handle, + feeder_handle, + } + } + + /// Crée un canal en construisant automatiquement le client pour ce descriptor. + pub async fn new( + descriptor: ChannelDescriptor, + config: ParadiseStreamChannelConfig, + ) -> Result { + let client = RadioParadiseClient::builder() + .channel(descriptor.id) + .build() + .await?; + Ok(Self::with_client(descriptor, client, config)) + } + + /// S'abonne au flux FLAC pur. + pub fn subscribe_flac(&self) -> ChannelFlacStream { + self.state.on_client_added(); + let inner = self.state.stream_handle.subscribe_flac(); + ChannelFlacStream::new(inner, self.state.clone()) + } + + /// S'abonne au flux FLAC + ICY metadata. + pub fn subscribe_icy(&self) -> ChannelIcyStream { + self.state.on_client_added(); + let inner = self.state.stream_handle.subscribe_icy(); + ChannelIcyStream::new(inner, self.state.clone()) + } + + /// S'abonne au flux OGG-FLAC. + pub fn subscribe_ogg(&self) -> ChannelOggStream { + self.state.on_client_added(); + let inner = self.state.ogg_handle.subscribe(); + ChannelOggStream::new(inner, self.state.clone()) + } + + /// Snapshot des métadonnées actuelles. + pub async fn metadata(&self) -> MetadataSnapshot { + self.state.stream_handle.get_metadata().await + } + + /// Nombre de clients actifs. + pub fn active_clients(&self) -> usize { + self.state.active_clients.load(Ordering::SeqCst) + } + + pub fn descriptor(&self) -> ChannelDescriptor { + self.descriptor + } +} + +impl Drop for ParadiseStreamChannel { + fn drop(&mut self) { + self.state.stop_token.cancel(); + self.pipeline_handle.abort(); + self.feeder_handle.abort(); + } +} + +struct ChannelState { + descriptor: ChannelDescriptor, + config: ParadiseStreamChannelConfig, + client: RadioParadiseClient, + block_handle: crate::radio_paradise_stream_source::BlockQueueHandle, + stream_handle: StreamHandle, + ogg_handle: OggFlacStreamHandle, + active_clients: AtomicUsize, + activity_notify: Notify, + stop_token: CancellationToken, +} + +impl ChannelState { + fn on_client_added(&self) { + if self.active_clients.fetch_add(1, Ordering::SeqCst) == 0 { + self.activity_notify.notify_one(); + } + } + + fn on_client_removed(&self) { + self.active_clients.fetch_sub(1, Ordering::SeqCst); + } + + async fn wait_for_clients(&self) -> bool { + while self.active_clients.load(Ordering::SeqCst) == 0 { + tokio::select! { + _ = self.stop_token.cancelled() => return false, + _ = self.activity_notify.notified() => {}, + } + } + true + } + + async fn run_scheduler(self: Arc) { + let mut backoff = Duration::from_secs(5); + loop { + if self.stop_token.is_cancelled() { + break; + } + + if !self.wait_for_clients().await { + break; + } + + match self.client.get_block(None).await { + Ok(block) => { + info!( + "Channel {} streaming block {}", + self.descriptor.display_name, block.event + ); + self.block_handle.enqueue(block.event); + let mut next_event = block.end_event; + + loop { + if self.stop_token.is_cancelled() { + return; + } + + if self.active_clients.load(Ordering::SeqCst) == 0 { + break; + } + + match self.client.get_block(Some(next_event)).await { + Ok(next_block) => { + self.block_handle.enqueue(next_block.event); + next_event = next_block.end_event; + backoff = Duration::from_secs(5); + } + Err(e) => { + warn!( + "Failed to fetch next block for channel {}: {}", + self.descriptor.display_name, e + ); + tokio::select! { + _ = self.stop_token.cancelled() => return, + _ = tokio::time::sleep(backoff) => {}, + } + backoff = (backoff * 2).min(Duration::from_secs(60)); + } + } + } + } + Err(e) => { + warn!( + "Failed to fetch current block for channel {}: {}", + self.descriptor.display_name, e + ); + tokio::select! { + _ = self.stop_token.cancelled() => break, + _ = tokio::time::sleep(backoff) => {}, + } + backoff = (backoff * 2).min(Duration::from_secs(60)); + } + } + } + } +} + +macro_rules! wrap_stream { + ($name:ident, $inner:ty) => { + pub struct $name { + inner: $inner, + state: Arc, + } + + impl $name { + fn new(inner: $inner, state: Arc) -> Self { + Self { inner, state } + } + } + + impl AsyncRead for $name { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } + } + + impl Drop for $name { + fn drop(&mut self) { + self.state.on_client_removed(); + } + } + }; +} + +wrap_stream!(ChannelFlacStream, FlacClientStream); +wrap_stream!(ChannelIcyStream, IcyClientStream); +wrap_stream!(ChannelOggStream, OggFlacClientStream); + +/// Gestionnaire multi-canaux. +pub struct ParadiseChannelManager { + channels: HashMap>, +} + +impl ParadiseChannelManager { + pub fn new(channels: HashMap>) -> Self { + Self { channels } + } + + pub async fn with_defaults() -> Result { + let mut map = HashMap::new(); + for descriptor in ALL_CHANNELS.iter().copied() { + let channel = + ParadiseStreamChannel::new(descriptor, ParadiseStreamChannelConfig::default()) + .await?; + map.insert(descriptor.id, Arc::new(channel)); + } + Ok(Self { channels: map }) + } + + pub fn get(&self, id: u8) -> Option> { + self.channels.get(&id).cloned() + } + + pub fn iter(&self) -> impl Iterator> { + self.channels.values() + } +}