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

Round 3

Browse Source
This commit is contained in:
Eric Coissac
2025-11-17 03:03:06 +01:00
parent cbe197da9c
commit 9bc0c2544d
8 changed files with 180 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -79,7 +79,7 @@ use pmometadata::TrackMetadata;
use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
use tokio::sync::{mpsc, RwLock};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, trace, warn};
use tracing::{debug, error, info, trace, warn};
/// Default ICY metadata interval (bytes of audio between metadata blocks).
/// Standard value used by most streaming servers.
@@ -306,7 +306,9 @@ impl AsyncRead for FlacClientStream {
self.state = FlacStreamState::Streaming;
continue; // Now copy header to output buffer
} else {
// Header not yet captured or can't acquire lock, skip to streaming
// Header not yet captured - client will receive it via broadcast
// Skip directly to streaming to avoid blocking
debug!("FLAC header not yet available, client will receive it via broadcast");
self.state = FlacStreamState::Streaming;
}
}
@@ -483,7 +485,9 @@ impl AsyncRead for IcyClientStream {
self.state = FlacStreamState::Streaming;
continue; // Now copy header to output buffer
} else {
// Header not yet captured or can't acquire lock, skip to streaming
// Header not yet captured - client will receive it via broadcast
// Skip directly to streaming to avoid blocking
debug!("FLAC header not yet available, ICY client will receive it via broadcast");
self.state = FlacStreamState::Streaming;
}
}
@@ -605,6 +609,7 @@ struct StreamingFlacSinkLogic {
encoder_state: Option<EncoderState>,
sample_rate: Option<u32>,
broadcast_max_lead_time: f64,
first_chunk_timestamp_checked: bool,
}
impl StreamingFlacSinkLogic {
@@ -748,6 +753,18 @@ impl NodeLogic for StreamingFlacSinkLogic {
Some(seg) => {
match &seg.segment {
_AudioSegment::Chunk(chunk) => {
if !self.first_chunk_timestamp_checked {
self.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.sample_rate.is_none() {
let sample_rate = chunk.sample_rate();
@@ -949,6 +966,8 @@ async fn broadcast_flac_stream(
// ║ Cela crée la backpressure vers TimerBufferNode tout en ║
// ║ permettant de dropper les chunks vraiment périmés. ║
// ╚═══════════════════════════════════════════════════════════════╝
// Calculer le timestamp de cette FLAC frame
let frame_start_samples = encoded_samples;
encoded_samples = encoded_samples.saturating_add(total_samples);
let audio_timestamp = frame_start_samples as f64 / sample_rate_f64;
@@ -1012,9 +1031,9 @@ async fn broadcast_flac_stream(
if !header_captured && bytes.len() >= 4 && &bytes[0..4] == b"fLaC" {
*header_cache.write().await = Some(bytes.clone());
header_captured = true;
trace!("FLAC header captured ({} bytes), not broadcasting", bytes.len());
// Skip broadcasting the header: clients prepend it locally on subscribe
continue;
trace!("FLAC header captured ({} bytes), will also broadcast it", bytes.len());
// Also broadcast the header so early-connecting clients receive it
// Later-connecting clients will get it from the cache
}
let num_receivers = broadcast_tx.receiver_count();
@@ -1143,6 +1162,7 @@ impl StreamingFlacSink {
encoder_state: None,
sample_rate: None,
broadcast_max_lead_time: broadcast_max_lead_time.max(0.0),
first_chunk_timestamp_checked: false,
};
let sink = Self {

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

@@ -17,6 +17,9 @@ use std::{
use tokio::sync::Notify;
use tracing::{trace, info, warn};
/// Tolérance pour détecter un timestamp à zéro (TopZero).
const TOP_ZERO_EPSILON: f64 = 1e-9;
/// Paquet diffusé contenant la charge utile + méta timing.
#[derive(Clone)]
pub struct TimedPacket<T> {
@@ -102,9 +105,7 @@ impl<T> State<T> {
}
}
fn purge_expired(&mut self) -> bool {
let now = Instant::now();
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) {
return false;
@@ -267,25 +268,27 @@ impl<T> Sender<T> {
return Err(SendError(payload.expect("payload already consumed")));
}
// Détecter si c'est un TopZero
let is_top_zero = audio_timestamp == 0.0;
// Capturer le temps UNE SEULE FOIS pour cohérence temporelle
let now = Instant::now();
// Détecter si c'est un TopZero (avec tolérance pour éviter erreurs d'arrondi)
let is_top_zero = audio_timestamp.abs() < TOP_ZERO_EPSILON;
// Gérer l'initialisation
if !state.initialized {
if !is_top_zero {
warn!(
"TimedBroadcast: First packet should have timestamp=0.0, got {:.3}s",
"TimedBroadcast: First packet has non-zero timestamp {:.3}s, treating as epoch start anyway",
audio_timestamp
);
}
let now = Instant::now();
// Initialiser TOUJOURS, quel que soit le timestamp du premier paquet
state.epoch_start = now;
state.epoch = 0;
state.initialized = true;
info!("TimedBroadcast: initialized (epoch=0)");
info!("TimedBroadcast: initialized (epoch=0, ts={:.3}s)", audio_timestamp);
} else if is_top_zero {
// TopZero = nouveau segment, toujours valide après l'initialisation
let now = Instant::now();
// Continuité temporelle : nouveau segment commence après le précédent
state.epoch_start = state.last_segment_end.unwrap_or(now);
state.epoch = state.epoch.wrapping_add(1);
@@ -296,26 +299,34 @@ impl<T> Sender<T> {
);
}
// 1. Calculer l'expiration du paquet actuel
let expires_at = state.epoch_start + Duration::from_secs_f64(audio_timestamp + segment_duration);
// 2. Vérifier que le paquet n'est pas déjà expiré
let now = Instant::now();
if expires_at <= now {
warn!(
"TimedBroadcast: packet already expired (ts={:.3}s, delta={}ms)",
audio_timestamp,
now.duration_since(expires_at).as_millis()
);
// On peut décider de l'ignorer ou de continuer
// Pour l'instant on continue pour ne pas bloquer le flux
// 1. Purger d'abord les paquets expirés et consommés pour libérer l'espace
// (skip pour le tout premier paquet)
if state.buffer.len() > 0 {
let consumed = state.prune_consumed();
let expired = state.purge_expired(now);
if consumed || expired {
self.inner.space_notify.notify_waiters();
}
}
// 3. Purger les paquets expirés et consommés
let consumed = state.prune_consumed();
let expired = state.purge_expired();
if consumed || expired {
self.inner.space_notify.notify_waiters();
// 2. Calculer l'expiration du paquet actuel
let expires_at = state.epoch_start + Duration::from_secs_f64(audio_timestamp + segment_duration);
// 3. Rejeter le paquet s'il est déjà expiré
// SAUF pour le premier paquet (initialisation) ou les paquets TopZero (nouveaux segments)
let is_first_packet = state.next_seq == 0;
if !is_first_packet && !is_top_zero && expires_at <= now {
// Tolérer une petite marge pour les latences d'initialisation
let grace_period = Duration::from_millis(50);
if now > expires_at + grace_period {
warn!(
"TimedBroadcast: rejecting already expired packet (ts={:.3}s, epoch={}, delta={}ms)",
audio_timestamp,
state.epoch,
now.duration_since(expires_at).as_millis()
);
return Err(SendError(payload.expect("payload already consumed")));
}
}
// 4. Vérifier la capacité et insérer
@@ -330,7 +341,8 @@ impl<T> Sender<T> {
state.next_seq += 1;
state.buffer.push_back(entry);
// 5. Stocker la fin du segment SEULEMENT pour les paquets non-TopZero
// 5. Mettre à jour la fin du segment SEULEMENT pour les paquets non-TopZero
// (pour que le prochain segment commence à la fin du dernier paquet de données)
if !is_top_zero {
state.last_segment_end = Some(expires_at);
}
@@ -429,7 +441,8 @@ where
return Err(TryRecvError::Closed);
}
if state.purge_expired() {
let now = Instant::now();
if state.purge_expired(now) {
self.inner.space_notify.notify_waiters();
}