From b3f22d1b61603d9bb3a7cfe5d73e0143bc5345dd Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 12 Nov 2025 10:18:25 +0000 Subject: [PATCH] Fix stream_block bug: wait for playback completion before closing channel Previously, RadioParadiseStreamSource would close its output channel as soon as the block finished downloading and decoding, causing TimerNode to receive EOF and terminate immediately, even if it still had audio chunks in its buffer waiting to be sent with proper timing. This fix makes RadioParadiseStreamSource wait for the actual playback duration to elapse before closing the channel, ensuring that TimerNode has enough time to broadcast all chunks at the correct pace. Changes: - Modified download_and_decode_block() to return (timestamp, Instant) instead of just timestamp, capturing the start time - Added wait logic in process() to sleep for remaining playback time after sending EndOfStream, before returning and closing the channel - Added Instant import to support timing calculations This ensures Radio Paradise blocks (~20 minutes each) stream completely instead of stopping prematurely when download completes. --- .../src/radio_paradise_stream_source.rs | 49 ++++++++++++++++--- 1 file changed, 41 insertions(+), 8 deletions(-) diff --git a/pmoparadise/src/radio_paradise_stream_source.rs b/pmoparadise/src/radio_paradise_stream_source.rs index ddd051d8..c1db0154 100644 --- a/pmoparadise/src/radio_paradise_stream_source.rs +++ b/pmoparadise/src/radio_paradise_stream_source.rs @@ -19,7 +19,7 @@ use pmometadata::{MemoryTrackMetadata, TrackMetadata}; use std::{ collections::VecDeque, sync::Arc, - time::Duration, + time::{Duration, Instant}, }; use tokio::io::AsyncReadExt; use tokio::sync::{mpsc, RwLock}; @@ -80,14 +80,14 @@ impl RadioParadiseStreamSourceLogic { } /// Télécharge et décode un bloc FLAC - /// Retourne le timestamp du dernier chunk audio envoyé + /// Retourne (timestamp_final, instant_debut) pour permettre le timing correct async fn download_and_decode_block( &mut self, block: &Block, output: &[mpsc::Sender>], stop_token: &CancellationToken, order: &mut u64, - ) -> Result { + ) -> Result<(f64, Instant), AudioError> { // Télécharger le FLAC tracing::debug!("Sending HTTP GET request for block FLAC"); let response = self.client.client @@ -130,6 +130,10 @@ impl RadioParadiseStreamSourceLogic { let mut total_samples = 0u64; tracing::debug!("Block has {} songs", songs.len()); + // Noter l'instant de début AVANT d'envoyer TopZeroSync + // Ceci permet de synchroniser la durée réelle du bloc + let start_instant = Instant::now(); + // Envoyer TopZeroSync au début du bloc tracing::debug!("Sending TopZeroSync to {} outputs", output.len()); let top_zero = Arc::new(AudioSegment { @@ -174,9 +178,9 @@ impl RadioParadiseStreamSourceLogic { loop { // Vérifier stop_token if stop_token.is_cancelled() { - // Retourner le timestamp actuel si on est interrompu + // Retourner le timestamp actuel et start_instant si on est interrompu let current_timestamp = total_samples as f64 / sample_rate as f64; - return Ok(current_timestamp); + return Ok((current_timestamp, start_instant)); } // Remplir le buffer @@ -245,11 +249,11 @@ impl RadioParadiseStreamSourceLogic { total_samples += chunk_len; } - // Retourner le timestamp du dernier chunk (durée totale du bloc) + // Retourner le timestamp du dernier chunk (durée totale du bloc) et l'instant de début let final_timestamp = total_samples as f64 / sample_rate as f64; tracing::debug!("Block decode complete: {} samples, {:.2}s duration", total_samples, final_timestamp); - Ok(final_timestamp) + Ok((final_timestamp, start_instant)) } /// Envoie un segment à tous les enfants @@ -449,6 +453,7 @@ impl NodeLogic for RadioParadiseStreamSourceLogic { let mut order = 0u64; let mut last_timestamp = 0.0; + let mut last_start_instant: Option = None; loop { // Attendre un block ID (timeout court pour une radio) @@ -502,9 +507,10 @@ impl NodeLogic for RadioParadiseStreamSourceLogic { // Télécharger et décoder le bloc tracing::info!("Starting download and decode for block {}...", event_id); - let block_duration = self.download_and_decode_block(&block, &output, &stop_token, &mut order) + let (block_duration, start_instant) = self.download_and_decode_block(&block, &output, &stop_token, &mut order) .await?; last_timestamp = block_duration; + last_start_instant = Some(start_instant); tracing::info!("Finished download and decode for block {} (duration: {:.2}s)", event_id, block_duration); } @@ -517,6 +523,33 @@ impl NodeLogic for RadioParadiseStreamSourceLogic { .map_err(|_| AudioError::ChildDied)?; } + // IMPORTANT: Attendre que la durée réelle du bloc soit écoulée avant de fermer le channel + // Sinon, le TimerNode reçoit EOF et se termine avant d'avoir fini de diffuser tous les chunks + if let Some(start_instant) = last_start_instant { + let elapsed = start_instant.elapsed().as_secs_f64(); + if elapsed < last_timestamp { + let remaining = last_timestamp - elapsed; + tracing::info!( + "Waiting {:.2}s for block playback to complete (elapsed={:.2}s, duration={:.2}s)", + remaining, elapsed, last_timestamp + ); + + tokio::select! { + _ = tokio::time::sleep(Duration::from_secs_f64(remaining)) => { + tracing::debug!("Block playback duration complete"); + } + _ = stop_token.cancelled() => { + tracing::debug!("Cancelled while waiting for playback completion"); + } + } + } else { + tracing::debug!( + "Block already played in real-time (elapsed={:.2}s >= duration={:.2}s)", + elapsed, last_timestamp + ); + } + } + Ok(()) } }