fin du debuggage des sink

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

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

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");

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,
+    collections::VecDeque, fmt, string, sync::{
-    fmt,
+        Arc, Mutex, Weak, atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}
-    sync::{
+    }, time::{Duration, Instant}
-        atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
-        Arc, Mutex, Weak,
-    },
-    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<T> {
 }
 
 struct State<T> {
+    name: String,
     buffer: VecDeque<Entry<T>>,
     head_seq: u64,
     next_seq: u64,
@@ -96,8 +93,9 @@ struct State<T> {
 }
 
 impl<T> State<T> {
-    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<T> State<T> {
     }
 
     fn purge_expired(&mut self, now: Instant) -> bool {
-        // Throttling : purger au maximum toutes les 100ms
+        // Throttling : purger au maximum toutes les 20ms
-        if now.duration_since(self.last_purge) < Duration::from_millis(100) {
+        if now.duration_since(self.last_purge) < Duration::from_millis(20) {
             return false;
         }
         self.last_purge = now;
@@ -122,7 +120,7 @@ impl<T> State<T> {
         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<T> State<T> {
         }
 
         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<T> {
 }
 
 impl<T> Inner<T> {
-    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<T> Inner<T> {
 }
 
 /// Créé un channel broadcast temporisé.
-pub fn channel<T>(capacity: usize) -> (Sender<T>, Receiver<T>) {
+pub fn channel<T>(name: &str, capacity: usize) -> (Sender<T>, Receiver<T>) {
     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<T> Sender<T> {
                             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={})",

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<AppState>) -> Result<Response, StatusCo
         .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
 }
 
+async fn stream_ogg(State(state): State<AppState>) -> Result<Response, StatusCode> {
+    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<String>) -> anyhow::Result<ChannelDescriptor> {
     if let Some(token) = arg {
         if let Some(desc) = ALL_CHANNELS.iter().find(|c| c.slug == token) {

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