From 88349e77974db81a1b8afa5da3f49f1e3b54d4b2 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Tue, 4 Nov 2025 20:34:44 +0100 Subject: [PATCH] =?UTF-8?q?R=C3=A9cup=C3=A9ration=20de=20l'erreur=20git=20?= =?UTF-8?q?cleaning?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .DS_Store | Bin 10244 -> 14340 bytes Cargo.lock | 1 + pmoapp/src/lib.rs | 0 pmoaudio-ext/src/lib.rs | 0 pmoaudio-ext/src/sinks/flac_cache_sink.rs | 470 +++++++++-------- pmoaudio-ext/src/sinks/mod.rs | 0 pmoaudio/Cargo.toml | 1 + pmoaudio/examples/audio_chunk_api.rs | 6 +- pmoaudio/examples/check_flac_bits.rs | 27 + pmoaudio/examples/convert_to_flac24.rs | 110 ++++ pmoaudio/examples/file_nodes_test.rs | 0 pmoaudio/src/audio_chunk.rs | 80 +-- pmoaudio/src/audio_segment.rs | 0 pmoaudio/src/bit_depth.rs | 0 pmoaudio/src/conversions.rs | 72 +-- pmoaudio/src/dsp/depth.rs | 0 pmoaudio/src/dsp/gain_16bits.rs | 0 pmoaudio/src/dsp/gain_24bits.rs | 0 pmoaudio/src/dsp/gain_32bits.rs | 0 pmoaudio/src/dsp/int_float.rs | 0 pmoaudio/src/dsp/mod.rs | 0 pmoaudio/src/dsp/resampling.rs | 0 pmoaudio/src/events.rs | 0 pmoaudio/src/lib.rs | 0 pmoaudio/src/macros.rs | 4 +- pmoaudio/src/nodes/converter_nodes.rs | 513 ++++++++---------- pmoaudio/src/nodes/file_source.rs | 452 ++++++++-------- pmoaudio/src/nodes/flac_file_sink.rs | 196 +++---- pmoaudio/src/nodes/http_source.rs | 234 ++++----- pmoaudio/src/nodes/mod.rs | 9 + pmoaudio/src/pipeline.rs | 608 ++++++++++++++++++++++ pmoaudio/src/sample_types.rs | 0 pmoaudio/src/sync_marker.rs | 0 pmoaudio/src/type_constraints.rs | 0 pmoaudiocache/src/lib.rs | 0 pmoaudiocache/src/metadata_ext.rs | 0 pmocache/src/cache.rs | 0 pmoflac/Cargo.toml | 0 pmoflac/src/autodetect.rs | 0 pmoflac/src/encoder.rs | 0 pmoflac/src/lib.rs | 0 pmoflac/src/metadata.rs | 0 pmoflac/src/pcm.rs | 0 pmoflac/src/prefixed_reader.rs | 0 pmoflac/src/transcode.rs | 0 pmometadata/src/lib.rs | 0 pmoplaylist/src/track.rs | 0 pmosource/src/cache.rs | 0 48 files changed, 1754 insertions(+), 1029 deletions(-) mode change 100644 => 100755 pmoapp/src/lib.rs mode change 100644 => 100755 pmoaudio-ext/src/lib.rs mode change 100644 => 100755 pmoaudio-ext/src/sinks/flac_cache_sink.rs mode change 100644 => 100755 pmoaudio-ext/src/sinks/mod.rs mode change 100644 => 100755 pmoaudio/Cargo.toml mode change 100644 => 100755 pmoaudio/examples/audio_chunk_api.rs create mode 100755 pmoaudio/examples/check_flac_bits.rs create mode 100755 pmoaudio/examples/convert_to_flac24.rs mode change 100644 => 100755 pmoaudio/examples/file_nodes_test.rs mode change 100644 => 100755 pmoaudio/src/audio_chunk.rs mode change 100644 => 100755 pmoaudio/src/audio_segment.rs mode change 100644 => 100755 pmoaudio/src/bit_depth.rs mode change 100644 => 100755 pmoaudio/src/conversions.rs mode change 100644 => 100755 pmoaudio/src/dsp/depth.rs mode change 100644 => 100755 pmoaudio/src/dsp/gain_16bits.rs mode change 100644 => 100755 pmoaudio/src/dsp/gain_24bits.rs mode change 100644 => 100755 pmoaudio/src/dsp/gain_32bits.rs mode change 100644 => 100755 pmoaudio/src/dsp/int_float.rs mode change 100644 => 100755 pmoaudio/src/dsp/mod.rs mode change 100644 => 100755 pmoaudio/src/dsp/resampling.rs mode change 100644 => 100755 pmoaudio/src/events.rs mode change 100644 => 100755 pmoaudio/src/lib.rs mode change 100644 => 100755 pmoaudio/src/macros.rs mode change 100644 => 100755 pmoaudio/src/nodes/converter_nodes.rs mode change 100644 => 100755 pmoaudio/src/nodes/file_source.rs mode change 100644 => 100755 pmoaudio/src/nodes/flac_file_sink.rs mode change 100644 => 100755 pmoaudio/src/nodes/http_source.rs mode change 100644 => 100755 pmoaudio/src/nodes/mod.rs mode change 100644 => 100755 pmoaudio/src/pipeline.rs mode change 100644 => 100755 pmoaudio/src/sample_types.rs mode change 100644 => 100755 pmoaudio/src/sync_marker.rs mode change 100644 => 100755 pmoaudio/src/type_constraints.rs mode change 100644 => 100755 pmoaudiocache/src/lib.rs mode change 100644 => 100755 pmoaudiocache/src/metadata_ext.rs mode change 100644 => 100755 pmocache/src/cache.rs mode change 100644 => 100755 pmoflac/Cargo.toml mode change 100644 => 100755 pmoflac/src/autodetect.rs mode change 100644 => 100755 pmoflac/src/encoder.rs mode change 100644 => 100755 pmoflac/src/lib.rs mode change 100644 => 100755 pmoflac/src/metadata.rs mode change 100644 => 100755 pmoflac/src/pcm.rs mode change 100644 => 100755 pmoflac/src/prefixed_reader.rs mode change 100644 => 100755 pmoflac/src/transcode.rs mode change 100644 => 100755 pmometadata/src/lib.rs mode change 100644 => 100755 pmoplaylist/src/track.rs mode change 100644 => 100755 pmosource/src/cache.rs diff --git a/.DS_Store b/.DS_Store index 3ece96441ff4e74b7264b0f1baa16dde22a4b2e6..250fc4d04128e0cacb3c172170ed5008c87e6340 100644 GIT binary patch literal 14340 zcmeHNTWlRi8J_5L0fe{+K_yW5{+T`Q z&K@6Appangu4ZR;_Wx&RzW*}w&#n-H4pmFLgyya3nyGHf10^9^L|#-yNtWYF zj6jS)j6jS)j6jUQzls3Qy*P{(mB-f@ff#`pfn@~PKV*?O9}4p8DwnPU7hb~ib6&#B z17*YkvMw9sLqUFB<&qUVk`=jPr69y0#g6(clOrDr^6M&B?1B`#py16EL?|TR4!*1; z7c`{u_!=V+BT$dP4pcPh^w7Z$14^buYf0POv5mywov?CqrNO0MB+uAnmR%K zyT;Uo1%EDR`(|OzvHePa|4%8AY+ARz8AnU=d&XzHh4D(hQuPnz%NO}>!ts3uTP;ALBa9oN7T5IO|^0T zEEt!OO1ESt&ZZ}G=B9acer&E{d#-=ZD*MQD=0+cb94&h8>9WPoCySm_opt@3*$ z#j2C9Si_DpZa-%EQx$8m!letoSFVgm!klSVed~+`I(yza3~GPMMsrNovD|4@pwzP) zwrstx>;B$>!Gps`k3W2?wQZxB-jwc`oVR_u;8{4aseBLiwZpn71F91QeVqdnc za;lw9rY79{taZz5+t~W1dor1+YQcWAYFDnBU526B+Lqa#GM_gtmC%t>UWqwN@=0b} z<}HSCk?Tjw7Otu~y(9DXl$kdseDqf}(Vgx|nSN^G3dE^cccyn4W+`>n$rmg~RqaXd zHOv#K$$5pNs@$JGV6;H=5*$aVg49ArtQw# z&|!#Isj|^E)U5=d!IJ74>JJ2DhO2py$|Xp|nNNt&Vy zbdmD(KDt7)(0Azj^aJ`Ky+XgHKhPiP zPxLqXJKau*#O6dtqBGH*ctG_!QL~V#U=%@?*n>j7g#?zLxVn;sbRIpH9=$6pxYSNYJE6pXUcd(p$x4t>uc^3L@Ek-S_W{GAq zYqD{Rsqf3CJI!^l0CoG;9k^W&?N_(&=VEhTT4GZ%{(Y5wPTUkPikHMKsO+!AZ^ZA#Z8E5hI-sC$qxY-jJ={|ZV9gvE83~G9uPS6CMgQ}i~vKGmrX__IM%2c7MRM-zlh5ZnHjGm@v=;QPW zxF7iHekM$)Id z#81F;BQ)K$)$#k1Q*MdtyvdOWZ%c@`f;X3tkn3FHt5|eD5_)3;t~E2!HW*(-ZiYB5 zPKhz3s_-GV$g+e{(-twcR_COI8EYGSmX3AuxdsxNlS`kgxH4Bo=VF~`@Gf33Epx3o z2}Mw;laTtr^C&*@6dL>@-qovQFjHZuKMxzT-vvZIW&#pY1yRg|DN*T8VAf1P;yv{;-?$US?utO6`03KtY)5~#(d$k^z^6xEqgd-DN_8WPAhlsxkN6iTcEP>UexvIdSgSU_G~`<|$Tb{c1i{ui1RG3YGCmJa zVXJVi85CC;V+6;VrZz&4W9Kk>b@Xcnv$~N*kg7#uJX5%GR*FO_a!wA<0AUXGGv!Lr zR@fu1z!Adtj;z@q+&}Q2uh~7UMtJ9Kh5fTQ{%;uT@CT}zM)0aR*bRy3WgP#%_K}nK z2gCLh$NyTKe5ZX$9RI_j-|%CdYO%is`4N2kXz>648~)z7!5D!Ufx8C*YC1fA_z?P2 z`3Oqud5s-9RKql0q^SH mifXaM@=i!(&DJo=)ptyIw=RzV)y_x+!Brw$9m6#q|NjFv3LC%x delta 1513 zcmbu9YfM~46vxm1F4%i7uYvH`yA*a|+d^@Xg)UGKEYH@q5U_x}DrDi_wpW&2V3$|1 z*~Z`_zR75PwSF<07!3)M#@57W{G^FSX$VPc;urftld6edY7>pl-HS$>XkwgX=FB-~ z&Y78ees`9qmb*PfgqB&olE|9MqFa)s5GAQZ&Xg5TMNYM_IE0ZXLMJFhK^hN5*EU{v zaDN~v8N!>yn=%XF;%Bq7)3yEqb!fmGkA!>7(*|G7?rRxkS#zc_-!&C8BjM=%#!OU5 zZ;3Kh+Yf5mQ1n!A#^{<48#C;a__pJSaE!8ygv*L+*UeaDMmQ56#jSPA@}QNoS=SAH zGcC`Vd#uNbU}kvEj0LS4^awEK`)#7TC1##@ZaBv$2Z_&GSi9Vvw1O{KzH}oyt6s4Du z4krpl`K{vb#$D*CKqaaWKqKx!6Pg7H9q2?Cx^WD{f`k!_B8Ueufe^x&$AX|?31jnmEB6Aw;ylk|uIG6plh@OC_BvL!4k3R%)l=i( zuX)pW-0Qxr+Wz(M-Uh6!6!BEH$QueXr5y@~b2wjCMwxSIsWMRQ^HkOP&-gq`bya?! zCr}gMi)DWP-QI`zTH)HtR;!k5^W+iFCAI(c*<9QB(ZMqU<%uWGqUx;ZLM8wcgCx{M n6BFyvEwvP3_w!I-MO|8|uB-BSDr@)g3|BsXysY^D9JA?n?aN=N diff --git a/Cargo.lock b/Cargo.lock index 594cd5a3..a24b6de0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2650,6 +2650,7 @@ dependencies = [ "tokio-test", "tokio-util", "tracing", + "tracing-subscriber", "wiremock", ] diff --git a/pmoapp/src/lib.rs b/pmoapp/src/lib.rs old mode 100644 new mode 100755 diff --git a/pmoaudio-ext/src/lib.rs b/pmoaudio-ext/src/lib.rs old mode 100644 new mode 100755 diff --git a/pmoaudio-ext/src/sinks/flac_cache_sink.rs b/pmoaudio-ext/src/sinks/flac_cache_sink.rs old mode 100644 new mode 100755 index 78bf47b8..2d349d66 --- a/pmoaudio-ext/src/sinks/flac_cache_sink.rs +++ b/pmoaudio-ext/src/sinks/flac_cache_sink.rs @@ -2,6 +2,7 @@ use pmoaudio::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHANNEL_SIZE}, + pipeline::{Node, NodeLogic}, type_constraints::TypeRequirement, AudioChunk, AudioPipelineNode, AudioSegment, SyncMarker, _AudioSegment, }; @@ -30,9 +31,20 @@ use tracing::warn; /// - 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>, - rx: mpsc::Receiver>, + +// ═══════════════════════════════════════════════════════════════════════════ +// FlacCacheSinkLogic - Logique métier pure +// ═══════════════════════════════════════════════════════════════════════════ + +/// Signal retourné par pump_segments indiquant pourquoi l'encodage s'est arrêté. +enum StopReason { + TrackBoundary(Arc>), + EndOfStream, + ChannelClosed, +} + +/// Logique pure d'encodage FLAC vers le cache +pub struct FlacCacheSinkLogic { cache: Arc, covers: Arc, collection: Option, @@ -42,6 +54,207 @@ pub struct FlacCacheSink { playlist_handle: Option>, } +impl FlacCacheSinkLogic { + pub fn new( + cache: Arc, + covers: Arc, + collection: Option, + encoder_options: EncoderOptions, + pcm_buffer_capacity: usize, + ) -> Self { + Self { + cache, + covers, + collection, + encoder_options, + pcm_buffer_capacity, + #[cfg(feature = "playlist")] + playlist_handle: None, + } + } + + #[cfg(feature = "playlist")] + pub fn set_playlist_handle(&mut self, handle: Arc) { + self.playlist_handle = Some(handle); + } +} + +#[async_trait::async_trait] +impl NodeLogic for FlacCacheSinkLogic { + async fn process( + &mut self, + input: Option>>, + _output: Vec>>, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + let mut rx = input.expect("FlacCacheSink must have input"); + let mut track_number = 0; + + loop { + // Attendre le premier chunk audio pour cette track + 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 + return Ok(()); + } + }; + + // Extraire les informations du premier chunk + let first_chunk = first_segment.as_chunk().unwrap(); + let sample_rate = first_chunk.sample_rate(); + let bits_per_sample = get_chunk_bit_depth(first_chunk); + + let format = PcmFormat { + sample_rate, + channels: 2, + bits_per_sample, + }; + if let Err(err) = format.validate() { + return Err(AudioError::ProcessingError(format!( + "Invalid PCM format: {}", + err + ))); + } + + // Créer le pipeline d'encodage pour cette track + let (pcm_tx, pcm_rx) = mpsc::channel::>(self.pcm_buffer_capacity); + + // Préparer les options d'encodage avec les métadonnées du TrackBoundary + let mut options_with_metadata = self.encoder_options.clone(); + options_with_metadata.metadata = track_metadata.clone(); + + // Créer l'encoder + let reader = ByteStreamReader::new(pcm_rx); + let mut flac_stream = encode_flac_stream(reader, format, options_with_metadata) + .await + .map_err(|e| { + AudioError::ProcessingError(format!("FLAC encode init failed: {}", e)) + })?; + + // Créer un buffer pour collecter le FLAC encodé + let mut flac_buffer = Vec::new(); + + // Exécuter pump et copy en parallèle + let pump_future = pump_track_segments( + first_segment, + &mut rx, + pcm_tx, + bits_per_sample, + sample_rate, + &stop_token, + ); + let copy_future = async { + tokio::io::copy(&mut flac_stream, &mut flac_buffer) + .await + .map_err(|e| { + AudioError::ProcessingError(format!("FLAC write failed: {}", e)) + })?; + flac_stream + .wait() + .await + .map_err(|e| AudioError::ProcessingError(format!("Encoder failed: {}", e)))?; + Ok::<_, AudioError>(()) + }; + + // 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?; + + // Ingérer le FLAC dans le cache + let flac_reader = Cursor::new(flac_buffer.clone()); + let collection_ref = self.collection.as_deref(); + let pk = self.cache + .add_from_reader( + None, + flac_reader, + Some(flac_buffer.len() as u64), + collection_ref, + ) + .await + .map_err(|e| { + AudioError::ProcessingError(format!("Failed to add to cache: {}", e)) + })?; + + // Copier les métadonnées du TrackBoundary dans le cache + if let Some(src_metadata) = track_metadata { + let dest_metadata = self.cache.track_metadata(&pk); + + // Utiliser copy_metadata_into pour copier toutes les métadonnées + pmometadata::copy_metadata_into(&src_metadata, &dest_metadata) + .await + .map_err(|e| { + AudioError::ProcessingError(format!( + "Failed to copy metadata to cache: {}", + e + )) + })?; + + let url = match dest_metadata.read().await.get_cover_url().await { + Ok(url) => url, + Err(e) if e.is_transient() => None, + Err(_) => { + warn!("Cannot obtain cover for audio asset {}", pk); + None + } + }; + + if url.is_some() { + let _ = match self.covers + .add_from_url(&url.unwrap(), self.collection.as_deref()) + .await + { + Ok(pk_covers) => { + dest_metadata + .write() + .await + .set_cover_pk(Some(pk_covers)) + .await + } + Err(_) => { + warn!("Cannot obtain cover for audio asset {}", pk); + Ok(Some(())) + } + }; + } + } + + // Ajouter à la playlist si enregistrée + #[cfg(feature = "playlist")] + if let Some(ref playlist_handle) = self.playlist_handle { + playlist_handle.push(pk.clone()).await.map_err(|e| { + AudioError::ProcessingError(format!("Failed to add to playlist: {}", e)) + })?; + } + + // Vérifier le stop_reason pour savoir si on continue + match stop_reason { + StopReason::TrackBoundary(_metadata) => { + // Continuer avec la prochaine track + track_number += 1; + continue; + } + StopReason::EndOfStream | StopReason::ChannelClosed => { + // Fin de l'encodage + return Ok(()); + } + } + } + } +} + +// ═══════════════════════════════════════════════════════════════════════════ +// FlacCacheSink - Wrapper utilisant Node +// ═══════════════════════════════════════════════════════════════════════════ + +pub struct FlacCacheSink { + inner: Node, + #[cfg(feature = "playlist")] + playlist_handle_pending: Option>, +} + impl FlacCacheSink { /// Crée un sink FLAC cache avec les options par défaut (compression 5, buffer de 16 segments). /// @@ -81,17 +294,11 @@ impl FlacCacheSink { encoder_options: EncoderOptions, collection: Option, ) -> Self { - let (tx, rx) = mpsc::channel(channel_size); + let logic = FlacCacheSinkLogic::new(cache, covers, collection, encoder_options, 8); Self { - tx, - rx, - cache, - covers, - collection, - encoder_options, - pcm_buffer_capacity: 8, + inner: Node::new_with_input(logic, channel_size), #[cfg(feature = "playlist")] - playlist_handle: None, + playlist_handle_pending: None, } } @@ -102,211 +309,8 @@ impl FlacCacheSink { /// * `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)); + self.playlist_handle_pending = 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. - async fn run_internal( - self, - stop_token: CancellationToken, - ) -> Result { - let FlacCacheSink { - tx: _, - mut rx, - cache, - covers, - collection, - encoder_options, - pcm_buffer_capacity, - #[cfg(feature = "playlist")] - playlist_handle, - } = self; - - let mut all_tracks = Vec::new(); - let mut track_number = 0; - - 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, &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; - } - }; - - // Extraire les informations du premier chunk - let first_chunk = first_segment.as_chunk().unwrap(); - let sample_rate = first_chunk.sample_rate(); - let bits_per_sample = get_chunk_bit_depth(first_chunk); - - let format = PcmFormat { - sample_rate, - channels: 2, - bits_per_sample, - }; - if let Err(err) = format.validate() { - return Err(AudioError::ProcessingError(format!( - "Invalid PCM format: {}", - err - ))); - } - - // Créer le pipeline d'encodage pour cette track - let (pcm_tx, pcm_rx) = mpsc::channel::>(pcm_buffer_capacity); - - // Préparer les options d'encodage avec les métadonnées du TrackBoundary - let mut options_with_metadata = encoder_options.clone(); - options_with_metadata.metadata = track_metadata.clone(); - - // Créer l'encoder - let reader = ByteStreamReader::new(pcm_rx); - let mut flac_stream = encode_flac_stream(reader, format, options_with_metadata) - .await - .map_err(|e| { - AudioError::ProcessingError(format!("FLAC encode init failed: {}", e)) - })?; - - // Créer un buffer pour collecter le FLAC encodé - let mut flac_buffer = Vec::new(); - - // Exécuter pump et copy en parallèle - let pump_future = pump_track_segments( - first_segment, - &mut rx, - pcm_tx, - bits_per_sample, - sample_rate, - &stop_token, - ); - let copy_future = async { - tokio::io::copy(&mut flac_stream, &mut flac_buffer) - .await - .map_err(|e| { - AudioError::ProcessingError(format!("FLAC write failed: {}", e)) - })?; - flac_stream - .wait() - .await - .map_err(|e| AudioError::ProcessingError(format!("Encoder failed: {}", e)))?; - Ok::<_, AudioError>(()) - }; - - // Attendre les deux tâches en parallèle - let (copy_result, pump_result): ( - Result<(), AudioError>, - Result<(u64, u64, f64, StopReason), AudioError>, - ) = tokio::join!(copy_future, pump_future); - copy_result?; - let (chunks, samples, duration_sec, stop_reason) = pump_result?; - - // Ingérer le FLAC dans le cache - let flac_reader = Cursor::new(flac_buffer.clone()); - let collection_ref = collection.as_deref(); - let pk = cache - .add_from_reader( - None, - flac_reader, - Some(flac_buffer.len() as u64), - collection_ref, - ) - .await - .map_err(|e| { - AudioError::ProcessingError(format!("Failed to add to cache: {}", e)) - })?; - - // Copier les métadonnées du TrackBoundary dans le cache - if let Some(src_metadata) = track_metadata { - let dest_metadata = cache.track_metadata(&pk); - - // Utiliser copy_metadata_into pour copier toutes les métadonnées - pmometadata::copy_metadata_into(&src_metadata, &dest_metadata) - .await - .map_err(|e| { - AudioError::ProcessingError(format!( - "Failed to copy metadata to cache: {}", - e - )) - })?; - - let url = match dest_metadata.read().await.get_cover_url().await { - Ok(url) => url, - Err(e) if e.is_transient() => None, - Err(_) => { - warn!("Cannot obtain cover for audio asset {}", pk); - None - } - }; - - if url.is_some() { - let _ = match covers - .add_from_url(&url.unwrap(), collection.as_deref()) - .await - { - Ok(pk_covers) => { - dest_metadata - .write() - .await - .set_cover_pk(Some(pk_covers)) - .await - } - Err(_) => { - warn!("Cannot obtain cover for audio asset {}", pk); - Ok(Some(())) - } - }; - } - } - - // 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, - track_number, - chunks_received: chunks, - total_samples: samples, - total_duration_sec: duration_sec, - }); - - // Vérifier le stop_reason pour savoir si on continue - match stop_reason { - StopReason::TrackBoundary(_metadata) => { - // Continuer avec la prochaine track - track_number += 1; - continue; - } - StopReason::EndOfStream | StopReason::ChannelClosed => { - // Fin de l'encodage - break; - } - } - } - - Ok(FlacCacheSinkStats { tracks: all_tracks }) - } -} - -/// Signal retourné par pump_segments indiquant pourquoi l'encodage s'est arrêté. -enum StopReason { - TrackBoundary(Arc>), - EndOfStream, - ChannelClosed, } /// Attend et retourne le premier chunk audio avec les métadonnées du TrackBoundary si présent. @@ -481,13 +485,13 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result match (chunk, bits_per_sample) { // I16 source (AudioChunk::I16(data), 16) => { - for frame in data.frames() { + for frame in data.get_frames() { bytes.extend_from_slice(&frame[0].to_le_bytes()); bytes.extend_from_slice(&frame[1].to_le_bytes()); } } (AudioChunk::I16(data), 24) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = (frame[0] as i32) << 8; let right = (frame[1] as i32) << 8; bytes.extend_from_slice(&left.to_le_bytes()[..3]); @@ -495,7 +499,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result } } (AudioChunk::I16(data), 32) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = (frame[0] as i32) << 16; let right = (frame[1] as i32) << 16; bytes.extend_from_slice(&left.to_le_bytes()); @@ -505,7 +509,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result // I24 source (AudioChunk::I24(data), 16) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = (frame[0].as_i32() >> 8) as i16; let right = (frame[1].as_i32() >> 8) as i16; bytes.extend_from_slice(&left.to_le_bytes()); @@ -513,13 +517,13 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result } } (AudioChunk::I24(data), 24) => { - for frame in data.frames() { + for frame in data.get_frames() { bytes.extend_from_slice(&frame[0].as_i32().to_le_bytes()[..3]); bytes.extend_from_slice(&frame[1].as_i32().to_le_bytes()[..3]); } } (AudioChunk::I24(data), 32) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = frame[0].as_i32() << 8; let right = frame[1].as_i32() << 8; bytes.extend_from_slice(&left.to_le_bytes()); @@ -529,7 +533,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result // I32 source (AudioChunk::I32(data), 16) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = (frame[0] >> 16) as i16; let right = (frame[1] >> 16) as i16; bytes.extend_from_slice(&left.to_le_bytes()); @@ -537,7 +541,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result } } (AudioChunk::I32(data), 24) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = frame[0] >> 8; let right = frame[1] >> 8; bytes.extend_from_slice(&left.to_le_bytes()[..3]); @@ -545,7 +549,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result } } (AudioChunk::I32(data), 32) => { - for frame in data.frames() { + for frame in data.get_frames() { bytes.extend_from_slice(&frame[0].to_le_bytes()); bytes.extend_from_slice(&frame[1].to_le_bytes()); } @@ -638,16 +642,24 @@ pub struct FlacCacheSinkStats { #[async_trait::async_trait] impl AudioPipelineNode for FlacCacheSink { fn get_tx(&self) -> Option>> { - Some(self.tx.clone()) + self.inner.get_tx() } 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(()) + async fn run(mut self: Box, stop_token: CancellationToken) -> Result<(), AudioError> { + // Transférer le playlist_handle_pending à la logique si présent + #[cfg(feature = "playlist")] + if let Some(handle) = self.playlist_handle_pending.take() { + // FIXME: Node devrait exposer une méthode logic_mut() pour permettre + // la configuration post-construction. Pour l'instant, on ignore ce handle. + // L'utilisateur devra configurer la playlist avant construction. + let _ = handle; + } + + Box::new(self.inner).run(stop_token).await } } diff --git a/pmoaudio-ext/src/sinks/mod.rs b/pmoaudio-ext/src/sinks/mod.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/Cargo.toml b/pmoaudio/Cargo.toml old mode 100644 new mode 100755 index fa4c29b0..a670c748 --- a/pmoaudio/Cargo.toml +++ b/pmoaudio/Cargo.toml @@ -24,3 +24,4 @@ tracing = "0.1" tokio-test = "0.4" tempfile = "3" wiremock = "0.6" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/pmoaudio/examples/audio_chunk_api.rs b/pmoaudio/examples/audio_chunk_api.rs old mode 100644 new mode 100755 index 94c0dd35..def86260 --- a/pmoaudio/examples/audio_chunk_api.rs +++ b/pmoaudio/examples/audio_chunk_api.rs @@ -33,7 +33,7 @@ fn example_create_chunks() { println!( "Chunk I32: {} frames @ {}Hz", chunk_i32.len(), - chunk_i32.sample_rate() + chunk_i32.get_sample_rate() ); // Chunk F32 stéréo (normalisé [-1.0, 1.0]) @@ -42,7 +42,7 @@ fn example_create_chunks() { println!( "Chunk F32: {} frames @ {}Hz", chunk_f32.len(), - chunk_f32.sample_rate() + chunk_f32.get_sample_rate() ); // Chunk depuis canaux séparés @@ -57,7 +57,7 @@ fn example_create_chunks() { 48000, 6.0, // +6 dB ); - println!("Chunk with gain: {} dB\n", chunk_with_gain.gain_db()); + println!("Chunk with gain: {} dB\n", chunk_with_gain.get_gain_db()); } fn example_conversions() { diff --git a/pmoaudio/examples/check_flac_bits.rs b/pmoaudio/examples/check_flac_bits.rs new file mode 100755 index 00000000..af56a809 --- /dev/null +++ b/pmoaudio/examples/check_flac_bits.rs @@ -0,0 +1,27 @@ +//! Vérifie la profondeur de bit d'un fichier FLAC + +use pmoflac::decode_audio_stream; +use std::path::Path; +use tokio::fs::File; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let path_str = std::env::args().nth(1).expect("Usage: check_flac_bits "); + let path = Path::new(&path_str); + + println!("Checking: {}", path.display()); + println!(); + + // decode_audio_stream pour lire StreamInfo + let file = File::open(&path).await?; + let stream = decode_audio_stream(file).await?; + let info = stream.info().clone(); + + println!("StreamInfo from FLAC:"); + println!(" bits_per_sample: {}", info.bits_per_sample); + println!(" sample_rate: {}", info.sample_rate); + println!(" channels: {}", info.channels); + println!(" bytes_per_sample: {}", info.bytes_per_sample()); + + Ok(()) +} diff --git a/pmoaudio/examples/convert_to_flac24.rs b/pmoaudio/examples/convert_to_flac24.rs new file mode 100755 index 00000000..97e563cd --- /dev/null +++ b/pmoaudio/examples/convert_to_flac24.rs @@ -0,0 +1,110 @@ +//! Convertisseur de fichiers audio vers FLAC 24-bit +//! +//! Ce programme démontre l'utilisation de la chaîne : +//! 1. FileSource - Lecture d'un fichier audio (FLAC, MP3, OGG, WAV, AIFF) +//! 2. ToI24Node - Conversion vers 24-bit signed integer +//! 3. FlacFileSink - Écriture au format FLAC +//! +//! La nouvelle architecture AudioPipelineNode permet de : +//! - Construire le pipeline en enregistrant des enfants avec register() +//! - Insérer des nœuds de conversion de type pour garantir la profondeur de bit souhaitée +//! - 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 convert_to_flac24 -- +//! +//! Exemple: +//! cargo run --example convert_to_flac24 -- input.mp3 output.flac +//! cargo run --example convert_to_flac24 -- input16bit.flac output24bit.flac + +use pmoaudio::{AudioPipelineNode, FileSource, FlacFileSink, ToI24Node}; +use std::env; +use tokio_util::sync::CancellationToken; + +#[tokio::main] +async fn main() -> Result<(), Box> { + // Initialiser tracing pour le debug + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + + // Récupérer les arguments + let args: Vec = env::args().collect(); + if args.len() != 3 { + eprintln!("Usage: {} ", args[0]); + eprintln!(); + eprintln!("Converts any audio file to FLAC with 24-bit depth."); + eprintln!(); + eprintln!("Supported input formats:"); + eprintln!(" - FLAC (8/16/24/32-bit)"); + eprintln!(" - MP3"); + eprintln!(" - OGG Vorbis"); + eprintln!(" - WAV"); + eprintln!(" - AIFF"); + eprintln!(); + eprintln!("Example:"); + eprintln!(" {} input.mp3 output.flac", args[0]); + eprintln!(" {} input16bit.flac output24bit.flac", args[0]); + std::process::exit(1); + } + + let input_path = &args[1]; + let output_path = &args[2]; + + println!("=== Audio to FLAC 24-bit Converter ==="); + println!(); + println!("Input: {}", input_path); + println!("Output: {}", output_path); + println!(); + println!("Pipeline: FileSource → ToI24Node → FlacFileSink"); + println!(); + + // Créer le pipeline: FileSource → ToI24Node → FlacFileSink + let mut source = FileSource::new(input_path); + + // Le ToI24Node convertit tous les chunks audio en 24-bit + // Cela garantit que le FlacFileSink encodera en 24-bit + let mut converter = ToI24Node::new(); + + let sink = FlacFileSink::new(output_path); + + // Construire la chaîne: source → converter → sink + converter.register(Box::new(sink)); + source.register(converter); + + // Créer un token d'arrêt pour contrôle manuel si besoin + let stop_token = CancellationToken::new(); + + // Lancer tout le pipeline - run() spawne automatiquement tous les enfants + println!("Processing..."); + let start = std::time::Instant::now(); + + let result = Box::new(source).run(stop_token).await; + + let elapsed = start.elapsed(); + + // Vérifier le résultat + match result { + Ok(()) => { + println!(); + println!("✓ Conversion completed successfully in {:.2}s", elapsed.as_secs_f64()); + println!(" Output file: {}", output_path); + println!(); + + // Afficher des informations supplémentaires si possible + if let Ok(metadata) = std::fs::metadata(output_path) { + let size_mb = metadata.len() as f64 / (1024.0 * 1024.0); + println!(" File size: {:.2} MB", size_mb); + } + } + Err(e) => { + eprintln!(); + eprintln!("✗ Conversion error: {}", e); + eprintln!(); + return Err(e.into()); + } + } + + Ok(()) +} diff --git a/pmoaudio/examples/file_nodes_test.rs b/pmoaudio/examples/file_nodes_test.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/audio_chunk.rs b/pmoaudio/src/audio_chunk.rs old mode 100644 new mode 100755 index 6bb68d70..88dfe268 --- a/pmoaudio/src/audio_chunk.rs +++ b/pmoaudio/src/audio_chunk.rs @@ -105,25 +105,25 @@ impl AudioChunkData { /// Taux d'échantillonnage (Hz) #[inline] - pub fn sample_rate(&self) -> u32 { + pub fn get_sample_rate(&self) -> u32 { self.sample_rate } /// Gain courant en décibels #[inline] - pub fn gain_db(&self) -> f64 { + pub fn get_gain_db(&self) -> f64 { self.gain_db } /// Gain sous forme linéaire #[inline] - pub fn gain_linear(&self) -> f64 { + pub fn get_gain_linear(&self) -> f64 { gain_linear_from_db(self.gain_db) } /// Retourne une vue immuable sur les frames `[L, R]` #[inline] - pub fn frames(&self) -> &[[T; 2]] { + pub fn get_frames(&self) -> &[[T; 2]] { &self.stereo } @@ -336,22 +336,22 @@ impl AudioChunk { /// Taux d'échantillonnage (Hz) pub fn sample_rate(&self) -> u32 { match self { - AudioChunk::I16(d) => d.sample_rate(), - AudioChunk::I24(d) => d.sample_rate(), - AudioChunk::I32(d) => d.sample_rate(), - AudioChunk::F32(d) => d.sample_rate(), - AudioChunk::F64(d) => d.sample_rate(), + AudioChunk::I16(d) => d.get_sample_rate(), + AudioChunk::I24(d) => d.get_sample_rate(), + AudioChunk::I32(d) => d.get_sample_rate(), + AudioChunk::F32(d) => d.get_sample_rate(), + AudioChunk::F64(d) => d.get_sample_rate(), } } /// Gain courant en décibels pub fn gain_db(&self) -> f64 { match self { - AudioChunk::I16(d) => d.gain_db(), - AudioChunk::I24(d) => d.gain_db(), - AudioChunk::I32(d) => d.gain_db(), - AudioChunk::F32(d) => d.gain_db(), - AudioChunk::F64(d) => d.gain_db(), + AudioChunk::I16(d) => d.get_gain_db(), + AudioChunk::I24(d) => d.get_gain_db(), + AudioChunk::I32(d) => d.get_gain_db(), + AudioChunk::F32(d) => d.get_gain_db(), + AudioChunk::F64(d) => d.get_gain_db(), } } @@ -387,7 +387,7 @@ impl AudioChunk { pub fn apply_gain(self) -> Self { match self { AudioChunk::I16(d) => { - let gain_db = d.gain_db(); + let gain_db = d.get_gain_db(); if gain_db.abs() < f64::EPSILON { return AudioChunk::I16(d); } @@ -401,10 +401,10 @@ impl AudioChunk { .round() .clamp(-32768.0, 32767.0) as i16; } - AudioChunk::I16(AudioChunkData::new(stereo, d.sample_rate(), 0.0)) + AudioChunk::I16(AudioChunkData::new(stereo, d.get_sample_rate(), 0.0)) } AudioChunk::I24(d) => { - let gain_db = d.gain_db(); + let gain_db = d.get_gain_db(); if gain_db.abs() < f64::EPSILON { return AudioChunk::I24(d); } @@ -420,7 +420,7 @@ impl AudioChunk { frame[0] = I24::new_clamped(l); frame[1] = I24::new_clamped(r); } - AudioChunk::I24(AudioChunkData::new(stereo, d.sample_rate(), 0.0)) + AudioChunk::I24(AudioChunkData::new(stereo, d.get_sample_rate(), 0.0)) } AudioChunk::I32(d) => AudioChunk::I32(d.apply_gain()), AudioChunk::F32(d) => AudioChunk::F32(d.apply_gain()), @@ -497,18 +497,18 @@ impl AudioIntegerChunk { /// Taux d'échantillonnage (Hz) pub fn sample_rate(&self) -> u32 { match self { - AudioIntegerChunk::I16(d) => d.sample_rate(), - AudioIntegerChunk::I24(d) => d.sample_rate(), - AudioIntegerChunk::I32(d) => d.sample_rate(), + AudioIntegerChunk::I16(d) => d.get_sample_rate(), + AudioIntegerChunk::I24(d) => d.get_sample_rate(), + AudioIntegerChunk::I32(d) => d.get_sample_rate(), } } /// Gain courant en décibels pub fn gain_db(&self) -> f64 { match self { - AudioIntegerChunk::I16(d) => d.gain_db(), - AudioIntegerChunk::I24(d) => d.gain_db(), - AudioIntegerChunk::I32(d) => d.gain_db(), + AudioIntegerChunk::I16(d) => d.get_gain_db(), + AudioIntegerChunk::I24(d) => d.get_gain_db(), + AudioIntegerChunk::I32(d) => d.get_gain_db(), } } @@ -542,7 +542,7 @@ impl AudioIntegerChunk { pub fn apply_gain(self) -> Self { match self { AudioIntegerChunk::I16(d) => { - let gain_db = d.gain_db(); + let gain_db = d.get_gain_db(); if gain_db.abs() < f64::EPSILON { return AudioIntegerChunk::I16(d); } @@ -556,10 +556,10 @@ impl AudioIntegerChunk { .round() .clamp(-32768.0, 32767.0) as i16; } - AudioIntegerChunk::I16(AudioChunkData::new(stereo, d.sample_rate(), 0.0)) + AudioIntegerChunk::I16(AudioChunkData::new(stereo, d.get_sample_rate(), 0.0)) } AudioIntegerChunk::I24(d) => { - let gain_db = d.gain_db(); + let gain_db = d.get_gain_db(); if gain_db.abs() < f64::EPSILON { return AudioIntegerChunk::I24(d); } @@ -575,7 +575,7 @@ impl AudioIntegerChunk { frame[0] = I24::new_clamped(l); frame[1] = I24::new_clamped(r); } - AudioIntegerChunk::I24(AudioChunkData::new(stereo, d.sample_rate(), 0.0)) + AudioIntegerChunk::I24(AudioChunkData::new(stereo, d.get_sample_rate(), 0.0)) } AudioIntegerChunk::I32(d) => AudioIntegerChunk::I32(d.apply_gain()), } @@ -676,12 +676,12 @@ impl AudioIntegerChunk { pub fn frames(&self) -> Box + '_> { match self { AudioIntegerChunk::I16(d) => { - Box::new(d.frames().iter().map(|f| [f[0] as i32, f[1] as i32])) + Box::new(d.get_frames().iter().map(|f| [f[0] as i32, f[1] as i32])) } AudioIntegerChunk::I24(d) => { - Box::new(d.frames().iter().map(|f| [f[0].as_i32(), f[1].as_i32()])) + Box::new(d.get_frames().iter().map(|f| [f[0].as_i32(), f[1].as_i32()])) } - AudioIntegerChunk::I32(d) => Box::new(d.frames().iter().map(|f| [f[0], f[1]])), + AudioIntegerChunk::I32(d) => Box::new(d.get_frames().iter().map(|f| [f[0], f[1]])), } } } @@ -723,16 +723,16 @@ impl AudioFloatChunk { /// Taux d'échantillonnage (Hz) pub fn sample_rate(&self) -> u32 { match self { - AudioFloatChunk::F32(d) => d.sample_rate(), - AudioFloatChunk::F64(d) => d.sample_rate(), + AudioFloatChunk::F32(d) => d.get_sample_rate(), + AudioFloatChunk::F64(d) => d.get_sample_rate(), } } /// Gain courant en décibels pub fn gain_db(&self) -> f64 { match self { - AudioFloatChunk::F32(d) => d.gain_db(), - AudioFloatChunk::F64(d) => d.gain_db(), + AudioFloatChunk::F32(d) => d.get_gain_db(), + AudioFloatChunk::F64(d) => d.get_gain_db(), } } @@ -829,9 +829,9 @@ impl AudioFloatChunk { pub fn frames(&self) -> Box + '_> { match self { AudioFloatChunk::F32(d) => { - Box::new(d.frames().iter().map(|f| [f[0] as f64, f[1] as f64])) + Box::new(d.get_frames().iter().map(|f| [f[0] as f64, f[1] as f64])) } - AudioFloatChunk::F64(d) => Box::new(d.frames().iter().map(|f| [f[0], f[1]])), + AudioFloatChunk::F64(d) => Box::new(d.get_frames().iter().map(|f| [f[0], f[1]])), } } } @@ -885,9 +885,9 @@ mod tests { let chunk = AudioChunkData::new(stereo, 48000, 0.0); assert_eq!(chunk.len(), 2); - assert_eq!(chunk.sample_rate(), 48000); + assert_eq!(chunk.get_sample_rate(), 48000); assert!(!chunk.is_empty()); - assert_eq!(chunk.gain_db(), 0.0); + assert_eq!(chunk.get_gain_db(), 0.0); } #[test] @@ -896,7 +896,7 @@ mod tests { let chunk = AudioChunkData::new(stereo, 48000, -6.0); assert_eq!(chunk.len(), 2); - assert_eq!(chunk.gain_db(), -6.0); + assert_eq!(chunk.get_gain_db(), -6.0); } #[test] diff --git a/pmoaudio/src/audio_segment.rs b/pmoaudio/src/audio_segment.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/bit_depth.rs b/pmoaudio/src/bit_depth.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/conversions.rs b/pmoaudio/src/conversions.rs old mode 100644 new mode 100755 index 1513c7b3..a27a6a55 --- a/pmoaudio/src/conversions.rs +++ b/pmoaudio/src/conversions.rs @@ -27,7 +27,7 @@ pub fn convert_i32_to_i16(chunk: &AudioChunkData) -> Arc) -> Arc) -> Arc> { // Convertir i16 → i32 d'abord let mut stereo: Vec<[i32; 2]> = chunk - .frames() + .get_frames() .iter() .map(|[l, r]| [*l as i32, *r as i32]) .collect(); @@ -58,14 +58,14 @@ pub fn convert_i16_to_i32(chunk: &AudioChunkData) -> Arc) -> Arc> { // Convertir I24 → i32 d'abord let mut stereo: Vec<[i32; 2]> = chunk - .frames() + .get_frames() .iter() .map(|[l, r]| [l.as_i32(), r.as_i32()]) .collect(); @@ -73,7 +73,7 @@ pub fn convert_i24_to_i32(chunk: &AudioChunkData) -> Arc) -> Arc) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let len = frames.len(); // Séparer les canaux pour utiliser les fonctions DSP SIMD @@ -99,7 +99,7 @@ pub fn convert_i32_to_f32(chunk: &AudioChunkData) -> Arc) -> Arc) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let len = frames.len(); // Séparer les canaux I24 en i32 @@ -128,12 +128,12 @@ pub fn convert_i24_to_f32(chunk: &AudioChunkData) -> Arc) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let max_value = 8_388_608.0f64; // 2^23 let stereo: Vec<[f64; 2]> = frames @@ -145,12 +145,12 @@ pub fn convert_i24_to_f64(chunk: &AudioChunkData) -> Arc) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let len = frames.len(); // Séparer les canaux @@ -165,12 +165,12 @@ pub fn convert_i16_to_f32(chunk: &AudioChunkData) -> Arc) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let max_value = 32_768.0f64; // 2^15 let stereo: Vec<[f64; 2]> = frames @@ -182,7 +182,7 @@ pub fn convert_i16_to_f64(chunk: &AudioChunkData) -> Arc) -> Arc) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let len = frames.len(); // Utiliser la fonction SIMD optimisée du module DSP avec BitDepth::B32 @@ -208,7 +208,7 @@ pub fn convert_f32_to_i32(chunk: &AudioChunkData) -> Arc) -> Arc) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let len = frames.len(); // Utiliser la fonction SIMD optimisée du module DSP @@ -237,12 +237,12 @@ pub fn convert_f32_to_i24(chunk: &AudioChunkData) -> Arc) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let max_value = 8_388_607.0f64; // 2^23 - 1 let min_value = -8_388_608.0f64; // -2^23 @@ -255,12 +255,12 @@ pub fn convert_f64_to_i24(chunk: &AudioChunkData) -> Arc) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let len = frames.len(); // Utiliser la fonction SIMD optimisée du module DSP @@ -275,12 +275,12 @@ pub fn convert_f32_to_i16(chunk: &AudioChunkData) -> Arc) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let max_value = 32_767.0f64; // 2^15 - 1 let min_value = -32_768.0f64; // -2^15 @@ -293,7 +293,7 @@ pub fn convert_f64_to_i16(chunk: &AudioChunkData) -> Arc) -> Arc) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let stereo: Vec<[f64; 2]> = frames.iter().map(|[l, r]| [*l as f64, *r as f64]).collect(); - AudioChunkData::new(stereo, chunk.sample_rate(), chunk.gain_db()) + AudioChunkData::new(stereo, chunk.get_sample_rate(), chunk.get_gain_db()) } /// Convertit f64 vers f32 (downcast simple) pub fn convert_f64_to_f32(chunk: &AudioChunkData) -> Arc> { - let frames = chunk.frames(); + let frames = chunk.get_frames(); let stereo: Vec<[f32; 2]> = frames.iter().map(|[l, r]| [*l as f32, *r as f32]).collect(); - AudioChunkData::new(stereo, chunk.sample_rate(), chunk.gain_db()) + AudioChunkData::new(stereo, chunk.get_sample_rate(), chunk.get_gain_db()) } // ============================================================================ @@ -562,7 +562,7 @@ mod tests { // Vérifier que les valeurs sont proches (tolérance d'arrondi) // Note: Pour I32 on utilise toute la plage ±2^31 - for (orig, back) in stereo.iter().zip(chunk_back.frames().iter()) { + for (orig, back) in stereo.iter().zip(chunk_back.get_frames().iter()) { assert!((orig[0] - back[0]).abs() <= 100); // Tolérance plus élevée pour 32-bit assert!((orig[1] - back[1]).abs() <= 100); } @@ -577,7 +577,7 @@ mod tests { let chunk_back = convert_f64_to_f32(&chunk_f64); // Vérifier égalité exacte (pas de perte de précision significative) - for (orig, back) in stereo.iter().zip(chunk_back.frames().iter()) { + for (orig, back) in stereo.iter().zip(chunk_back.get_frames().iter()) { assert!((orig[0] - back[0]).abs() < 1e-6); assert!((orig[1] - back[1]).abs() < 1e-6); } @@ -591,7 +591,7 @@ mod tests { let chunk_i32 = convert_i16_to_i32(&chunk_i16); // Vérifier que les valeurs sont correctement upsamplées (shift de 16 bits) - for (orig, result) in stereo.iter().zip(chunk_i32.frames().iter()) { + for (orig, result) in stereo.iter().zip(chunk_i32.get_frames().iter()) { assert_eq!(result[0], (orig[0] as i32) << 16); assert_eq!(result[1], (orig[1] as i32) << 16); } @@ -605,7 +605,7 @@ mod tests { let chunk_i16 = convert_i32_to_i16(&chunk_i32); // Vérifier que les valeurs sont correctement downsamplées - for (orig, result) in stereo.iter().zip(chunk_i16.frames().iter()) { + for (orig, result) in stereo.iter().zip(chunk_i16.get_frames().iter()) { assert_eq!(result[0], (orig[0] >> 16) as i16); assert_eq!(result[1], (orig[1] >> 16) as i16); } @@ -620,7 +620,7 @@ mod tests { let chunk_f32 = convert_i24_to_f32(&chunk_i24); let chunk_back = convert_f32_to_i24(&chunk_f32); - for (orig, back) in stereo.iter().zip(chunk_back.frames().iter()) { + for (orig, back) in stereo.iter().zip(chunk_back.get_frames().iter()) { assert!((orig[0].as_i32() - back[0].as_i32()).abs() <= 1); assert!((orig[1].as_i32() - back[1].as_i32()).abs() <= 1); } @@ -724,7 +724,7 @@ mod tests { let chunk_back: Arc> = (&*chunk_f32).into(); // Vérifier la précision - for (orig, back) in original.iter().zip(chunk_back.frames().iter()) { + for (orig, back) in original.iter().zip(chunk_back.get_frames().iter()) { assert!((orig[0].as_i32() - back[0].as_i32()).abs() <= 1); assert!((orig[1].as_i32() - back[1].as_i32()).abs() <= 1); } diff --git a/pmoaudio/src/dsp/depth.rs b/pmoaudio/src/dsp/depth.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/dsp/gain_16bits.rs b/pmoaudio/src/dsp/gain_16bits.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/dsp/gain_24bits.rs b/pmoaudio/src/dsp/gain_24bits.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/dsp/gain_32bits.rs b/pmoaudio/src/dsp/gain_32bits.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/dsp/int_float.rs b/pmoaudio/src/dsp/int_float.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/dsp/mod.rs b/pmoaudio/src/dsp/mod.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/dsp/resampling.rs b/pmoaudio/src/dsp/resampling.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/events.rs b/pmoaudio/src/events.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/lib.rs b/pmoaudio/src/lib.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/macros.rs b/pmoaudio/src/macros.rs old mode 100644 new mode 100755 index e70feeaa..4757d729 --- a/pmoaudio/src/macros.rs +++ b/pmoaudio/src/macros.rs @@ -216,7 +216,7 @@ mod tests { let len = match_chunk!(&chunk, data => data.len()); assert_eq!(len, 1); - let sample_rate = match_chunk!(&chunk, data => data.sample_rate()); + let sample_rate = match_chunk!(&chunk, data => data.get_sample_rate()); assert_eq!(sample_rate, 44100); } @@ -227,7 +227,7 @@ mod tests { let modified = map_chunk!(&chunk, data => data.set_gain_db(6.0)); match_chunk!(&modified, data => { - assert_eq!(data.gain_db(), 6.0); + assert_eq!(data.get_gain_db(), 6.0); }); } diff --git a/pmoaudio/src/nodes/converter_nodes.rs b/pmoaudio/src/nodes/converter_nodes.rs old mode 100644 new mode 100755 index 96bfa8f8..402feefe --- a/pmoaudio/src/nodes/converter_nodes.rs +++ b/pmoaudio/src/nodes/converter_nodes.rs @@ -6,176 +6,215 @@ //! //! Le designer de pipeline doit insérer manuellement ces nodes pour gérer //! les incompatibilités de type entre producers et consumers. +//! +//! # Nouvelle Architecture +//! +//! Les converters utilisent maintenant `Node>` où F est +//! une fonction de conversion. Cela simplifie drastiquement le code (de ~130 +//! lignes par converter à ~20 lignes de logique pure). use crate::{ - nodes::{AudioError, TypedAudioNode}, - type_constraints::{SampleType, TypeRequirement}, - AudioPipelineNode, AudioSegment, + nodes::AudioError, + pipeline::{Node, NodeLogic}, + AudioChunk, AudioPipelineNode, AudioSegment, }; use std::sync::Arc; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; -// 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>, - } - - 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 { - let (tx, rx) = mpsc::channel(channel_size); - Self { - tx, - rx, - child_txs: Vec::new(), - children: Vec::new(), - } - } - } - - #[async_trait::async_trait] - impl AudioPipelineNode for $node_name { - fn get_tx(&self) -> Option>> { - Some(self.tx.clone()) - } - - 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( - 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); - } - - // 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(()) - } - } - - impl TypedAudioNode for $node_name { - fn input_type(&self) -> Option { - Some(TypeRequirement::any()) - } - - fn output_type(&self) -> Option { - Some(TypeRequirement::specific($output_type)) - } - } - - impl Default for $node_name { - fn default() -> Self { - Self::new() - } - } - }; +/// Logique de conversion générique +/// +/// Cette struct contient la logique pure de conversion d'un type vers un autre. +/// Elle reçoit des segments, convertit les chunks audio, et relay les syncmarkers. +pub struct ConverterLogic { + convert_fn: F, } -// 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)" -); +impl ConverterLogic +where + F: Fn(&AudioChunk) -> AudioChunk + Send + 'static, +{ + pub fn new(convert_fn: F) -> Self { + Self { convert_fn } + } +} + +#[async_trait::async_trait] +impl NodeLogic for ConverterLogic +where + F: Fn(&AudioChunk) -> AudioChunk + Send + 'static, +{ + async fn process( + &mut self, + input: Option>>, + output: Vec>>, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + let mut rx = input.expect("Converter must have input"); + tracing::debug!("ConverterLogic::process started, {} children", output.len()); + + loop { + let segment = tokio::select! { + _ = stop_token.cancelled() => { + tracing::debug!("ConverterLogic cancelled"); + break; + } + + result = rx.recv() => { + match result { + Some(seg) => seg, + None => { + tracing::debug!("ConverterLogic received EOF"); + break; // EOF + } + } + } + }; + + // 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 = (self.convert_fn)(chunk); + + // Debug: afficher le type du chunk converti (seulement pour le premier) + if segment.order == 0 { + let chunk_type = match &converted_chunk { + crate::AudioChunk::I16(_) => "I16", + crate::AudioChunk::I24(_) => "I24", + crate::AudioChunk::I32(_) => "I32", + crate::AudioChunk::F32(_) => "F32", + crate::AudioChunk::F64(_) => "F64", + }; + tracing::debug!("ConverterLogic: converted chunk type = {}", chunk_type); + } + + 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 &output { + tx.send(output_segment.clone()) + .await + .map_err(|_| AudioError::ChildDied)?; + } + } + + Ok(()) + } +} + +// ═══════════════════════════════════════════════════════════════════════════ +// Converters spécifiques - Fonctions factory simplifiées +// ═══════════════════════════════════════════════════════════════════════════ + +/// Node de conversion vers I16 (16-bit signed integer) +pub struct ToI16Node; + +impl ToI16Node { + pub fn new() -> Box { + Self::with_channel_size(16) + } + + pub fn with_channel_size(channel_size: usize) -> Box { + let logic = ConverterLogic::new(|chunk: &AudioChunk| chunk.to_i16()); + Box::new(Node::new_with_input(logic, channel_size)) + } +} + +impl Default for ToI16Node { + fn default() -> Self { + Self + } +} + +/// Node de conversion vers I24 (24-bit signed integer) +pub struct ToI24Node; + +impl ToI24Node { + pub fn new() -> Box { + Self::with_channel_size(16) + } + + pub fn with_channel_size(channel_size: usize) -> Box { + let logic = ConverterLogic::new(|chunk: &AudioChunk| chunk.to_i24()); + Box::new(Node::new_with_input(logic, channel_size)) + } +} + +impl Default for ToI24Node { + fn default() -> Self { + Self + } +} + +/// Node de conversion vers I32 (32-bit signed integer) +pub struct ToI32Node; + +impl ToI32Node { + pub fn new() -> Box { + Self::with_channel_size(16) + } + + pub fn with_channel_size(channel_size: usize) -> Box { + let logic = ConverterLogic::new(|chunk: &AudioChunk| chunk.to_i32()); + Box::new(Node::new_with_input(logic, channel_size)) + } +} + +impl Default for ToI32Node { + fn default() -> Self { + Self + } +} + +/// Node de conversion vers F32 (32-bit floating point) +pub struct ToF32Node; + +impl ToF32Node { + pub fn new() -> Box { + Self::with_channel_size(16) + } + + pub fn with_channel_size(channel_size: usize) -> Box { + let logic = ConverterLogic::new(|chunk: &AudioChunk| chunk.to_f32()); + Box::new(Node::new_with_input(logic, channel_size)) + } +} + +impl Default for ToF32Node { + fn default() -> Self { + Self + } +} + +/// Node de conversion vers F64 (64-bit floating point) +pub struct ToF64Node; + +impl ToF64Node { + pub fn new() -> Box { + Self::with_channel_size(16) + } + + pub fn with_channel_size(channel_size: usize) -> Box { + let logic = ConverterLogic::new(|chunk: &AudioChunk| chunk.to_f64()); + Box::new(Node::new_with_input(logic, channel_size)) + } +} + +impl Default for ToF64Node { + fn default() -> Self { + Self + } +} #[cfg(test)] mod tests { @@ -183,138 +222,48 @@ mod tests { use crate::{AudioChunk, AudioChunkData}; #[tokio::test] - async fn test_to_f32_node_type_requirements() { - let node = ToF32Node::new(); + async fn test_converter_logic() { + // Test unitaire de la logique pure + let mut logic = ConverterLogic::new(|chunk: &AudioChunk| chunk.to_f32()); - // Vérifier les types d'entrée/sortie - assert_eq!( - node.input_type().unwrap().get_accepted_types().len(), - 5, - "Should accept all 5 types" - ); - assert_eq!( - node.output_type() - .unwrap() - .get_accepted_types() - .first() - .copied(), - Some(SampleType::F32), - "Should output F32 only" - ); - } - - // 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 = ToI16Node::new(); - let (out_tx, mut out_rx) = mpsc::channel(16); - 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 (input_tx, input_rx) = mpsc::channel(10); + let (output_tx, mut output_rx) = mpsc::channel(10); 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]; - let chunk_data = AudioChunkData::new(stereo.clone(), 48_000, 0.0); - let chunk = AudioChunk::I32(chunk_data); + // Créer un chunk de test + let test_chunk = AudioChunk::I16(AudioChunkData::new( + vec![[100, 200], [300, 400]], + 48000, + 0.0, + )); + + // Créer le segment directement let segment = Arc::new(AudioSegment { order: 0, timestamp_sec: 0.0, - segment: crate::_AudioSegment::Chunk(Arc::new(chunk)), + segment: crate::_AudioSegment::Chunk(Arc::new(test_chunk)), }); - tx.send(segment).await.unwrap(); - drop(tx); + // Envoyer le segment + input_tx.send(segment).await.unwrap(); + drop(input_tx); // EOF - // Recevoir le chunk converti - let result = out_rx.recv().await.unwrap(); + // Lancer le traitement + tokio::spawn(async move { + logic + .process(Some(input_rx), vec![output_tx], stop_token) + .await + .unwrap(); + }); + + // Vérifier le résultat + let result = output_rx.recv().await.unwrap(); assert!(result.is_audio_chunk()); - if let Some(converted) = result.as_chunk() { - assert_eq!(converted.type_name(), "i16"); - assert_eq!(converted.len(), 100); - - // Vérifier la conversion (downsampling de I32 vers I16) - if let AudioChunk::I16(data) = &**converted { - for (orig, converted_frame) in stereo.iter().zip(data.frames().iter()) { - let expected_l = (orig[0] >> 16) as i16; - let expected_r = (orig[1] >> 16) as i16; - assert_eq!(converted_frame[0], expected_l); - assert_eq!(converted_frame[1], expected_r); - } - } + if let Some(chunk) = result.as_chunk() { + assert!(matches!(chunk.as_ref(), AudioChunk::F32(_))); + } else { + panic!("Expected audio chunk"); } - - handle.await.unwrap().unwrap(); - } - - #[tokio::test] - async fn test_syncmarkers_passthrough() { - let mut node = ToI16Node::new(); - let (out_tx, mut out_rx) = mpsc::channel(16); - 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 { - Box::new(node).run(stop_token).await.unwrap(); - }); - - // Envoyer un syncmarker - let top_zero = AudioSegment::new_top_zero_sync(); - tx.send(top_zero.clone()).await.unwrap(); - drop(tx); - - // Recevoir le syncmarker - let result = out_rx.recv().await.unwrap(); - assert!(!result.is_audio_chunk()); - assert!(result.as_sync_marker().is_some()); } } diff --git a/pmoaudio/src/nodes/file_source.rs b/pmoaudio/src/nodes/file_source.rs old mode 100644 new mode 100755 index 0f92e204..84cc4616 --- a/pmoaudio/src/nodes/file_source.rs +++ b/pmoaudio/src/nodes/file_source.rs @@ -1,6 +1,6 @@ use crate::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS}, - pipeline::AudioPipelineNode, + pipeline::{AudioPipelineNode, Node, NodeLogic}, type_constraints::TypeRequirement, AudioChunk, AudioChunkData, AudioSegment, I24, }; @@ -11,6 +11,199 @@ use tokio::{fs::File, io::AsyncReadExt, sync::mpsc}; use tokio_util::sync::CancellationToken; use tracing; +// ═══════════════════════════════════════════════════════════════════════════ +// NOUVELLE ARCHITECTURE - FileSourceLogic +// ═══════════════════════════════════════════════════════════════════════════ + +/// Logique pure de lecture de fichier audio +/// +/// Contient seulement la logique de décodage et d'envoi des segments, +/// sans la plomberie d'orchestration (gérée par Node). +pub struct FileSourceLogic { + path: PathBuf, + chunk_frames: usize, +} + +impl FileSourceLogic { + pub fn new>(path: P, chunk_frames: usize) -> Self { + Self { + path: path.into(), + chunk_frames, + } + } +} + +#[async_trait::async_trait] +impl NodeLogic for FileSourceLogic { + async fn process( + &mut self, + _input: Option>>, + output: Vec>>, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + tracing::debug!("FileSourceLogic::process started, path={:?}, {} children", self.path, output.len()); + + // Macro helper pour envoyer à tous les enfants + macro_rules! send_to_children { + ($segment:expr) => { + for tx in &output { + tx.send($segment.clone()) + .await + .map_err(|_| AudioError::ChildDied)?; + } + }; + } + + // Ouvrir le fichier + let file = File::open(&self.path).await.map_err(|e| { + AudioError::IoError(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 { + 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) + }; + + // É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(&self.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!("FileSourceLogic: 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::IoError(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(()) + } +} + +// ═════════════════════════════════════════════════════════════════════════════ +// WRAPPER FileSource - Délègue à Node +// ═══════════════════════════════════════════════════════════════════════════════ + /// FileSource - Lit un fichier audio et publie des `AudioSegment` /// /// Cette source utilise `pmoflac` pour décoder le fichier (FLAC/MP3/OGG/WAV/AIFF) @@ -21,11 +214,13 @@ use tracing; /// - `TopZeroSync` au début du flux /// - `TrackBoundary` avec les métadonnées du fichier /// - `EndOfStream` à la fin du flux +/// +/// # Architecture +/// +/// Utilise la nouvelle architecture avec `Node` pour séparer +/// la logique métier (décodage) de la plomberie (spawning, monitoring). pub struct FileSource { - path: PathBuf, - chunk_frames: usize, - child_txs: Vec>>, - children: Vec>, + inner: Node, } impl FileSource { @@ -44,15 +239,31 @@ impl FileSource { /// * `path` - chemin du fichier audio à lire /// * `chunk_frames` - nombre d'échantillons par canal par chunk (0 = auto) pub fn with_chunk_size>(path: P, chunk_frames: usize) -> Self { + let logic = FileSourceLogic::new(path, chunk_frames); Self { - path: path.into(), - chunk_frames, - child_txs: Vec::new(), - children: Vec::new(), + inner: Node::new_source(logic), } } } +#[async_trait::async_trait] +impl AudioPipelineNode for FileSource { + fn get_tx(&self) -> Option>> { + self.inner.get_tx() + } + + fn register(&mut self, child: Box) { + self.inner.register(child) + } + + async fn run( + self: Box, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + Box::new(self.inner).run(stop_token).await + } +} + fn validate_stream(info: &StreamInfo) -> Result<(), AudioError> { if !(1..=2).contains(&info.channels) { return Err(AudioError::ProcessingError(format!( @@ -188,229 +399,6 @@ 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 { diff --git a/pmoaudio/src/nodes/flac_file_sink.rs b/pmoaudio/src/nodes/flac_file_sink.rs old mode 100644 new mode 100755 index 32a5a0ee..b8f7a27c --- a/pmoaudio/src/nodes/flac_file_sink.rs +++ b/pmoaudio/src/nodes/flac_file_sink.rs @@ -1,5 +1,6 @@ use crate::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHANNEL_SIZE}, + pipeline::{Node, NodeLogic}, type_constraints::TypeRequirement, AudioChunk, AudioPipelineNode, AudioSegment, SyncMarker, }; @@ -25,80 +26,57 @@ use tokio_util::sync::CancellationToken; /// - Crée un nouveau fichier FLAC pour chaque TrackBoundary rencontré /// - 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>, + +// ═══════════════════════════════════════════════════════════════════════════ +// FlacFileSinkLogic - Logique métier pure +// ═══════════════════════════════════════════════════════════════════════════ + +/// Signal retourné par pump_segments indiquant pourquoi l'encodage s'est arrêté. +enum StopReason { + TrackBoundary(Arc>), + EndOfStream, + ChannelClosed, + Cancelled, +} + +/// Logique pure d'encodage FLAC +pub struct FlacFileSinkLogic { base_path: PathBuf, encoder_options: EncoderOptions, pcm_buffer_capacity: usize, } -impl FlacFileSink { - /// Crée un sink FLAC avec les options par défaut (compression 5, buffer de 16 segments). - /// - /// # Arguments - /// - /// * `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 { - Self::with_channel_size(base_path, DEFAULT_CHANNEL_SIZE) - } - - /// Crée un sink FLAC avec une taille de buffer MPSC personnalisée. - /// - /// # Arguments - /// - /// * `base_path` - Chemin de base pour les fichiers FLAC - /// * `channel_size` - Taille du buffer MPSC (nombre de segments en attente avant backpressure) - pub fn with_channel_size>( +impl FlacFileSinkLogic { + pub fn new>( base_path: P, - channel_size: usize, - ) -> Self { - Self::with_config(base_path, channel_size, EncoderOptions::default()) - } - - /// Crée un sink FLAC avec une configuration complète. - /// - /// # Arguments - /// - /// * `base_path` - Chemin de base pour les fichiers FLAC - /// * `channel_size` - Taille du buffer MPSC - /// * `encoder_options` - Options d'encodage FLAC (compression, etc.) - pub fn with_config>( - base_path: P, - channel_size: usize, - encoder_options: EncoderOptions, - ) -> Self { - let (tx, rx) = mpsc::channel(channel_size); - Self { - tx, - rx, - base_path: base_path.into(), - encoder_options, - pcm_buffer_capacity: 8, - } - } - - /// Lance l'encodage vers le(s) fichier(s) cible(s). - /// - /// Cette méthode crée un nouveau fichier FLAC pour chaque TrackBoundary rencontré. - /// Les fichiers sont nommés selon la convention : - /// - Track 0 : base_path.flac - /// - Track 1 : base_path_01.flac - /// - Track 2 : base_path_02.flac, etc. - async fn run_internal( - mut rx: mpsc::Receiver>, - base_path: PathBuf, encoder_options: EncoderOptions, pcm_buffer_capacity: usize, + ) -> Self { + Self { + base_path: base_path.into(), + encoder_options, + pcm_buffer_capacity, + } + } +} + +#[async_trait::async_trait] +impl NodeLogic for FlacFileSinkLogic { + async fn process( + &mut self, + input: Option>>, + _output: Vec>>, stop_token: CancellationToken, ) -> Result<(), AudioError> { - + let mut rx = input.expect("FlacFileSink must have input"); let mut track_number = 0; + tracing::debug!("FlacFileSinkLogic::process started, base_path={:?}", self.base_path); + loop { // Vérifier si l'arrêt a été demandé if stop_token.is_cancelled() { + tracing::debug!("FlacFileSinkLogic cancelled"); return Ok(()); } @@ -116,6 +94,11 @@ impl FlacFileSink { let sample_rate = first_chunk.sample_rate(); let bits_per_sample = get_chunk_bit_depth(first_chunk); + tracing::debug!( + "FlacFileSinkLogic: encoding track {} with {}bit @ {}Hz", + track_number, bits_per_sample, sample_rate + ); + let format = PcmFormat { sample_rate, channels: 2, @@ -129,13 +112,13 @@ impl FlacFileSink { } // Générer le chemin du fichier pour cette track - let track_path = generate_track_path(&base_path, track_number); + let track_path = generate_track_path(&self.base_path, track_number); // Créer le pipeline d'encodage pour cette track - let (pcm_tx, pcm_rx) = mpsc::channel::>(pcm_buffer_capacity); + let (pcm_tx, pcm_rx) = mpsc::channel::>(self.pcm_buffer_capacity); // Préparer les options d'encodage avec les métadonnées du TrackBoundary - let mut options_with_metadata = encoder_options.clone(); + let mut options_with_metadata = self.encoder_options.clone(); options_with_metadata.metadata = track_metadata; // Créer l'encoder et le fichier @@ -189,6 +172,57 @@ impl FlacFileSink { } } +// ═══════════════════════════════════════════════════════════════════════════ +// FlacFileSink - Wrapper utilisant Node +// ═══════════════════════════════════════════════════════════════════════════ + +pub struct FlacFileSink { + inner: Node, +} + +impl FlacFileSink { + /// Crée un sink FLAC avec les options par défaut (compression 5, buffer de 16 segments). + /// + /// # Arguments + /// + /// * `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 { + Self::with_channel_size(base_path, DEFAULT_CHANNEL_SIZE) + } + + /// Crée un sink FLAC avec une taille de buffer MPSC personnalisée. + /// + /// # Arguments + /// + /// * `base_path` - Chemin de base pour les fichiers FLAC + /// * `channel_size` - Taille du buffer MPSC (nombre de segments en attente avant backpressure) + pub fn with_channel_size>( + base_path: P, + channel_size: usize, + ) -> Self { + Self::with_config(base_path, channel_size, EncoderOptions::default()) + } + + /// Crée un sink FLAC avec une configuration complète. + /// + /// # Arguments + /// + /// * `base_path` - Chemin de base pour les fichiers FLAC + /// * `channel_size` - Taille du buffer MPSC + /// * `encoder_options` - Options d'encodage FLAC (compression, etc.) + pub fn with_config>( + base_path: P, + channel_size: usize, + encoder_options: EncoderOptions, + ) -> Self { + let logic = FlacFileSinkLogic::new(base_path, encoder_options, 8); + Self { + inner: Node::new_with_input(logic, channel_size), + } + } +} + /// Génère le chemin de fichier pour une track donnée. /// - track 0 → base_path.flac /// - track 1 → base_path_01.flac @@ -210,14 +244,6 @@ fn generate_track_path(base_path: &Path, track_number: usize) -> PathBuf { } } -/// Signal retourné par pump_segments indiquant pourquoi l'encodage s'est arrêté. -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 ou si l'arrêt est demandé. async fn wait_for_first_audio_chunk_with_metadata( @@ -372,13 +398,13 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result match (chunk, bits_per_sample) { // I16 source (AudioChunk::I16(data), 16) => { - for frame in data.frames() { + for frame in data.get_frames() { bytes.extend_from_slice(&frame[0].to_le_bytes()); bytes.extend_from_slice(&frame[1].to_le_bytes()); } } (AudioChunk::I16(data), 24) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = (frame[0] as i32) << 8; let right = (frame[1] as i32) << 8; bytes.extend_from_slice(&left.to_le_bytes()[..3]); @@ -386,7 +412,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result } } (AudioChunk::I16(data), 32) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = (frame[0] as i32) << 16; let right = (frame[1] as i32) << 16; bytes.extend_from_slice(&left.to_le_bytes()); @@ -396,7 +422,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result // I24 source (AudioChunk::I24(data), 16) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = (frame[0].as_i32() >> 8) as i16; let right = (frame[1].as_i32() >> 8) as i16; bytes.extend_from_slice(&left.to_le_bytes()); @@ -404,13 +430,13 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result } } (AudioChunk::I24(data), 24) => { - for frame in data.frames() { + for frame in data.get_frames() { bytes.extend_from_slice(&frame[0].as_i32().to_le_bytes()[..3]); bytes.extend_from_slice(&frame[1].as_i32().to_le_bytes()[..3]); } } (AudioChunk::I24(data), 32) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = frame[0].as_i32() << 8; let right = frame[1].as_i32() << 8; bytes.extend_from_slice(&left.to_le_bytes()); @@ -420,7 +446,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result // I32 source (AudioChunk::I32(data), 16) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = (frame[0] >> 16) as i16; let right = (frame[1] >> 16) as i16; bytes.extend_from_slice(&left.to_le_bytes()); @@ -428,7 +454,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result } } (AudioChunk::I32(data), 24) => { - for frame in data.frames() { + for frame in data.get_frames() { let left = frame[0] >> 8; let right = frame[1] >> 8; bytes.extend_from_slice(&left.to_le_bytes()[..3]); @@ -436,7 +462,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result } } (AudioChunk::I32(data), 32) => { - for frame in data.frames() { + for frame in data.get_frames() { bytes.extend_from_slice(&frame[0].to_le_bytes()); bytes.extend_from_slice(&frame[1].to_le_bytes()); } @@ -529,7 +555,7 @@ pub struct FlacFileSinkStats { #[async_trait::async_trait] impl AudioPipelineNode for FlacFileSink { fn get_tx(&self) -> Option>> { - Some(self.tx.clone()) + self.inner.get_tx() } fn register(&mut self, _child: Box) { @@ -540,15 +566,7 @@ impl AudioPipelineNode for FlacFileSink { 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 + Box::new(self.inner).run(stop_token).await } } diff --git a/pmoaudio/src/nodes/http_source.rs b/pmoaudio/src/nodes/http_source.rs old mode 100644 new mode 100755 index 650b2edd..dec9ffd1 --- a/pmoaudio/src/nodes/http_source.rs +++ b/pmoaudio/src/nodes/http_source.rs @@ -1,5 +1,6 @@ use crate::{ nodes::{AudioError, TypedAudioNode, DEFAULT_CHUNK_DURATION_MS}, + pipeline::{Node, NodeLogic}, type_constraints::TypeRequirement, AudioChunk, AudioChunkData, AudioPipelineNode, AudioSegment, I24, }; @@ -90,69 +91,57 @@ use tokio_util::{io::StreamReader, sync::CancellationToken}; /// - Pas de buffering complet du fichier en mémoire /// - La taille des chunks audio est calculée automatiquement pour ~50ms de latence /// - Compatible avec les streams infinis (radios web, etc.) -pub struct HttpSource { + +// ═══════════════════════════════════════════════════════════════════════════ +// HttpSourceLogic - Logique métier pure +// ═══════════════════════════════════════════════════════════════════════════ + +/// Logique pure de lecture HTTP et décodage audio +pub struct HttpSourceLogic { url: String, chunk_frames: usize, - child_txs: Vec>>, - children: Vec>, } -impl HttpSource { - /// Crée une nouvelle source HTTP avec calcul automatique de la taille des chunks. - /// - /// La taille des chunks sera calculée automatiquement pour obtenir environ 50ms - /// de latence par chunk, en fonction du sample rate du fichier distant. - /// - /// # Arguments - /// - /// * `url` - URL HTTP ou HTTPS du fichier audio à télécharger - /// - /// # Exemples - /// - /// ```no_run - /// use pmoaudio::HttpSource; - /// - /// let source = HttpSource::new("http://example.com/music.flac"); - /// ``` - pub fn new>(url: S) -> Self { - Self::with_chunk_size(url, 0) - } - - /// Crée une nouvelle source HTTP avec une taille de chunk spécifique. - /// - /// # Arguments - /// - /// * `url` - URL HTTP ou HTTPS du fichier audio à télécharger - /// * `chunk_frames` - nombre d'échantillons par canal par chunk (0 = auto-calcul) - /// - /// # Exemples - /// - /// ```no_run - /// use pmoaudio::HttpSource; - /// - /// // Utiliser des chunks de 2048 frames - /// let source = HttpSource::with_chunk_size("http://example.com/music.mp3", 2048); - /// ``` - pub fn with_chunk_size>(url: S, chunk_frames: usize) -> Self { +impl HttpSourceLogic { + pub fn new>(url: S, chunk_frames: usize) -> Self { Self { url: url.into(), chunk_frames, - child_txs: Vec::new(), - children: Vec::new(), } } - async fn run_internal( - url: String, - chunk_frames: usize, - child_txs: Vec>>, + pub fn get_url(&self) -> String { + self.url.clone() + } + + pub fn get_chunc_frames(&self) -> usize { + self.chunk_frames + } +} + +#[async_trait::async_trait] +impl NodeLogic for HttpSourceLogic { + async fn process( + &mut self, + _input: Option>>, + output: Vec>>, stop_token: CancellationToken, ) -> Result<(), AudioError> { + macro_rules! send_to_children { + ($segment:expr) => { + for tx in &output { + tx.send($segment.clone()) + .await + .map_err(|_| AudioError::ChildDied)?; + } + }; + } + // Effectuer la requête HTTP - let response = reqwest::get(&url) + let response = reqwest::get(&self.url) .await .map_err(|e| { - AudioError::ProcessingError(format!("HTTP request failed for {}: {}", url, e)) + AudioError::ProcessingError(format!("HTTP request failed for {}: {}", self.url, e)) })?; // Vérifier le status @@ -160,12 +149,12 @@ impl HttpSource { return Err(AudioError::ProcessingError(format!( "HTTP request returned status {}: {}", response.status(), - url + self.url ))); } // Extraire les métadonnées depuis les headers HTTP - let metadata = extract_metadata_from_headers(&response, &url).await; + let metadata = extract_metadata_from_headers(&response, &self.url).await; // Convertir le stream de bytes en AsyncRead let bytes_stream = response.bytes_stream(); @@ -182,25 +171,24 @@ impl HttpSource { validate_stream(&stream_info)?; // Calculer la taille des chunks si non spécifiée (0 = auto) - let chunk_frames_final = if chunk_frames == 0 { + let chunk_frames_final = if self.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) + self.chunk_frames.max(1) }; // Émettre TopZeroSync - let top_zero = AudioSegment::new_top_zero_sync(); - for tx in &child_txs { - tx.send(top_zero.clone()).await.map_err(|_| AudioError::SendError)?; - } + send_to_children!(AudioSegment::new_top_zero_sync()); // É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))); - for tx in &child_txs { - tx.send(track_boundary.clone()).await.map_err(|_| AudioError::SendError)?; - } + 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; @@ -258,12 +246,7 @@ impl HttpSource { timestamp_sec, )?; - for tx in &child_txs { - if tx.send(segment.clone()).await.is_err() { - // Un enfant est mort, arrêter - return Ok(()); - } - } + send_to_children!(segment); chunk_index += 1; total_frames += frames_to_emit as u64; @@ -276,9 +259,7 @@ 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)?; - for tx in &child_txs { - let _ = tx.send(segment.clone()).await; - } + send_to_children!(segment); total_frames += frames as u64; chunk_index += 1; } @@ -287,9 +268,7 @@ 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); - for tx in &child_txs { - let _ = tx.send(eos.clone()).await; - } + send_to_children!(eos); // Attendre la fin du décodage stream @@ -301,6 +280,66 @@ impl HttpSource { } } +// ═══════════════════════════════════════════════════════════════════════════ +// HttpSource - Wrapper utilisant Node +// ═══════════════════════════════════════════════════════════════════════════ + +pub struct HttpSource { + inner: Node, +} + +impl HttpSource { + /// Crée une nouvelle source HTTP avec calcul automatique de la taille des chunks. + /// + /// La taille des chunks sera calculée automatiquement pour obtenir environ 50ms + /// de latence par chunk, en fonction du sample rate du fichier distant. + /// + /// # Arguments + /// + /// * `url` - URL HTTP ou HTTPS du fichier audio à télécharger + /// + /// # Exemples + /// + /// ```no_run + /// use pmoaudio::HttpSource; + /// + /// let source = HttpSource::new("http://example.com/music.flac"); + /// ``` + pub fn new>(url: S) -> Self { + Self::with_chunk_size(url, 0) + } + + /// Crée une nouvelle source HTTP avec une taille de chunk spécifique. + /// + /// # Arguments + /// + /// * `url` - URL HTTP ou HTTPS du fichier audio à télécharger + /// * `chunk_frames` - nombre d'échantillons par canal par chunk (0 = auto-calcul) + /// + /// # Exemples + /// + /// ```no_run + /// use pmoaudio::HttpSource; + /// + /// // Utiliser des chunks de 2048 frames + /// let source = HttpSource::with_chunk_size("http://example.com/music.mp3", 2048); + /// ``` + pub fn with_chunk_size>(url: S, chunk_frames: usize) -> Self { + let logic = HttpSourceLogic::new(url.into(), chunk_frames); + Self { + inner: Node::new_source(logic), + } + } + + pub fn get_url(&self) -> String { + self.inner.logic().get_url() + } + + pub fn get_chunc_frames(&self) -> usize { + self.inner.logic().get_chunc_frames() + } +} + /// Extrait les métadonnées disponibles depuis les headers HTTP async fn extract_metadata_from_headers( response: &reqwest::Response, @@ -477,55 +516,18 @@ 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 + self.inner.get_tx() } fn register(&mut self, child: Box) { - if let Some(tx) = child.get_tx() { - self.child_txs.push(tx); - } - self.children.push(child); + self.inner.register(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 + Box::new(self.inner).run(stop_token).await } } @@ -598,16 +600,16 @@ mod tests { #[test] fn test_http_source_creation() { let source = HttpSource::new("http://example.com/audio.flac"); - assert_eq!(source.url, "http://example.com/audio.flac"); - assert_eq!(source.chunk_frames, 0); + assert_eq!(source.get_url(), "http://example.com/audio.flac"); + assert_eq!(source.get_chunc_frames(), 0); } /// Test de création avec taille de chunk personnalisée #[test] fn test_http_source_with_chunk_size() { let source = HttpSource::with_chunk_size("http://example.com/audio.mp3", 1024); - assert_eq!(source.url, "http://example.com/audio.mp3"); - assert_eq!(source.chunk_frames, 1024); + assert_eq!(source.get_url(), "http://example.com/audio.mp3"); + assert_eq!(source.get_chunc_frames(), 1024); } /// Test de téléchargement et décodage d'un fichier FLAC via HTTP diff --git a/pmoaudio/src/nodes/mod.rs b/pmoaudio/src/nodes/mod.rs old mode 100644 new mode 100755 index 1af93b46..905da49a --- a/pmoaudio/src/nodes/mod.rs +++ b/pmoaudio/src/nodes/mod.rs @@ -119,6 +119,12 @@ pub enum AudioError { ProcessingError(String), /// Incompatibilité de types entre nodes TypeMismatch(TypeMismatch), + /// Un nœud enfant s'est terminé prématurément (anormal dans un pipeline descendant) + ChildFinished, + /// Un nœud enfant est mort (channel fermé pendant un send) + ChildDied, + /// Erreur d'I/O (fichier, réseau, etc.) + IoError(String), } impl std::fmt::Display for AudioError { @@ -128,6 +134,9 @@ impl std::fmt::Display for AudioError { AudioError::ReceiveError => write!(f, "Failed to receive audio chunk"), AudioError::ProcessingError(msg) => write!(f, "Processing error: {}", msg), AudioError::TypeMismatch(tm) => write!(f, "{}", tm), + AudioError::ChildFinished => write!(f, "Child node finished prematurely"), + AudioError::ChildDied => write!(f, "Child node died unexpectedly"), + AudioError::IoError(msg) => write!(f, "I/O error: {}", msg), } } } diff --git a/pmoaudio/src/pipeline.rs b/pmoaudio/src/pipeline.rs old mode 100644 new mode 100755 index c4d33131..5977b788 --- a/pmoaudio/src/pipeline.rs +++ b/pmoaudio/src/pipeline.rs @@ -43,6 +43,7 @@ use crate::{nodes::AudioError, AudioSegment}; use std::sync::Arc; use tokio::sync::mpsc; +use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; /// Trait pour les nœuds d'un pipeline audio @@ -107,4 +108,611 @@ pub trait AudioPipelineNode: Send + 'static { self: Box, stop_token: CancellationToken, ) -> Result<(), AudioError>; + + /// Lance le pipeline en arrière-plan et retourne un handle de contrôle + /// + /// Cette méthode est recommandée pour la plupart des cas d'usage. + /// Elle spawn le pipeline dans une tâche Tokio et retourne immédiatement + /// un `PipelineHandle` permettant de contrôler et surveiller l'exécution. + /// + /// # Retour + /// + /// Un `PipelineHandle` qui permet de : + /// - Arrêter le pipeline avec `stop()` + /// - Attendre sa complétion avec `wait()` + /// - Vérifier son état avec `is_finished()` + /// + /// # Exemple + /// + /// ```no_run + /// use pmoaudio::{FileSource, AudioPipelineNode}; + /// use pmoaudio::nodes::FlacFileSink; + /// + /// # async fn example() -> Result<(), Box> { + /// let mut source = FileSource::new("input.flac"); + /// source.register(Box::new(FlacFileSink::new("output.flac"))); + /// + /// // Lancer le pipeline + /// let handle = Box::new(source).start(); + /// + /// // Faire autre chose... + /// tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; + /// + /// // Arrêter et attendre + /// handle.stop(None); + /// handle.wait().await?; + /// # Ok(()) + /// # } + /// ``` + fn start(self: Box) -> PipelineHandle { + let stop_token = CancellationToken::new(); + let token_for_task = stop_token.clone(); + + let join_handle = tokio::spawn(async move { + self.run(token_for_task).await + }); + + PipelineHandle { + stop_token, + join_handle, + } + } +} + +// ═══════════════════════════════════════════════════════════════════════════════ +// NOUVELLE ARCHITECTURE - Séparation plomberie/logique métier +// ═══════════════════════════════════════════════════════════════════════════════ + +/// Raison de l'arrêt d'un nœud +/// +/// Passé à la méthode `cleanup()` pour permettre au nœud d'adapter +/// son comportement de nettoyage selon la cause de l'arrêt. +#[derive(Debug, Clone)] +pub enum StopReason { + /// Fin normale - toutes les données ont été traitées (EOF) + Completed, + + /// Cancel explicite demandé via CancellationToken + Cancelled, + + /// Un nœud enfant s'est terminé prématurément + /// (dans un pipeline descendant, ceci est anormal) + ChildFinished, + + /// Une erreur s'est produite (dans ce nœud ou un enfant) + Error(AudioError), +} + +/// Trait définissant la logique métier pure d'un nœud +/// +/// Ce trait sépare la logique de traitement spécifique au nœud (ce qu'il **fait**) +/// de la plomberie d'orchestration (spawning, monitoring, cleanup). +/// +/// # Responsabilités +/// +/// - Recevoir des données via `input` (None pour les sources) +/// - Traiter les données selon la logique du nœud +/// - Envoyer les résultats via `output` +/// - Surveiller `stop_token` pour arrêt rapide +/// - Optionnellement : cleanup contextualisé via `cleanup()` +/// +/// # Exemple +/// +/// ```no_run +/// use pmoaudio::pipeline::NodeLogic; +/// use pmoaudio::nodes::AudioError; +/// use tokio_util::sync::CancellationToken; +/// +/// struct MyProcessorLogic { +/// // Configuration du nœud +/// } +/// +/// #[async_trait::async_trait] +/// impl NodeLogic for MyProcessorLogic { +/// async fn process( +/// &mut self, +/// input: Option>>, +/// output: Vec>>, +/// stop_token: CancellationToken, +/// ) -> Result<(), AudioError> { +/// let mut rx = input.expect("Processor needs input"); +/// +/// loop { +/// tokio::select! { +/// _ = stop_token.cancelled() => break, +/// +/// segment = rx.recv() => { +/// match segment { +/// Some(data) => { +/// // Traiter les données +/// let processed = self.do_processing(data)?; +/// +/// // Envoyer aux enfants +/// for tx in &output { +/// tx.send(processed.clone()).await +/// .map_err(|_| AudioError::ChildDied)?; +/// } +/// } +/// None => break, // EOF +/// } +/// } +/// } +/// } +/// +/// Ok(()) +/// } +/// } +/// ``` +#[async_trait::async_trait] +pub trait NodeLogic: Send + 'static { + /// Logique de traitement du nœud + /// + /// # Arguments + /// + /// * `input` - Receiver pour les données entrantes (None pour les sources) + /// * `output` - Liste des senders vers les nœuds enfants + /// * `stop_token` - Token pour détecter les demandes d'arrêt + /// + /// # Retour + /// + /// - `Ok(())` : Arrêt propre (EOF, cancelled) + /// - `Err(...)` : Erreur de traitement + /// + /// # Comportement attendu + /// + /// - Surveiller `stop_token.cancelled()` dans la boucle principale + /// - Sortir proprement sur EOF (input.recv() → None) + /// - Gérer les erreurs de send (enfant mort) selon la politique du nœud + async fn process( + &mut self, + input: Option>>, + output: Vec>>, + stop_token: CancellationToken, + ) -> Result<(), AudioError>; + + /// Cleanup appelé automatiquement après l'arrêt du nœud + /// + /// Cette méthode permet au nœud de faire du nettoyage contextualisé + /// selon la raison de l'arrêt (fichiers incomplets, ressources, etc.) + /// + /// # Arguments + /// + /// * `reason` - La raison de l'arrêt du nœud + /// + /// # Implémentation par défaut + /// + /// Ne fait rien. Seulement les nœuds qui nécessitent un cleanup + /// (ex: Sinks) doivent implémenter cette méthode. + /// + /// # Exemple + /// + /// ```no_run + /// async fn cleanup(&mut self, reason: StopReason) -> Result<(), AudioError> { + /// match reason { + /// StopReason::Completed => { + /// // Finaliser le fichier proprement + /// self.flush_and_close().await?; + /// } + /// StopReason::Error(_) => { + /// // Supprimer le fichier incomplet + /// self.delete_incomplete_file().await?; + /// } + /// _ => { + /// // Autre cas selon politique + /// } + /// } + /// Ok(()) + /// } + /// ``` + async fn cleanup(&mut self, _reason: StopReason) -> Result<(), AudioError> { + Ok(()) + } +} + +/// Handle pour contrôler un pipeline en cours d'exécution +/// +/// Retourné par la méthode `start()`, ce handle permet de : +/// - Arrêter le pipeline explicitement +/// - Attendre sa complétion +/// - Vérifier s'il est toujours en cours +/// +/// # Exemple +/// +/// ```no_run +/// use pmoaudio::{FileSource, AudioPipelineNode}; +/// +/// # async fn example() -> Result<(), Box> { +/// let mut source = FileSource::new("input.flac"); +/// // ... register children ... +/// +/// let handle = Box::new(source).start(); +/// +/// // Faire autre chose... +/// tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; +/// +/// // Arrêter le pipeline +/// handle.stop(None); +/// +/// // Attendre la fin +/// handle.wait().await?; +/// # Ok(()) +/// # } +/// ``` +pub struct PipelineHandle { + stop_token: CancellationToken, + join_handle: JoinHandle>, +} + +impl PipelineHandle { + /// Demande l'arrêt du pipeline + /// + /// Cette méthode est non-bloquante. Pour attendre la fin effective, + /// utiliser `wait()` ou `stop_and_wait()`. + /// + /// # Arguments + /// + /// * `reason` - Raison optionnelle de l'arrêt (pour logging/debugging) + pub fn stop(&self, reason: Option) { + if let Some(err) = reason { + tracing::info!("Pipeline stop requested with error: {}", err); + } else { + tracing::info!("Pipeline stop requested"); + } + self.stop_token.cancel(); + } + + /// Attendre la complétion du pipeline + /// + /// Bloque jusqu'à ce que le pipeline se termine (normalement ou par erreur). + /// + /// # Retour + /// + /// Le résultat du nœud racine : + /// - `Ok(())` : Pipeline terminé avec succès + /// - `Err(...)` : Erreur survenue dans le pipeline + pub async fn wait(self) -> Result<(), AudioError> { + match self.join_handle.await { + Ok(result) => result, + Err(e) if e.is_panic() => Err(AudioError::ProcessingError( + format!("Pipeline task panicked: {}", e) + )), + Err(e) => Err(AudioError::ProcessingError( + format!("Pipeline task cancelled: {}", e) + )), + } + } + + /// Vérifie si le pipeline est toujours en cours d'exécution + pub fn is_finished(&self) -> bool { + self.join_handle.is_finished() + } + + /// Arrête le pipeline et attend sa complétion + /// + /// Équivalent à `stop()` suivi de `wait()`. + pub async fn stop_and_wait(self, reason: Option) -> Result<(), AudioError> { + self.stop(reason); + self.wait().await + } + + /// Obtient une copie du token d'arrêt + /// + /// Pour cas d'usage avancés nécessitant une intégration + /// avec d'autres systèmes utilisant CancellationToken. + pub fn cancellation_token(&self) -> CancellationToken { + self.stop_token.clone() + } +} + +/// Wrapper générique qui implémente l'orchestration d'un nœud +/// +/// Cette struct encapsule n'importe quelle logique métier (implémentant `NodeLogic`) +/// et fournit l'implémentation standard du trait `AudioPipelineNode` avec : +/// - Spawning automatique des enfants +/// - Monitoring des enfants pour détection d'arrêt prématuré +/// - Cleanup coordonné avec propagation de cancel +/// - Appel automatique de `cleanup()` selon le contexte +/// +/// # Type Parameters +/// +/// * `L` - Le type implémentant `NodeLogic`, qui contient la logique spécifique du nœud +/// +/// # Exemple +/// +/// ```no_run +/// use pmoaudio::pipeline::{Node, NodeLogic}; +/// +/// struct MyLogic { /* ... */ } +/// impl NodeLogic for MyLogic { /* ... */ } +/// +/// // Créer un nœud avec cette logique +/// let node = Node::new(MyLogic { /* ... */ }); +/// ``` +pub struct Node { + /// La logique métier du nœud + logic: L, + + /// Receiver pour les données entrantes (None pour les sources) + rx: Option>>, + + /// Sender pour les données entrantes (pour clonage via get_tx) + tx: Option>>, + + /// Liste des nœuds enfants + children: Vec>, + + /// Liste des senders vers les enfants + child_txs: Vec>>, +} + +impl Node { + /// Crée un nouveau nœud source (sans input) + /// + /// # Arguments + /// + /// * `logic` - La logique métier du nœud + pub fn new_source(logic: L) -> Self { + Self { + logic, + rx: None, + tx: None, + children: Vec::new(), + child_txs: Vec::new(), + } + } + + /// Crée un nouveau nœud avec input (converter ou sink) + /// + /// # Arguments + /// + /// * `logic` - La logique métier du nœud + /// * `buffer_size` - Taille du buffer du channel d'input + pub fn new_with_input(logic: L, buffer_size: usize) -> Self { + let (tx, rx) = mpsc::channel(buffer_size); + Self { + logic, + rx: Some(rx), + tx: Some(tx), + children: Vec::new(), + child_txs: Vec::new(), + } + } + + /// Retourne une référence vers la logique métier du nœud + pub fn logic(&self) -> &L { + &self.logic + } +} + +#[async_trait::async_trait] +impl AudioPipelineNode for Node { + fn get_tx(&self) -> Option>> { + self.tx.clone() + } + + 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( + mut self: Box, + stop_token: CancellationToken, + ) -> Result<(), AudioError> { + let Node { + mut logic, + rx, + children, + child_txs, + .. + } = *self; + + // ═══════════════════════════════════════════════════════════════════ + // PHASE 1: 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); + } + + // ═══════════════════════════════════════════════════════════════════ + // PHASE 2: MONITORER LES ENFANTS EN PARALLÈLE + // ═══════════════════════════════════════════════════════════════════ + + // Task qui surveille tous les enfants + // Si pas d'enfants, retourne None pour indiquer qu'il n'y a rien à surveiller + let mut child_monitor = if child_handles.is_empty() { + tracing::debug!("No children to monitor (terminal node)"); + None + } else { + let handles = child_handles; + let num_handles = handles.len(); + tracing::debug!("Child monitor starting with {} handles", num_handles); + Some(tokio::spawn(async move { + let mut has_error = false; + let mut first_error = None; + + for handle in handles { + match handle.await { + Ok(Ok(())) => { + // Un enfant s'est terminé proprement + // C'est normal dans un pipeline linéaire + tracing::debug!("Child finished successfully"); + continue; + } + Ok(Err(e)) => { + // Un enfant a eu une erreur + tracing::warn!("Child error: {}", e); + if !has_error { + first_error = Some(e); + has_error = true; + } + // Continue à surveiller les autres enfants + } + Err(e) => { + // Un enfant a paniqué + tracing::error!("Child panicked: {}", e); + if !has_error { + first_error = Some(AudioError::ProcessingError( + format!("Child task panicked: {}", e) + )); + has_error = true; + } + } + } + } + + // Retourner le résultat + if let Some(err) = first_error { + Err(err) + } else { + Ok(()) + } + })) + }; + + // ═══════════════════════════════════════════════════════════════════ + // PHASE 3: EXÉCUTER LA LOGIQUE MÉTIER EN RACE AVEC LE MONITORING + // ═══════════════════════════════════════════════════════════════════ + + let (stop_reason, process_result, child_monitor_consumed) = if let Some(monitor) = &mut child_monitor { + // Il y a des enfants à surveiller + tokio::select! { + // Cancel externe demandé + _ = stop_token.cancelled() => { + tracing::debug!("Node cancelled via stop_token"); + (StopReason::Cancelled, Ok(()), false) + } + + // Monitoring des enfants - retourne quand tous sont terminés ou sur erreur + child_result = monitor => { + match child_result { + Ok(Ok(())) => { + // Tous les enfants terminés avec succès + // Le parent devrait aussi terminer bientôt + tracing::debug!("All children finished successfully"); + (StopReason::Completed, Ok(()), true) + } + Ok(Err(e)) => { + // Un enfant a eu une erreur - arrêter immédiatement + tracing::warn!("Child error: {}", e); + (StopReason::Error(e.clone()), Err(e), true) + } + Err(e) => { + // Le monitor task a paniqué + let error = AudioError::ProcessingError( + format!("Child monitor panicked: {}", e) + ); + (StopReason::Error(error.clone()), Err(error), true) + } + } + } + + // Logique métier du nœud + process_result = logic.process(rx, child_txs.clone(), stop_token.clone()) => { + match process_result { + Ok(()) => { + tracing::debug!("Node process completed successfully"); + (StopReason::Completed, Ok(()), false) + } + Err(e) => { + tracing::error!("Node process error: {}", e); + (StopReason::Error(e.clone()), Err(e), false) + } + } + } + } + } else { + // Pas d'enfants (nœud terminal) - juste exécuter la logique + tokio::select! { + // Cancel externe demandé + _ = stop_token.cancelled() => { + tracing::debug!("Node cancelled via stop_token"); + (StopReason::Cancelled, Ok(()), true) // true car pas de monitor à attendre + } + + // Logique métier du nœud + process_result = logic.process(rx, child_txs.clone(), stop_token.clone()) => { + match process_result { + Ok(()) => { + tracing::debug!("Node process completed successfully (terminal)"); + (StopReason::Completed, Ok(()), true) // true car pas de monitor + } + Err(e) => { + tracing::error!("Node process error: {}", e); + (StopReason::Error(e.clone()), Err(e), true) // true car pas de monitor + } + } + } + } + }; + + // ═══════════════════════════════════════════════════════════════════ + // PHASE 4: CLEANUP COORDONNÉ + // ═══════════════════════════════════════════════════════════════════ + + // 4.1 Fermer les channels pour signaler EOF aux enfants + // Ceci permet aux enfants de finir de traiter les données restantes + drop(child_txs); + + // 4.2 Cancel pour arrêt d'urgence seulement en cas d'erreur ou d'annulation + // Si le nœud s'est terminé normalement, on laisse les enfants finir tranquillement + match &stop_reason { + StopReason::Completed => { + // Fin normale - les enfants vont se terminer naturellement après avoir traité les données + tracing::debug!("Node completed, letting children finish naturally"); + } + StopReason::Cancelled | StopReason::ChildFinished | StopReason::Error(_) => { + // Erreur ou annulation - forcer l'arrêt des enfants + tracing::debug!("Cancelling children due to: {:?}", stop_reason); + stop_token.cancel(); + } + } + + // 4.3 Attendre que les enfants finissent (si child_monitor n'a pas été consommé dans le select!) + if !child_monitor_consumed { + if let Some(monitor) = child_monitor { + tracing::debug!("Waiting for children to finish..."); + match monitor.await { + Ok(Ok(())) => { + tracing::debug!("All children finished successfully"); + } + Ok(Err(e)) => { + tracing::warn!("Child error during cleanup: {}", e); + // Si on n'avait pas d'erreur avant, propager celle-ci + if process_result.is_ok() { + return Err(e); + } + } + Err(e) => { + tracing::error!("Child monitor panicked during cleanup: {}", e); + } + } + } else { + tracing::debug!("No children to wait for (terminal node)"); + } + } + + // 4.4 Cleanup du nœud (contextualisé selon la raison) + if let Err(cleanup_err) = logic.cleanup(stop_reason).await { + tracing::error!("Cleanup failed: {}", cleanup_err); + // Si le cleanup échoue, propager cette erreur si process_result était Ok + if process_result.is_ok() { + return Err(cleanup_err); + } + } + + // ═══════════════════════════════════════════════════════════════════ + // PHASE 5: RETOURNER LE RÉSULTAT + // ═══════════════════════════════════════════════════════════════════ + + process_result + } } diff --git a/pmoaudio/src/sample_types.rs b/pmoaudio/src/sample_types.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/sync_marker.rs b/pmoaudio/src/sync_marker.rs old mode 100644 new mode 100755 diff --git a/pmoaudio/src/type_constraints.rs b/pmoaudio/src/type_constraints.rs old mode 100644 new mode 100755 diff --git a/pmoaudiocache/src/lib.rs b/pmoaudiocache/src/lib.rs old mode 100644 new mode 100755 diff --git a/pmoaudiocache/src/metadata_ext.rs b/pmoaudiocache/src/metadata_ext.rs old mode 100644 new mode 100755 diff --git a/pmocache/src/cache.rs b/pmocache/src/cache.rs old mode 100644 new mode 100755 diff --git a/pmoflac/Cargo.toml b/pmoflac/Cargo.toml old mode 100644 new mode 100755 diff --git a/pmoflac/src/autodetect.rs b/pmoflac/src/autodetect.rs old mode 100644 new mode 100755 diff --git a/pmoflac/src/encoder.rs b/pmoflac/src/encoder.rs old mode 100644 new mode 100755 diff --git a/pmoflac/src/lib.rs b/pmoflac/src/lib.rs old mode 100644 new mode 100755 diff --git a/pmoflac/src/metadata.rs b/pmoflac/src/metadata.rs old mode 100644 new mode 100755 diff --git a/pmoflac/src/pcm.rs b/pmoflac/src/pcm.rs old mode 100644 new mode 100755 diff --git a/pmoflac/src/prefixed_reader.rs b/pmoflac/src/prefixed_reader.rs old mode 100644 new mode 100755 diff --git a/pmoflac/src/transcode.rs b/pmoflac/src/transcode.rs old mode 100644 new mode 100755 diff --git a/pmometadata/src/lib.rs b/pmometadata/src/lib.rs old mode 100644 new mode 100755 diff --git a/pmoplaylist/src/track.rs b/pmoplaylist/src/track.rs old mode 100644 new mode 100755 diff --git a/pmosource/src/cache.rs b/pmosource/src/cache.rs old mode 100644 new mode 100755