diff --git a/.DS_Store b/.DS_Store index 5f1c4e87..009731bc 100644 Binary files a/.DS_Store and b/.DS_Store differ diff --git a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs index b696afa8..bb354b7f 100644 --- a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs @@ -228,6 +228,9 @@ impl NodeLogic for StreamingFlacSinkLogic { self.ctx.sample_rate = Some(sample_rate); debug!("Detected sample rate: {} Hz", sample_rate); + // If a duration is already known for this track, fill total_samples now. + self.ctx.refresh_total_samples_with_sample_rate(); + // Initialize the FLAC encoder now (first track starts at 0.0) self.ctx .initialize_encoder( @@ -306,6 +309,15 @@ impl NodeLogic for StreamingFlacSinkLogic { _AudioSegment::Sync(marker) => { match marker.as_ref() { SyncMarker::TrackBoundary { metadata } => { + // Prepare encoder options (metadata + duration) for the upcoming track. + if let Err(e) = + self.ctx.prepare_encoder_options_for_track(metadata).await + { + error!("Failed to prepare encoder options for new track: {}", e); + } + + debug!("StreamingFlacSink: SyncMarker::TrackBoundary {:?}",metadata.read().await.get_duration().await); + // Only restart encoder if it's already initialized (not the first track) if self.ctx.sample_rate.is_some() && self.ctx.encoder_state.is_some() { // Restart encoder to emit new header and reset timestamps @@ -464,6 +476,7 @@ impl StreamingFlacSink { first_chunk_timestamp_checked: false, timestamp_offset_sec: 0.0, current_timestamp: Arc::new(RwLock::new(0.0)), + pending_track_duration: None, }, }; diff --git a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs index 845dc981..533b4067 100644 --- a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs @@ -196,6 +196,9 @@ impl NodeLogic for StreamingOggFlacSinkLogic { self.ctx.sample_rate = Some(sample_rate); debug!("Detected sample rate: {} Hz", sample_rate); + // Populate total_samples if we already know the track duration. + self.ctx.refresh_total_samples_with_sample_rate(); + // Initialize the FLAC encoder now (first track starts at 0.0) self.ctx .initialize_encoder( @@ -263,6 +266,13 @@ impl NodeLogic for StreamingOggFlacSinkLogic { _AudioSegment::Sync(marker) => { match marker.as_ref() { SyncMarker::TrackBoundary { metadata } => { + // Inject per-track metadata and duration into the next FLAC header. + if let Err(e) = + self.ctx.prepare_encoder_options_for_track(metadata).await + { + error!("Failed to prepare encoder options for new track: {}", e); + } + // Only restart encoder if it's already initialized (not the first track) if self.ctx.sample_rate.is_some() && self.ctx.encoder_state.is_some() { // Restart encoder to emit new OGG stream header and reset timestamps @@ -421,6 +431,7 @@ impl StreamingOggFlacSink { first_chunk_timestamp_checked: false, timestamp_offset_sec: 0.0, current_timestamp: Arc::new(RwLock::new(0.0)), + pending_track_duration: None, }, }; @@ -515,8 +526,9 @@ async fn broadcast_ogg_flac_stream( let bos_page = ogg_writer.create_page(&ogg_flac_id, true, false, false); let bos_bytes = Bytes::from(bos_page); - // Step 3: Create Vorbis Comment page (empty for now, metadata comes from /metadata endpoint) - let vorbis_comment = create_empty_vorbis_comment(); + // Step 3: Create Vorbis Comment page (reuse FLAC metadata blocks when available) + let vorbis_comment = extract_comment_packet_from_flac_header(&flac_header) + .unwrap_or_else(create_empty_vorbis_comment); let comment_page = ogg_writer.create_page(&vorbis_comment, false, false, false); let comment_bytes = Bytes::from(comment_page); @@ -857,6 +869,27 @@ fn create_ogg_flac_identification(flac_header: &[u8]) -> Result, AudioEr Ok(packet) } +/// Extract the concatenated FLAC metadata blocks after STREAMINFO to use as the OGG comment packet. +/// Returns `None` if the FLAC header only contains STREAMINFO. +fn extract_comment_packet_from_flac_header(flac_header: &[u8]) -> Option> { + if flac_header.len() < 8 { + return None; + } + + // STREAMINFO block length is stored in bytes 5-7 (after type byte at 4) + let block_length = + u32::from_be_bytes([0, flac_header[5], flac_header[6], flac_header[7]]) as usize; + let streaminfo_total = 4 + block_length; // block header + data + + // Skip "fLaC" + STREAMINFO block. + let offset = 4 + streaminfo_total; + if flac_header.len() <= offset { + return None; + } + + Some(flac_header[offset..].to_vec()) +} + /// Create empty Vorbis Comment block as a proper FLAC metadata block fn create_empty_vorbis_comment() -> Vec { let mut vorbis_data = Vec::new(); diff --git a/pmoaudio-ext/src/sinks/streaming_sink_common.rs b/pmoaudio-ext/src/sinks/streaming_sink_common.rs index 1d6199cd..dc142126 100644 --- a/pmoaudio-ext/src/sinks/streaming_sink_common.rs +++ b/pmoaudio-ext/src/sinks/streaming_sink_common.rs @@ -213,6 +213,7 @@ pub struct SharedSinkContext { pub first_chunk_timestamp_checked: bool, pub timestamp_offset_sec: f64, pub current_timestamp: Arc>, + pub pending_track_duration: Option, } impl SharedSinkContext { @@ -299,6 +300,50 @@ impl SharedSinkContext { Ok(()) } + /// Prepare encoder options for a new track so the next FLAC header embeds up-to-date metadata + /// and duration (total_samples) when available. + pub async fn prepare_encoder_options_for_track( + &mut self, + metadata_lock: &Arc>, + ) -> Result<(), AudioError> { + debug!("Encoder metadata: preparing metadata"); + // Always pass the metadata handle to the encoder so Vorbis comments are emitted. + self.encoder_options.metadata = Some(metadata_lock.clone()); + + // Capture duration (if any) to set total_samples. + let duration_opt = { + let metadata = metadata_lock.read().await; + metadata.get_duration().await.ok().flatten() + }; + self.pending_track_duration = duration_opt; + + // Compute total_samples only when we know the sample rate. + if let (Some(duration), Some(sr)) = (self.pending_track_duration, self.sample_rate) { + debug!("Encoder metadata: duration: {:3}s - rate: {}Hz", Duration::as_secs_f64(&duration),sr); + let samples = (duration.as_secs_f64() * sr as f64).round() as u64; + self.encoder_options.total_samples = Some(samples); + } else { + if self.pending_track_duration.is_none() { + debug!("Encoder metadata: duration: None"); + } + if self.sample_rate.is_none() { + debug!("Encoder metadata: sample rate: None"); + } + // Avoid leaking the previous track's length. + self.encoder_options.total_samples = None; + } + + Ok(()) + } + + /// Refresh total_samples when the sample rate is learned after metadata was already set. + pub fn refresh_total_samples_with_sample_rate(&mut self) { + if let (Some(duration), Some(sr)) = (self.pending_track_duration, self.sample_rate) { + let samples = (duration.as_secs_f64() * sr as f64).round() as u64; + self.encoder_options.total_samples = Some(samples); + } + } + pub async fn restart_encoder_for_new_track( &mut self, broadcaster: F, diff --git a/pmoaudio-ext/src/sinks/timed_broadcast.rs b/pmoaudio-ext/src/sinks/timed_broadcast.rs index ff4dd273..65662c95 100644 --- a/pmoaudio-ext/src/sinks/timed_broadcast.rs +++ b/pmoaudio-ext/src/sinks/timed_broadcast.rs @@ -5,13 +5,17 @@ //! - Propagation d’un compteur `epoch` incrémenté sur chaque TopZeroSync. use std::{ - collections::VecDeque, fmt, string, sync::{ - Arc, Mutex, Weak, atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering} - }, time::{Duration, Instant} + collections::VecDeque, + fmt, string, + sync::{ + atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, + Arc, Mutex, Weak, + }, + time::{Duration, Instant}, }; use tokio::sync::Notify; -use tracing::{info, debug, trace, warn}; +use tracing::{debug, info, trace, warn}; /// Tolérance pour détecter un timestamp à zéro (TopZero). const TOP_ZERO_EPSILON: f64 = 1e-9; @@ -93,9 +97,9 @@ struct State { } impl State { - fn new(name: &str,capacity: usize, epoch_start: Instant) -> Self { + fn new(name: &str, capacity: usize, epoch_start: Instant) -> Self { Self { - name: name.to_string() , + name: name.to_string(), buffer: VecDeque::with_capacity(capacity), head_seq: 0, next_seq: 0, @@ -120,7 +124,13 @@ impl State { while let Some(entry) = self.buffer.front() { if entry.expires_at <= now { let delta = now - entry.expires_at; - debug!("TimedBroadcast[{}]: purging expired packet (@{} epoch={},delta={})", self.name,entry.seq, entry.epoch, delta.as_millis()); + trace!( + "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; @@ -130,7 +140,8 @@ impl State { } if purged > 0 { trace!( - "TimedBroadcast: purged {} expired packet(s) (head_seq={})", + "TimedBroadcast[{}]: purged {} expired packet(s) (head_seq={})", + self.name, purged, self.head_seq ); @@ -167,8 +178,11 @@ impl State { for _ in 0..removable { 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); + let entry = oentry.unwrap(); + trace!( + "TimedBroadcast[{}]: pruning played packet (@{} epoch={})", + self.name, entry.seq, entry.epoch + ); self.head_seq += 1; } @@ -188,9 +202,9 @@ struct Inner { } impl Inner { - fn new(name: &str,capacity: usize) -> Self { + fn new(name: &str, capacity: usize) -> Self { Self { - state: Mutex::new(State::new(name,capacity, Instant::now())), + state: Mutex::new(State::new(name, capacity, Instant::now())), data_notify: Notify::new(), space_notify: Notify::new(), capacity, @@ -214,7 +228,7 @@ impl Inner { /// Créé un channel broadcast temporisé. pub fn channel(name: &str, capacity: usize) -> (Sender, Receiver) { assert!(capacity > 0, "capacity must be > 0"); - let inner = Arc::new(Inner::new(name,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 @@ -297,44 +311,62 @@ impl Sender { } // 2. Vérifier si un slot est disponible et insérer - let is_top_zero = audio_timestamp.abs() < TOP_ZERO_EPSILON; + let is_top_zero = + audio_timestamp.abs() < TOP_ZERO_EPSILON + && segment_duration >= TOP_ZERO_EPSILON; + let is_zero_header = + audio_timestamp.abs() < TOP_ZERO_EPSILON + && segment_duration < TOP_ZERO_EPSILON; if state.buffer.len() < self.inner.capacity { - if !state.initialized { - if !is_top_zero { + if !state.initialized { + if !is_top_zero && segment_duration >= TOP_ZERO_EPSILON { warn!( - "TimedBroadcast: First packet has non-zero timestamp {:.3}s, treating as epoch start anyway", - audio_timestamp + "TimedBroadcast[{}]: First packet has non-zero timestamp {:.1}ms - Duration={:.1}ms, treating as epoch start anyway", + state.name, + audio_timestamp*1000.0, + segment_duration*1000.0 ); } state.epoch_start = now; state.epoch = 0; state.initialized = true; info!( - "TimedBroadcast: initialized (epoch=0, ts={:.3}s)", - audio_timestamp + "TimedBroadcast[{}]: initialized (epoch=0, ts={:.1}ms - Duration={:.1}ms)", + state.name, + audio_timestamp*1000.0, + segment_duration*1000.0 ); - } else if is_top_zero { + } else if is_top_zero || is_zero_header { // 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); + // expired packets when there's a long gap between tracks. Also trigger + // on zero-duration headers (OGG BOS/comment) so the epoch is reset + // before testing expiration. + state.epoch_start = state + .last_segment_end + .map(|end| end.max(now)) + .unwrap_or(now); // state.epoch_start = now; state.epoch = state.epoch.wrapping_add(1); info!( - "TimedBroadcast: new epoch={} (continuous={})", + "TimedBroadcast[{}]: new epoch={} (continuous={} - Duration={}ms)", + state.name, state.epoch, - state.last_segment_end.is_some() + state.last_segment_end.is_some(), + segment_duration*1000.0 ); } - let expires_at = - state.epoch_start + Duration::from_secs_f64(audio_timestamp + segment_duration); + let expires_at = state.epoch_start + + Duration::from_secs_f64(audio_timestamp + + segment_duration); let is_first_packet = state.next_seq == 0; - if !is_first_packet && !is_top_zero && expires_at <= now { + if !is_first_packet && !is_top_zero && !is_zero_header && expires_at <= now { 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)", + "TimedBroadcast[{}]: rejecting already expired packet (ts={:.3}s, epoch={}, delta={}ms)", + state.name, audio_timestamp, state.epoch, now.duration_since(expires_at).as_millis() @@ -355,10 +387,13 @@ impl Sender { state.next_seq += 1; state.buffer.push_back(entry); - // 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); + // 5. Only advance segment end for real audio (skip 0-duration metadata) + if segment_duration >= TOP_ZERO_EPSILON { + let new_end = expires_at; + state.last_segment_end = Some(match state.last_segment_end.take() { + Some(prev) => prev.max(new_end), + None => new_end, + }); } let receivers = self.inner.receiver_count.load(Ordering::SeqCst); diff --git a/pmoaudiocache/src/track_metadata.rs b/pmoaudiocache/src/track_metadata.rs index af9f2e92..47ae25ae 100644 --- a/pmoaudiocache/src/track_metadata.rs +++ b/pmoaudiocache/src/track_metadata.rs @@ -186,6 +186,45 @@ impl TrackMetadata for AudioCacheTrackMetadata { Ok(Some(())) } + async fn get_sample_rate(&self) -> MetadataResult { + Ok(match self.read_number("sample_rate")? { + Some(n) => n.as_i64().and_then(|v| u32::try_from(v).ok()), + None => None, + }) + } + + async fn set_sample_rate(&mut self, value: Option) -> MetadataResult<()> { + self.write_number("sample_rate", value.map(|v| v as i64))?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_total_samples(&self) -> MetadataResult { + Ok(match self.read_number("total_samples")? { + Some(n) => n.as_i64().and_then(|v| u64::try_from(v).ok()), + None => None, + }) + } + + async fn set_total_samples(&mut self, value: Option) -> MetadataResult<()> { + self.write_u64("total_samples", value)?; + let _ = self.touch().await?; + Ok(Some(())) + } + + async fn get_bits_per_sample(&self) -> MetadataResult { + Ok(match self.read_number("bits_per_sample")? { + Some(n) => n.as_i64().and_then(|v| u8::try_from(v).ok()), + None => None, + }) + } + + async fn set_bits_per_sample(&mut self, value: Option) -> MetadataResult<()> { + self.write_number("bits_per_sample", value.map(|v| v as i64))?; + let _ = self.touch().await?; + Ok(Some(())) + } + async fn get_track_id(&self) -> MetadataResult { Ok(self.read_string("track_id")?) } @@ -286,6 +325,9 @@ mod tests { meta.set_duration(Some(Duration::from_secs(90))) .await .unwrap(); + meta.set_sample_rate(Some(44100)).await.unwrap(); + meta.set_total_samples(Some(9_999_999)).await.unwrap(); + meta.set_bits_per_sample(Some(16)).await.unwrap(); meta.set_track_id(Some("trk".into())).await.unwrap(); meta.set_channel_id(Some("chn".into())).await.unwrap(); meta.set_event(Some("event".into())).await.unwrap(); @@ -306,6 +348,9 @@ mod tests { meta.get_duration().await.unwrap(), Some(Duration::from_secs(90)) ); + assert_eq!(meta.get_sample_rate().await.unwrap(), Some(44100)); + assert_eq!(meta.get_total_samples().await.unwrap(), Some(9_999_999)); + assert_eq!(meta.get_bits_per_sample().await.unwrap(), Some(16)); assert_eq!(meta.get_track_id().await.unwrap(), Some("trk".into())); assert_eq!(meta.get_channel_id().await.unwrap(), Some("chn".into())); assert_eq!(meta.get_event().await.unwrap(), Some("event".into())); diff --git a/pmometadata/src/lib.rs b/pmometadata/src/lib.rs index 4c9c6876..eb42dd20 100755 --- a/pmometadata/src/lib.rs +++ b/pmometadata/src/lib.rs @@ -194,6 +194,30 @@ pub trait TrackMetadata: Send + Sync { Err(MetadataError::NotImplemented) } + async fn get_sample_rate(&self) -> MetadataResult { + Err(MetadataError::NotImplemented) + } + + async fn set_sample_rate(&mut self, _value: Option) -> MetadataResult<()> { + Err(MetadataError::NotImplemented) + } + + async fn get_total_samples(&self) -> MetadataResult { + Err(MetadataError::NotImplemented) + } + + async fn set_total_samples(&mut self, _value: Option) -> MetadataResult<()> { + Err(MetadataError::NotImplemented) + } + + async fn get_bits_per_sample(&self) -> MetadataResult { + Err(MetadataError::NotImplemented) + } + + async fn set_bits_per_sample(&mut self, _value: Option) -> MetadataResult<()> { + Err(MetadataError::NotImplemented) + } + async fn get_track_id(&self) -> MetadataResult { Err(MetadataError::NotImplemented) } @@ -257,6 +281,15 @@ pub trait TrackMetadata: Send + Sync { async fn touch(&mut self) -> MetadataResult<()> { Err(MetadataError::NotImplemented) } + + async fn set_sample_rate(&mut self, _value: Option) -> MetadataResult<()> { + Err(MetadataError::NotImplemented) + } + + async fn get_sample_rate(self) -> MetadataResult { + Err(MetadataError::NotImplemented) + } + } /// Copies all available metadata from one implementation to another. @@ -318,8 +351,8 @@ where let src_guard = src.read().await; copy_metadata!( - src_guard, dest, title, artist, album, year, duration, track_id, channel_id, event, rating, - cover_url, cover_pk, extra + src_guard, dest, title, artist, album, year, duration, sample_rate, total_samples, + bits_per_sample, track_id, channel_id, event, rating, cover_url, cover_pk, extra ); // Try to update the timestamp, but ignore transient errors @@ -338,6 +371,9 @@ pub struct MemoryTrackMetadata { album: Option, year: Option, duration: Option, + sample_rate: Option, + total_samples: Option, + bits_per_sample: Option, track_id: Option, channel_id: Option, event: Option, @@ -406,6 +442,36 @@ impl TrackMetadata for MemoryTrackMetadata { Ok(Some(())) } + async fn get_sample_rate(&self) -> MetadataResult { + Ok(self.sample_rate) + } + + async fn set_sample_rate(&mut self, value: Option) -> MetadataResult<()> { + self.sample_rate = value; + self.touch().await?; + Ok(Some(())) + } + + async fn get_total_samples(&self) -> MetadataResult { + Ok(self.total_samples) + } + + async fn set_total_samples(&mut self, value: Option) -> MetadataResult<()> { + self.total_samples = value; + self.touch().await?; + Ok(Some(())) + } + + async fn get_bits_per_sample(&self) -> MetadataResult { + Ok(self.bits_per_sample) + } + + async fn set_bits_per_sample(&mut self, value: Option) -> MetadataResult<()> { + self.bits_per_sample = value; + self.touch().await?; + Ok(Some(())) + } + async fn get_track_id(&self) -> MetadataResult { Ok(self.track_id.clone()) } @@ -484,6 +550,8 @@ impl TrackMetadata for MemoryTrackMetadata { self.updated_at = Some(SystemTime::now()); Ok(Some(())) } + + } #[cfg(test)] @@ -499,6 +567,9 @@ mod tests { assert_eq!(metadata.get_album().await.unwrap(), None); assert_eq!(metadata.get_year().await.unwrap(), None); assert_eq!(metadata.get_duration().await.unwrap(), None); + assert_eq!(metadata.get_sample_rate().await.unwrap(), None); + assert_eq!(metadata.get_total_samples().await.unwrap(), None); + assert_eq!(metadata.get_bits_per_sample().await.unwrap(), None); assert_eq!(metadata.get_updated_at().await.unwrap(), None); }