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

Refactor WebRenderer to use server-side streaming with OGG-FLAC sink

Browse Source 
This commit refactors the WebRenderer to use a server-side streaming architecture with OGG-FLAC sink instead of the previous WebSocket-based approach. The changes include:

- Replaced WebSocket communication with HTTP streaming using DirectOggFlacSink
- Implemented a new pipeline architecture with dedicated handlers for UPnP commands
- Added new modules for registration, registry, and streaming
- Updated the renderer to work with a pipeline control system
- Removed old WebSocket session management
- Added support for HTTP streaming with gapless playback
- Updated dependencies and features for the new architecture

The WebRenderer now acts as a MediaRenderer UPnP device that serves audio streams via HTTP endpoints, with commands relayed to the audio pipeline through a new control system.
This commit is contained in:
Eric Coissac
2026-02-26 19:58:37 +01:00
parent 4b3e8a961c
commit 306e691c61
32 changed files with 2797 additions and 1653 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
pmoaudio-ext/src/lib.rs
View File

@@ -28,7 +28,7 @@ pub mod sinks;
#[cfg(any(feature = "cache-sink", feature = "http-stream"))]
pub mod nodes;
#[cfg(feature = "playlist")]
#[cfg(any(feature = "playlist", feature = "http-stream"))]
pub mod sources;
// Re-exports pour faciliter l'utilisation
@@ -39,4 +39,7 @@ pub use sinks::*;
pub use nodes::*;
#[cfg(feature = "playlist")]
pub use sources::*;
pub use sources::PlaylistSource;
#[cfg(feature = "http-stream")]
pub use sources::UriSource;

44
pmoaudio-ext/src/sinks/broadcast_pacing.rs
View File

@@ -5,7 +5,7 @@
//! - Drops frames that are late (audio_ts < elapsed)
//! - Paces broadcast to match audio playback rate
use std::time::Instant;
use std::time::{Duration, Instant};
use tracing::trace;
/// Error returned when a frame should be skipped (too late)
@@ -13,7 +13,6 @@ use tracing::trace;
pub struct SkipFrame;
/// Manages broadcast pacing with TopZeroSync detection
#[allow(dead_code)]
pub struct BroadcastPacer {
/// Start time (reset on TopZeroSync)
start_time: Instant,
@@ -21,8 +20,6 @@ pub struct BroadcastPacer {
max_lead_time: f64,
/// Label for logging (e.g., "FLAC" or "OGG")
label: String,
/// Pending reset flag - will reset timer on next chunk
pending_reset: bool,
}
impl BroadcastPacer {
@@ -37,24 +34,37 @@ impl BroadcastPacer {
start_time: Instant::now(),
max_lead_time: max_lead_time.max(0.0),
label: label.into(),
pending_reset: false,
}
}
/// Check timing and apply pacing - NO-OP VERSION
/// Reset the pacer clock (call when audio timestamp resets to 0).
pub fn reset(&mut self) {
self.start_time = Instant::now();
trace!("{} broadcaster: pacer reset", self.label);
}
/// Check timing and apply pacing.
///
/// Pacing is now handled entirely by the expiration-based system in
/// TimedBroadcast. This method is kept for backward compatibility
/// but always returns Ok(()).
///
/// # Returns
///
/// - Always returns `Ok(())`
/// If the audio is ahead of real time by more than `max_lead_time`, sleeps
/// until the lead is within bounds. Returns `Err(SkipFrame)` if the chunk
/// is already late (audio_ts < elapsed - 1s grace).
pub async fn check_and_pace(&mut self, audio_timestamp: f64) -> Result<(), SkipFrame> {
trace!(
"{} broadcaster: check_and_pace called with audio_ts={:.3}s (no-op - pacing handled by TimedBroadcast)",
self.label, audio_timestamp
);
if self.max_lead_time <= 0.0 {
return Ok(());
}
let elapsed = self.start_time.elapsed().as_secs_f64();
let lead = audio_timestamp - elapsed;
if lead > self.max_lead_time {
let sleep_secs = lead - self.max_lead_time;
trace!(
"{} broadcaster: audio ahead by {:.3}s, sleeping {:.3}s",
self.label, lead, sleep_secs
);
tokio::time::sleep(Duration::from_secs_f64(sleep_secs)).await;
}
Ok(())
}
}

367
pmoaudio-ext/src/sinks/direct_flac_sink.rs Normal file
View File

