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

Ajout d'un noeud puis vers le cache audio

Browse Source
This commit is contained in:
Eric Coissac
2025-11-03 06:55:45 +01:00
parent 7d907e7429
commit 157baadcfd
22 changed files with 1507 additions and 1162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
.DS_Store vendored
View File

Binary file not shown.

713
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

1
PMOMusic/Cargo.toml
View File

@@ -12,6 +12,7 @@ pmosource = { path = "../pmosource", features = ["server"] }
pmoserver = { path = "../pmoserver" }
pmocovers = { path = "../pmocovers", features = ["pmoserver"] }
pmoaudiocache = { path = "../pmoaudiocache", features = ["pmoserver"]}
pmoaudio-ext = { path = "../pmoaudio-ext", features = ["all"] }
pmoapp = { path = "../pmoapp", features = ["pmoserver"] }
tokio = { version = "1.35", features = ["rt-multi-thread", "macros", "sync", "time","signal"] }

2
pmoapp/src/lib.rs
View File

@@ -238,8 +238,6 @@
//! - [Vite Documentation](https://vitejs.dev/)
use rust_embed::RustEmbed;
use std::future::Future;
use std::pin::Pin;
/// Structure représentant l'application web embarquée.
///

30
pmoaudio-ext/Cargo.toml Normal file
View File

@@ -0,0 +1,30 @@
[package]
name = "pmoaudio-ext"
version = "0.1.0"
edition = "2021"
[dependencies]
# Core audio types
pmoaudio = { path = "../pmoaudio" }
# Optional dependencies for cache-sink feature
pmoaudiocache = { path = "../pmoaudiocache", optional = true }
pmoflac = { path = "../pmoflac", optional = true }
pmometadata = { path = "../pmometadata", optional = true }
# Optional dependency for playlist integration
pmoplaylist = { path = "../pmoplaylist", optional = true }
# Async runtime
tokio = { version = "1.0", features = ["full"] }
tokio-util = { version = "0.7" }
async-trait = "0.1"
# Utilities
tracing = "0.1"
[features]
default = []
cache-sink = ["dep:pmoaudiocache", "dep:pmoflac", "dep:pmometadata"]
playlist = ["dep:pmoplaylist"]
all = ["cache-sink", "playlist"]

30
pmoaudio-ext/src/lib.rs Normal file
View File

@@ -0,0 +1,30 @@
//! Extensions pour pmoaudio
//!
//! Cette crate fournit des nodes d'extension pour pmoaudio qui dépendent
//! d'autres crates du projet. Elle permet d'éviter les dépendances cycliques
//! en plaçant ces extensions en "bout de chaîne" de dépendances.
//!
//! # Features
//!
//! - `cache-sink` : Active le `FlacCacheSink` qui encode l'audio en FLAC et le stocke dans pmoaudiocache
//! - `playlist` : Active l'intégration avec pmoplaylist pour les sinks
//! - `all` : Active toutes les features d'un coup
//!
//! # Architecture
//!
//! Cette crate dépend de :
//! - `pmoaudio` : Types de base (AudioSegment, AudioError, etc.)
//! - `pmoaudiocache` (optionnel) : Cache audio pour le stockage FLAC
//! - `pmoflac` (optionnel) : Encodage FLAC
//! - `pmometadata` (optionnel) : Gestion des métadonnées
//! - `pmoplaylist` (optionnel) : Intégration playlist
//!
//! Aucune des crates ci-dessus ne dépend de `pmoaudio-ext`, évitant ainsi
//! tout cycle de dépendances.
#[cfg(feature = "cache-sink")]
pub mod sinks;
// Re-exports pour faciliter l'utilisation
#[cfg(feature = "cache-sink")]
pub use sinks::*;

111
pmoaudiocache/src/nodes/flac_cache_sink.rs → pmoaudio-ext/src/sinks/flac_cache_sink.rs
View File

