From 09fcad06e18087c518e8c9990c1b81cffdd16984 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Mon, 3 Nov 2025 06:55:45 +0100 Subject: [PATCH] Ajout d'un noeud puis vers le cache audio --- .DS_Store | Bin 10244 -> 10244 bytes Cargo.lock | 713 ++++++++++-------- PMOMusic/Cargo.toml | 1 + pmoapp/src/lib.rs | 2 - pmoaudio-ext/Cargo.toml | 30 + pmoaudio-ext/src/lib.rs | 30 + .../src/sinks}/flac_cache_sink.rs | 115 ++- pmoaudio-ext/src/sinks/mod.rs | 11 + pmoaudio/Cargo.toml | 1 + pmoaudio/examples/file_nodes_test.rs | 65 +- pmoaudio/src/lib.rs | 6 +- pmoaudio/src/nodes/converter_nodes.rs | 554 +++++--------- pmoaudio/src/nodes/file_source.rs | 444 +++++++---- pmoaudio/src/nodes/flac_file_sink.rs | 168 +++-- pmoaudio/src/nodes/http_source.rs | 282 ++++--- pmoaudio/src/nodes/mod.rs | 89 --- pmoaudio/src/pipeline.rs | 110 +++ pmoaudiocache/Cargo.toml | 1 - pmoaudiocache/src/lib.rs | 2 - pmoaudiocache/src/nodes/mod.rs | 8 - pmoplaylist/Cargo.toml | 3 + pmoplaylist/src/track.rs | 34 +- 22 files changed, 1507 insertions(+), 1162 deletions(-) create mode 100644 pmoaudio-ext/Cargo.toml create mode 100644 pmoaudio-ext/src/lib.rs rename {pmoaudiocache/src/nodes => pmoaudio-ext/src/sinks}/flac_cache_sink.rs (84%) create mode 100644 pmoaudio-ext/src/sinks/mod.rs create mode 100644 pmoaudio/src/pipeline.rs delete mode 100644 pmoaudiocache/src/nodes/mod.rs diff --git a/.DS_Store b/.DS_Store index eb3eed6f94fdab2864329514a575aa0be28a52d9..f9d43b4635642e3ce52bfd5ef79cd3f7cab3ba65 100644 GIT binary patch delta 1652 zcmb`ITWl0n7{|Z=?6#emF81_(nJlgBRVw8f&gDDj z`@T76&hJ0x#OR69vyDbP>K?_dgS_JW-`V@TVg5Gu$@TcYY0AnS$Tyh$(71PRGx|&@uaoSIIhpIDtk7W zAL8a{M$Z{~ej=VUvkA+dVbyWfjAu>N(fdr(b+)69e$;V$Q+BeDar#T1GmK?zeXq_P zjhsHf-Fa+{n`$KG7)i_CuV-xDhPFodYWX8w4%7!%-E~i7!=^0*L%a4Ythl3!$u)A_ zM9#2{n5mEF4b7RIRPDHKX@+%p%FuMnF%B7eMykfJG-jy@y|BO<8k?G16yFmZQCgLDQJi+i`!hPPyG!{MjJ5#UYM7DHP*yrU_7S*gx5 zmTQnm&9sVosFy}*k~GTGnF2je=jZ}mrg!KAxclrtr?vRK62YtEnD!5Ag z_xk!K;3}%RrnatrWoS)j*S!yHT}IA>bcxGbfiE%sCuMkymzcO6PEUGraGZ-kKl)D! zWEC#%;)3hl+9n5C8JCG?hTFMW&L!fRcXtGXOyJV=%ysL7GV`&5DoUl`?hbBa(b#yO zEHgjXpf}gMnVaPP7d*Rf8&Bm5O8ie;m$1G>Z_&H-QGveTTz^kL(2sNzgmMHp$1Bms zSq`I}vm8Mey3vEp*n&Q6<75wGH%74sdohhiFoQ>-a;{AzU?GoVob4w#-6wGhr*Q^P z;~ZYZdAx*|OO(HXH$6N{(AK;ChrEHM8XwNT(2y#FG9O^4%jb(NDKwXPlWqJKq3!&_ bSiVi>(!cnMRpFuHf$)X--|NnU`zP=R>;9Tq delta 135 zcmZn(XbG4g&d4*dP;8=}>, rx: mpsc::Receiver>, - cache: Arc, + cache: Arc, collection: Option, encoder_options: EncoderOptions, pcm_buffer_capacity: usize, + #[cfg(feature = "playlist")] + playlist_handle: Option>, } 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) -> (Self, mpsc::Sender>) { + pub fn new(cache: Arc) -> 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, - channel_size: usize, - ) -> (Self, mpsc::Sender>) { + pub fn with_channel_size(cache: Arc, 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, + cache: Arc, channel_size: usize, encoder_options: EncoderOptions, collection: Option, - ) -> (Self, mpsc::Sender>) { + ) -> 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 { + async fn run_internal( + self, + stop_token: CancellationToken, + ) -> Result { 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>, + stop_token: &CancellationToken, ) -> Result< ( Arc, @@ -248,10 +279,14 @@ async fn wait_for_first_audio_chunk_with_metadata( let mut track_metadata: Option>> = 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>, 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,9 +344,17 @@ async fn pump_track_segments( // Boucle sur les segments suivants loop { - let segment = match rx.recv().await { - Some(seg) => seg, - None => { + 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)); } @@ -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, } +#[async_trait::async_trait] +impl AudioPipelineNode for FlacCacheSink { + fn get_tx(&self) -> Option>> { + Some(self.tx.clone()) + } + + fn register(&mut self, _child: Box) { + panic!("FlacCacheSink is a terminal sink and cannot have children"); + } + + async fn run( + self: Box, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + self.run_internal(stop_token).await?; + Ok(()) + } +} + impl TypedAudioNode for FlacCacheSink { fn input_type(&self) -> Option { // FlacCacheSink accepte n'importe quel type entier (I16, I24, I32) diff --git a/pmoaudio-ext/src/sinks/mod.rs b/pmoaudio-ext/src/sinks/mod.rs new file mode 100644 index 00000000..9cb71261 --- /dev/null +++ b/pmoaudio-ext/src/sinks/mod.rs @@ -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}; diff --git a/pmoaudio/Cargo.toml b/pmoaudio/Cargo.toml index 02f186df..fa4c29b0 100644 --- a/pmoaudio/Cargo.toml +++ b/pmoaudio/Cargo.toml @@ -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" diff --git a/pmoaudio/examples/file_nodes_test.rs b/pmoaudio/examples/file_nodes_test.rs index ac5909dd..e40f1d5e 100644 --- a/pmoaudio/examples/file_nodes_test.rs +++ b/pmoaudio/examples/file_nodes_test.rs @@ -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 -- -use pmoaudio::{FileSource, FlacFileSink}; +use pmoaudio::{AudioPipelineNode, FileSource, FlacFileSink}; use std::env; +use tokio_util::sync::CancellationToken; #[tokio::main] async fn main() -> Result<(), Box> { @@ -28,48 +34,35 @@ async fn main() -> Result<(), Box> { 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(()) } diff --git a/pmoaudio/src/lib.rs b/pmoaudio/src/lib.rs index 1d2004c8..ce9eaee1 100644 --- a/pmoaudio/src/lib.rs +++ b/pmoaudio/src/lib.rs @@ -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 diff --git a/pmoaudio/src/nodes/converter_nodes.rs b/pmoaudio/src/nodes/converter_nodes.rs index bf06dc88..96bfa8f8 100644 --- a/pmoaudio/src/nodes/converter_nodes.rs +++ b/pmoaudio/src/nodes/converter_nodes.rs @@ -8,376 +8,174 @@ //! 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 { - rx: mpsc::Receiver>, - subscribers: MultiSubscriberNode, -} - -impl ToI16Node { - /// Crée un nouveau node de conversion vers I16 - pub fn new() -> (Self, mpsc::Sender>) { - 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>) { - 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>) { - 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(); - 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?; +// 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>, + rx: mpsc::Receiver>, + child_txs: Vec>>, + children: Vec>, } - Ok(()) - } -} - -impl TypedAudioNode for ToI16Node { - fn input_type(&self) -> Option { - // Accepte n'importe quel type - Some(TypeRequirement::any()) - } - - fn output_type(&self) -> Option { - // 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>, - subscribers: MultiSubscriberNode, -} - -impl ToI24Node { - /// Crée un nouveau node de conversion vers I24 - pub fn new() -> (Self, mpsc::Sender>) { - 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>) { - 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>) { - 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; + impl $node_name { + /// Crée un nouveau node de conversion + pub fn new() -> Self { + Self::with_channel_size(16) } - 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?; + /// Crée un nouveau node avec une taille de buffer spécifique + pub fn with_channel_size(channel_size: usize) -> Self { + let (tx, rx) = mpsc::channel(channel_size); + Self { + tx, + rx, + child_txs: Vec::new(), + children: Vec::new(), + } + } } - Ok(()) - } -} - -impl TypedAudioNode for ToI24Node { - fn input_type(&self) -> Option { - Some(TypeRequirement::any()) - } - - fn output_type(&self) -> Option { - Some(TypeRequirement::specific(SampleType::I24)) - } -} - -impl Default for ToI24Node { - fn default() -> Self { - Self::new().0 - } -} - -/// 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>, - subscribers: MultiSubscriberNode, -} - -impl ToI32Node { - /// Crée un nouveau node de conversion vers I32 - pub fn new() -> (Self, mpsc::Sender>) { - 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>) { - 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>) { - 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; + #[async_trait::async_trait] + impl AudioPipelineNode for $node_name { + fn get_tx(&self) -> Option>> { + Some(self.tx.clone()) } - 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 { - Some(TypeRequirement::any()) - } - - fn output_type(&self) -> Option { - 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>, - subscribers: MultiSubscriberNode, -} - -impl ToF32Node { - /// Crée un nouveau node de conversion vers F32 - pub fn new() -> (Self, mpsc::Sender>) { - 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>) { - 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>) { - 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; + fn register(&mut self, child: Box) { + if let Some(tx) = child.get_tx() { + self.child_txs.push(tx); + } + self.children.push(child); } - 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 - }; + async fn run( + mut self: Box, + 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); + } - self.subscribers.push(converted_segment).await?; + // Boucle de traitement + loop { + let segment = tokio::select! { + result = self.rx.recv() => { + match result { + Some(seg) => seg, + None => break, + } + } + _ = stop_token.cancelled() => { + break; + } + }; + + // 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, + segment: crate::_AudioSegment::Chunk(Arc::new(converted_chunk)), + }) + } else { + segment + } + } else { + segment + }; + + // 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(()) + } } - Ok(()) - } -} - -impl TypedAudioNode for ToF32Node { - fn input_type(&self) -> Option { - Some(TypeRequirement::any()) - } - - fn output_type(&self) -> Option { - 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>, - subscribers: MultiSubscriberNode, -} - -impl ToF64Node { - /// Crée un nouveau node de conversion vers F64 - pub fn new() -> (Self, mpsc::Sender>) { - 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>) { - 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>) { - 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; + impl TypedAudioNode for $node_name { + fn input_type(&self) -> Option { + Some(TypeRequirement::any()) } - 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?; + fn output_type(&self) -> Option { + Some(TypeRequirement::specific($output_type)) + } } - Ok(()) - } + impl Default for $node_name { + fn default() -> Self { + Self::new() + } + } + }; } -impl TypedAudioNode for ToF64Node { - fn input_type(&self) -> Option { - Some(TypeRequirement::any()) - } - - fn output_type(&self) -> Option { - 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>, + input_rx: mpsc::Receiver>, + output_tx: mpsc::Sender>, + } + + impl TestCollectorNode { + fn new(output_tx: mpsc::Sender>) -> 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>> { + Some(self.input_tx.clone()) + } + + fn register(&mut self, _child: Box) { + panic!("TestCollectorNode is a sink"); + } + + async fn run( + mut self: Box, + _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 diff --git a/pmoaudio/src/nodes/file_source.rs b/pmoaudio/src/nodes/file_source.rs index 510c5d0a..0f92e204 100644 --- a/pmoaudio/src/nodes/file_source.rs +++ b/pmoaudio/src/nodes/file_source.rs @@ -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>>, + children: Vec>, } 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>) { - 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::>(); - - // 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>> { + // FileSource est une source, elle n'a pas d'input + None + } + + fn register(&mut self, child: Box) { + // 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, + 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::>(); + + // 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 { // 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>, + rx: mpsc::Receiver>, + test_tx: mpsc::Sender>, + } + + impl TestCollectorNode { + fn new(test_tx: mpsc::Sender>) -> 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>> { + Some(self.tx.clone()) + } + + fn register(&mut self, _child: Box) { + panic!("TestCollectorNode is a terminal node"); + } + + async fn run( + mut self: Box, + 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; diff --git a/pmoaudio/src/nodes/flac_file_sink.rs b/pmoaudio/src/nodes/flac_file_sink.rs index 5068911f..32a5a0ee 100644 --- a/pmoaudio/src/nodes/flac_file_sink.rs +++ b/pmoaudio/src/nodes/flac_file_sink.rs @@ -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>, rx: mpsc::Receiver>, 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>(base_path: P) -> (Self, mpsc::Sender>) { + pub fn new>(base_path: P) -> Self { Self::with_channel_size(base_path, DEFAULT_CHANNEL_SIZE) } @@ -51,7 +53,7 @@ impl FlacFileSink { pub fn with_channel_size>( base_path: P, channel_size: usize, - ) -> (Self, mpsc::Sender>) { + ) -> 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>) { + ) -> 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 { - let FlacFileSink { - mut rx, - base_path, - encoder_options, - pcm_buffer_capacity, - } = self; + async fn run_internal( + mut rx: mpsc::Receiver>, + 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>), 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>, + stop_token: &CancellationToken, ) -> Result<(Arc, Option>>), AudioError> { let mut track_metadata: Option>> = 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>, 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 { // 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 { - Some(seg) => seg, - None => { - drop(pcm_tx); // Fermer le channel PCM - return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed)); + let segment = tokio::select! { + result = rx.recv() => { + match result { + Some(seg) => seg, + None => { + 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, } +#[async_trait::async_trait] +impl AudioPipelineNode for FlacFileSink { + fn get_tx(&self) -> Option>> { + Some(self.tx.clone()) + } + + fn register(&mut self, _child: Box) { + panic!("FlacFileSink is a terminal node and cannot have children"); + } + + async fn run( + self: Box, + 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 { // 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(); diff --git a/pmoaudio/src/nodes/http_source.rs b/pmoaudio/src/nodes/http_source.rs index 38bb8c4c..12c7f3c9 100644 --- a/pmoaudio/src/nodes/http_source.rs +++ b/pmoaudio/src/nodes/http_source.rs @@ -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>>, + children: Vec>, } 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>) { - 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> { - /// 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>>, + 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| { - AudioError::ProcessingError(format!("I/O error while decoding: {}", 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::>(); @@ -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>> { + None // HttpSource est une source, pas d'input + } + + fn register(&mut self, child: Box) { + if let Some(tx) = child.get_tx() { + self.child_txs.push(tx); + } + self.children.push(child); + } + + async fn run( + self: Box, + 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 { // 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>, + input_rx: mpsc::Receiver>, + output_tx: mpsc::Sender>, + } + + impl TestCollectorNode { + fn new(output_tx: mpsc::Sender>) -> 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>> { + Some(self.input_tx.clone()) + } + + fn register(&mut self, _child: Box) { + panic!("TestCollectorNode is a sink and cannot have children"); + } + + async fn run( + mut self: Box, + _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; diff --git a/pmoaudio/src/nodes/mod.rs b/pmoaudio/src/nodes/mod.rs index abc378f7..1af93b46 100644 --- a/pmoaudio/src/nodes/mod.rs +++ b/pmoaudio/src/nodes/mod.rs @@ -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>, -} - -impl SingleSubscriberNode { - pub fn new(tx: mpsc::Sender>) -> Self { - Self { tx } - } - - pub async fn push(&self, chunk: Arc) -> 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`, 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>>, -} - -impl MultiSubscriberNode { - pub fn new() -> Self { - Self { - subscribers: Vec::new(), - } - } - - pub fn add_subscriber(&mut self, tx: mpsc::Sender>) { - self.subscribers.push(tx); - } - - pub async fn push(&self, chunk: Arc) -> 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) -> 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 { diff --git a/pmoaudio/src/pipeline.rs b/pmoaudio/src/pipeline.rs new file mode 100644 index 00000000..c4d33131 --- /dev/null +++ b/pmoaudio/src/pipeline.rs @@ -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>>; + + /// 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); + + /// 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, + stop_token: CancellationToken, + ) -> Result<(), AudioError>; +} diff --git a/pmoaudiocache/Cargo.toml b/pmoaudiocache/Cargo.toml index f61dda24..c04f343a 100644 --- a/pmoaudiocache/Cargo.toml +++ b/pmoaudiocache/Cargo.toml @@ -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"] } diff --git a/pmoaudiocache/src/lib.rs b/pmoaudiocache/src/lib.rs index dffb8602..16e8888a 100644 --- a/pmoaudiocache/src/lib.rs +++ b/pmoaudiocache/src/lib.rs @@ -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")] diff --git a/pmoaudiocache/src/nodes/mod.rs b/pmoaudiocache/src/nodes/mod.rs deleted file mode 100644 index 7d2d365d..00000000 --- a/pmoaudiocache/src/nodes/mod.rs +++ /dev/null @@ -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}; diff --git a/pmoplaylist/Cargo.toml b/pmoplaylist/Cargo.toml index eb5b37f3..b842ea27 100644 --- a/pmoplaylist/Cargo.toml +++ b/pmoplaylist/Cargo.toml @@ -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" } diff --git a/pmoplaylist/src/track.rs b/pmoplaylist/src/track.rs index d4439b83..94bd493c 100644 --- a/pmoplaylist/src/track.rs +++ b/pmoplaylist/src/track.rs @@ -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>> { + 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