@@ -0,0 +1,367 @@
//! DirectFlacSink — nœud puits FLAC pour un seul client HTTP.
//!
//! Encode l'audio en FLAC (format fixe : 96 kHz / stéréo / 24 bits).
//!
//! # Cycle de vie
//!
//! - **Play** : le navigateur appelle `GET /stream`. `connect()` crée un nouveau
//! canal PCM + pipe duplex + encodeur FLAC, installe le sender dans le sink,
//! et notifie le sink via `client_notify`. Le flux reste ouvert : les morceaux
//! s'enchaînent en gapless.
//! - **Stop** : le navigateur ferme la connexion. Le pipe se rompt, l'encodeur
//! s'arrête. Le sink voit `pcm_tx.send()` échouer, passe le sender à `None`,
//! et **bloque** sur `client_notify` jusqu'au prochain Play.
//! Cela bloque la source et préserve la backpressure.
//! - **Play suivant** : `connect()` → nouveau pipe → `client_notify.notify_one()`
//! → le sink se débloque et reprend la consommation des segments.
//!
//! # Architecture
//!
//! ```text
//! AudioSegment I24 @ 96 kHz
//! ↓ NodeLogic::process() [bloque si pas de client]
//! chunk_to_pcm_bytes() → PCM 24-bit LE
//! ↓ Arc<Mutex<Option<mpsc::Sender<PcmChunk>>>>
//! ByteStreamReader (AsyncRead)
//! ↓ encode_flac_stream()
//! ↓ tokio::io::copy()
//! ↓ tokio::io::duplex pipe (256 KB)
//! ↓ DirectFlacStream (AsyncRead) → Body HTTP
//! ```
use std::io;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use async_trait::async_trait;
use pmoaudio::{
pipeline::{AudioPipelineNode, Node, NodeLogic, PipelineHandle, StopReason},
AudioError, AudioSegment, SyncMarker, TypeRequirement, TypedAudioNode, _AudioSegment,
};
use pmoflac::{EncoderOptions, PcmFormat};
use tokio::io::{AsyncRead, ReadBuf};
use tokio::sync::{mpsc, watch, Mutex};
use tokio_util::sync::CancellationToken;
use tracing::debug;
use crate::sinks::byte_stream_reader::{ByteStreamReader, PcmChunk};
use crate::sinks::chunk_to_pcm::chunk_to_pcm_bytes;
/// Format de sortie fixe du sink.
pub const DIRECT_FLAC_SAMPLE_RATE: u32 = 96_000;
pub const DIRECT_FLAC_CHANNELS: u8 = 2;
pub const DIRECT_FLAC_BITS_PER_SAMPLE: u8 = 24;
/// Capacité du pipe duplex (~256 KB ≈ 0.35s à 96 kHz/24 bits/stéréo).
const PIPE_CAPACITY: usize = 256 * 1024;
// ─── Shared state ─────────────────────────────────────────────────────────────
type SharedPcmTx = Arc<Mutex<Option<mpsc::Sender<PcmChunk>>>>;
// ─── Handle public ────────────────────────────────────────────────────────────
/// Handle vers le sink, cloneable, reconnectable à chaque Play.
#[derive(Clone)]
pub struct DirectFlacHandle {
pcm_tx: SharedPcmTx,
/// Compteur de connexions : incrémenté à chaque connect().
/// Utiliser un watch channel pour éviter les notifications perdues (vs Notify).
client_connect_tx: Arc<watch::Sender<u64>>,
/// Notifie le sink interne qu'un client vient de se connecter (edge-triggered,
/// usage interne uniquement — le sink tourne dans le même contexte que connect()).
client_notify_internal: Arc<tokio::sync::Notify>,
/// Signale que le premier byte FLAC a été lu par le client HTTP.
/// `false` au démarrage / après connect(), `true` dès le premier poll_read non-vide.
first_byte_tx: Arc<watch::Sender<bool>>,
encoder_options: EncoderOptions,
}
impl DirectFlacHandle {
/// Crée un nouveau pipe + encodeur FLAC et retourne le flux côté lecture.
/// Débloque le sink s'il attendait un client.
pub async fn connect(&self) -> DirectFlacStream {
let connect_count_before = *self.client_connect_tx.borrow();
debug!("DirectFlacHandle::connect() called, connect_count={}", connect_count_before);
let (pcm_tx, pcm_rx) = mpsc::channel::<PcmChunk>(8);
let current_ts = Arc::new(tokio::sync::RwLock::new(0.0f64));
let current_dur = Arc::new(tokio::sync::RwLock::new(0.0f64));
let pcm_reader = ByteStreamReader::new(pcm_rx, current_ts, current_dur);
let (pipe_writer, pipe_reader) = tokio::io::duplex(PIPE_CAPACITY);
// Réinitialiser le signal "premier byte" AVANT de notifier le sink,
// pour éviter qu'une notification précédente ne se propage.
let _ = self.first_byte_tx.send(false);
debug!("DirectFlacHandle::connect() first_byte reset to false");
// Installer le nouveau sender (remplace l'éventuel ancien)
*self.pcm_tx.lock().await = Some(pcm_tx);
debug!("DirectFlacHandle::connect() pcm_tx installed");
// Incrémenter le compteur de connexions (mémorisé dans watch — pas de perte)
let new_count = connect_count_before.wrapping_add(1);
let _ = self.client_connect_tx.send(new_count);
debug!("DirectFlacHandle::connect() client_connect_count -> {}", new_count);
// Débloquer le sink interne (même contexte async → pas de race)
self.client_notify_internal.notify_one();
// Lancer l'encodeur en background
let options = self.encoder_options.clone();
tokio::spawn(async move {
debug!("DirectFlacHandle: encoder task started");
if let Err(e) = run_encoder(pcm_reader, pipe_writer, options).await {
debug!("DirectFlacStream encoder stopped: {}", e);
}
debug!("DirectFlacHandle: encoder task ended");
});
debug!("DirectFlacHandle::connect() returning DirectFlacStream");
DirectFlacStream {
inner: pipe_reader,
first_byte_tx: Some(self.first_byte_tx.clone()),
}
}
/// Retourne un receiver qui passe à `true` quand le premier byte FLAC
/// a été effectivement lu par le client HTTP.
pub fn first_byte_ready(&self) -> watch::Receiver<bool> {
self.first_byte_tx.subscribe()
}
/// Attend qu'un client HTTP se connecte (i.e. que `connect()` soit appelé).
/// Utilisé par `stream_source` pour retarder l'ouverture de la source
/// jusqu'à ce que le navigateur soit prêt à recevoir des données.
///
/// Mémorise la valeur du compteur au moment de l'appel et attend qu'elle
/// augmente — ce qui garantit qu'on attend bien UNE NOUVELLE connexion,
/// même si `connect()` a déjà été appelé lors d'une lecture précédente.
pub async fn wait_for_client(&self) {
let seen = *self.client_connect_tx.borrow();
debug!("DirectFlacHandle::wait_for_client() called, seen connect_count={}", seen);
// subscribe() retourne un receiver dont la valeur courante est marquée "changed"
// donc wait_for() retourne immédiatement si la condition est déjà vraie.
let mut rx = self.client_connect_tx.subscribe();
let result = rx.wait_for(|v| {
debug!("DirectFlacHandle::wait_for_client() checking v={} > seen={}: {}", v, seen, *v > seen);
*v > seen
}).await;
debug!("DirectFlacHandle::wait_for_client() unblocked, result ok={}", result.is_ok());
}
}
// ─── Stream public ────────────────────────────────────────────────────────────
/// Flux FLAC exposé au handler HTTP.
///
/// Transmet les bytes du pipe directement au client HTTP.
/// Intercepte le premier `poll_read` non-vide pour signaler via `first_byte_tx`
/// que des données FLAC ont effectivement été transmises au client.
pub struct DirectFlacStream {
inner: tokio::io::DuplexStream,
/// Présent jusqu'au premier byte reçu, puis consommé (set à None).
first_byte_tx: Option<Arc<watch::Sender<bool>>>,
}
impl AsyncRead for DirectFlacStream {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
let filled_before = buf.filled().len();
let result = Pin::new(&mut self.inner).poll_read(cx, buf);
if let Poll::Ready(Ok(())) = &result {
let filled_after = buf.filled().len();
if filled_after > filled_before {
if let Some(tx) = self.first_byte_tx.take() {
debug!("DirectFlacStream: first {} bytes sent to HTTP client", filled_after - filled_before);
let _ = tx.send(true);
}
}
}
result
}
}
// ─── Logique du nœud ─────────────────────────────────────────────────────────
struct DirectFlacSinkLogic {
pcm_tx: SharedPcmTx,
client_notify: Arc<tokio::sync::Notify>,
}
#[async_trait]
impl NodeLogic for DirectFlacSinkLogic {
async fn process(
&mut self,
input: Option<mpsc::Receiver<Arc<AudioSegment>>>,
_output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
let mut input = input.ok_or_else(|| {
AudioError::ProcessingError("DirectFlacSink requires an input".into())
})?;
loop {
tokio::select! {
_ = stop_token.cancelled() => {
debug!("DirectFlacSink: cancelled");
break;
}
segment = input.recv() => {
match segment {
None => {
debug!("DirectFlacSink: input channel closed");
break;
}
Some(seg) => match &seg.segment {
_AudioSegment::Chunk(chunk) => {
// Attendre un client si nécessaire (backpressure quand pas de Play)
loop {
let tx_opt = self.pcm_tx.lock().await.clone();
if tx_opt.is_some() {
break;
}
// Pas de client : bloquer jusqu'à connect() ou stop
debug!("DirectFlacSink: no pcm_tx, waiting for client_notify...");
tokio::select! {
_ = stop_token.cancelled() => {
debug!("DirectFlacSink: cancelled while waiting for client");
return Ok(());
}
_ = self.client_notify.notified() => {
debug!("DirectFlacSink: client_notify received, rechecking pcm_tx");
}
}
}
let tx = self.pcm_tx.lock().await.clone().unwrap();
let pcm_bytes = chunk_to_pcm_bytes(chunk, DIRECT_FLAC_BITS_PER_SAMPLE)?;
let duration_sec = chunk.len() as f64 / DIRECT_FLAC_SAMPLE_RATE as f64;
let pcm_chunk = PcmChunk {
bytes: pcm_bytes,
timestamp_sec: seg.timestamp_sec,
duration_sec,
};
if tx.send(pcm_chunk).await.is_err() {
debug!("DirectFlacSink: pcm_tx send failed (client disconnected), clearing pcm_tx");
*self.pcm_tx.lock().await = None;
}
}
_AudioSegment::Sync(marker) => match marker.as_ref() {
SyncMarker::EndOfStream => {
debug!("DirectFlacSink: EndOfStream");
}
_ => {}
},
},
}
}
}
}
Ok(())
}
async fn cleanup(&mut self, _reason: StopReason) -> Result<(), AudioError> {
Ok(())
}
}
// ─── Encodeur FLAC ────────────────────────────────────────────────────────────
async fn run_encoder(
pcm_reader: ByteStreamReader,
mut pipe_writer: tokio::io::DuplexStream,
options: EncoderOptions,
) -> Result<(), AudioError> {
let format = PcmFormat {
sample_rate: DIRECT_FLAC_SAMPLE_RATE,
channels: DIRECT_FLAC_CHANNELS,
bits_per_sample: DIRECT_FLAC_BITS_PER_SAMPLE,
};
let mut flac_stream = pmoflac::encode_flac_stream(pcm_reader, format, options)
.await
.map_err(|e| AudioError::ProcessingError(format!("FLAC encoder init: {}", e)))?;
tokio::io::copy(&mut flac_stream, &mut pipe_writer)
.await
.map_err(|e| AudioError::IoError(format!("FLAC pipe copy: {}", e)))?;
flac_stream
.wait()
.await
.map_err(|e| AudioError::ProcessingError(format!("FLAC encoder wait: {}", e)))?;
Ok(())
}
// ─── Nœud public ─────────────────────────────────────────────────────────────
pub struct DirectFlacSink {
inner: Node<DirectFlacSinkLogic>,
}
impl DirectFlacSink {
pub fn new(encoder_options: EncoderOptions) -> (Self, DirectFlacHandle) {
let pcm_tx: SharedPcmTx = Arc::new(Mutex::new(None));
let client_notify_internal = Arc::new(tokio::sync::Notify::new());
let (client_connect_tx, _) = watch::channel(0u64);
let client_connect_tx = Arc::new(client_connect_tx);
let (first_byte_tx, _) = watch::channel(false);
let first_byte_tx = Arc::new(first_byte_tx);
let logic = DirectFlacSinkLogic {
pcm_tx: pcm_tx.clone(),
client_notify: client_notify_internal.clone(),
};
let sink = Self {
inner: Node::new_with_input(logic, 16),
};
let handle = DirectFlacHandle {
pcm_tx,
client_connect_tx,
client_notify_internal,
first_byte_tx,
encoder_options,
};
(sink, handle)
}
}
#[async_trait]
impl AudioPipelineNode for DirectFlacSink {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
self.inner.get_tx()
}
fn register(&mut self, _child: Box<dyn AudioPipelineNode>) {
panic!("DirectFlacSink is a terminal sink and cannot have children");
}
async fn run(self: Box<Self>, stop_token: CancellationToken) -> Result<(), AudioError> {
Box::new(self.inner).run(stop_token).await
}
fn start(self: Box<Self>) -> PipelineHandle {
Box::new(self.inner).start()
}
}
impl TypedAudioNode for DirectFlacSink {
fn input_type(&self) -> Option<TypeRequirement> {
Some(TypeRequirement::any_integer())
}
fn output_type(&self) -> Option<TypeRequirement> {
None
}
}