@@ -1,11 +1,11 @@
//! Sink qui encode les AudioSegment au format FLAC et les stocke dans le cache audio
use crate::metadata_ext::AudioTrackMetadataExt;
use pmoaudio::{
nodes::{AudioError, TypedAudioNode, DEFAULT_CHANNEL_SIZE},
type_constraints::TypeRequirement,
AudioChunk, AudioSegment, SyncMarker, _AudioSegment,
AudioChunk, AudioPipelineNode, AudioSegment, SyncMarker, _AudioSegment,
};
use pmoaudiocache::AudioTrackMetadataExt;
use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat};
use std::{
collections::VecDeque,
@@ -18,6 +18,7 @@ use tokio::{
io::{self, AsyncRead, ReadBuf},
sync::{mpsc, RwLock},
};
use tokio_util::sync::CancellationToken;
/// Sink qui encode les `AudioSegment` reçus au format FLAC et les stocke dans le cache audio.
///
@@ -26,13 +27,17 @@ use tokio::{
/// - Crée une nouvelle entrée de cache pour chaque TrackBoundary rencontré
/// - Adapte automatiquement l'encodage FLAC selon la profondeur de bit du chunk (8/16/24/32-bit)
/// - Copie les métadonnées du TrackBoundary dans le cache après ingestion
/// - Peut optionnellement ajouter les tracks à une playlist via `register_playlist()`
/// - Termine l'encodage proprement quand il reçoit EndOfStream
pub struct FlacCacheSink {
tx: mpsc::Sender<Arc<AudioSegment>>,
rx: mpsc::Receiver<Arc<AudioSegment>>,
cache: Arc<crate::Cache>,
cache: Arc<pmoaudiocache::Cache>,
collection: Option<String>,
encoder_options: EncoderOptions,
pcm_buffer_capacity: usize,
#[cfg(feature = "playlist")]
playlist_handle: Option<Arc<pmoplaylist::WriteHandle>>,
}
impl FlacCacheSink {
@@ -41,7 +46,7 @@ impl FlacCacheSink {
/// # Arguments
///
/// * `cache` - Arc vers le cache audio où stocker les fichiers FLAC encodés
pub fn new(cache: Arc<crate::Cache>) -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
pub fn new(cache: Arc<pmoaudiocache::Cache>) -> Self {
Self::with_channel_size(cache, DEFAULT_CHANNEL_SIZE)
}
@@ -51,10 +56,7 @@ impl FlacCacheSink {
///
/// * `cache` - Arc vers le cache audio
/// * `channel_size` - Taille du buffer MPSC (nombre de segments en attente avant backpressure)
pub fn with_channel_size(
cache: Arc<crate::Cache>,
channel_size: usize,
) -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
pub fn with_channel_size(cache: Arc<pmoaudiocache::Cache>, channel_size: usize) -> Self {
Self::with_config(cache, channel_size, EncoderOptions::default(), None)
}
@@ -67,33 +69,51 @@ impl FlacCacheSink {
/// * `encoder_options` - Options d'encodage FLAC (compression, etc.)
/// * `collection` - Collection optionnelle à laquelle appartiennent les fichiers
pub fn with_config(
cache: Arc<crate::Cache>,
cache: Arc<pmoaudiocache::Cache>,
channel_size: usize,
encoder_options: EncoderOptions,
collection: Option<String>,
) -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
) -> Self {
let (tx, rx) = mpsc::channel(channel_size);
let sink = Self {
Self {
tx,
rx,
cache,
collection,
encoder_options,
pcm_buffer_capacity: 8,
};
(sink, tx)
#[cfg(feature = "playlist")]
playlist_handle: None,
}
}
/// Lance l'encodage et l'ingestion dans le cache.
/// Enregistre une playlist pour recevoir automatiquement les tracks sauvées dans le cache.
///
/// # Arguments
///
/// * `handle` - WriteHandle de la playlist qui recevra les pk des tracks
#[cfg(feature = "playlist")]
pub fn register_playlist(&mut self, handle: pmoplaylist::WriteHandle) {
self.playlist_handle = Some(Arc::new(handle));
}
/// Lance l'encodage et l'ingestion dans le cache (version interne).
///
/// Cette méthode crée une nouvelle entrée de cache pour chaque TrackBoundary rencontré.
/// Les métadonnées du TrackBoundary sont copiées dans le cache après l'ingestion.
pub async fn run(self) -> Result<FlacCacheSinkStats, AudioError> {
async fn run_internal(
self,
stop_token: CancellationToken,
) -> Result<FlacCacheSinkStats, AudioError> {
let FlacCacheSink {
tx: _,
mut rx,
cache,
collection,
encoder_options,
pcm_buffer_capacity,
#[cfg(feature = "playlist")]
playlist_handle,
} = self;
let mut all_tracks = Vec::new();
@@ -102,7 +122,7 @@ impl FlacCacheSink {
loop {
// Attendre le premier chunk audio pour cette track, en capturant les métadonnées du TrackBoundary
let (first_segment, track_metadata) =
match wait_for_first_audio_chunk_with_metadata(&mut rx).await {
match wait_for_first_audio_chunk_with_metadata(&mut rx, &stop_token).await {
Ok(result) => result,
Err(_) => {
// Plus d'audio disponible
@@ -157,6 +177,7 @@ impl FlacCacheSink {
pcm_tx,
bits_per_sample,
sample_rate,
&stop_token,
);
let copy_future = async {
tokio::io::copy(&mut flac_stream, &mut flac_buffer)
@@ -200,6 +221,15 @@ impl FlacCacheSink {
})?;
}
// Ajouter à la playlist si enregistrée
#[cfg(feature = "playlist")]
if let Some(ref playlist_handle) = playlist_handle {
playlist_handle.push(pk.clone()).await
.map_err(|e| AudioError::ProcessingError(
format!("Failed to add to playlist: {}", e)
))?;
}
// Ajouter les stats de cette track
all_tracks.push(TrackStats {
pk,
@@ -238,6 +268,7 @@ enum StopReason {
/// Retourne une erreur si EndOfStream est reçu avant tout audio.
async fn wait_for_first_audio_chunk_with_metadata(
rx: &mut mpsc::Receiver<Arc<AudioSegment>>,
stop_token: &CancellationToken,
) -> Result<
(
Arc<AudioSegment>,
@@ -248,10 +279,14 @@ async fn wait_for_first_audio_chunk_with_metadata(
let mut track_metadata: Option<Arc<RwLock<dyn pmometadata::TrackMetadata>>> = None;
loop {
let segment = rx
.recv()
.await
.ok_or_else(|| AudioError::ProcessingError("No audio data received".into()))?;
let segment = tokio::select! {
result = rx.recv() => {
result.ok_or_else(|| AudioError::ProcessingError("No audio data received".into()))?
}
_ = stop_token.cancelled() => {
return Err(AudioError::ProcessingError("Cancelled".into()));
}
};
match &segment.segment {
_AudioSegment::Chunk(chunk) => {
@@ -260,8 +295,8 @@ async fn wait_for_first_audio_chunk_with_metadata(
}
return Ok((segment, track_metadata));
}
_AudioSegment::Sync(marker) => match **marker {
SyncMarker::TrackBoundary { ref metadata, .. } => {
_AudioSegment::Sync(marker) => match &**marker {
SyncMarker::TrackBoundary { metadata, .. } => {
// Capturer les métadonnées du TrackBoundary
track_metadata = Some(metadata.clone());
continue;
@@ -287,6 +322,7 @@ async fn pump_track_segments(
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;
@@ -308,12 +344,20 @@ async fn pump_track_segments(
// Boucle sur les segments suivants
loop {
let segment = match rx.recv().await {
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 {
@@ -327,7 +371,7 @@ async fn pump_track_segments(
)));
}
let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?;
let pcm_bytes = chunk_to_pcm_bytes(&chunk, bits_per_sample)?;
if pcm_bytes.is_empty() {
continue;
}
@@ -546,6 +590,25 @@ pub struct FlacCacheSinkStats {
pub tracks: Vec<TrackStats>,
}
#[async_trait::async_trait]
impl AudioPipelineNode for FlacCacheSink {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
Some(self.tx.clone())
}
fn register(&mut self, _child: Box<dyn AudioPipelineNode>) {
panic!("FlacCacheSink is a terminal sink and cannot have children");
}
async fn run(
self: Box<Self>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
self.run_internal(stop_token).await?;
Ok(())
}
}
impl TypedAudioNode for FlacCacheSink {
fn input_type(&self) -> Option<TypeRequirement> {
// FlacCacheSink accepte n'importe quel type entier (I16, I24, I32)

11
pmoaudio-ext/src/sinks/mod.rs Normal file
View File

@@ -0,0 +1,11 @@
//! Sinks d'extension pour pmoaudio
//!
//! Ce module contient des sinks qui dépendent de multiples crates
//! et ne peuvent pas être placés directement dans pmoaudio sans créer
//! de dépendances cycliques.
#[cfg(feature = "cache-sink")]
mod flac_cache_sink;
#[cfg(feature = "cache-sink")]
pub use flac_cache_sink::{FlacCacheSink, FlacCacheSinkStats, TrackStats};

1
pmoaudio/Cargo.toml
View File

@@ -18,6 +18,7 @@ paste = "1"
soxr = "0.6.0"
bytemuck = "1.24.0"
reqwest = { version = "0.12", features = ["stream"] }
tracing = "0.1"
[dev-dependencies]
tokio-test = "0.4"

65
pmoaudio/examples/file_nodes_test.rs
View File

@@ -1,14 +1,20 @@
//! Test d'intégration pour FileSource et FlacFileSink
//! Test d'intégration pour FileSource et FlacFileSink avec la nouvelle architecture AudioPipelineNode
//!
//! Ce programme teste la chaîne complète :
//! 1. Lecture d'un fichier audio avec FileSource
//! 2. Écriture vers FLAC avec FlacFileSink
//!
//! La nouvelle architecture permet de :
//! - Construire le pipeline en enregistrant des enfants avec register()
//! - Lancer tout le pipeline avec un seul appel à run() sur la racine
//! - Arrêter proprement tout le pipeline avec un CancellationToken
//!
//! Usage:
//! cargo run --example file_nodes_test -- <input_file> <output_file>
use pmoaudio::{FileSource, FlacFileSink};
use pmoaudio::{AudioPipelineNode, FileSource, FlacFileSink};
use std::env;
use tokio_util::sync::CancellationToken;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -28,48 +34,35 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
println!();
// Créer le pipeline: FileSource → FlacFileSink
let mut source = FileSource::new(input_path); // Calcul automatique de la taille des chunks (~50ms)
let (sink, tx) = FlacFileSink::new(output_path); // Utilise le buffer par défaut (16 segments)
let mut source = FileSource::new(input_path);
let sink = FlacFileSink::new(output_path);
source.add_subscriber(tx);
// Enregistrer le sink comme enfant de la source
source.register(Box::new(sink));
// Lancer le sink dans une tâche séparée
let sink_handle = tokio::spawn(async move {
println!("FlacFileSink started");
let result = sink.run().await;
println!("FlacFileSink finished");
result
});
// Créer un token d'arrêt pour contrôle manuel si besoin
let stop_token = CancellationToken::new();
// Lancer le source
println!("FileSource started");
let source_result = source.run().await;
println!("FileSource finished");
// Lancer tout le pipeline - run() spawne automatiquement tous les enfants
println!("Pipeline started");
println!(" FileSource: reading from {}", input_path);
println!(" FlacFileSink: writing to {}", output_path);
// Vérifier les résultats
match source_result {
Ok(()) => println!("✓ FileSource completed successfully"),
let result = Box::new(source).run(stop_token).await;
// Vérifier le résultat
match result {
Ok(()) => {
println!();
println!("✓ Pipeline completed successfully");
println!(" Output file: {}", output_path);
}
Err(e) => {
eprintln!("✗ FileSource error: {}", e);
eprintln!();
eprintln!("✗ Pipeline error: {}", e);
return Err(e.into());
}
}
let stats = sink_handle.await??;
println!("✓ FlacFileSink completed successfully");
println!();
println!("Statistics:");
println!(" Tracks written: {}", stats.tracks.len());
for (i, track) in stats.tracks.iter().enumerate() {
println!(" Track {}:", i);
println!(" Output file: {:?}", track.path);
println!(" Chunks received: {}", track.chunks_received);
println!(" Total samples: {}", track.total_samples);
println!(
" Duration: {:.2} seconds",
track.total_duration_sec
);
}
Ok(())
}

6
pmoaudio/src/lib.rs
View File

@@ -85,6 +85,7 @@ mod audio_segment;
pub mod conversions;
pub mod events;
pub mod nodes;
pub mod pipeline;
mod sample_types;
mod sync_marker;
pub mod type_constraints;
@@ -112,13 +113,16 @@ pub use events::{
VolumeChangeEvent,
};
// Export du trait de pipeline
pub use pipeline::AudioPipelineNode;
// Exports publics des nodes
pub use nodes::{
converter_nodes::{ToF32Node, ToF64Node, ToI16Node, ToI24Node, ToI32Node},
file_source::FileSource,
flac_file_sink::{FlacFileSink, FlacFileSinkStats},
http_source::HttpSource,
AudioError, AudioNode, MultiSubscriberNode, SingleSubscriberNode, TypedAudioNode,
AudioError, AudioNode, TypedAudioNode,
};
// Nodes temporairement désactivés

512
pmoaudio/src/nodes/converter_nodes.rs
View File

@@ -8,55 +8,86 @@
//! les incompatibilités de type entre producers et consumers.
use crate::{
nodes::{AudioError, MultiSubscriberNode, TypedAudioNode},
nodes::{AudioError, TypedAudioNode},
type_constraints::{SampleType, TypeRequirement},
AudioSegment,
AudioPipelineNode, AudioSegment,
};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
/// Node de conversion vers I16
///
/// Convertit n'importe quel type de chunk audio vers I16 (16-bit signed integer).
/// Utilise les conversions DSP SIMD optimisées.
pub struct ToI16Node {
// Macro pour générer les converter nodes avec la nouvelle architecture AudioPipelineNode
macro_rules! converter_node {
($node_name:ident, $convert_method:ident, $output_type:expr, $doc:expr) => {
#[doc = $doc]
pub struct $node_name {
tx: mpsc::Sender<Arc<AudioSegment>>,
rx: mpsc::Receiver<Arc<AudioSegment>>,
subscribers: MultiSubscriberNode,
}
child_txs: Vec<mpsc::Sender<Arc<AudioSegment>>>,
children: Vec<Box<dyn AudioPipelineNode>>,
}
impl ToI16Node {
/// Crée un nouveau node de conversion vers I16
pub fn new() -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
impl $node_name {
/// Crée un nouveau node de conversion
pub fn new() -> Self {
Self::with_channel_size(16)
}
/// Crée un nouveau node avec une taille de buffer spécifique
pub fn with_channel_size(channel_size: usize) -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
pub fn with_channel_size(channel_size: usize) -> Self {
let (tx, rx) = mpsc::channel(channel_size);
let node = Self {
Self {
tx,
rx,
subscribers: MultiSubscriberNode::new(),
child_txs: Vec::new(),
children: Vec::new(),
}
}
}
#[async_trait::async_trait]
impl AudioPipelineNode for $node_name {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
Some(self.tx.clone())
}
fn register(&mut self, child: Box<dyn AudioPipelineNode>) {
if let Some(tx) = child.get_tx() {
self.child_txs.push(tx);
}
self.children.push(child);
}
async fn run(
mut self: Box<Self>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
// Spawner tous les enfants
let mut child_handles = Vec::new();
for child in self.children {
let child_token = stop_token.child_token();
let handle = tokio::spawn(async move { child.run(child_token).await });
child_handles.push(handle);
}
// Boucle de traitement
loop {
let segment = tokio::select! {
result = self.rx.recv() => {
match result {
Some(seg) => seg,
None => break,
}
}
_ = stop_token.cancelled() => {
break;
}
};
(node, tx)
}
/// Ajoute un abonné qui recevra les segments audio convertis
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioSegment>>) {
self.subscribers.add_subscriber(tx);
}
/// Lance le traitement de conversion
pub async fn run(mut self) -> Result<(), AudioError> {
while let Some(segment) = self.rx.recv().await {
// Si c'est un syncmarker, passer directement
if !segment.is_audio_chunk() {
self.subscribers.push(segment).await?;
continue;
}
// Convertir le chunk audio vers I16
let converted_segment = if let Some(chunk) = segment.as_chunk() {
let converted_chunk = chunk.to_i16();
// Convertir si c'est un chunk audio, sinon passer tel quel
let output_segment = if segment.is_audio_chunk() {
if let Some(chunk) = segment.as_chunk() {
let converted_chunk = chunk.$convert_method();
Arc::new(AudioSegment {
order: segment.order,
timestamp_sec: segment.timestamp_sec,
@@ -64,320 +95,87 @@ impl ToI16Node {
})
} else {
segment
};
self.subscribers.push(converted_segment).await?;
}
Ok(())
}
}
impl TypedAudioNode for ToI16Node {
fn input_type(&self) -> Option<TypeRequirement> {
// Accepte n'importe quel type
Some(TypeRequirement::any())
}
fn output_type(&self) -> Option<TypeRequirement> {
// Produit uniquement I16
Some(TypeRequirement::specific(SampleType::I16))
}
}
impl Default for ToI16Node {
fn default() -> Self {
Self::new().0
}
}
/// Node de conversion vers I24
///
/// Convertit n'importe quel type de chunk audio vers I24 (24-bit signed integer).
/// Utilise les conversions DSP SIMD optimisées.
pub struct ToI24Node {
rx: mpsc::Receiver<Arc<AudioSegment>>,
subscribers: MultiSubscriberNode,
}
impl ToI24Node {
/// Crée un nouveau node de conversion vers I24
pub fn new() -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
Self::with_channel_size(16)
}
/// Crée un nouveau node avec une taille de buffer spécifique
pub fn with_channel_size(channel_size: usize) -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
let (tx, rx) = mpsc::channel(channel_size);
let node = Self {
rx,
subscribers: MultiSubscriberNode::new(),
};
(node, tx)
}
/// Ajoute un abonné qui recevra les segments audio convertis
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioSegment>>) {
self.subscribers.add_subscriber(tx);
}
/// Lance le traitement de conversion
pub async fn run(mut self) -> Result<(), AudioError> {
while let Some(segment) = self.rx.recv().await {
if !segment.is_audio_chunk() {
self.subscribers.push(segment).await?;
continue;
}
let converted_segment = if let Some(chunk) = segment.as_chunk() {
let converted_chunk = chunk.to_i24();
Arc::new(AudioSegment {
order: segment.order,
timestamp_sec: segment.timestamp_sec,
segment: crate::_AudioSegment::Chunk(Arc::new(converted_chunk)),
})
} else {
segment
};
self.subscribers.push(converted_segment).await?;
// Envoyer à tous les enfants
for tx in &self.child_txs {
if tx.send(output_segment.clone()).await.is_err() {
// Un enfant est mort, arrêter
break;
}
}
}
// Attendre que tous les enfants se terminent
for handle in child_handles {
match handle.await {
Ok(Ok(())) => {}
Ok(Err(e)) => return Err(e),
Err(e) => {
return Err(AudioError::ProcessingError(format!(
"Child task panicked: {}",
e
)))
}
}
}
Ok(())
}
}
}
impl TypedAudioNode for ToI24Node {
impl TypedAudioNode for $node_name {
fn input_type(&self) -> Option<TypeRequirement> {
Some(TypeRequirement::any())
}
fn output_type(&self) -> Option<TypeRequirement> {
Some(TypeRequirement::specific(SampleType::I24))
Some(TypeRequirement::specific($output_type))
}
}
}
impl Default for ToI24Node {
impl Default for $node_name {
fn default() -> Self {
Self::new().0
Self::new()
}
}
/// Node de conversion vers I32
///
/// Convertit n'importe quel type de chunk audio vers I32 (32-bit signed integer).
/// Utilise les conversions DSP SIMD optimisées.
pub struct ToI32Node {
rx: mpsc::Receiver<Arc<AudioSegment>>,
subscribers: MultiSubscriberNode,
}
impl ToI32Node {
/// Crée un nouveau node de conversion vers I32
pub fn new() -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
Self::with_channel_size(16)
}
/// Crée un nouveau node avec une taille de buffer spécifique
pub fn with_channel_size(channel_size: usize) -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
let (tx, rx) = mpsc::channel(channel_size);
let node = Self {
rx,
subscribers: MultiSubscriberNode::new(),
};
(node, tx)
}
/// Ajoute un abonné qui recevra les segments audio convertis
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioSegment>>) {
self.subscribers.add_subscriber(tx);
}
/// Lance le traitement de conversion
pub async fn run(mut self) -> Result<(), AudioError> {
while let Some(segment) = self.rx.recv().await {
if !segment.is_audio_chunk() {
self.subscribers.push(segment).await?;
continue;
}
let converted_segment = if let Some(chunk) = segment.as_chunk() {
let converted_chunk = chunk.to_i32();
Arc::new(AudioSegment {
order: segment.order,
timestamp_sec: segment.timestamp_sec,
segment: crate::_AudioSegment::Chunk(Arc::new(converted_chunk)),
})
} else {
segment
};
self.subscribers.push(converted_segment).await?;
}
Ok(())
}
}
impl TypedAudioNode for ToI32Node {
fn input_type(&self) -> Option<TypeRequirement> {
Some(TypeRequirement::any())
}
fn output_type(&self) -> Option<TypeRequirement> {
Some(TypeRequirement::specific(SampleType::I32))
}
}
impl Default for ToI32Node {
fn default() -> Self {
Self::new().0
}
}
/// Node de conversion vers F32
///
/// Convertit n'importe quel type de chunk audio vers F32 (32-bit floating point).
/// Utilise les conversions DSP SIMD optimisées.
pub struct ToF32Node {
rx: mpsc::Receiver<Arc<AudioSegment>>,
subscribers: MultiSubscriberNode,
}
impl ToF32Node {
/// Crée un nouveau node de conversion vers F32
pub fn new() -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
Self::with_channel_size(16)
}
/// Crée un nouveau node avec une taille de buffer spécifique
pub fn with_channel_size(channel_size: usize) -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
let (tx, rx) = mpsc::channel(channel_size);
let node = Self {
rx,
subscribers: MultiSubscriberNode::new(),
};
(node, tx)
}
/// Ajoute un abonné qui recevra les segments audio convertis
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioSegment>>) {
self.subscribers.add_subscriber(tx);
}
/// Lance le traitement de conversion
pub async fn run(mut self) -> Result<(), AudioError> {
while let Some(segment) = self.rx.recv().await {
if !segment.is_audio_chunk() {
self.subscribers.push(segment).await?;
continue;
}
let converted_segment = if let Some(chunk) = segment.as_chunk() {
let converted_chunk = chunk.to_f32();
Arc::new(AudioSegment {
order: segment.order,
timestamp_sec: segment.timestamp_sec,
segment: crate::_AudioSegment::Chunk(Arc::new(converted_chunk)),
})
} else {
segment
};
self.subscribers.push(converted_segment).await?;
}
Ok(())
}
}
impl TypedAudioNode for ToF32Node {
fn input_type(&self) -> Option<TypeRequirement> {
Some(TypeRequirement::any())
}
fn output_type(&self) -> Option<TypeRequirement> {
Some(TypeRequirement::specific(SampleType::F32))
}
}
impl Default for ToF32Node {
fn default() -> Self {
Self::new().0
}
}
/// Node de conversion vers F64
///
/// Convertit n'importe quel type de chunk audio vers F64 (64-bit floating point).
/// Utilise les conversions DSP SIMD optimisées.
pub struct ToF64Node {
rx: mpsc::Receiver<Arc<AudioSegment>>,
subscribers: MultiSubscriberNode,
}
impl ToF64Node {
/// Crée un nouveau node de conversion vers F64
pub fn new() -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
Self::with_channel_size(16)
}
/// Crée un nouveau node avec une taille de buffer spécifique
pub fn with_channel_size(channel_size: usize) -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
let (tx, rx) = mpsc::channel(channel_size);
let node = Self {
rx,
subscribers: MultiSubscriberNode::new(),
};
(node, tx)
}
/// Ajoute un abonné qui recevra les segments audio convertis
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioSegment>>) {
self.subscribers.add_subscriber(tx);
}
/// Lance le traitement de conversion
pub async fn run(mut self) -> Result<(), AudioError> {
while let Some(segment) = self.rx.recv().await {
if !segment.is_audio_chunk() {
self.subscribers.push(segment).await?;
continue;
}
let converted_segment = if let Some(chunk) = segment.as_chunk() {
let converted_chunk = chunk.to_f64();
Arc::new(AudioSegment {
order: segment.order,
timestamp_sec: segment.timestamp_sec,
segment: crate::_AudioSegment::Chunk(Arc::new(converted_chunk)),
})
} else {
segment
};
self.subscribers.push(converted_segment).await?;
}
Ok(())
}
}
impl TypedAudioNode for ToF64Node {
fn input_type(&self) -> Option<TypeRequirement> {
Some(TypeRequirement::any())
}
fn output_type(&self) -> Option<TypeRequirement> {
Some(TypeRequirement::specific(SampleType::F64))
}
}
impl Default for ToF64Node {
fn default() -> Self {
Self::new().0
}
}
// Générer les 5 converter nodes
converter_node!(
ToI16Node,
to_i16,
SampleType::I16,
"Node de conversion vers I16 (16-bit signed integer)"
);
converter_node!(
ToI24Node,
to_i24,
SampleType::I24,
"Node de conversion vers I24 (24-bit signed integer)"
);
converter_node!(
ToI32Node,
to_i32,
SampleType::I32,
"Node de conversion vers I32 (32-bit signed integer)"
);
converter_node!(
ToF32Node,
to_f32,
SampleType::F32,
"Node de conversion vers F32 (32-bit floating point)"
);
converter_node!(
ToF64Node,
to_f64,
SampleType::F64,
"Node de conversion vers F64 (64-bit floating point)"
);
#[cfg(test)]
mod tests {
@@ -386,7 +184,7 @@ mod tests {
#[tokio::test]
async fn test_to_f32_node_type_requirements() {
let (node, _tx) = ToF32Node::new();
let node = ToF32Node::new();
// Vérifier les types d'entrée/sortie
assert_eq!(
@@ -405,14 +203,60 @@ mod tests {
);
}
// Nœud de test simple qui collecte les segments
struct TestCollectorNode {
input_tx: mpsc::Sender<Arc<AudioSegment>>,
input_rx: mpsc::Receiver<Arc<AudioSegment>>,
output_tx: mpsc::Sender<Arc<AudioSegment>>,
}
impl TestCollectorNode {
fn new(output_tx: mpsc::Sender<Arc<AudioSegment>>) -> Self {
let (input_tx, input_rx) = mpsc::channel(16);
Self {
input_tx,
input_rx,
output_tx,
}
}
}
#[async_trait::async_trait]
impl AudioPipelineNode for TestCollectorNode {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
Some(self.input_tx.clone())
}
fn register(&mut self, _child: Box<dyn AudioPipelineNode>) {
panic!("TestCollectorNode is a sink");
}
async fn run(
mut self: Box<Self>,
_stop_token: CancellationToken,
) -> Result<(), AudioError> {
while let Some(segment) = self.input_rx.recv().await {
if self.output_tx.send(segment).await.is_err() {
break;
}
}
Ok(())
}
}
#[tokio::test]
async fn test_to_i16_node_converts_from_i32() {
let (mut node, tx) = ToI16Node::new();
let mut node = ToI16Node::new();
let (out_tx, mut out_rx) = mpsc::channel(16);
node.add_subscriber(out_tx);
let collector = TestCollectorNode::new(out_tx);
node.register(Box::new(collector));
// Récupérer le tx du node
let tx = node.get_tx().unwrap();
// Lancer le node dans une tâche
let handle = tokio::spawn(async move { node.run().await });
let stop_token = CancellationToken::new();
let handle = tokio::spawn(async move { Box::new(node).run(stop_token).await });
// Créer et envoyer un chunk I32
let stereo = vec![[1_000_000i32 << 16, -500_000i32 << 16]; 100];
@@ -451,12 +295,16 @@ mod tests {
#[tokio::test]
async fn test_syncmarkers_passthrough() {
let (mut node, tx) = ToI16Node::new();
let mut node = ToI16Node::new();
let (out_tx, mut out_rx) = mpsc::channel(16);
node.add_subscriber(out_tx);
let collector = TestCollectorNode::new(out_tx);
node.register(Box::new(collector));
let tx = node.get_tx().unwrap();
let stop_token = CancellationToken::new();
tokio::spawn(async move {
node.run().await.unwrap();
Box::new(node).run(stop_token).await.unwrap();
});
// Envoyer un syncmarker

444
pmoaudio/src/nodes/file_source.rs
View File

@@ -1,5 +1,6 @@
use crate::{
nodes::{AudioError, MultiSubscriberNode, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS},
nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS},
pipeline::AudioPipelineNode,
type_constraints::TypeRequirement,
AudioChunk, AudioChunkData, AudioSegment, I24,
};
@@ -7,6 +8,8 @@ use pmoflac::{decode_audio_stream, AudioFileMetadata, StreamInfo};
use pmometadata::{MemoryTrackMetadata, TrackMetadata};
use std::{path::PathBuf, sync::Arc, time::Duration};
use tokio::{fs::File, io::AsyncReadExt, sync::mpsc};
use tokio_util::sync::CancellationToken;
use tracing;
/// FileSource - Lit un fichier audio et publie des `AudioSegment`
///
@@ -21,7 +24,8 @@ use tokio::{fs::File, io::AsyncReadExt, sync::mpsc};
pub struct FileSource {
path: PathBuf,
chunk_frames: usize,
subscribers: MultiSubscriberNode,
child_txs: Vec<mpsc::Sender<Arc<AudioSegment>>>,
children: Vec<Box<dyn AudioPipelineNode>>,
}
impl FileSource {
@@ -43,156 +47,10 @@ impl FileSource {
Self {
path: path.into(),
chunk_frames,
subscribers: MultiSubscriberNode::new(),
child_txs: Vec::new(),
children: Vec::new(),
}
}
/// Ajoute un abonné qui recevra les segments audio.
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioSegment>>) {
self.subscribers.add_subscriber(tx);
}
/// Lance la lecture du fichier et diffuse les segments audio.
pub async fn run(self) -> Result<(), AudioError> {
// Ouvrir le fichier
let file = File::open(&self.path).await.map_err(|e| {
AudioError::ProcessingError(format!("Failed to open {:?}: {}", self.path, e))
})?;
// Décoder le flux audio
let mut 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)?;
// Calculer la taille des chunks si non spécifiée (0 = auto)
let chunk_frames = if self.chunk_frames == 0 {
// Calculer pour obtenir DEFAULT_CHUNK_DURATION_MS millisecondes
let frames =
(stream_info.sample_rate as f64 * DEFAULT_CHUNK_DURATION_MS / 1000.0) as usize;
// Arrondir à la puissance de 2 la plus proche pour optimiser les buffers
frames.next_power_of_two().max(256)
} else {
self.chunk_frames.max(1)
};
// Émettre TopZeroSync
let top_zero = AudioSegment::new_top_zero_sync();
self.subscribers.push(top_zero).await?;
// Extraire et émettre les métadonnées du fichier
match AudioFileMetadata::from_file(&self.path) {
Ok(file_metadata) => {
let mut metadata = MemoryTrackMetadata::new();
// Convertir AudioFileMetadata vers MemoryTrackMetadata
if let Some(title) = file_metadata.title {
let _ = metadata.set_title(Some(title)).await;
}
if let Some(artist) = file_metadata.artist {
let _ = metadata.set_artist(Some(artist)).await;
}
if let Some(album) = file_metadata.album {
let _ = metadata.set_album(Some(album)).await;
}
if let Some(year) = file_metadata.year {
let _ = metadata.set_year(Some(year)).await;
}
if let Some(duration_secs) = file_metadata.duration_secs {
let _ = metadata
.set_duration(Some(Duration::from_secs(duration_secs)))
.await;
}
// Émettre TrackBoundary
let track_boundary = AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata)));
self.subscribers.push(track_boundary).await?;
}
Err(e) => {
eprintln!(
"Warning: Failed to extract metadata from {:?}: {}",
self.path, e
);
// Continuer sans métadonnées
}
}
// Préparer la lecture des chunks audio
let frame_bytes = stream_info.bytes_per_sample() * stream_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;
// Lire et émettre les chunks audio
loop {
// Remplir le buffer
if pending.len() < chunk_byte_len {
let read = stream.read(&mut read_buf).await.map_err(|e| {
AudioError::ProcessingError(format!("I/O error while decoding: {}", e))
})?;
if read == 0 {
break;
}
pending.extend_from_slice(&read_buf[..read]);
}
if pending.is_empty() {
break;
}
// Extraire un chunk
let frames_in_pending = pending.len() / frame_bytes;
let frames_to_emit = frames_in_pending.min(chunk_frames);
let take_bytes = frames_to_emit * frame_bytes;
let chunk_bytes = pending.drain(..take_bytes).collect::<Vec<u8>>();
// Calculer le timestamp
let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64;
// Créer le segment audio
let segment = bytes_to_segment(
&chunk_bytes,
&stream_info,
frames_to_emit,
chunk_index,
timestamp_sec,
)?;
self.subscribers.push(segment).await?;
chunk_index += 1;
total_frames += frames_to_emit as u64;
}
// Traiter le reste éventuel (moins qu'un chunk complet)
if !pending.is_empty() {
let frames = pending.len() / frame_bytes;
if frames > 0 {
let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64;
let segment =
bytes_to_segment(&pending, &stream_info, frames, chunk_index, timestamp_sec)?;
self.subscribers.push(segment).await?;
total_frames += frames as u64;
chunk_index += 1;
}
}
// Émettre EndOfStream
let final_timestamp = total_frames as f64 / stream_info.sample_rate as f64;
let eos = AudioSegment::new_end_of_stream(chunk_index, final_timestamp);
self.subscribers.push(eos).await?;
// Attendre la fin du décodage
stream
.wait()
.await
.map_err(|e| AudioError::ProcessingError(format!("Decode task failed: {}", e)))?;
Ok(())
}
}
fn validate_stream(info: &StreamInfo) -> Result<(), AudioError> {
@@ -330,6 +188,230 @@ fn bytes_to_segment(
}))
}
#[async_trait::async_trait]
impl AudioPipelineNode for FileSource {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
// FileSource est une source, elle n'a pas d'input
None
}
fn register(&mut self, child: Box<dyn AudioPipelineNode>) {
// Extraire le tx du child avant de le stocker
if let Some(tx) = child.get_tx() {
self.child_txs.push(tx.clone());
}
self.children.push(child);
}
async fn run(
self: Box<Self>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
let Self {
path,
chunk_frames,
child_txs,
children,
} = *self;
// 1. Spawner tous les enfants AVANT de commencer à lire
let mut child_handles = Vec::new();
for child in children {
let child_token = stop_token.child_token();
let handle = tokio::spawn(async move {
child.run(child_token).await
});
child_handles.push(handle);
}
// 2. Faire le travail de lecture du fichier
let work_result: Result<(), AudioError> = async {
// Macro helper pour envoyer à tous les enfants
macro_rules! send_to_children {
($segment:expr) => {
for tx in &child_txs {
if tx.send($segment.clone()).await.is_err() {
tracing::warn!("FileSource: child died during send");
return Err(AudioError::SendError);
}
}
};
}
// Ouvrir le fichier
let file = File::open(&path).await.map_err(|e| {
AudioError::ProcessingError(format!("Failed to open {:?}: {}", path, e))
})?;
// Décoder le flux audio
let mut 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)?;
// Calculer la taille des chunks si non spécifiée (0 = auto)
let chunk_frames = if chunk_frames == 0 {
let frames =
(stream_info.sample_rate as f64 * DEFAULT_CHUNK_DURATION_MS / 1000.0) as usize;
frames.next_power_of_two().max(256)
} else {
chunk_frames.max(1)
};
// Émettre TopZeroSync
let top_zero = AudioSegment::new_top_zero_sync();
send_to_children!(top_zero);
// Extraire et émettre les métadonnées du fichier
if let Ok(file_metadata) = AudioFileMetadata::from_file(&path) {
let mut metadata = MemoryTrackMetadata::new();
if let Some(title) = file_metadata.title {
let _ = metadata.set_title(Some(title)).await;
}
if let Some(artist) = file_metadata.artist {
let _ = metadata.set_artist(Some(artist)).await;
}
if let Some(album) = file_metadata.album {
let _ = metadata.set_album(Some(album)).await;
}
if let Some(year) = file_metadata.year {
let _ = metadata.set_year(Some(year)).await;
}
if let Some(duration_secs) = file_metadata.duration_secs {
let _ = metadata
.set_duration(Some(Duration::from_secs(duration_secs)))
.await;
}
let track_boundary = AudioSegment::new_track_boundary(
0,
0.0,
Arc::new(tokio::sync::RwLock::new(metadata)),
);
send_to_children!(track_boundary);
}
// Préparer la lecture des chunks audio
let frame_bytes = stream_info.bytes_per_sample() * stream_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;
// Lire et émettre les chunks audio
loop {
tokio::select! {
_ = stop_token.cancelled() => {
tracing::info!("FileSource: stop requested");
break;
}
read_result = stream.read(&mut read_buf) => {
// Remplir le buffer
if pending.len() < chunk_byte_len {
let read = read_result.map_err(|e| {
AudioError::ProcessingError(format!("I/O error while decoding: {}", e))
})?;
if read == 0 && pending.is_empty() {
break;
}
if read > 0 {
pending.extend_from_slice(&read_buf[..read]);
}
}
if pending.is_empty() {
break;
}
// Extraire un chunk
let frames_in_pending = pending.len() / frame_bytes;
let frames_to_emit = frames_in_pending.min(chunk_frames);
if frames_to_emit == 0 {
break;
}
let take_bytes = frames_to_emit * frame_bytes;
let chunk_bytes = pending.drain(..take_bytes).collect::<Vec<u8>>();
// Calculer le timestamp
let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64;
// Créer et envoyer le segment audio
let segment = bytes_to_segment(
&chunk_bytes,
&stream_info,
frames_to_emit,
chunk_index,
timestamp_sec,
)?;
send_to_children!(segment);
chunk_index += 1;
total_frames += frames_to_emit as u64;
}
}
}
// Traiter le reste éventuel (moins qu'un chunk complet)
if !pending.is_empty() {
let frames = pending.len() / frame_bytes;
if frames > 0 {
let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64;
let segment =
bytes_to_segment(&pending, &stream_info, frames, chunk_index, timestamp_sec)?;
send_to_children!(segment);
total_frames += frames as u64;
chunk_index += 1;
}
}
// Émettre EndOfStream
let final_timestamp = total_frames as f64 / stream_info.sample_rate as f64;
let eos = AudioSegment::new_end_of_stream(chunk_index, final_timestamp);
send_to_children!(eos);
// Attendre la fin du décodage
stream
.wait()
.await
.map_err(|e| AudioError::ProcessingError(format!("Decode task failed: {}", e)))?;
Ok(())
}.await;
// 3. Arrêter les enfants qui tournent encore (descendant uniquement)
stop_token.cancel();
// 4. Fermer les channels pour signaler EOF
drop(child_txs);
// 5. Attendre que TOUS les enfants se terminent
for handle in child_handles {
match handle.await {
Ok(Ok(())) => {
// Enfant terminé normalement
}
Ok(Err(e)) => {
// Enfant en erreur → propager
tracing::error!("FileSource: child error: {}", e);
return Err(e);
}
Err(e) => {
// Panic dans l'enfant
tracing::error!("FileSource: child panic: {}", e);
return Err(AudioError::ProcessingError(format!("Child panic: {}", e)));
}
}
}
// 6. Retourner notre propre résultat (montant vers le parent)
work_result
}
}
impl TypedAudioNode for FileSource {
fn input_type(&self) -> Option<TypeRequirement> {
// FileSource est une source, elle ne consomme pas d'audio
@@ -384,12 +466,68 @@ mod tests {
file.flush().await.expect("flush file");
flac_stream.wait().await.unwrap();
// Créer un collecteur simple qui transmet les segments à un channel de test
struct TestCollectorNode {
tx: mpsc::Sender<Arc<AudioSegment>>,
rx: mpsc::Receiver<Arc<AudioSegment>>,
test_tx: mpsc::Sender<Arc<AudioSegment>>,
}
impl TestCollectorNode {
fn new(test_tx: mpsc::Sender<Arc<AudioSegment>>) -> Self {
let (tx, rx) = mpsc::channel(16);
Self {
tx,
rx,
test_tx,
}
}
}
#[async_trait::async_trait]
impl AudioPipelineNode for TestCollectorNode {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
Some(self.tx.clone())
}
fn register(&mut self, _child: Box<dyn AudioPipelineNode>) {
panic!("TestCollectorNode is a terminal node");
}
async fn run(
mut self: Box<Self>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
// Transférer tous les segments au test
loop {
tokio::select! {
_ = stop_token.cancelled() => {
break;
}
segment = self.rx.recv() => {
match segment {
Some(seg) => {
if self.test_tx.send(seg).await.is_err() {
break;
}
}
None => break,
}
}
}
}
Ok(())
}
}
let (test_tx, mut rx) = mpsc::channel(1024);
let mut source = FileSource::with_chunk_size(&flac_path, 64);
let (tx, mut rx) = mpsc::channel(16);
source.add_subscriber(tx);
let collector = TestCollectorNode::new(test_tx);
source.register(Box::new(collector));
tokio::spawn(async move {
source.run().await.unwrap();
let token = CancellationToken::new();
Box::new(source).run(token).await.unwrap();
});
let mut received_frames = 0usize;

164
pmoaudio/src/nodes/flac_file_sink.rs
View File

@@ -1,7 +1,7 @@
use crate::{
nodes::{AudioError, TypedAudioNode, DEFAULT_CHANNEL_SIZE},
type_constraints::TypeRequirement,
AudioChunk, AudioSegment, SyncMarker,
AudioChunk, AudioPipelineNode, AudioSegment, SyncMarker,
};
use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat};
use std::{
@@ -16,6 +16,7 @@ use tokio::{
io::{self, AsyncRead, AsyncWriteExt, ReadBuf},
sync::mpsc,
};
use tokio_util::sync::CancellationToken;
/// Sink qui encode les `AudioSegment` reçus au format FLAC.
///
@@ -25,6 +26,7 @@ use tokio::{
/// - Adapte automatiquement l'encodage FLAC selon la profondeur de bit du chunk (8/16/24/32-bit)
/// - Termine l'encodage proprement quand il reçoit EndOfStream
pub struct FlacFileSink {
tx: mpsc::Sender<Arc<AudioSegment>>,
rx: mpsc::Receiver<Arc<AudioSegment>>,
base_path: PathBuf,
encoder_options: EncoderOptions,
@@ -38,7 +40,7 @@ impl FlacFileSink {
///
/// * `base_path` - Chemin de base pour les fichiers FLAC. Si des TrackBoundary sont reçus,
/// des fichiers seront créés avec des suffixes (_01, _02, etc.)
pub fn new<P: Into<PathBuf>>(base_path: P) -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
pub fn new<P: Into<PathBuf>>(base_path: P) -> Self {
Self::with_channel_size(base_path, DEFAULT_CHANNEL_SIZE)
}
@@ -51,7 +53,7 @@ impl FlacFileSink {
pub fn with_channel_size<P: Into<PathBuf>>(
base_path: P,
channel_size: usize,
) -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
) -> Self {
Self::with_config(base_path, channel_size, EncoderOptions::default())
}
@@ -66,15 +68,15 @@ impl FlacFileSink {
base_path: P,
channel_size: usize,
encoder_options: EncoderOptions,
) -> (Self, mpsc::Sender<Arc<AudioSegment>>) {
) -> Self {
let (tx, rx) = mpsc::channel(channel_size);
let sink = Self {
Self {
tx,
rx,
base_path: base_path.into(),
encoder_options,
pcm_buffer_capacity: 8,
};
(sink, tx)
}
}
/// Lance l'encodage vers le(s) fichier(s) cible(s).
@@ -84,27 +86,28 @@ impl FlacFileSink {
/// - Track 0 : base_path.flac
/// - Track 1 : base_path_01.flac
/// - Track 2 : base_path_02.flac, etc.
pub async fn run(self) -> Result<FlacFileSinkStats, AudioError> {
let FlacFileSink {
mut rx,
base_path,
encoder_options,
pcm_buffer_capacity,
} = self;
async fn run_internal(
mut rx: mpsc::Receiver<Arc<AudioSegment>>,
base_path: PathBuf,
encoder_options: EncoderOptions,
pcm_buffer_capacity: usize,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
let mut all_tracks = Vec::new();
let mut track_number = 0;
loop {
// Vérifier si l'arrêt a été demandé
if stop_token.is_cancelled() {
return Ok(());
}
// Attendre le premier chunk audio pour cette track, en capturant les métadonnées du TrackBoundary
let (first_segment, track_metadata) = match wait_for_first_audio_chunk_with_metadata(&mut rx).await {
let (first_segment, track_metadata) = match wait_for_first_audio_chunk_with_metadata(&mut rx, &stop_token).await {
Ok(result) => result,
Err(_) => {
// Plus d'audio disponible
if all_tracks.is_empty() {
return Err(AudioError::ProcessingError("No audio data received".into()));
}
break;
// Plus d'audio disponible ou arrêt demandé
return Ok(());
}
};
@@ -149,7 +152,7 @@ impl FlacFileSink {
// Exécuter pump et copy en parallèle avec tokio::select! en boucle
let pump_future =
pump_track_segments(first_segment, &mut rx, pcm_tx, bits_per_sample, sample_rate);
pump_track_segments(first_segment, &mut rx, pcm_tx, bits_per_sample, sample_rate, &stop_token);
let copy_future = async {
let copy_result = tokio::io::copy(&mut flac_stream, &mut output).await;
let flush_result = output.flush().await;
@@ -168,16 +171,7 @@ impl FlacFileSink {
// Attendre les deux tâches en parallèle
let (copy_result, pump_result) = tokio::join!(copy_future, pump_future);
copy_result?;
let (chunks, samples, duration_sec, stop_reason) = pump_result?;
// Ajouter les stats de cette track
all_tracks.push(TrackStats {
path: track_path,
track_number,
chunks_received: chunks,
total_samples: samples,
total_duration_sec: duration_sec,
});
let stop_reason = pump_result?;
// Vérifier le stop_reason pour savoir si on continue
match stop_reason {
@@ -186,14 +180,12 @@ impl FlacFileSink {
track_number += 1;
continue;
}
StopReason::EndOfStream | StopReason::ChannelClosed => {
StopReason::EndOfStream | StopReason::ChannelClosed | StopReason::Cancelled => {
// Fin de l'encodage
break;
return Ok(());
}
}
}
Ok(FlacFileSinkStats { tracks: all_tracks })
}
}
@@ -223,20 +215,26 @@ enum StopReason {
TrackBoundary(Arc<tokio::sync::RwLock<dyn pmometadata::TrackMetadata>>),
EndOfStream,
ChannelClosed,
Cancelled,
}
/// Attend et retourne le premier chunk audio avec les métadonnées du TrackBoundary si présent.
/// Retourne une erreur si EndOfStream est reçu avant tout audio.
/// Retourne une erreur si EndOfStream est reçu avant tout audio ou si l'arrêt est demandé.
async fn wait_for_first_audio_chunk_with_metadata(
rx: &mut mpsc::Receiver<Arc<AudioSegment>>,
stop_token: &CancellationToken,
) -> Result<(Arc<AudioSegment>, Option<Arc<tokio::sync::RwLock<dyn pmometadata::TrackMetadata>>>), AudioError> {
let mut track_metadata: Option<Arc<tokio::sync::RwLock<dyn pmometadata::TrackMetadata>>> = None;
loop {
let segment = rx
.recv()
.await
.ok_or_else(|| AudioError::ProcessingError("No audio data received".into()))?;
let segment = tokio::select! {
result = rx.recv() => {
result.ok_or_else(|| AudioError::ProcessingError("No audio data received".into()))?
}
_ = stop_token.cancelled() => {
return Err(AudioError::ProcessingError("Cancelled".into()));
}
};
match &segment.segment {
crate::_AudioSegment::Chunk(chunk) => {
@@ -274,11 +272,8 @@ async fn pump_track_segments(
pcm_tx: mpsc::Sender<Vec<u8>>,
bits_per_sample: u8,
expected_rate: u32,
) -> Result<(u64, u64, f64, StopReason), AudioError> {
let mut chunks = 0u64;
let mut samples = 0u64;
let mut duration_sec = 0.0f64;
stop_token: &CancellationToken,
) -> Result<StopReason, AudioError> {
// Traiter le premier segment
if let Some(chunk) = first_segment.as_chunk() {
let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?;
@@ -287,19 +282,24 @@ async fn pump_track_segments(
.send(pcm_bytes)
.await
.map_err(|_| AudioError::SendError)?;
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 = match rx.recv().await {
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));
drop(pcm_tx);
return Ok(StopReason::ChannelClosed);
}
}
}
_ = stop_token.cancelled() => {
drop(pcm_tx);
return Ok(StopReason::Cancelled);
}
};
@@ -323,25 +323,16 @@ async fn pump_track_segments(
.send(pcm_bytes)
.await
.map_err(|_| AudioError::SendError)?;
chunks += 1;
samples += chunk.len() as u64;
duration_sec += chunk.len() as f64 / expected_rate as f64;
}
crate::_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()),
));
return Ok(StopReason::TrackBoundary(metadata.clone()));
}
SyncMarker::EndOfStream => {
drop(pcm_tx); // Fermer le channel PCM
return Ok((chunks, samples, duration_sec, StopReason::EndOfStream));
return Ok(StopReason::EndOfStream);
}
_ => {} // Ignorer les autres syncmarkers
}
@@ -535,6 +526,32 @@ pub struct FlacFileSinkStats {
pub tracks: Vec<TrackStats>,
}
#[async_trait::async_trait]
impl AudioPipelineNode for FlacFileSink {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
Some(self.tx.clone())
}
fn register(&mut self, _child: Box<dyn AudioPipelineNode>) {
panic!("FlacFileSink is a terminal node and cannot have children");
}
async fn run(
self: Box<Self>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
let FlacFileSink {
tx: _tx,
rx,
base_path,
encoder_options,
pcm_buffer_capacity,
} = *self;
Self::run_internal(rx, base_path, encoder_options, pcm_buffer_capacity, stop_token).await
}
}
impl TypedAudioNode for FlacFileSink {
fn input_type(&self) -> Option<TypeRequirement> {
// FlacFileSink accepte n'importe quel type entier (I16, I24, I32)
@@ -565,8 +582,12 @@ mod tests {
let frames = 256;
// Créer le sink
let (sink, tx) = FlacFileSink::with_channel_size(&output_path, 16);
let sink_handle = tokio::spawn(async move { sink.run().await.unwrap() });
let sink = FlacFileSink::with_channel_size(&output_path, 16);
let tx = sink.get_tx().unwrap();
let stop_token = CancellationToken::new();
let sink_handle = tokio::spawn(async move {
Box::new(sink).run(stop_token).await.unwrap()
});
// Envoyer des segments avec métadonnées
tokio::spawn(async move {
@@ -682,8 +703,12 @@ mod tests {
flac_stream.wait().await.unwrap();
// Maintenant utiliser FlacFileSink pour réécrire le fichier
let (sink, tx) = FlacFileSink::with_channel_size(&output_path, 16);
let sink_handle = tokio::spawn(async move { sink.run().await.unwrap() });
let sink = FlacFileSink::with_channel_size(&output_path, 16);
let tx = sink.get_tx().unwrap();
let stop_token = CancellationToken::new();
let sink_handle = tokio::spawn(async move {
Box::new(sink).run(stop_token).await.unwrap()
});
// Lire le fichier input et envoyer les segments au sink
tokio::spawn(async move {
@@ -745,10 +770,7 @@ mod tests {
decode_stream.wait().await.unwrap();
});
let stats = sink_handle.await.unwrap();
assert_eq!(stats.tracks.len(), 1);
assert!(stats.tracks[0].chunks_received > 0);
assert_eq!(stats.tracks[0].total_samples, frames as u64);
sink_handle.await.unwrap();
// Vérifier que le fichier de sortie est valide
let file = File::open(&output_path).await.unwrap();

280
pmoaudio/src/nodes/http_source.rs
View File

@@ -1,14 +1,14 @@
use crate::{
nodes::{AudioError, MultiSubscriberNode, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS},
nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS},
type_constraints::TypeRequirement,
AudioChunk, AudioChunkData, AudioSegment, I24,
AudioChunk, AudioChunkData, AudioPipelineNode, AudioSegment, I24,
};
use futures_util::StreamExt;
use pmoflac::{decode_audio_stream, StreamInfo};
use pmometadata::{MemoryTrackMetadata, TrackMetadata};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio_util::io::StreamReader;
use tokio_util::{io::StreamReader, sync::CancellationToken};
/// HttpSource - Récupère un fichier audio via HTTP et publie des `AudioSegment`
///
@@ -93,7 +93,8 @@ use tokio_util::io::StreamReader;
pub struct HttpSource {
url: String,
chunk_frames: usize,
subscribers: MultiSubscriberNode,
child_txs: Vec<mpsc::Sender<Arc<AudioSegment>>>,
children: Vec<Box<dyn AudioPipelineNode>>,
}
impl HttpSource {
@@ -136,85 +137,22 @@ impl HttpSource {
Self {
url: url.into(),
chunk_frames,
subscribers: MultiSubscriberNode::new(),
child_txs: Vec::new(),
children: Vec::new(),
}
}
/// Ajoute un abonné qui recevra les segments audio.
///
/// Chaque abonné recevra une copie (via `Arc`) de tous les segments audio
/// produits par cette source, y compris les syncmarkers.
///
/// # Arguments
///
/// * `tx` - Channel sender pour recevoir les `AudioSegment`
///
/// # Exemples
///
/// ```no_run
/// use pmoaudio::HttpSource;
/// use tokio::sync::mpsc;
///
/// let mut source = HttpSource::new("http://example.com/audio.flac");
/// let (tx, rx) = mpsc::channel(16);
/// source.add_subscriber(tx);
/// ```
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioSegment>>) {
self.subscribers.add_subscriber(tx);
}
/// Lance le téléchargement et la lecture du flux audio.
///
/// Cette méthode consomme `self` et exécute le pipeline complet:
/// 1. Effectue la requête HTTP GET vers l'URL spécifiée
/// 2. Vérifie le status HTTP (doit être 2xx)
/// 3. Extrait les métadonnées des headers HTTP
/// 4. Décode le flux audio en streaming
/// 5. Émet les syncmarkers et chunks audio vers les abonnés
///
/// La méthode se termine quand le flux est complètement lu ou en cas d'erreur.
///
/// # Erreurs
///
/// Retourne `AudioError::ProcessingError` si:
/// - La requête HTTP échoue (réseau, DNS, etc.)
/// - Le serveur retourne un status code non-2xx
/// - Le format audio n'est pas supporté
/// - Le décodage échoue
/// - Le fichier a un nombre de canaux non supporté (doit être 1 ou 2)
/// - La profondeur de bit n'est pas supportée (doit être 8, 16, 24 ou 32 bits)
///
/// # Exemples
///
/// ```no_run
/// use pmoaudio::HttpSource;
/// use tokio::sync::mpsc;
///
/// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let mut source = HttpSource::new("http://example.com/audio.flac");
/// let (tx, mut rx) = mpsc::channel(16);
/// source.add_subscriber(tx);
///
/// let handle = tokio::spawn(async move {
/// source.run().await
/// });
///
/// // Traiter les segments
/// while let Some(segment) = rx.recv().await {
/// println!("Segment reçu: order={}", segment.order);
/// }
///
/// handle.await??;
/// Ok(())
/// }
/// ```
pub async fn run(self) -> Result<(), AudioError> {
async fn run_internal(
url: String,
chunk_frames: usize,
child_txs: Vec<mpsc::Sender<Arc<AudioSegment>>>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
// Effectuer la requête HTTP
let response = reqwest::get(&self.url)
let response = reqwest::get(&url)
.await
.map_err(|e| {
AudioError::ProcessingError(format!("HTTP request failed for {}: {}", self.url, e))
AudioError::ProcessingError(format!("HTTP request failed for {}: {}", url, e))
})?;
// Vérifier le status
@@ -222,12 +160,12 @@ impl HttpSource {
return Err(AudioError::ProcessingError(format!(
"HTTP request returned status {}: {}",
response.status(),
self.url
url
)));
}
// Extraire les métadonnées depuis les headers HTTP
let metadata = extract_metadata_from_headers(&response, &self.url).await;
let metadata = extract_metadata_from_headers(&response, &url).await;
// Convertir le stream de bytes en AsyncRead
let bytes_stream = response.bytes_stream();
@@ -244,38 +182,54 @@ impl HttpSource {
validate_stream(&stream_info)?;
// Calculer la taille des chunks si non spécifiée (0 = auto)
let chunk_frames = if self.chunk_frames == 0 {
let chunk_frames_final = if chunk_frames == 0 {
let frames =
(stream_info.sample_rate as f64 * DEFAULT_CHUNK_DURATION_MS / 1000.0) as usize;
frames.next_power_of_two().max(256)
} else {
self.chunk_frames.max(1)
chunk_frames.max(1)
};
// Émettre TopZeroSync
let top_zero = AudioSegment::new_top_zero_sync();
self.subscribers.push(top_zero).await?;
for tx in &child_txs {
tx.send(top_zero.clone()).await.map_err(|_| AudioError::SendError)?;
}
// Émettre TrackBoundary avec les métadonnées HTTP
let track_boundary = AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata)));
self.subscribers.push(track_boundary).await?;
for tx in &child_txs {
tx.send(track_boundary.clone()).await.map_err(|_| AudioError::SendError)?;
}
// Préparer la lecture des chunks audio
let frame_bytes = stream_info.bytes_per_sample() * stream_info.channels as usize;
let chunk_byte_len = chunk_frames * frame_bytes;
let chunk_byte_len = chunk_frames_final * frame_bytes;
let mut pending = Vec::new();
let mut read_buf = vec![0u8; frame_bytes * 512.max(chunk_frames)];
let mut read_buf = vec![0u8; frame_bytes * 512.max(chunk_frames_final)];
let mut chunk_index = 0u64;
let mut total_frames = 0u64;
// Lire et émettre les chunks audio
loop {
// Vérifier l'arrêt
if stop_token.is_cancelled() {
return Ok(());
}
// Remplir le buffer
if pending.len() < chunk_byte_len {
use tokio::io::AsyncReadExt;
let read = stream.read(&mut read_buf).await.map_err(|e| {
let read = tokio::select! {
result = stream.read(&mut read_buf) => {
result.map_err(|e| {
AudioError::ProcessingError(format!("I/O error while decoding: {}", e))
})?;
})?
}
_ = stop_token.cancelled() => {
return Ok(());
}
};
if read == 0 {
break;
}
@@ -288,7 +242,7 @@ impl HttpSource {
// Extraire un chunk
let frames_in_pending = pending.len() / frame_bytes;
let frames_to_emit = frames_in_pending.min(chunk_frames);
let frames_to_emit = frames_in_pending.min(chunk_frames_final);
let take_bytes = frames_to_emit * frame_bytes;
let chunk_bytes = pending.drain(..take_bytes).collect::<Vec<u8>>();
@@ -303,7 +257,13 @@ impl HttpSource {
chunk_index,
timestamp_sec,
)?;
self.subscribers.push(segment).await?;
for tx in &child_txs {
if tx.send(segment.clone()).await.is_err() {
// Un enfant est mort, arrêter
return Ok(());
}
}
chunk_index += 1;
total_frames += frames_to_emit as u64;
@@ -316,7 +276,9 @@ impl HttpSource {
let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64;
let segment =
bytes_to_segment(&pending, &stream_info, frames, chunk_index, timestamp_sec)?;
self.subscribers.push(segment).await?;
for tx in &child_txs {
let _ = tx.send(segment.clone()).await;
}
total_frames += frames as u64;
chunk_index += 1;
}
@@ -325,7 +287,9 @@ impl HttpSource {
// Émettre EndOfStream
let final_timestamp = total_frames as f64 / stream_info.sample_rate as f64;
let eos = AudioSegment::new_end_of_stream(chunk_index, final_timestamp);
self.subscribers.push(eos).await?;
for tx in &child_txs {
let _ = tx.send(eos.clone()).await;
}
// Attendre la fin du décodage
stream
@@ -510,6 +474,61 @@ fn bytes_to_segment(
}))
}
#[async_trait::async_trait]
impl AudioPipelineNode for HttpSource {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
None // HttpSource est une source, pas d'input
}
fn register(&mut self, child: Box<dyn AudioPipelineNode>) {
if let Some(tx) = child.get_tx() {
self.child_txs.push(tx);
}
self.children.push(child);
}
async fn run(
self: Box<Self>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
let HttpSource {
url,
chunk_frames,
child_txs,
children,
} = *self;
// Spawner tous les enfants
let mut child_handles = Vec::new();
for child in children {
let child_token = stop_token.child_token();
let handle = tokio::spawn(async move {
child.run(child_token).await
});
child_handles.push(handle);
}
// Lancer la logique interne
let work_result = Self::run_internal(url, chunk_frames, child_txs, stop_token.clone()).await;
// Attendre que tous les enfants se terminent
for handle in child_handles {
match handle.await {
Ok(Ok(())) => {},
Ok(Err(e)) => return Err(e),
Err(e) => {
return Err(AudioError::ProcessingError(format!(
"Child task panicked: {}",
e
)))
}
}
}
work_result
}
}
impl TypedAudioNode for HttpSource {
fn input_type(&self) -> Option<TypeRequirement> {
// HttpSource est une source, elle ne consomme pas d'audio
@@ -534,6 +553,47 @@ mod tests {
Mock, MockServer, ResponseTemplate,
};
/// Nœud de test qui collecte tous les segments et les envoie à un channel de test
struct TestCollectorNode {
input_tx: mpsc::Sender<Arc<AudioSegment>>,
input_rx: mpsc::Receiver<Arc<AudioSegment>>,
output_tx: mpsc::Sender<Arc<AudioSegment>>,
}
impl TestCollectorNode {
fn new(output_tx: mpsc::Sender<Arc<AudioSegment>>) -> Self {
let (input_tx, input_rx) = mpsc::channel(16);
Self {
input_tx,
input_rx,
output_tx,
}
}
}
#[async_trait::async_trait]
impl AudioPipelineNode for TestCollectorNode {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
Some(self.input_tx.clone())
}
fn register(&mut self, _child: Box<dyn AudioPipelineNode>) {
panic!("TestCollectorNode is a sink and cannot have children");
}
async fn run(
mut self: Box<Self>,
_stop_token: CancellationToken,
) -> Result<(), AudioError> {
while let Some(segment) = self.input_rx.recv().await {
if self.output_tx.send(segment).await.is_err() {
break;
}
}
Ok(())
}
}
/// Test de création basique de HttpSource
#[test]
fn test_http_source_creation() {
@@ -599,12 +659,18 @@ mod tests {
// Créer la source HTTP pointant vers le mock
let url = format!("{}/test.flac", mock_server.uri());
let mut source = HttpSource::with_chunk_size(&url, 64);
let (tx, mut rx) = mpsc::channel(16);
source.add_subscriber(tx);
// Lancer le téléchargement et le décodage
// Créer un noeud collecteur pour recevoir les segments
let (tx, mut rx) = mpsc::channel(16);
let collector = TestCollectorNode::new(tx);
// Construire le pipeline
source.register(Box::new(collector));
// Lancer le pipeline
let stop_token = CancellationToken::new();
tokio::spawn(async move {
source.run().await.unwrap();
Box::new(source).run(stop_token).await.unwrap();
});
// Vérifier les segments reçus
@@ -683,10 +749,12 @@ mod tests {
let url = format!("{}/stream", mock_server.uri());
let mut source = HttpSource::new(&url);
let (tx, mut rx) = mpsc::channel(16);
source.add_subscriber(tx);
let collector = TestCollectorNode::new(tx);
source.register(Box::new(collector));
let stop_token = CancellationToken::new();
tokio::spawn(async move {
source.run().await.unwrap();
Box::new(source).run(stop_token).await.unwrap();
});
// Chercher le TrackBoundary pour vérifier les métadonnées
@@ -720,9 +788,11 @@ mod tests {
let url = format!("{}/notfound.flac", mock_server.uri());
let mut source = HttpSource::new(&url);
let (tx, _rx) = mpsc::channel(16);
source.add_subscriber(tx);
let collector = TestCollectorNode::new(tx);
source.register(Box::new(collector));
let result = source.run().await;
let stop_token = CancellationToken::new();
let result = Box::new(source).run(stop_token).await;
assert!(result.is_err(), "Doit retourner une erreur pour HTTP 404");
if let Err(AudioError::ProcessingError(msg)) = result {
@@ -751,9 +821,11 @@ mod tests {
let url = format!("{}/invalid.flac", mock_server.uri());
let mut source = HttpSource::new(&url);
let (tx, _rx) = mpsc::channel(16);
source.add_subscriber(tx);
let collector = TestCollectorNode::new(tx);
source.register(Box::new(collector));
let result = source.run().await;
let stop_token = CancellationToken::new();
let result = Box::new(source).run(stop_token).await;
assert!(
result.is_err(),
"Doit retourner une erreur pour un format invalide"
@@ -804,10 +876,12 @@ mod tests {
let url = format!("{}/my-song.flac", mock_server.uri());
let mut source = HttpSource::new(&url);
let (tx, mut rx) = mpsc::channel(16);
source.add_subscriber(tx);
let collector = TestCollectorNode::new(tx);
source.register(Box::new(collector));
let stop_token = CancellationToken::new();
tokio::spawn(async move {
source.run().await.unwrap();
Box::new(source).run(stop_token).await.unwrap();
});
let mut found_title = false;

89
pmoaudio/src/nodes/mod.rs
View File

@@ -4,7 +4,6 @@
//! un pipeline audio, ainsi que les traits et structures de support.
use std::sync::Arc;
use tokio::sync::mpsc;
use crate::type_constraints::{TypeMismatch, TypeRequirement};
use crate::AudioSegment;
@@ -109,94 +108,6 @@ pub trait TypedAudioNode {
}
}
/// Node avec un seul abonné (pas de clone inutile)
///
/// Optimisé pour les cas où un node n'a qu'un seul destinataire.
/// Le Arc du chunk est simplement transféré sans clonage supplémentaire.
///
/// # Exemples
///
/// ```
/// use pmoaudio::SingleSubscriberNode;
/// use tokio::sync::mpsc;
///
/// let (tx, rx) = mpsc::channel(10);
/// let node = SingleSubscriberNode::new(tx);
/// ```
pub struct SingleSubscriberNode {
tx: mpsc::Sender<Arc<AudioSegment>>,
}
impl SingleSubscriberNode {
pub fn new(tx: mpsc::Sender<Arc<AudioSegment>>) -> Self {
Self { tx }
}
pub async fn push(&self, chunk: Arc<AudioSegment>) -> Result<(), AudioError> {
self.tx.send(chunk).await.map_err(|_| AudioError::SendError)
}
}
/// Node avec plusieurs abonnés (partage le même Arc)
///
/// Permet de broadcaster un chunk à plusieurs destinations.
/// Tous les abonnés reçoivent le même `Arc<AudioSegment>`, donc pas de copie
/// des données audio - seul le compteur de référence Arc est incrémenté.
///
/// # Exemples
///
/// ```
/// use pmoaudio::MultiSubscriberNode;
/// use tokio::sync::mpsc;
///
/// let mut node = MultiSubscriberNode::new();
/// let (tx1, rx1) = mpsc::channel(10);
/// let (tx2, rx2) = mpsc::channel(10);
///
/// node.add_subscriber(tx1);
/// node.add_subscriber(tx2);
/// // Les deux abonnés recevront les mêmes chunks
/// ```
pub struct MultiSubscriberNode {
subscribers: Vec<mpsc::Sender<Arc<AudioSegment>>>,
}
impl MultiSubscriberNode {
pub fn new() -> Self {
Self {
subscribers: Vec::new(),
}
}
pub fn add_subscriber(&mut self, tx: mpsc::Sender<Arc<AudioSegment>>) {
self.subscribers.push(tx);
}
pub async fn push(&self, chunk: Arc<AudioSegment>) -> Result<(), AudioError> {
for tx in &self.subscribers {
// On partage le même Arc avec tous les abonnés
tx.send(chunk.clone())
.await
.map_err(|_| AudioError::SendError)?;
}
Ok(())
}
pub async fn try_push(&self, chunk: Arc<AudioSegment>) -> Result<(), AudioError> {
for tx in &self.subscribers {
// try_send non-bloquant, ignore si saturé
let _ = tx.try_send(chunk.clone());
}
Ok(())
}
}
impl Default for MultiSubscriberNode {
fn default() -> Self {
Self::new()
}
}
/// Erreurs possibles dans le pipeline audio
#[derive(Debug, Clone)]
pub enum AudioError {

110
pmoaudio/src/pipeline.rs Normal file
View File

@@ -0,0 +1,110 @@
//! Architecture de pipeline audio avec propagation automatique du run et gestion d'arrêt
//!
//! Ce module définit le trait `AudioPipelineNode` qui permet de construire des arbres
//! de traitement audio avec :
//! - Démarrage automatique de tous les enfants lors du run de la tête
//! - Arrêt coordonné sur EOF ou erreur
//! - Propagation bidirectionnelle sans boucle infinie
//!
//! # Architecture
//!
//! Les pipelines forment des **arbres** (pas de DAG) où :
//! - Les sources n'ont pas d'input (get_tx retourne None)
//! - Les sinks n'ont pas d'enfants (register panic)
//! - Les convertisseurs ont à la fois un input et des enfants
//!
//! # Mécanisme d'arrêt
//!
//! - **Descendant** : `stop_token.cancel()` propage l'arrêt vers les fils
//! - **Montant** : Le retour de `run()` informe le parent
//! - **Détection** : Un enfant mort → parent voit `send().is_err()` ou `await handle`
//!
//! # Exemple
//!
//! ```no_run
//! use pmoaudio::{FileSource, AudioPipelineNode};
//! use pmoaudio::nodes::FlacFileSink;
//! use tokio_util::sync::CancellationToken;
//!
//! # async fn example() -> Result<(), pmoaudio::nodes::AudioError> {
//! // Construire le pipeline
//! let mut source = FileSource::new("input.flac");
//! let sink = FlacFileSink::new("output.flac");
//!
//! source.register(Box::new(sink));
//!
//! // Lancer avec contrôle d'arrêt
//! let stop_token = CancellationToken::new();
//! Box::new(source).run(stop_token).await?;
//! # Ok(())
//! # }
//! ```
use crate::{nodes::AudioError, AudioSegment};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
/// Trait pour les nœuds d'un pipeline audio
///
/// Permet la construction d'arbres de traitement avec:
/// - Démarrage automatique de tous les enfants
/// - Arrêt coordonné sur EOF ou erreur
/// - Propagation bidirectionnelle sans boucle
#[async_trait::async_trait]
pub trait AudioPipelineNode: Send + 'static {
/// Retourne un clone du sender pour recevoir des segments
///
/// # Retourne
///
/// - `Some(tx)` pour les nœuds qui ont un input (sinks, convertisseurs)
/// - `None` pour les sources qui génèrent des données
///
/// Le sender retourné est un clone, permettant au parent de l'extraire
/// avant de consommer le nœud dans `run()`.
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>>;
/// Enregistre un nœud enfant dans l'arbre
///
/// # Arguments
///
/// * `child` - Le nœud enfant à enregistrer
///
/// # Comportement
///
/// - Pour les sources et convertisseurs : enregistre l'enfant et clone son tx
/// - Pour les sinks : panic (nœuds terminaux)
///
/// Le parent extrait le tx via `child.get_tx()` avant de stocker le child.
fn register(&mut self, child: Box<dyn AudioPipelineNode>);
/// Lance le nœud et tous ses enfants
///
/// # Arguments
///
/// * `stop_token` - Token d'arrêt partagé pour coordination
///
/// # Comportement
///
/// 1. **Spawn enfants** : Tous les enfants sont spawned avant le traitement
/// 2. **Traitement** : Le nœud fait son travail (lecture, conversion, écriture)
/// 3. **Détection** : Si un enfant meurt, le parent le détecte via send().is_err()
/// 4. **Arrêt** : Sur EOF/erreur, appel de `stop_token.cancel()` pour les enfants
/// 5. **Attente** : Attend que tous les enfants se terminent
/// 6. **Retour** : Retourne pour informer le parent (propagation montante)
///
/// # Propagation d'erreur
///
/// - Erreur du nœud → propagée vers les enfants (cancel) puis vers le parent (return)
/// - Erreur d'un enfant → détectée à l'await du handle, propagée vers le parent
///
/// # Arrêt sans boucle
///
/// - Un seul `cancel()` par nœud (en sortant de la boucle de travail)
/// - L'enfant ne cancel JAMAIS le parent
/// - `cancel()` est idempotent (pas de problème si appelé plusieurs fois)
async fn run(
self: Box<Self>,
stop_token: CancellationToken,
) -> Result<(), AudioError>;
}

1
pmoaudiocache/Cargo.toml
View File

@@ -13,7 +13,6 @@ pmodidl = { path = "../pmodidl" }
# Streaming FLAC asynchrone
pmoflac = { path = "../pmoflac" }
pmometadata = { path = "../pmometadata" }
pmoaudio = { path = "../pmoaudio" }
# Base de données
rusqlite = { version = "0.37", features = ["bundled"] }

2
pmoaudiocache/src/lib.rs
View File

@@ -79,7 +79,6 @@
pub mod cache;
pub mod metadata;
pub mod metadata_ext;
pub mod nodes;
pub mod streaming;
pub mod track_metadata;
@@ -93,7 +92,6 @@ pub mod config_ext;
pub use cache::{add_with_metadata_extraction, get_metadata, new_cache, AudioConfig, Cache};
pub use metadata::AudioMetadata;
pub use metadata_ext::{AudioMetadataExt, AudioTrackMetadataExt};
pub use nodes::{FlacCacheSink, FlacCacheSinkStats};
pub use track_metadata::AudioCacheTrackMetadata;
#[cfg(feature = "pmoconfig")]

8
pmoaudiocache/src/nodes/mod.rs
View File

@@ -1,8 +0,0 @@
//! Nodes audio pour pmoaudiocache
//!
//! Ce module fournit des nodes audio spécialisés qui étendent pmoaudio
//! pour intégrer le cache audio.
pub mod flac_cache_sink;
pub use flac_cache_sink::{FlacCacheSink, FlacCacheSinkStats};

3
pmoplaylist/Cargo.toml
View File

@@ -8,6 +8,9 @@ edition = "2021"
pmoaudiocache = { path = "../pmoaudiocache" }
pmocache = { path = "../pmocache" }
# Métadonnées
pmometadata = { path = "../pmometadata" }
# DIDL-Lite pour UPnP
pmodidl = { path = "../pmodidl" }

34
pmoplaylist/src/track.rs
View File

@@ -1,9 +1,11 @@
//! PlaylistTrack : résultat d'un pop() avec helpers pour accéder au cache
use crate::Result;
use pmoaudiocache::AudioMetadataExt;
use pmoaudiocache::{AudioMetadataExt, AudioTrackMetadataExt};
use pmocache::cache_trait::FileCache;
use std::path::PathBuf;
use pmometadata::TrackMetadata;
use std::{path::PathBuf, sync::Arc};
use tokio::sync::RwLock;
/// Un morceau récupéré depuis une playlist
///
@@ -87,6 +89,34 @@ impl PlaylistTrack {
.map_err(|e| crate::Error::CacheError(e.to_string()))
}
/// Retourne une instance de TrackMetadata pour ce morceau
///
/// Cette méthode fournit un accès unifié aux métadonnées via le trait `TrackMetadata`.
/// L'instance retournée implémente toutes les méthodes du trait et permet un accès
/// asynchrone thread-safe aux métadonnées via RwLock.
///
/// # Exemples
///
/// ```no_run
/// # use pmoplaylist::*;
/// # async fn example(track: PlaylistTrack) -> Result<()> {
/// // Obtenir l'instance TrackMetadata
/// let metadata = track.track_metadata()?;
///
/// // Accéder aux métadonnées via le trait
/// let title = metadata.read().await.get_title().await?;
/// let artist = metadata.read().await.get_artist().await?;
///
/// println!("Titre: {:?}", title);
/// println!("Artiste: {:?}", artist);
/// # Ok(())
/// # }
/// ```
pub fn track_metadata(&self) -> Result<Arc<RwLock<dyn TrackMetadata>>> {
let cache = crate::manager::audio_cache()?;
Ok(cache.track_metadata(&self.cache_pk))
}
/// Récupère uniquement le titre du morceau (méthode légère)
///
/// Utilise l'extension trait `AudioMetadataExt` pour récupérer qu'une seule valeur