eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Fabriquans un object Channel dans RadioRaradise

Browse Source
This commit is contained in:
Eric Coissac
2025-11-15 14:02:22 +01:00
parent 4e27255305
commit d9ad056933
12 changed files with 712 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pmoaudio-ext/src/lib.rs
View File

@@ -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")]

5
pmoaudio-ext/src/sinks/broadcast_pacing.rs
View File

@@ -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);
}

18
pmoaudio-ext/src/sinks/streaming_flac_sink.rs
View File

@@ -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<RwLock<Option<Bytes>>>,
auto_stop: Arc<AtomicBool>,
}
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 {
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 {
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 {

13
pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs
View File

@@ -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<RwLock<Option<Bytes>>>,
auto_stop: Arc<AtomicBool>,
}
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 {
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 {

37
pmoaudio-ext/src/sinks/timed_broadcast.rs
View File

@@ -99,7 +99,7 @@ impl<T> State<T> {
}
}
if purged > 0 {
warn!(
tracing::debug!(
"TimedBroadcast: purged {} expired packet(s) (head_seq={})",
purged,
self.head_seq
@@ -167,10 +167,7 @@ impl<T> Inner<T> {
}
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<T>(capacity: usize) -> (Sender<T>, Receiver<T>) {
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<T>(capacity: usize) -> (Sender<T>, Receiver<T>) {
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<T> Sender<T> {
}
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<T> Clone for Receiver<T> {
impl<T> Drop for Receiver<T> {
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();

1
pmoaudio/src/nodes/timer_node.rs
View File

@@ -340,4 +340,3 @@ impl TypedAudioNode for TimerNode {
Some(TypeRequirement::any())
}
}

157
pmoparadise/examples/serve_channels.rs Normal file
View File

@@ -0,0 +1,157 @@
//! Minimal HTTP server exposing all four Radio Paradise channels.
//!
//! Routes:
//! - `/radioparadise/stream/<slug>/flac`
//! - `/radioparadise/stream/<slug>/ogg`
//! - `/radioparadise/stream/<slug>/icy`
//! - `/radioparadise/metadata/<slug>`
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<ParadiseChannelManager>,
}
#[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<Arc<AppState>>| {
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<Arc<AppState>>| {
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<Arc<AppState>>| {
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<Arc<AppState>>| {
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<ParadiseChannelManager>,
channel_id: u8,
) -> Result<Response, StatusCode> {
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<ParadiseChannelManager>,
channel_id: u8,
) -> Result<Response, StatusCode> {
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<ParadiseChannelManager>,
channel_id: u8,
) -> Result<Response, StatusCode> {
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<ParadiseChannelManager>,
channel_id: u8,
) -> Result<impl IntoResponse, StatusCode> {
let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?;
let metadata = channel.metadata().await;
Ok(Json(metadata))
}

37
pmoparadise/examples/stream_block.rs
View File

@@ -248,9 +248,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
});
// 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<dyn std::error::Error>> {
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!("");

8
pmoparadise/src/lib.rs
View File

@@ -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,

6
pmoparadise/src/pmoserver_ext.rs
View File

@@ -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

105
pmoparadise/src/radio_paradise_stream_source.rs
View File

@@ -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<Mutex<VecDeque<EventId>>>,
notify: Arc<Notify>,
}
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<EventId> {
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<EventId> {
let queue = self.queue.lock().expect("block queue poisoned");
queue.iter().copied().collect()
}
fn front(&self) -> Option<EventId> {
let queue = self.queue.lock().expect("block queue poisoned");
queue.front().copied()
}
fn back(&self) -> Option<EventId> {
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<EventId>,
block_queue: VecDeque<EventId>,
block_queue: BlockQueueHandle,
stats: Arc<NodeStats>,
}
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<RadioParadiseStreamSourceLogic>,
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));
}
}

381
pmoparadise/src/stream_channel.rs Normal file
View File

@@ -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::<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
}
}
_ => {
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<ChannelState>,
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<Self> {
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<Self>) {
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<ChannelState>,
}
impl $name {
fn new(inner: $inner, state: Arc<ChannelState>) -> Self {
Self { inner, state }
}
}
impl AsyncRead for $name {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
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<u8, Arc<ParadiseStreamChannel>>,
}
impl ParadiseChannelManager {
pub fn new(channels: HashMap<u8, Arc<ParadiseStreamChannel>>) -> Self {
Self { channels }
}
pub async fn with_defaults() -> Result<Self> {
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<Arc<ParadiseStreamChannel>> {
self.channels.get(&id).cloned()
}
pub fn iter(&self) -> impl Iterator<Item = &Arc<ParadiseStreamChannel>> {
self.channels.values()
}
}