564
pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs Normal file
View File

@@ -0,0 +1,564 @@
//! DirectOggFlacSink — nœud puits OGG-FLAC pour un seul client HTTP.
//!
//! Combine la logique de backpressure/reconnexion de `DirectFlacSink`
//! avec l'encodage OGG-FLAC de `StreamingOggFlacSink`.
//!
//! # Cycle de vie
//!
//! - **Play** : le navigateur appelle `GET /stream`. `connect()` crée un nouveau
//! canal PCM + pipe duplex + encodeur FLAC + wrapper OGG, installe le sender
//! dans le sink, et notifie le sink via `client_notify`. Le flux reste ouvert :
//! les morceaux s'enchaînent en gapless.
//! - **Stop** : le navigateur ferme la connexion. Le pipe se rompt, l'encodeur
//! s'arrête. Le sink voit `pcm_tx.send()` échouer, passe le sender à `None`,
//! et **bloque** sur `client_notify` jusqu'au prochain Play.
//! - **Play suivant** : `connect()` → nouveau pipe → `client_notify.notify_one()`
//! → le sink se débloque et reprend la consommation des segments.
//!
//! # Architecture
//!
//! ```text
//! AudioSegment I24 @ 96 kHz
//! ↓ NodeLogic::process() [bloque si pas de client]
//! chunk_to_pcm_bytes() → PCM 24-bit LE
//! ↓ Arc<Mutex<Option<mpsc::Sender<PcmChunk>>>>
//! ByteStreamReader (AsyncRead)
//! ↓ encode_flac_stream()
//! ↓ broadcast_ogg_flac_stream() → wrapping OGG pages
//! ↓ tokio::io::duplex pipe (256 KB)
//! ↓ DirectOggFlacStream (AsyncRead) → Body HTTP
//! ```
use std::io;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use async_trait::async_trait;
use bytes::Bytes;
use pmoaudio::{
pipeline::{AudioPipelineNode, Node, NodeLogic, PipelineHandle, StopReason},
AudioError, AudioSegment, SyncMarker, TypeRequirement, TypedAudioNode, _AudioSegment,
};
use pmoflac::{EncoderOptions, PcmFormat};
use tokio::io::{AsyncRead, ReadBuf};
use tokio::sync::{mpsc, watch, Mutex};
use tokio_util::sync::CancellationToken;
use tracing::debug;
use crate::sinks::byte_stream_reader::{ByteStreamReader, PcmChunk};
use crate::sinks::chunk_to_pcm::chunk_to_pcm_bytes;
use crate::sinks::flac_frame_utils::{extract_sample_rate_from_streaminfo, read_flac_header};
/// Format de sortie fixe du sink.
pub const DIRECT_OGG_FLAC_SAMPLE_RATE: u32 = 96_000;
pub const DIRECT_OGG_FLAC_CHANNELS: u8 = 2;
pub const DIRECT_OGG_FLAC_BITS_PER_SAMPLE: u8 = 24;
/// Capacité du pipe duplex (~256 KB).
const PIPE_CAPACITY: usize = 256 * 1024;
// ─── Shared state ─────────────────────────────────────────────────────────────
type SharedPcmTx = Arc<Mutex<Option<mpsc::Sender<PcmChunk>>>>;
// ─── Handle public ────────────────────────────────────────────────────────────
/// Handle vers le sink, cloneable, reconnectable à chaque Play.
#[derive(Clone)]
pub struct DirectOggFlacHandle {
pcm_tx: SharedPcmTx,
client_connect_tx: Arc<watch::Sender<u64>>,
client_notify_internal: Arc<tokio::sync::Notify>,
first_byte_tx: Arc<watch::Sender<bool>>,
encoder_options: EncoderOptions,
/// Position de lecture courante (mise à jour par ByteStreamReader).
current_timestamp: Arc<tokio::sync::RwLock<f64>>,
}
impl DirectOggFlacHandle {
/// Crée un nouveau pipe OGG-FLAC et retourne le flux côté lecture.
/// Débloque le sink s'il attendait un client.
pub async fn connect(&self) -> DirectOggFlacStream {
let connect_count_before = *self.client_connect_tx.borrow();
debug!("DirectOggFlacHandle::connect() called, connect_count={}", connect_count_before);
let (pcm_tx, pcm_rx) = mpsc::channel::<PcmChunk>(8);
// Réinitialiser le timestamp à 0 pour la nouvelle connexion
*self.current_timestamp.write().await = 0.0;
let current_dur = Arc::new(tokio::sync::RwLock::new(0.0f64));
// Partager current_timestamp avec ByteStreamReader : il sera mis à jour
// avec le timestamp absolu du segment audio (position dans le fichier source).
let pcm_reader = ByteStreamReader::new(pcm_rx, self.current_timestamp.clone(), current_dur);
let (pipe_writer, pipe_reader) = tokio::io::duplex(PIPE_CAPACITY);
let _ = self.first_byte_tx.send(false);
debug!("DirectOggFlacHandle::connect() first_byte reset to false");
*self.pcm_tx.lock().await = Some(pcm_tx);
debug!("DirectOggFlacHandle::connect() pcm_tx installed");
let new_count = connect_count_before.wrapping_add(1);
let _ = self.client_connect_tx.send(new_count);
debug!("DirectOggFlacHandle::connect() client_connect_count -> {}", new_count);
self.client_notify_internal.notify_one();
let options = self.encoder_options.clone();
let current_timestamp = self.current_timestamp.clone();
tokio::spawn(async move {
debug!("DirectOggFlacHandle: encoder+ogg task started");
if let Err(e) = run_ogg_encoder(pcm_reader, pipe_writer, options, current_timestamp).await {
debug!("DirectOggFlacStream encoder stopped: {}", e);
}
debug!("DirectOggFlacHandle: encoder+ogg task ended");
});
debug!("DirectOggFlacHandle::connect() returning DirectOggFlacStream");
DirectOggFlacStream {
inner: pipe_reader,
first_byte_tx: Some(self.first_byte_tx.clone()),
}
}
pub fn first_byte_ready(&self) -> watch::Receiver<bool> {
self.first_byte_tx.subscribe()
}
/// Retourne la position de lecture courante en secondes.
pub async fn current_position_sec(&self) -> f64 {
*self.current_timestamp.read().await
}
pub async fn wait_for_client(&self) {
let seen = *self.client_connect_tx.borrow();
debug!("DirectOggFlacHandle::wait_for_client() called, seen connect_count={}", seen);
let mut rx = self.client_connect_tx.subscribe();
let _ = rx.wait_for(|v| *v > seen).await;
debug!("DirectOggFlacHandle::wait_for_client() unblocked");
}
}
// ─── Stream public ────────────────────────────────────────────────────────────
pub struct DirectOggFlacStream {
inner: tokio::io::DuplexStream,
first_byte_tx: Option<Arc<watch::Sender<bool>>>,
}
impl AsyncRead for DirectOggFlacStream {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
let filled_before = buf.filled().len();
let result = Pin::new(&mut self.inner).poll_read(cx, buf);
if let Poll::Ready(Ok(())) = &result {
let filled_after = buf.filled().len();
if filled_after > filled_before {
if let Some(tx) = self.first_byte_tx.take() {
debug!("DirectOggFlacStream: first {} bytes sent to HTTP client", filled_after - filled_before);
let _ = tx.send(true);
}
}
}
result
}
}
// ─── Logique du nœud ─────────────────────────────────────────────────────────
struct DirectOggFlacSinkLogic {
pcm_tx: SharedPcmTx,
client_notify: Arc<tokio::sync::Notify>,
}
#[async_trait]
impl NodeLogic for DirectOggFlacSinkLogic {
async fn process(
&mut self,
input: Option<mpsc::Receiver<Arc<AudioSegment>>>,
_output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
let mut input = input.ok_or_else(|| {
AudioError::ProcessingError("DirectOggFlacSink requires an input".into())
})?;
loop {
tokio::select! {
_ = stop_token.cancelled() => {
debug!("DirectOggFlacSink: cancelled");
break;
}
segment = input.recv() => {
match segment {
None => {
debug!("DirectOggFlacSink: input channel closed");
break;
}
Some(seg) => match &seg.segment {
_AudioSegment::Chunk(chunk) => {
// Attendre un client si nécessaire (backpressure quand pas de Play)
loop {
let tx_opt = self.pcm_tx.lock().await.clone();
if tx_opt.is_some() {
break;
}
debug!("DirectOggFlacSink: no pcm_tx, waiting for client_notify...");
tokio::select! {
_ = stop_token.cancelled() => {
debug!("DirectOggFlacSink: cancelled while waiting for client");
return Ok(());
}
_ = self.client_notify.notified() => {
debug!("DirectOggFlacSink: client_notify received, rechecking pcm_tx");
}
}
}
let tx = self.pcm_tx.lock().await.clone().unwrap();
let pcm_bytes = chunk_to_pcm_bytes(chunk, DIRECT_OGG_FLAC_BITS_PER_SAMPLE)?;
let duration_sec = chunk.len() as f64 / DIRECT_OGG_FLAC_SAMPLE_RATE as f64;
let pcm_chunk = PcmChunk {
bytes: pcm_bytes,
timestamp_sec: seg.timestamp_sec,
duration_sec,
};
if tx.send(pcm_chunk).await.is_err() {
debug!("DirectOggFlacSink: pcm_tx send failed (client disconnected), clearing pcm_tx");
*self.pcm_tx.lock().await = None;
}
}
_AudioSegment::Sync(marker) => match marker.as_ref() {
SyncMarker::EndOfStream => {
debug!("DirectOggFlacSink: EndOfStream");
}
_ => {}
},
},
}
}
}
}
Ok(())
}
async fn cleanup(&mut self, _reason: StopReason) -> Result<(), AudioError> {
Ok(())
}
}
// ─── Encodeur FLAC + wrapper OGG ─────────────────────────────────────────────
async fn run_ogg_encoder(
pcm_reader: ByteStreamReader,
mut pipe_writer: tokio::io::DuplexStream,
options: EncoderOptions,
_current_timestamp: Arc<tokio::sync::RwLock<f64>>,
) -> Result<(), AudioError> {
let format = PcmFormat {
sample_rate: DIRECT_OGG_FLAC_SAMPLE_RATE,
channels: DIRECT_OGG_FLAC_CHANNELS,
bits_per_sample: DIRECT_OGG_FLAC_BITS_PER_SAMPLE,
};
let mut flac_stream = pmoflac::encode_flac_stream(pcm_reader, format, options)
.await
.map_err(|e| AudioError::ProcessingError(format!("FLAC encoder init: {}", e)))?;
// Lire le header FLAC et construire les pages OGG d'en-tête
let flac_header = read_flac_header(&mut flac_stream).await?;
let sample_rate = extract_sample_rate_from_streaminfo(&flac_header)?;
let stream_serial: u32 = rand::random();
let mut ogg = OggPageWriter::new(stream_serial);
// Page BOS (identification OGG-FLAC)
let ogg_flac_id = create_ogg_flac_identification(&flac_header)?;
let bos_page = Bytes::from(ogg.create_page(&ogg_flac_id, true, false, false));
// Page Vorbis Comment
let vorbis_comment = create_empty_vorbis_comment();
let comment_page = Bytes::from(ogg.create_page(&vorbis_comment, false, false, false));
pipe_writer.write_all(&bos_page).await
.map_err(|e| AudioError::IoError(format!("OGG BOS write: {}", e)))?;
pipe_writer.write_all(&comment_page).await
.map_err(|e| AudioError::IoError(format!("OGG comment write: {}", e)))?;
// Lire les frames FLAC et les encapsuler dans des pages OGG
let sample_rate_f64 = sample_rate as f64;
let mut encoded_samples = 0u64;
let mut read_buffer = vec![0u8; 16384];
let mut accumulator: Vec<u8> = Vec::with_capacity(32768);
use tokio::io::{AsyncReadExt, AsyncWriteExt};
loop {
match flac_stream.read(&mut read_buffer).await {
Ok(0) => {
// EOF : page EOS finale
let eos_page = Bytes::from(ogg.create_page(&accumulator, false, true, false));
let _ = pipe_writer.write_all(&eos_page).await;
break;
}
Ok(n) => {
accumulator.extend_from_slice(&read_buffer[..n]);
loop {
if accumulator.len() < 4 {
break;
}
// Trouver les positions de sync FLAC
let mut sync_data: Vec<(usize, u32)> = Vec::new();
for i in 0..accumulator.len() - 1 {
let b1 = accumulator[i];
let b2 = accumulator[i + 1];
if b1 == 0xFF && b2 >= 0xF8 && b2 <= 0xFE {
use crate::sinks::flac_frame_utils::{validate_frame_header_crc, parse_flac_block_size};
if validate_frame_header_crc(&accumulator, i) {
if let Some(samples) = parse_flac_block_size(&accumulator, i) {
sync_data.push((i, samples));
}
}
}
}
if sync_data.len() < 2 {
break;
}
let first_start = sync_data[0].0;
let first_samples = sync_data[0].1;
let second_start = sync_data[1].0;
if first_start != 0 {
accumulator.drain(0..first_start);
continue;
}
let frame: Vec<u8> = accumulator.drain(0..second_start).collect();
encoded_samples = encoded_samples.saturating_add(first_samples as u64);
ogg.add_samples(first_samples as u64);
let ogg_page = Bytes::from(ogg.create_page(&frame, false, false, false));
if pipe_writer.write_all(&ogg_page).await.is_err() {
// Client déconnecté
return Ok(());
}
}
}
Err(e) => {
return Err(AudioError::ProcessingError(format!("FLAC read: {}", e)));
}
}
}
flac_stream.wait().await
.map_err(|e| AudioError::ProcessingError(format!("FLAC encoder wait: {}", e)))?;
Ok(())
}
// ─── OGG helpers (copiés de streaming_ogg_flac_sink) ─────────────────────────
struct OggPageWriter {
stream_serial: u32,
page_sequence: u32,
granule_position: u64,
}
impl OggPageWriter {
fn new(stream_serial: u32) -> Self {
Self { stream_serial, page_sequence: 0, granule_position: 0 }
}
fn add_samples(&mut self, samples: u64) {
self.granule_position += samples;
}
fn create_page(&mut self, packet_data: &[u8], is_bos: bool, is_eos: bool, is_continuation: bool) -> Vec<u8> {
use std::io::Write;
let mut segments = Vec::new();
let mut remaining = packet_data.len();
while remaining > 0 {
let seg = remaining.min(255);
segments.push(seg as u8);
remaining -= seg;
}
if !packet_data.is_empty() && packet_data.len() % 255 == 0 && !is_continuation {
segments.push(0);
}
let segment_count = segments.len();
let total_size = 27 + segment_count + packet_data.len();
let mut page = Vec::with_capacity(total_size);
page.write_all(b"OggS").unwrap();
page.write_all(&[0]).unwrap();
let mut header_type = 0u8;
if is_continuation { header_type |= 0x01; }
if is_bos { header_type |= 0x02; }
if is_eos { header_type |= 0x04; }
page.write_all(&[header_type]).unwrap();
page.write_all(&self.granule_position.to_le_bytes()).unwrap();
page.write_all(&self.stream_serial.to_le_bytes()).unwrap();
page.write_all(&self.page_sequence.to_le_bytes()).unwrap();
self.page_sequence += 1;
let crc_offset = page.len();
page.write_all(&[0, 0, 0, 0]).unwrap();
page.write_all(&[segment_count as u8]).unwrap();
page.write_all(&segments).unwrap();
page.write_all(packet_data).unwrap();
let crc = calculate_ogg_crc(&page);
page[crc_offset..crc_offset + 4].copy_from_slice(&crc.to_le_bytes());
page
}
}
fn calculate_ogg_crc(data: &[u8]) -> u32 {
const CRC_TABLE: [u32; 256] = generate_crc_table();
let mut crc: u32 = 0;
for &byte in data {
crc = (crc << 8) ^ CRC_TABLE[((crc >> 24) ^ (byte as u32)) as usize];
}
crc
}
const fn generate_crc_table() -> [u32; 256] {
let mut table = [0u32; 256];
let mut i = 0usize;
while i < 256 {
let mut r = (i as u32) << 24;
let mut j = 0;
while j < 8 {
if (r & 0x80000000) != 0 { r = (r << 1) ^ 0x04c11db7; } else { r <<= 1; }
j += 1;
}
table[i] = r;
i += 1;
}
table
}
fn create_ogg_flac_identification(flac_header: &[u8]) -> Result<Vec<u8>, AudioError> {
if flac_header.len() < 8 || &flac_header[0..4] != b"fLaC" {
return Err(AudioError::ProcessingError("Invalid FLAC header".into()));
}
let first_block_type = flac_header[4] & 0x7F;
if first_block_type != 0 {
return Err(AudioError::ProcessingError("First FLAC block is not STREAMINFO".into()));
}
let block_length = u32::from_be_bytes([0, flac_header[5], flac_header[6], flac_header[7]]) as usize;
let streaminfo_size = 4 + block_length;
if flac_header.len() < 4 + streaminfo_size {
return Err(AudioError::ProcessingError("FLAC header truncated".into()));
}
let streaminfo = &flac_header[4..4 + streaminfo_size];
let mut packet = Vec::new();
packet.push(0x7F);
packet.extend_from_slice(b"FLAC");
packet.push(0x01);
packet.push(0x00);
packet.extend_from_slice(&1u16.to_be_bytes());
packet.extend_from_slice(b"fLaC");
packet.extend_from_slice(streaminfo);
Ok(packet)
}
fn create_empty_vorbis_comment() -> Vec<u8> {
let vendor = "pmoaudio DirectOggFlacSink";
let vendor_bytes = vendor.as_bytes();
let mut vorbis_data = Vec::new();
vorbis_data.extend_from_slice(&(vendor_bytes.len() as u32).to_le_bytes());
vorbis_data.extend_from_slice(vendor_bytes);
vorbis_data.extend_from_slice(&0u32.to_le_bytes());
let mut block = Vec::new();
block.push(0x84); // last-block + VORBIS_COMMENT type
let length = vorbis_data.len() as u32;
block.push((length >> 16) as u8);
block.push((length >> 8) as u8);
block.push(length as u8);
block.extend_from_slice(&vorbis_data);
block
}
// ─── Nœud public ─────────────────────────────────────────────────────────────
pub struct DirectOggFlacSink {
inner: Node<DirectOggFlacSinkLogic>,
}
impl DirectOggFlacSink {
pub fn new(encoder_options: EncoderOptions) -> (Self, DirectOggFlacHandle) {
let pcm_tx: SharedPcmTx = Arc::new(Mutex::new(None));
let client_notify_internal = Arc::new(tokio::sync::Notify::new());
let (client_connect_tx, _) = watch::channel(0u64);
let client_connect_tx = Arc::new(client_connect_tx);
let (first_byte_tx, _) = watch::channel(false);
let first_byte_tx = Arc::new(first_byte_tx);
let current_timestamp = Arc::new(tokio::sync::RwLock::new(0.0f64));
let logic = DirectOggFlacSinkLogic {
pcm_tx: pcm_tx.clone(),
client_notify: client_notify_internal.clone(),
};
let sink = Self {
inner: Node::new_with_input(logic, 16),
};
let handle = DirectOggFlacHandle {
pcm_tx,
client_connect_tx,
client_notify_internal,
first_byte_tx,
encoder_options,
current_timestamp,
};
(sink, handle)
}
}
#[async_trait]
impl AudioPipelineNode for DirectOggFlacSink {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
self.inner.get_tx()
}
fn register(&mut self, _child: Box<dyn AudioPipelineNode>) {
panic!("DirectOggFlacSink is a terminal sink and cannot have children");
}
async fn run(self: Box<Self>, stop_token: CancellationToken) -> Result<(), AudioError> {
Box::new(self.inner).run(stop_token).await
}
fn start(self: Box<Self>) -> PipelineHandle {
Box::new(self.inner).start()
}
}
impl TypedAudioNode for DirectOggFlacSink {
fn input_type(&self) -> Option<TypeRequirement> {
Some(TypeRequirement::any_integer())
}
fn output_type(&self) -> Option<TypeRequirement> {
None
}
}

