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

push-pqqsxyupswry #21

Merged
eric merged 76 commits from push-pqqsxyupswry into main 2025-12-06 12:49:47 +01:00
Conversation 0 Commits 76 Files Changed 241 +148 -213
6 changed files with 148 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 3957e17f64 - Show all commits

240
pmoaudio-ext/src/sinks/flac_cache_sink.rs
View File

@@ -8,7 +8,6 @@ use pmoaudio::{
};
use pmoaudiocache::AudioTrackMetadataExt;
use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat};
use serde_json::{Number, Value};
use std::{
collections::VecDeque,
pin::Pin,
@@ -35,13 +34,6 @@ use tokio_util::sync::CancellationToken;
// FlacCacheSinkLogic - Logique métier pure
// ═══════════════════════════════════════════════════════════════════════════
/// Signal retourné par pump_segments indiquant pourquoi l'encodage s'est arrêté.
enum StopReason {
TrackBoundary(Arc<RwLock<dyn pmometadata::TrackMetadata>>),
EndOfStream,
ChannelClosed,
}
/// Logique pure d'encodage FLAC vers le cache
pub struct FlacCacheSinkLogic {
cache: Arc<pmoaudiocache::Cache>,
@@ -291,7 +283,64 @@ impl NodeLogic for FlacCacheSinkLogic {
};
if let Some(transform) = self.cache.transform_metadata(&pk).await {
persist_transform_streaminfo(&self.cache, &pk, &transform);
tracing::debug!(
"FlacCacheSink: Got transform metadata for pk {}: sr={:?}, bps={:?}, ch={:?}, ts={:?}",
pk,
transform.sample_rate,
transform.bits_per_sample,
transform.channels,
transform.total_samples
);
// Persister les métadonnées techniques via l'interface TrackMetadata
let track_meta = self.cache.track_metadata(&pk);
let mut meta = track_meta.write().await;
if let Some(sr) = transform.sample_rate {
if let Err(e) = meta.set_sample_rate(Some(sr)).await {
tracing::error!("FlacCacheSink: Failed to set sample_rate for pk {}: {:?}", pk, e);
} else {
tracing::debug!("FlacCacheSink: Set sample_rate={} for pk {}", sr, pk);
}
}
if let Some(bps) = transform.bits_per_sample {
if let Err(e) = meta.set_bits_per_sample(Some(bps)).await {
tracing::error!("FlacCacheSink: Failed to set bits_per_sample for pk {}: {:?}", pk, e);
} else {
tracing::debug!("FlacCacheSink: Set bits_per_sample={} for pk {}", bps, pk);
}
}
if let Some(ch) = transform.channels {
if let Err(e) = meta.set_channels(Some(ch)).await {
tracing::error!("FlacCacheSink: Failed to set channels for pk {}: {:?}", pk, e);
} else {
tracing::debug!("FlacCacheSink: Set channels={} for pk {}", ch, pk);
}
}
if let Some(ts) = transform.total_samples {
if let Err(e) = meta.set_total_samples(Some(ts)).await {
tracing::error!("FlacCacheSink: Failed to set total_samples for pk {}: {:?}", pk, e);
} else {
tracing::debug!("FlacCacheSink: Set total_samples={} for pk {}", ts, pk);
}
// Calculer la durée à partir de total_samples et sample_rate
if let Some(sr) = transform.sample_rate {
if sr > 0 {
use std::time::Duration;
let secs = (ts as f64 / sr as f64).round() as u64;
if let Err(e) = meta.set_duration(Some(Duration::from_secs(secs))).await {
tracing::error!("FlacCacheSink: Failed to set duration for pk {}: {:?}", pk, e);
} else {
tracing::debug!("FlacCacheSink: Set duration={} secs for pk {}", secs, pk);
}
}
}
}
drop(meta); // Libérer le lock explicitement
} else {
tracing::warn!("FlacCacheSink: No transform metadata available for pk {}", pk);
}
// Phase 2: Prebuffer terminé! Copier les métadonnées et pusher à la playlist
@@ -701,144 +750,6 @@ async fn wait_for_first_audio_chunk_with_metadata(
}
}
/// Draine tous les segments jusqu'au prochain TrackBoundary ou EndOfStream
///
/// Cette fonction est utilisée quand le fichier était déjà en cache et que
/// nous devons ignorer les segments restants pour rester synchronisé avec la source.
async fn drain_until_track_boundary(
rx: &mut mpsc::Receiver<Arc<AudioSegment>>,
stop_token: &CancellationToken,
) -> Result<StopReason, AudioError> {
loop {
let segment = tokio::select! {
result = rx.recv() => {
match result {
Some(seg) => seg,
None => {
return Ok(StopReason::ChannelClosed);
}
}
}
_ = stop_token.cancelled() => {
return Ok(StopReason::ChannelClosed);
}
};
match &segment.segment {
_AudioSegment::Chunk(_) => {
// Ignorer les chunks audio
continue;
}
_AudioSegment::Sync(marker) => match &**marker {
SyncMarker::TrackBoundary { metadata, .. } => {
return Ok(StopReason::TrackBoundary(metadata.clone()));
}
SyncMarker::EndOfStream => {
return Ok(StopReason::EndOfStream);
}
_ => {
// Ignorer les autres syncmarkers
continue;
}
},
}
}
}
/// Pompe les segments pour une seule track (s'arrête au TrackBoundary).
async fn pump_track_segments(
first_segment: Arc<AudioSegment>,
rx: &mut mpsc::Receiver<Arc<AudioSegment>>,
pcm_tx: mpsc::Sender<Vec<u8>>,
bits_per_sample: u8,
expected_rate: u32,
stop_token: &CancellationToken,
) -> Result<(u64, u64, f64, StopReason), AudioError> {
let mut chunks = 0u64;
let mut samples = 0u64;
let mut duration_sec = 0.0f64;
// Traiter le premier segment
if let Some(chunk) = first_segment.as_chunk() {
let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?;
if !pcm_bytes.is_empty() {
// Si le send échoue, c'est que le receiver est fermé
// (par exemple, le fichier était déjà en cache et add_from_reader a retourné immédiatement)
if pcm_tx.send(pcm_bytes).await.is_err() {
drop(pcm_tx);
return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed));
}
chunks += 1;
samples += chunk.len() as u64;
duration_sec += chunk.len() as f64 / expected_rate as f64;
}
}
// Boucle sur les segments suivants
loop {
let segment = tokio::select! {
result = rx.recv() => {
match result {
Some(seg) => seg,
None => {
drop(pcm_tx); // Fermer le channel PCM
return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed));
}
}
}
_ = stop_token.cancelled() => {
drop(pcm_tx); // Fermer le channel PCM
return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed));
}
};
match &segment.segment {
_AudioSegment::Chunk(chunk) => {
// Vérifier la cohérence du sample rate
if chunk.sample_rate() != expected_rate {
return Err(AudioError::ProcessingError(format!(
"FlacCacheSink: inconsistent sample rate ({} vs {})",
chunk.sample_rate(),
expected_rate
)));
}
let pcm_bytes = chunk_to_pcm_bytes(&chunk, bits_per_sample)?;
if pcm_bytes.is_empty() {
continue;
}
// Si le send échoue, c'est que le receiver est fermé
// (par exemple, le fichier était déjà en cache et add_from_reader a retourné immédiatement)
if pcm_tx.send(pcm_bytes).await.is_err() {
drop(pcm_tx);
return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed));
}
chunks += 1;
samples += chunk.len() as u64;
duration_sec += chunk.len() as f64 / expected_rate as f64;
}
_AudioSegment::Sync(marker) => match &**marker {
SyncMarker::TrackBoundary { metadata, .. } => {
drop(pcm_tx); // Fermer le channel PCM
return Ok((
chunks,
samples,
duration_sec,
StopReason::TrackBoundary(metadata.clone()),
));
}
SyncMarker::EndOfStream => {
drop(pcm_tx); // Fermer le channel PCM
return Ok((chunks, samples, duration_sec, StopReason::EndOfStream));
}
_ => {} // Ignorer les autres syncmarkers
},
}
}
}
/// Pompe les segments pour une seule track depuis un channel dédié.
///
/// Cette version permet d'avoir plusieurs pumps en parallèle (pour cache progressif),
@@ -915,41 +826,6 @@ async fn pump_track_segments_from_channel(
}
}
fn persist_transform_streaminfo(
cache: &pmoaudiocache::Cache,
pk: &str,
tmeta: &pmocache::download::TransformMetadata,
) {
if let Some(sr) = tmeta.sample_rate {
let _ = cache
.db
.set_a_metadata(pk, "sample_rate", Value::Number(Number::from(sr)));
}
if let Some(bps) = tmeta.bits_per_sample {
let _ = cache
.db
.set_a_metadata(pk, "bits_per_sample", Value::Number(Number::from(bps)));
}
if let Some(ch) = tmeta.channels {
let _ = cache
.db
.set_a_metadata(pk, "channels", Value::Number(Number::from(ch)));
}
if let Some(ts) = tmeta.total_samples {
let _ = cache
.db
.set_a_metadata(pk, "total_samples", Value::Number(Number::from(ts)));
if let Some(sr) = tmeta.sample_rate {
if sr > 0 {
let secs = (ts as f64 / sr as f64).round() as u64;
let _ = cache
.db
.set_a_metadata(pk, "duration_secs", Value::Number(Number::from(secs)));
}
}
}
}
/// Détermine la profondeur de bit d'un chunk audio
fn get_chunk_bit_depth(chunk: &AudioChunk) -> u8 {
match chunk {

23
pmoaudio-ext/src/sinks/streaming_sink_common.rs
View File

@@ -376,15 +376,6 @@ impl SharedSinkContext {
// Always pass the metadata handle to the encoder so Vorbis comments are emitted.
self.encoder_options.metadata = Some(metadata_lock.clone());
// In raw FLAC live streaming we must NOT advertise a total_samples value,
// otherwise players think the stream ends after the first track.
if !self.enable_total_samples {
self.pending_track_duration = None;
self.pending_total_samples = None;
self.encoder_options.total_samples = None;
return Ok(());
}
// Capture duration (if any) to set total_samples.
let duration_opt = {
let metadata = metadata_lock.read().await;
@@ -396,6 +387,20 @@ impl SharedSinkContext {
metadata.get_total_samples().await.ok().flatten()
};
debug!(
"Encoder metadata: from TrackBoundary - duration={:?}s, total_samples={:?}",
self.pending_track_duration.as_ref().map(|d| d.as_secs_f64()),
self.pending_total_samples
);
// In raw FLAC live streaming we must NOT advertise a total_samples value,
// otherwise players think the stream ends after the first track.
if !self.enable_total_samples {
self.encoder_options.total_samples = None;
debug!("Encoder metadata: total_samples disabled for this sink (enable_total_samples=false)");
return Ok(());
}
// Compute total_samples only when we know the sample rate.
self.refresh_total_samples_with_sample_rate();

46
pmoaudiocache/src/streaming.rs
View File

@@ -33,25 +33,33 @@ pub fn create_streaming_flac_transformer() -> StreamTransformer {
"transcode"
};
context
.set_metadata(TransformMetadata {
mode: Some(mode.to_string()),
input_codec: Some(codec_to_string(codec)),
details: Some(
json!({
"sample_rate": info.sample_rate,
"bits_per_sample": info.bits_per_sample,
"channels": info.channels,
"total_samples": info.total_samples,
})
.to_string(),
),
sample_rate: Some(info.sample_rate),
bits_per_sample: Some(info.bits_per_sample),
channels: Some(info.channels),
total_samples: info.total_samples,
})
.await;
let metadata = TransformMetadata {
mode: Some(mode.to_string()),
input_codec: Some(codec_to_string(codec)),
details: Some(
json!({
"sample_rate": info.sample_rate,
"bits_per_sample": info.bits_per_sample,
"channels": info.channels,
"total_samples": info.total_samples,
})
.to_string(),
),
sample_rate: Some(info.sample_rate),
bits_per_sample: Some(info.bits_per_sample),
channels: Some(info.channels),
total_samples: info.total_samples,
};
tracing::debug!(
"Transformer setting metadata: sr={:?}, bps={:?}, ch={:?}, ts={:?}",
metadata.sample_rate,
metadata.bits_per_sample,
metadata.channels,
metadata.total_samples
);
context.set_metadata(metadata).await;
let mut flac_stream = transcode.into_stream();
let mut buffer = vec![0u8; 64 * 1024];

35
pmocache/src/cache.rs
View File

@@ -189,6 +189,41 @@ impl<C: CacheConfig> Cache<C> {
self.db.set_origin_url(pk, url)?;
}
// Sauvegarder les métadonnées techniques du transformer
if let Some(transform) = download.transform_metadata().await {
tracing::debug!(
"Cache: Got transform metadata for pk {}: sr={:?}, bps={:?}, ch={:?}, ts={:?}",
pk,
transform.sample_rate,
transform.bits_per_sample,
transform.channels,
transform.total_samples
);
if let Some(sr) = transform.sample_rate {
self.db.set_a_metadata(pk, "sample_rate", serde_json::json!(sr))?;
}
if let Some(bps) = transform.bits_per_sample {
self.db.set_a_metadata(pk, "bits_per_sample", serde_json::json!(bps))?;
}
if let Some(ch) = transform.channels {
self.db.set_a_metadata(pk, "channels", serde_json::json!(ch))?;
}
if let Some(ts) = transform.total_samples {
self.db.set_a_metadata(pk, "total_samples", serde_json::json!(ts))?;
// Calculer la durée à partir de total_samples et sample_rate
if let Some(sr) = transform.sample_rate {
if sr > 0 {
let secs = (ts as f64 / sr as f64).round() as u64;
self.db.set_a_metadata(pk, "duration_secs", serde_json::json!(secs))?;
}
}
}
} else {
tracing::debug!("Cache: No transform metadata available for pk {}", pk);
}
if let Err(e) = self.enforce_limit().await {
tracing::warn!("Error enforcing cache limit: {}", e);
}

6
pmoflac/src/encoder.rs
View File

@@ -525,10 +525,16 @@ fn run_encoder(
"set_verify failed",
)?;
if let Some(total) = options.total_samples {
tracing::debug!(
"FLAC encoder: setting total_samples_estimate = {} before init",
total
);
ensure(
FLAC__stream_encoder_set_total_samples_estimate(encoder, total),
"set_total_samples_estimate failed",
)?;
} else {
tracing::warn!("FLAC encoder: total_samples is None, STREAMINFO will have total_samples=0");
}
if let Some(block_size) = options.block_size {
ensure(

11
pmoparadise/examples/single_channel_server.rs
View File

@@ -64,10 +64,15 @@ async fn main() -> anyhow::Result<()> {
let history_opts = history_builder.build_for_channel(&descriptor).await?;
let mut channel_config = ParadiseStreamChannelConfig::default();
channel_config.flac_options = StreamingSinkOptions::flac_defaults()
// Configuration commune pour FLAC et OGG
let common_options = StreamingSinkOptions::flac_defaults()
.with_default_artist(Some("Radio Paradise".to_string()))
.with_default_title(descriptor.display_name.to_string())
.with_only_default_metadata(true);
.with_default_title(descriptor.display_name.to_string());
channel_config.flac_options = common_options.clone();
channel_config.ogg_options = StreamingSinkOptions::ogg_defaults()
.with_default_artist(Some("Radio Paradise".to_string()))
.with_default_title(descriptor.display_name.to_string());
let channel = Arc::new(
ParadiseStreamChannel::new(