diff --git a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs index f3939f64..b696afa8 100644 --- a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs @@ -430,7 +430,7 @@ impl StreamingFlacSink { ); // Broadcast channel for FLAC bytes - let (broadcast, _) = timed_broadcast::channel(broadcast_capacity); + let (broadcast, _) = timed_broadcast::channel("Flac", broadcast_capacity); // FLAC header cache let 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 870ff5bf..845dc981 100644 --- a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs @@ -178,6 +178,18 @@ impl NodeLogic for StreamingOggFlacSinkLogic { Some(seg) => { match &seg.segment { _AudioSegment::Chunk(chunk) => { + if !self.ctx.first_chunk_timestamp_checked { + self.ctx.first_chunk_timestamp_checked = true; + if seg.timestamp_sec.abs() > 1e-6 { + warn!( + "StreamingFlacSink: first chunk timestamp is {:.6}s (expected 0.0)", + seg.timestamp_sec + ); + } else { + trace!("StreamingFlacSink: first chunk timestamp verified at 0.0s"); + } + } + // Detect sample rate from first chunk and initialize encoder if self.ctx.sample_rate.is_none() { let sample_rate = chunk.sample_rate(); @@ -375,7 +387,7 @@ impl StreamingOggFlacSink { "Streaming Sink: using broadcast capacity of {} items (max_lead_time={:.1}s)", broadcast_capacity, broadcast_max_lead_time ); - let (broadcast, _) = timed_broadcast::channel(broadcast_capacity); + let (broadcast, _) = timed_broadcast::channel("Ogg-Flac",broadcast_capacity); // OGG-FLAC header cache let header = Arc::new(RwLock::new(None)); diff --git a/pmoaudio-ext/src/sinks/streaming_sink_common.rs b/pmoaudio-ext/src/sinks/streaming_sink_common.rs index ce968956..1d6199cd 100644 --- a/pmoaudio-ext/src/sinks/streaming_sink_common.rs +++ b/pmoaudio-ext/src/sinks/streaming_sink_common.rs @@ -347,7 +347,10 @@ impl SharedSinkContext { self.pcm_tx = Some(pcm_tx); self.pcm_rx = Some(pcm_rx); - self.initialize_encoder(sample_rate, self.timestamp_offset_sec, broadcaster) + // self.initialize_encoder(sample_rate, self.timestamp_offset_sec, broadcaster) + // .await?; + + self.initialize_encoder(sample_rate, 0.0, broadcaster) .await?; debug!("FLAC encoder restarted successfully for new track"); diff --git a/pmoaudio-ext/src/sinks/timed_broadcast.rs b/pmoaudio-ext/src/sinks/timed_broadcast.rs index e4930325..ff4dd273 100644 --- a/pmoaudio-ext/src/sinks/timed_broadcast.rs +++ b/pmoaudio-ext/src/sinks/timed_broadcast.rs @@ -5,17 +5,13 @@ //! - 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}, + collections::VecDeque, fmt, string, sync::{ + Arc, Mutex, Weak, atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering} + }, time::{Duration, Instant} }; use tokio::sync::Notify; -use tracing::{info, trace, warn}; +use tracing::{info, debug, trace, warn}; /// Tolérance pour détecter un timestamp à zéro (TopZero). const TOP_ZERO_EPSILON: f64 = 1e-9; @@ -83,6 +79,7 @@ struct Entry { } struct State { + name: String, buffer: VecDeque>, head_seq: u64, next_seq: u64, @@ -96,8 +93,9 @@ struct State { } impl State { - fn new(capacity: usize, epoch_start: Instant) -> Self { + fn new(name: &str,capacity: usize, epoch_start: Instant) -> Self { Self { + name: name.to_string() , buffer: VecDeque::with_capacity(capacity), head_seq: 0, next_seq: 0, @@ -112,8 +110,8 @@ impl State { } fn purge_expired(&mut self, now: Instant) -> bool { - // Throttling : purger au maximum toutes les 100ms - if now.duration_since(self.last_purge) < Duration::from_millis(100) { + // Throttling : purger au maximum toutes les 20ms + if now.duration_since(self.last_purge) < Duration::from_millis(20) { return false; } self.last_purge = now; @@ -122,7 +120,7 @@ impl State { while let Some(entry) = self.buffer.front() { if entry.expires_at <= now { let delta = now - entry.expires_at; - // info!("TimedBroadcast: purging expired packet (epoch={},delta={})", entry.epoch, delta.as_millis()); + debug!("TimedBroadcast[{}]: purging expired packet (@{} epoch={},delta={})", self.name,entry.seq, entry.epoch, delta.as_millis()); self.buffer.pop_front(); self.head_seq += 1; purged += 1; @@ -167,7 +165,11 @@ impl State { } for _ in 0..removable { - if self.buffer.pop_front().is_some() { + let oentry = self.buffer.pop_front(); + if oentry.is_some() { + let entry= oentry.unwrap(); + debug!("TimedBroadcast[{}]: pruning played packet (@{} epoch={})", self.name,entry.seq, entry.epoch); + self.head_seq += 1; } } @@ -186,9 +188,9 @@ struct Inner { } impl Inner { - fn new(capacity: usize) -> Self { + fn new(name: &str,capacity: usize) -> Self { Self { - state: Mutex::new(State::new(capacity, Instant::now())), + state: Mutex::new(State::new(name,capacity, Instant::now())), data_notify: Notify::new(), space_notify: Notify::new(), capacity, @@ -210,9 +212,9 @@ impl Inner { } /// Créé un channel broadcast temporisé. -pub fn channel(capacity: usize) -> (Sender, Receiver) { +pub fn channel(name: &str, capacity: usize) -> (Sender, Receiver) { assert!(capacity > 0, "capacity must be > 0"); - let inner = Arc::new(Inner::new(capacity)); + let inner = Arc::new(Inner::new(name,capacity)); let next_seq = { let state = inner.state.lock().expect("timed broadcast mutex poisoned"); state.next_seq @@ -312,7 +314,10 @@ impl Sender { audio_timestamp ); } else if is_top_zero { + // Restart epoch on TopZero relative to current wall-clock time to avoid + // expired packets when there's a long gap between tracks. state.epoch_start = state.last_segment_end.unwrap_or(now); + // state.epoch_start = now; state.epoch = state.epoch.wrapping_add(1); info!( "TimedBroadcast: new epoch={} (continuous={})", diff --git a/pmoparadise/examples/single_channel_server.rs b/pmoparadise/examples/single_channel_server.rs index 4d9859f1..c6de215a 100644 --- a/pmoparadise/examples/single_channel_server.rs +++ b/pmoparadise/examples/single_channel_server.rs @@ -79,11 +79,13 @@ async fn main() -> anyhow::Result<()> { let app = Router::new() .route("/stream/flac", get(stream_flac)) + .route("/stream/ogg", get(stream_ogg)) .with_state(state); let addr: SocketAddr = ([0, 0, 0, 0], 8080).into(); - info!("HTTP server listening on http://{addr}/stream/flac"); + info!("HTTP server listening on http://{addr}/stream/flac and /stream/ogg"); info!("Connect with a FLAC player (e.g. ffplay http://localhost:8080/stream/flac)"); + info!("Connect with an OGG/OGG-FLAC player (e.g. ffplay http://localhost:8080/stream/ogg)"); let listener = TcpListener::bind(addr).await?; axum::serve(listener, app.into_make_service()).await?; @@ -108,6 +110,23 @@ async fn stream_flac(State(state): State) -> Result) -> Result { + let stream = state.channel.subscribe_ogg(); + let body = Body::from_stream(ReaderStream::new(stream)); + Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/ogg") + .header( + "X-PMO-Channel", + format!( + "{} ({})", + state.descriptor.display_name, state.descriptor.slug + ), + ) + .body(body) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR) +} + fn pick_descriptor(arg: Option) -> anyhow::Result { if let Some(token) = arg { if let Some(desc) = ALL_CHANNELS.iter().find(|c| c.slug == token) { diff --git a/pmoparadise/src/models.rs b/pmoparadise/src/models.rs index cdabea3b..79d7c75a 100644 --- a/pmoparadise/src/models.rs +++ b/pmoparadise/src/models.rs @@ -376,6 +376,11 @@ mod tests { cover: None, rating: None, extra: HashMap::new(), + gapless_url: todo!(), + sched_time_millis: todo!(), + song_id: todo!(), + artist_id: todo!(), + cover_large: todo!(), }; assert_eq!(song.end_time_ms(), 6000);