18
pmoaudio-ext/src/sinks/mod.rs
View File

@@ -23,6 +23,24 @@ mod flac_frame_utils;
#[cfg(feature = "http-stream")]
mod timed_broadcast;
#[cfg(feature = "http-stream")]
mod direct_flac_sink;
#[cfg(feature = "http-stream")]
pub use direct_flac_sink::{
DirectFlacHandle, DirectFlacSink, DirectFlacStream,
DIRECT_FLAC_BITS_PER_SAMPLE, DIRECT_FLAC_CHANNELS, DIRECT_FLAC_SAMPLE_RATE,
};
#[cfg(feature = "http-stream")]
mod direct_ogg_flac_sink;
#[cfg(feature = "http-stream")]
pub use direct_ogg_flac_sink::{
DirectOggFlacHandle, DirectOggFlacSink, DirectOggFlacStream,
DIRECT_OGG_FLAC_BITS_PER_SAMPLE, DIRECT_OGG_FLAC_CHANNELS, DIRECT_OGG_FLAC_SAMPLE_RATE,
};
#[cfg(feature = "http-stream")]
mod streaming_flac_sink;

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

@@ -215,7 +215,13 @@ impl AsyncRead for SharedClientStream {
self.state = StreamState::Streaming;
continue;
} else {
self.state = StreamState::Streaming;
// Header not yet available (encoder not yet started): wait and retry
let waker = cx.waker().clone();
tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(10)).await;
waker.wake();
});
return Poll::Pending;
}
}

11
pmoaudio-ext/src/sources/mod.rs
View File

@@ -3,8 +3,19 @@
//! Ce module contient des sources audio qui dépendent d'autres crates
//! du projet PMO (pmoplaylist, pmoaudiocache, etc.)
// Helpers partagés (conversion PCM → AudioSegment)
// Disponibles dès que l'une des deux features qui en dépend est activée
#[cfg(any(feature = "playlist", feature = "http-stream"))]
pub(crate) mod pcm_decode;
#[cfg(feature = "playlist")]
mod playlist_source;
#[cfg(feature = "playlist")]
pub use playlist_source::PlaylistSource;
#[cfg(feature = "http-stream")]
mod uri_source;
#[cfg(feature = "http-stream")]
pub use uri_source::UriSource;

145
pmoaudio-ext/src/sources/pcm_decode.rs Normal file
View File

@@ -0,0 +1,145 @@
//! Helpers partagés pour le décodage PCM → AudioSegment
//!
//! Utilisés par `PlaylistSource` et `UriSource`.
//!
//! Les fonctions `bytes_to_segment` et `validate_stream` ont été extraites
//! de `playlist_source.rs` sans modification pour éviter toute duplication.
use std::sync::Arc;
use pmoaudio::{AudioChunk, AudioChunkData, AudioSegment, I24, _AudioSegment, nodes::AudioError};
use pmoflac::StreamInfo;
pub(crate) fn validate_stream(info: &StreamInfo) -> Result<(), AudioError> {
if !(1..=2).contains(&info.channels) {
return Err(AudioError::ProcessingError(format!(
"Unsupported channel count: {}",
info.channels
)));
}
match info.bits_per_sample {
8 | 16 | 24 | 32 => Ok(()),
other => Err(AudioError::ProcessingError(format!(
"Unsupported bit depth: {}",
other
))),
}
}
/// Convertit des bytes PCM en AudioSegment avec le type approprié
pub(crate) fn bytes_to_segment(
chunk_bytes: &[u8],
info: &StreamInfo,
frames: usize,
order: u64,
timestamp_sec: f64,
) -> Result<Arc<AudioSegment>, AudioError> {
let bytes_per_sample = info.bytes_per_sample();
let channels = info.channels as usize;
let frame_bytes = bytes_per_sample * channels;
// Créer le chunk du bon type selon la profondeur de bit
let chunk = match info.bits_per_sample {
16 => {
// Type I16
let mut stereo = Vec::with_capacity(frames);
for frame_idx in 0..frames {
let base = frame_idx * frame_bytes;
let l = i16::from_le_bytes(
chunk_bytes[base..base + bytes_per_sample]
.try_into()
.unwrap(),
);
let r = if channels == 1 {
l
} else {
i16::from_le_bytes(
chunk_bytes[base + bytes_per_sample..base + 2 * bytes_per_sample]
.try_into()
.unwrap(),
)
};
stereo.push([l, r]);
}
let chunk_data = AudioChunkData::new(stereo, info.sample_rate, 0.0);
AudioChunk::I16(chunk_data)
}
24 => {
// Type I24
let mut stereo = Vec::with_capacity(frames);
for frame_idx in 0..frames {
let base = frame_idx * frame_bytes;
let l_i32 = {
let mut buf = [0u8; 4];
buf[..3].copy_from_slice(&chunk_bytes[base..base + 3]);
// Sign extend
if chunk_bytes[base + 2] & 0x80 != 0 {
buf[3] = 0xFF;
}
i32::from_le_bytes(buf)
};
let l = I24::new(l_i32).ok_or_else(|| {
AudioError::ProcessingError(format!("Invalid I24 value: {}", l_i32))
})?;
let r = if channels == 1 {
l
} else {
let r_i32 = {
let mut buf = [0u8; 4];
buf[..3].copy_from_slice(
&chunk_bytes[base + bytes_per_sample..base + bytes_per_sample + 3],
);
// Sign extend
if chunk_bytes[base + bytes_per_sample + 2] & 0x80 != 0 {
buf[3] = 0xFF;
}
i32::from_le_bytes(buf)
};
I24::new(r_i32).ok_or_else(|| {
AudioError::ProcessingError(format!("Invalid I24 value: {}", r_i32))
})?
};
stereo.push([l, r]);
}
let chunk_data = AudioChunkData::new(stereo, info.sample_rate, 0.0);
AudioChunk::I24(chunk_data)
}
32 => {
// Type I32
let mut stereo = Vec::with_capacity(frames);
for frame_idx in 0..frames {
let base = frame_idx * frame_bytes;
let l = i32::from_le_bytes(
chunk_bytes[base..base + bytes_per_sample]
.try_into()
.unwrap(),
);
let r = if channels == 1 {
l
} else {
i32::from_le_bytes(
chunk_bytes[base + bytes_per_sample..base + 2 * bytes_per_sample]
.try_into()
.unwrap(),
)
};
stereo.push([l, r]);
}
let chunk_data = AudioChunkData::new(stereo, info.sample_rate, 0.0);
AudioChunk::I32(chunk_data)
}
_ => {
return Err(AudioError::ProcessingError(format!(
"Unsupported bit depth: {}",
info.bits_per_sample
)))
}
};
Ok(Arc::new(AudioSegment {
order,
timestamp_sec,
segment: _AudioSegment::Chunk(Arc::new(chunk)),
}))
}

138
pmoaudio-ext/src/sources/playlist_source.rs
View File

@@ -113,11 +113,13 @@ use pmoaudio::{
nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS},
pipeline::{send_to_children, AudioPipelineNode, Node, NodeLogic},
type_constraints::TypeRequirement,
AudioChunk, AudioChunkData, AudioSegment, I24,
AudioSegment,
};
use pmoaudiocache::Cache as AudioCache;
use pmoflac::{decode_audio_stream, StreamInfo};
use pmoplaylist::{PlaylistRole, ReadHandle};
use super::pcm_decode::{bytes_to_segment, validate_stream};
use std::{path::PathBuf, sync::Arc, time::Duration};
use tokio::{fs::File, io::AsyncReadExt, sync::mpsc};
use tokio_util::sync::CancellationToken;
@@ -563,139 +565,6 @@ async fn decode_and_emit_track(
Ok(())
}
fn validate_stream(info: &StreamInfo) -> Result<(), AudioError> {
if !(1..=2).contains(&info.channels) {
return Err(AudioError::ProcessingError(format!(
"Unsupported channel count: {}",
info.channels
)));
}
match info.bits_per_sample {
8 | 16 | 24 | 32 => Ok(()),
other => Err(AudioError::ProcessingError(format!(
"Unsupported bit depth: {}",
other
))),
}
}
/// Convertit des bytes PCM en AudioSegment avec le type approprié
fn bytes_to_segment(
chunk_bytes: &[u8],
info: &StreamInfo,
frames: usize,
order: u64,
timestamp_sec: f64,
) -> Result<Arc<AudioSegment>, AudioError> {
let bytes_per_sample = info.bytes_per_sample();
let channels = info.channels as usize;
let frame_bytes = bytes_per_sample * channels;
// Créer le chunk du bon type selon la profondeur de bit
let chunk = match info.bits_per_sample {
16 => {
// Type I16
let mut stereo = Vec::with_capacity(frames);
for frame_idx in 0..frames {
let base = frame_idx * frame_bytes;
let l = i16::from_le_bytes(
chunk_bytes[base..base + bytes_per_sample]
.try_into()
.unwrap(),
);
let r = if channels == 1 {
l
} else {
i16::from_le_bytes(
chunk_bytes[base + bytes_per_sample..base + 2 * bytes_per_sample]
.try_into()
.unwrap(),
)
};
stereo.push([l, r]);
}
let chunk_data = AudioChunkData::new(stereo, info.sample_rate, 0.0);
AudioChunk::I16(chunk_data)
}
24 => {
// Type I24
let mut stereo = Vec::with_capacity(frames);
for frame_idx in 0..frames {
let base = frame_idx * frame_bytes;
let l_i32 = {
let mut buf = [0u8; 4];
buf[..3].copy_from_slice(&chunk_bytes[base..base + 3]);
// Sign extend
if chunk_bytes[base + 2] & 0x80 != 0 {
buf[3] = 0xFF;
}
i32::from_le_bytes(buf)
};
let l = I24::new(l_i32).ok_or_else(|| {
AudioError::ProcessingError(format!("Invalid I24 value: {}", l_i32))
})?;
let r = if channels == 1 {
l
} else {
let r_i32 = {
let mut buf = [0u8; 4];
buf[..3].copy_from_slice(
&chunk_bytes[base + bytes_per_sample..base + bytes_per_sample + 3],
);
// Sign extend
if chunk_bytes[base + bytes_per_sample + 2] & 0x80 != 0 {
buf[3] = 0xFF;
}
i32::from_le_bytes(buf)
};
I24::new(r_i32).ok_or_else(|| {
AudioError::ProcessingError(format!("Invalid I24 value: {}", r_i32))
})?
};
stereo.push([l, r]);
}
let chunk_data = AudioChunkData::new(stereo, info.sample_rate, 0.0);
AudioChunk::I24(chunk_data)
}
32 => {
// Type I32
let mut stereo = Vec::with_capacity(frames);
for frame_idx in 0..frames {
let base = frame_idx * frame_bytes;
let l = i32::from_le_bytes(
chunk_bytes[base..base + bytes_per_sample]
.try_into()
.unwrap(),
);
let r = if channels == 1 {
l
} else {
i32::from_le_bytes(
chunk_bytes[base + bytes_per_sample..base + 2 * bytes_per_sample]
.try_into()
.unwrap(),
)
};
stereo.push([l, r]);
}
let chunk_data = AudioChunkData::new(stereo, info.sample_rate, 0.0);
AudioChunk::I32(chunk_data)
}
_ => {
return Err(AudioError::ProcessingError(format!(
"Unsupported bit depth: {}",
info.bits_per_sample
)))
}
};
Ok(Arc::new(AudioSegment {
order,
timestamp_sec,
segment: pmoaudio::_AudioSegment::Chunk(Arc::new(chunk)),
}))
}
// ═══════════════════════════════════════════════════════════════════════════
// WRAPPER PlaylistSource - Délègue à Node<PlaylistSourceLogic>
@@ -795,6 +664,7 @@ impl TypedAudioNode for PlaylistSource {
#[cfg(test)]
mod tests {
use super::*;
use pmoaudio::AudioChunk;
// ═══════════════════════════════════════════════════════════════════════════
// Tests unitaires pour les fonctions helper

254
pmoaudio-ext/src/sources/uri_source.rs Normal file
View File

@@ -0,0 +1,254 @@
//! UriSource - Source audio depuis une URI arbitraire
//!
//! Ouvre une URI (fichier local ou HTTP/HTTPS), décode l'audio (tout format
//! supporté par `pmoflac` : FLAC, MP3, OGG, WAV, AIFF) et émet des `AudioSegment`
//! vers un sender tokio.
//!
//! # Usage
//!
//! ```rust,no_run
//! use pmoaudio_ext::sources::UriSource;
//! use tokio_util::sync::CancellationToken;
//! use tokio::sync::mpsc;
//! use std::sync::Arc;
//! use pmoaudio::AudioSegment;
//!
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! let (tx, mut rx) = mpsc::channel::<Arc<AudioSegment>>(64);
//! let stop = CancellationToken::new();
//!
//! let source = UriSource::open("/music/track.flac", 0.0, stop.clone()).await?;
//! println!("Duration: {:?}", source.duration_sec());
//!
//! let eof = source.emit_to_channel(&tx, &stop).await?;
//! # Ok(())
//! # }
//! ```
//!
//! # Seek
//!
//! Implémenté par skip des frames initiales. Pour les formats sans seek natif
//! (MP3, stream HTTP), tout le contenu est lu mais les frames avant `seek_sec`
//! ne sont pas émises.
use std::sync::Arc;
use pmoaudio::{AudioSegment, nodes::AudioError};
use pmoflac::{StreamInfo, decode_audio_stream};
use tokio::io::AsyncReadExt;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info};
use super::pcm_decode::{bytes_to_segment, validate_stream};
const CHUNK_FRAMES: usize = 2048; // ~46ms @ 44.1kHz
/// Source audio ouverte depuis une URI, prête à émettre des segments.
pub struct UriSource {
reader: Box<dyn tokio::io::AsyncRead + Send + Unpin>,
stream_info: StreamInfo,
frames_to_skip: u64,
}
impl UriSource {
/// Ouvre une URI et prépare la source.
///
/// - Chemin absolu ou `file://...` → fichier local
/// - `http://...` / `https://...` → streaming HTTP
pub async fn open(
uri: &str,
seek_sec: f64,
stop_token: CancellationToken,
) -> Result<Self, AudioError> {
if uri.starts_with("http://") || uri.starts_with("https://") {
Self::open_http(uri, seek_sec, &stop_token).await
} else {
let path = uri.strip_prefix("file://").unwrap_or(uri);
Self::open_file(path, seek_sec).await
}
}
/// Durée totale en secondes, si connue.
pub fn duration_sec(&self) -> Option<f64> {
let info = &self.stream_info;
info.total_samples
.filter(|&s| s > 0)
.map(|s| s as f64 / info.sample_rate as f64)
}
/// Nombre total de samples à 96 kHz après resampling, si connu.
/// Utilisé pour renseigner STREAMINFO.total_samples dans le FLAC de sortie.
pub fn total_samples_at(&self, output_sample_rate: u32) -> Option<u64> {
let info = &self.stream_info;
info.total_samples.filter(|&s| s > 0).map(|s| {
// Convertir le nombre de samples source vers le sample rate de sortie
let ratio = output_sample_rate as f64 / info.sample_rate as f64;
(s as f64 * ratio).round() as u64
})
}
/// Émet les chunks audio vers `tx`.
///
/// Retourne `Ok(true)` si EOF naturel, `Ok(false)` si annulé ou receiver fermé.
pub async fn emit_to_channel(
mut self,
tx: &mpsc::Sender<Arc<AudioSegment>>,
stop_token: &CancellationToken,
) -> Result<bool, AudioError> {
let info = self.stream_info.clone();
let bytes_per_sample = info.bytes_per_sample();
let frame_bytes = bytes_per_sample * info.channels as usize;
let chunk_byte_len = CHUNK_FRAMES * frame_bytes;
let mut pending = Vec::new();
let mut read_buf = vec![0u8; frame_bytes * 512.max(CHUNK_FRAMES)];
let mut chunk_index = 0u64;
let mut total_frames = 0u64;
loop {
tokio::select! {
_ = stop_token.cancelled() => {
debug!("UriSource: cancelled");
return Ok(false);
}
read_result = self.reader.read(&mut read_buf) => {
let read = read_result
.map_err(|e| AudioError::IoError(e.to_string()))?;
if read == 0 {
break; // EOF
}
pending.extend_from_slice(&read_buf[..read]);
while pending.len() >= chunk_byte_len {
let chunk_bytes = pending.drain(..chunk_byte_len).collect::<Vec<_>>();
let frames = CHUNK_FRAMES;
// Seek : ignorer les frames avant la position demandée
if total_frames + frames as u64 <= self.frames_to_skip {
total_frames += frames as u64;
chunk_index += 1;
continue;
}
let timestamp_sec = total_frames as f64 / info.sample_rate as f64;
let segment = bytes_to_segment(&chunk_bytes, &info, frames, chunk_index, timestamp_sec)?;
if tx.send(segment).await.is_err() {
debug!("UriSource: receiver dropped");
return Ok(false);
}
total_frames += frames as u64;
chunk_index += 1;
}
}
}
}
// Émettre le reste (< un chunk complet)
if !pending.is_empty() {
let frames = pending.len() / frame_bytes;
if frames > 0 && total_frames >= self.frames_to_skip {
let timestamp_sec = total_frames as f64 / info.sample_rate as f64;
if let Ok(seg) = bytes_to_segment(
&pending[..frames * frame_bytes],
&info,
frames,
chunk_index,
timestamp_sec,
) {
let _ = tx.send(seg).await;
}
}
}
info!(
"UriSource: EOF after {} frames ({:.1}s)",
total_frames,
total_frames as f64 / info.sample_rate.max(1) as f64
);
Ok(true)
}
// ── Constructeurs internes ────────────────────────────────────────────────
async fn open_file(path: &str, seek_sec: f64) -> Result<Self, AudioError> {
let file = tokio::fs::File::open(path)
.await
.map_err(|e| AudioError::IoError(format!("Cannot open {:?}: {}", path, e)))?;
let stream = decode_audio_stream(file)
.await
.map_err(|e| AudioError::ProcessingError(format!("Decode error: {}", e)))?;
let stream_info = stream.info().clone();
validate_stream(&stream_info)?;
let frames_to_skip = (seek_sec * stream_info.sample_rate as f64) as u64;
info!(
"UriSource: opened file {} Hz {} ch {} bps {:.1}s",
stream_info.sample_rate,
stream_info.channels,
stream_info.bits_per_sample,
stream_info.total_samples
.map(|s| s as f64 / stream_info.sample_rate as f64)
.unwrap_or(0.0),
);
let (_, reader) = stream.into_reader();
Ok(Self { reader: Box::new(reader), stream_info, frames_to_skip })
}
async fn open_http(
url: &str,
seek_sec: f64,
stop_token: &CancellationToken,
) -> Result<Self, AudioError> {
let response = tokio::select! {
_ = stop_token.cancelled() => {
return Err(AudioError::IoError("Cancelled before HTTP connect".into()));
}
result = reqwest::get(url) => {
result.map_err(|e| AudioError::IoError(format!("HTTP error: {}", e)))?
}
};
if !response.status().is_success() {
return Err(AudioError::IoError(format!(
"HTTP {} for {}",
response.status(),
url
)));
}
use futures::TryStreamExt;
use tokio_util::io::StreamReader;
let byte_stream = response
.bytes_stream()
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e));
let reader = StreamReader::new(byte_stream);
let stream = decode_audio_stream(reader)
.await
.map_err(|e| AudioError::ProcessingError(format!("Decode error: {}", e)))?;
let stream_info = stream.info().clone();
validate_stream(&stream_info)?;
let frames_to_skip = (seek_sec * stream_info.sample_rate as f64) as u64;
info!(
"UriSource: opened HTTP {} Hz {} ch {} bps",
stream_info.sample_rate, stream_info.channels, stream_info.bits_per_sample,
);
let (_, reader) = stream.into_reader();
Ok(Self { reader: Box::new(reader), stream_info, frames_to_skip })
}
}