Récupération de l'erreur git cleaning

Cargo.lock

@@ -2650,6 +2650,7 @@ dependencies = [
  "tokio-test",
  "tokio-util",
  "tracing",
+ "tracing-subscriber",
  "wiremock",
 ]
 

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<Arc<AudioSegment>>,
+// ═══════════════════════════════════════════════════════════════════════════
-    rx: mpsc::Receiver<Arc<AudioSegment>>,
+// FlacCacheSinkLogic - Logique métier pure
+// ═══════════════════════════════════════════════════════════════════════════
+
+/// Signal retourné par pump_segments indiquant pourquoi l'encodage s'est arrêté.
+enum StopReason {
+    TrackBoundary(Arc<RwLock<dyn pmometadata::TrackMetadata>>),
+    EndOfStream,
+    ChannelClosed,
+}
+
+/// Logique pure d'encodage FLAC vers le cache
+pub struct FlacCacheSinkLogic {
     cache: Arc<pmoaudiocache::Cache>,
     covers: Arc<pmocovers::Cache>,
     collection: Option<String>,
@@ -42,6 +54,207 @@ pub struct FlacCacheSink {
     playlist_handle: Option<Arc<pmoplaylist::WriteHandle>>,
 }
 
+impl FlacCacheSinkLogic {
+    pub fn new(
+        cache: Arc<pmoaudiocache::Cache>,
+        covers: Arc<pmocovers::Cache>,
+        collection: Option<String>,
+        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<pmoplaylist::WriteHandle>) {
+        self.playlist_handle = Some(handle);
+    }
+}
+
+#[async_trait::async_trait]
+impl NodeLogic for FlacCacheSinkLogic {
+    async fn process(
+        &mut self,
+        input: Option<mpsc::Receiver<Arc<AudioSegment>>>,
+        _output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
+        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::<Vec<u8>>(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<FlacCacheSinkLogic>
+// ═══════════════════════════════════════════════════════════════════════════
+
+pub struct FlacCacheSink {
+    inner: Node<FlacCacheSinkLogic>,
+    #[cfg(feature = "playlist")]
+    playlist_handle_pending: Option<Arc<pmoplaylist::WriteHandle>>,
+}
+
 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<String>,
     ) -> Self {
-        let (tx, rx) = mpsc::channel(channel_size);
+        let logic = FlacCacheSinkLogic::new(cache, covers, collection, encoder_options, 8);
         Self {
-            tx,
+            inner: Node::new_with_input(logic, channel_size),
-            rx,
-            cache,
-            covers,
-            collection,
-            encoder_options,
-            pcm_buffer_capacity: 8,
             #[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<FlacCacheSinkStats, AudioError> {
-        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::<Vec<u8>>(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<RwLock<dyn pmometadata::TrackMetadata>>),
-    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<Vec<u8>
     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<Vec<u8>
             }
         }
         (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<Vec<u8>
 
         // 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<Vec<u8>
             }
         }
         (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<Vec<u8>
 
         // 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<Vec<u8>
             }
         }
         (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<Vec<u8>
             }
         }
         (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<mpsc::Sender<Arc<AudioSegment>>> {
-        Some(self.tx.clone())
+        self.inner.get_tx()
     }
 
     fn register(&mut self, _child: Box<dyn AudioPipelineNode>) {
         panic!("FlacCacheSink is a terminal sink and cannot have children");
     }
 
-    async fn run(self: Box<Self>, stop_token: CancellationToken) -> Result<(), AudioError> {
+    async fn run(mut self: Box<Self>, stop_token: CancellationToken) -> Result<(), AudioError> {
-        self.run_internal(stop_token).await?;
+        // Transférer le playlist_handle_pending à la logique si présent
-        Ok(())
+        #[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
     }
 }
 

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"] }

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() {

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<dyn std::error::Error>> {
+    let path_str = std::env::args().nth(1).expect("Usage: check_flac_bits <file.flac>");
+    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(())
+}

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 -- <input_file> <output_file>
+//!
+//! 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<dyn std::error::Error>> {
+    // 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<String> = env::args().collect();
+    if args.len() != 3 {
+        eprintln!("Usage: {} <input_file> <output_file>", 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(())
+}

pmoaudio/src/audio_chunk.rs

@@ -105,25 +105,25 @@ impl<T: Sample> AudioChunkData<T> {
 
     /// 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::I16(d) => d.get_sample_rate(),
-            AudioChunk::I24(d) => d.sample_rate(),
+            AudioChunk::I24(d) => d.get_sample_rate(),
-            AudioChunk::I32(d) => d.sample_rate(),
+            AudioChunk::I32(d) => d.get_sample_rate(),
-            AudioChunk::F32(d) => d.sample_rate(),
+            AudioChunk::F32(d) => d.get_sample_rate(),
-            AudioChunk::F64(d) => d.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::I16(d) => d.get_gain_db(),
-            AudioChunk::I24(d) => d.gain_db(),
+            AudioChunk::I24(d) => d.get_gain_db(),
-            AudioChunk::I32(d) => d.gain_db(),
+            AudioChunk::I32(d) => d.get_gain_db(),
-            AudioChunk::F32(d) => d.gain_db(),
+            AudioChunk::F32(d) => d.get_gain_db(),
-            AudioChunk::F64(d) => d.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::I16(d) => d.get_sample_rate(),
-            AudioIntegerChunk::I24(d) => d.sample_rate(),
+            AudioIntegerChunk::I24(d) => d.get_sample_rate(),
-            AudioIntegerChunk::I32(d) => d.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::I16(d) => d.get_gain_db(),
-            AudioIntegerChunk::I24(d) => d.gain_db(),
+            AudioIntegerChunk::I24(d) => d.get_gain_db(),
-            AudioIntegerChunk::I32(d) => d.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<dyn Iterator<Item = [i32; 2]> + '_> {
         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::F32(d) => d.get_sample_rate(),
-            AudioFloatChunk::F64(d) => d.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::F32(d) => d.get_gain_db(),
-            AudioFloatChunk::F64(d) => d.gain_db(),
+            AudioFloatChunk::F64(d) => d.get_gain_db(),
         }
     }
 
@@ -829,9 +829,9 @@ impl AudioFloatChunk {
     pub fn frames(&self) -> Box<dyn Iterator<Item = [f64; 2]> + '_> {
         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]

pmoaudio/src/conversions.rs

@@ -27,7 +27,7 @@ pub fn convert_i32_to_i16(chunk: &AudioChunkData<i32>) -> Arc<AudioChunkData<i16
         .map(|[l, r]| [l as i16, r as i16])
         .collect();
 
-    AudioChunkData::new(stereo_i16, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(stereo_i16, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 /// Convertit i32 vers I24 (downsampling via bit depth change)
@@ -43,14 +43,14 @@ pub fn convert_i32_to_i24(chunk: &AudioChunkData<i32>) -> Arc<AudioChunkData<I24
         .map(|[l, r]| [I24::new_clamped(l), I24::new_clamped(r)])
         .collect();
 
-    AudioChunkData::new(stereo_i24, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(stereo_i24, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 /// Convertit i16 vers i32 (upsampling via bit depth change)
 pub fn convert_i16_to_i32(chunk: &AudioChunkData<i16>) -> Arc<AudioChunkData<i32>> {
     // 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<i16>) -> Arc<AudioChunkData<i32
     // Utiliser la fonction DSP optimisée pour passer de B16 → B32
     dsp::bitdepth_change_stereo(&mut stereo, BitDepth::B16, BitDepth::B32);
 
-    AudioChunkData::new(stereo, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(stereo, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 /// Convertit I24 vers i32 (upsampling via bit depth change)
 pub fn convert_i24_to_i32(chunk: &AudioChunkData<I24>) -> Arc<AudioChunkData<i32>> {
     // 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<I24>) -> Arc<AudioChunkData<i32
     // Utiliser la fonction DSP optimisée pour passer de B24 → B32
     dsp::bitdepth_change_stereo(&mut stereo, BitDepth::B24, BitDepth::B32);
 
-    AudioChunkData::new(stereo, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(stereo, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 // ============================================================================
@@ -84,7 +84,7 @@ pub fn convert_i24_to_i32(chunk: &AudioChunkData<I24>) -> Arc<AudioChunkData<i32
 ///
 /// I32 = 32 bits complets, donc normalisation par 2^31
 pub fn convert_i32_to_f32(chunk: &AudioChunkData<i32>) -> Arc<AudioChunkData<f32>> {
-    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<i32>) -> Arc<AudioChunkData<f32
     let mut out_pairs = vec![[0.0f32; 2]; len];
     dsp::i32_stereo_to_pairs_f32(&left, &right, &mut out_pairs, BitDepth::B32);
 
-    AudioChunkData::new(out_pairs, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(out_pairs, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 /// Convertit i32 vers f64
@@ -113,7 +113,7 @@ pub fn convert_i32_to_f64(chunk: &AudioChunkData<i32>) -> Arc<AudioChunkData<f64
 
 /// Convertit I24 vers f32 via les fonctions DSP optimisées SIMD
 pub fn convert_i24_to_f32(chunk: &AudioChunkData<I24>) -> Arc<AudioChunkData<f32>> {
-    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<I24>) -> Arc<AudioChunkData<f32
     let mut out_pairs = vec![[0.0f32; 2]; len];
     dsp::i24_as_i32_stereo_to_pairs_f32(&left, &right, &mut out_pairs);
 
-    AudioChunkData::new(out_pairs, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(out_pairs, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 /// Convertit I24 vers f64
 pub fn convert_i24_to_f64(chunk: &AudioChunkData<I24>) -> Arc<AudioChunkData<f64>> {
-    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<I24>) -> Arc<AudioChunkData<f64
         })
         .collect();
 
-    AudioChunkData::new(stereo, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(stereo, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 /// Convertit i16 vers f32 via les fonctions DSP optimisées SIMD
 pub fn convert_i16_to_f32(chunk: &AudioChunkData<i16>) -> Arc<AudioChunkData<f32>> {
-    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<i16>) -> Arc<AudioChunkData<f32
     let mut out_pairs = vec![[0.0f32; 2]; len];
     dsp::i16_stereo_to_pairs_f32(&left, &right, &mut out_pairs);
 
-    AudioChunkData::new(out_pairs, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(out_pairs, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 /// Convertit i16 vers f64
 pub fn convert_i16_to_f64(chunk: &AudioChunkData<i16>) -> Arc<AudioChunkData<f64>> {
-    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<i16>) -> Arc<AudioChunkData<f64
         })
         .collect();
 
-    AudioChunkData::new(stereo, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(stereo, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 // ============================================================================
@@ -193,7 +193,7 @@ pub fn convert_i16_to_f64(chunk: &AudioChunkData<i16>) -> Arc<AudioChunkData<f64
 ///
 /// I32 = 32 bits complets, donc quantization vers ±2^31
 pub fn convert_f32_to_i32(chunk: &AudioChunkData<f32>) -> Arc<AudioChunkData<i32>> {
-    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<f32>) -> Arc<AudioChunkData<i32
         .map(|(l, r)| [l, r])
         .collect();
 
-    AudioChunkData::new(stereo, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(stereo, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 /// Convertit f64 vers i32 (via f32)
@@ -222,7 +222,7 @@ pub fn convert_f64_to_i32(chunk: &AudioChunkData<f64>) -> Arc<AudioChunkData<i32
 
 /// Convertit f32 vers I24 via les fonctions DSP optimisées SIMD
 pub fn convert_f32_to_i24(chunk: &AudioChunkData<f32>) -> Arc<AudioChunkData<I24>> {
-    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<f32>) -> Arc<AudioChunkData<I24
         .map(|(l, r)| [I24::new_clamped(l), I24::new_clamped(r)])
         .collect();
 
-    AudioChunkData::new(stereo, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(stereo, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 /// Convertit f64 vers I24
 pub fn convert_f64_to_i24(chunk: &AudioChunkData<f64>) -> Arc<AudioChunkData<I24>> {
-    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<f64>) -> Arc<AudioChunkData<I24
         })
         .collect();
 
-    AudioChunkData::new(stereo, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(stereo, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 /// Convertit f32 vers i16 via les fonctions DSP optimisées SIMD
 pub fn convert_f32_to_i16(chunk: &AudioChunkData<f32>) -> Arc<AudioChunkData<i16>> {
-    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<f32>) -> Arc<AudioChunkData<i16
         .map(|(l, r)| [l, r])
         .collect();
 
-    AudioChunkData::new(stereo, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(stereo, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 /// Convertit f64 vers i16
 pub fn convert_f64_to_i16(chunk: &AudioChunkData<f64>) -> Arc<AudioChunkData<i16>> {
-    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<f64>) -> Arc<AudioChunkData<i16
         })
         .collect();
 
-    AudioChunkData::new(stereo, chunk.sample_rate(), chunk.gain_db())
+    AudioChunkData::new(stereo, chunk.get_sample_rate(), chunk.get_gain_db())
 }
 
 // ============================================================================
@@ -302,18 +302,18 @@ pub fn convert_f64_to_i16(chunk: &AudioChunkData<f64>) -> Arc<AudioChunkData<i16
 
 /// Convertit f32 vers f64 (upcast simple)
 pub fn convert_f32_to_f64(chunk: &AudioChunkData<f32>) -> Arc<AudioChunkData<f64>> {
-    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<f64>) -> Arc<AudioChunkData<f32>> {
-    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<AudioChunkData<I24>> = (&*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);
         }

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);
         });
     }
 

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<ConverterLogic<F>>` 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},
+    nodes::AudioError,
-    type_constraints::{SampleType, TypeRequirement},
+    pipeline::{Node, NodeLogic},
-    AudioPipelineNode, AudioSegment,
+    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
+/// Logique de conversion générique
-macro_rules! converter_node {
+///
-    ($node_name:ident, $convert_method:ident, $output_type:expr, $doc:expr) => {
+/// Cette struct contient la logique pure de conversion d'un type vers un autre.
-        #[doc = $doc]
+/// Elle reçoit des segments, convertit les chunks audio, et relay les syncmarkers.
-        pub struct $node_name {
+pub struct ConverterLogic<F> {
-            tx: mpsc::Sender<Arc<AudioSegment>>,
+    convert_fn: F,
-            rx: mpsc::Receiver<Arc<AudioSegment>>,
-            child_txs: Vec<mpsc::Sender<Arc<AudioSegment>>>,
-            children: Vec<Box<dyn AudioPipelineNode>>,
-        }
-
-        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<mpsc::Sender<Arc<AudioSegment>>> {
-                Some(self.tx.clone())
-            }
-
-            fn register(&mut self, child: Box<dyn AudioPipelineNode>) {
-                if let Some(tx) = child.get_tx() {
-                    self.child_txs.push(tx);
-                }
-                self.children.push(child);
-            }
-
-            async fn run(
-                mut self: Box<Self>,
-                stop_token: CancellationToken,
-            ) -> Result<(), AudioError> {
-                // Spawner tous les enfants
-                let mut child_handles = Vec::new();
-                for child in self.children {
-                    let child_token = stop_token.child_token();
-                    let handle = tokio::spawn(async move { child.run(child_token).await });
-                    child_handles.push(handle);
-                }
-
-                // Boucle de traitement
-                loop {
-                    let segment = tokio::select! {
-                        result = self.rx.recv() => {
-                            match result {
-                                Some(seg) => seg,
-                                None => break,
-                            }
-                        }
-                        _ = stop_token.cancelled() => {
-                            break;
-                        }
-                    };
-
-                    // 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<TypeRequirement> {
-                Some(TypeRequirement::any())
-            }
-
-            fn output_type(&self) -> Option<TypeRequirement> {
-                Some(TypeRequirement::specific($output_type))
-            }
-        }
-
-        impl Default for $node_name {
-            fn default() -> Self {
-                Self::new()
-            }
-        }
-    };
 }
 
-// Générer les 5 converter nodes
+impl<F> ConverterLogic<F>
-converter_node!(
+where
-    ToI16Node,
+    F: Fn(&AudioChunk) -> AudioChunk + Send + 'static,
-    to_i16,
+{
-    SampleType::I16,
+    pub fn new(convert_fn: F) -> Self {
-    "Node de conversion vers I16 (16-bit signed integer)"
+        Self { convert_fn }
-);
+    }
-converter_node!(
+}
-    ToI24Node,
+
-    to_i24,
+#[async_trait::async_trait]
-    SampleType::I24,
+impl<F> NodeLogic for ConverterLogic<F>
-    "Node de conversion vers I24 (24-bit signed integer)"
+where
-);
+    F: Fn(&AudioChunk) -> AudioChunk + Send + 'static,
-converter_node!(
+{
-    ToI32Node,
+    async fn process(
-    to_i32,
+        &mut self,
-    SampleType::I32,
+        input: Option<mpsc::Receiver<Arc<AudioSegment>>>,
-    "Node de conversion vers I32 (32-bit signed integer)"
+        output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
-);
+        stop_token: CancellationToken,
-converter_node!(
+    ) -> Result<(), AudioError> {
-    ToF32Node,
+        let mut rx = input.expect("Converter must have input");
-    to_f32,
+        tracing::debug!("ConverterLogic::process started, {} children", output.len());
-    SampleType::F32,
+
-    "Node de conversion vers F32 (32-bit floating point)"
+        loop {
-);
+            let segment = tokio::select! {
-converter_node!(
+                _ = stop_token.cancelled() => {
-    ToF64Node,
+                    tracing::debug!("ConverterLogic cancelled");
-    to_f64,
+                    break;
-    SampleType::F64,
+                }
-    "Node de conversion vers F64 (64-bit floating point)"
+
-);
+                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<dyn AudioPipelineNode> {
+        Self::with_channel_size(16)
+    }
+
+    pub fn with_channel_size(channel_size: usize) -> Box<dyn AudioPipelineNode> {
+        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<dyn AudioPipelineNode> {
+        Self::with_channel_size(16)
+    }
+
+    pub fn with_channel_size(channel_size: usize) -> Box<dyn AudioPipelineNode> {
+        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<dyn AudioPipelineNode> {
+        Self::with_channel_size(16)
+    }
+
+    pub fn with_channel_size(channel_size: usize) -> Box<dyn AudioPipelineNode> {
+        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<dyn AudioPipelineNode> {
+        Self::with_channel_size(16)
+    }
+
+    pub fn with_channel_size(channel_size: usize) -> Box<dyn AudioPipelineNode> {
+        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<dyn AudioPipelineNode> {
+        Self::with_channel_size(16)
+    }
+
+    pub fn with_channel_size(channel_size: usize) -> Box<dyn AudioPipelineNode> {
+        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() {
+    async fn test_converter_logic() {
-        let node = ToF32Node::new();
+        // Test unitaire de la logique pure
+        let mut logic = ConverterLogic::new(|chunk: &AudioChunk| chunk.to_f32());
 
-        // Vérifier les types d'entrée/sortie
+        let (input_tx, input_rx) = mpsc::channel(10);
-        assert_eq!(
+        let (output_tx, mut output_rx) = mpsc::channel(10);
-            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<Arc<AudioSegment>>,
-        input_rx: mpsc::Receiver<Arc<AudioSegment>>,
-        output_tx: mpsc::Sender<Arc<AudioSegment>>,
-    }
-
-    impl TestCollectorNode {
-        fn new(output_tx: mpsc::Sender<Arc<AudioSegment>>) -> Self {
-            let (input_tx, input_rx) = mpsc::channel(16);
-            Self {
-                input_tx,
-                input_rx,
-                output_tx,
-            }
-        }
-    }
-
-    #[async_trait::async_trait]
-    impl AudioPipelineNode for TestCollectorNode {
-        fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
-            Some(self.input_tx.clone())
-        }
-
-        fn register(&mut self, _child: Box<dyn AudioPipelineNode>) {
-            panic!("TestCollectorNode is a sink");
-        }
-
-        async fn run(
-            mut self: Box<Self>,
-            _stop_token: CancellationToken,
-        ) -> Result<(), AudioError> {
-            while let Some(segment) = self.input_rx.recv().await {
-                if self.output_tx.send(segment).await.is_err() {
-                    break;
-                }
-            }
-            Ok(())
-        }
-    }
-
-    #[tokio::test]
-    async fn test_to_i16_node_converts_from_i32() {
-        let mut node = 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 stop_token = CancellationToken::new();
-        let handle = tokio::spawn(async move { Box::new(node).run(stop_token).await });
 
-        // Créer et envoyer un chunk I32
+        // Créer un chunk de test
-        let stereo = vec![[1_000_000i32 << 16, -500_000i32 << 16]; 100];
+        let test_chunk = AudioChunk::I16(AudioChunkData::new(
-        let chunk_data = AudioChunkData::new(stereo.clone(), 48_000, 0.0);
+            vec![[100, 200], [300, 400]],
-        let chunk = AudioChunk::I32(chunk_data);
+            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();
+        // Envoyer le segment
-        drop(tx);
+        input_tx.send(segment).await.unwrap();
+        drop(input_tx); // EOF
 
-        // Recevoir le chunk converti
+        // Lancer le traitement
-        let result = out_rx.recv().await.unwrap();
+        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() {
+        if let Some(chunk) = result.as_chunk() {
-            assert_eq!(converted.type_name(), "i16");
+            assert!(matches!(chunk.as_ref(), AudioChunk::F32(_)));
-            assert_eq!(converted.len(), 100);
+        } else {
-
+            panic!("Expected audio chunk");
-            // 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);
-                }
-            }
         }
-
-        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());
     }
 }

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<FileSourceLogic>).
+pub struct FileSourceLogic {
+    path: PathBuf,
+    chunk_frames: usize,
+}
+
+impl FileSourceLogic {
+    pub fn new<P: Into<PathBuf>>(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<mpsc::Receiver<Arc<AudioSegment>>>,
+        output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
+        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::<Vec<u8>>();
+
+                    // Calculer le timestamp
+                    let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64;
+
+                    // Créer et envoyer le segment audio
+                    let segment = bytes_to_segment(
+                        &chunk_bytes,
+                        &stream_info,
+                        frames_to_emit,
+                        chunk_index,
+                        timestamp_sec,
+                    )?;
+
+                    send_to_children!(segment);
+
+                    chunk_index += 1;
+                    total_frames += frames_to_emit as u64;
+                }
+            }
+        }
+
+        // Traiter le reste éventuel (moins qu'un chunk complet)
+        if !pending.is_empty() {
+            let frames = pending.len() / frame_bytes;
+            if frames > 0 {
+                let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64;
+                let segment =
+                    bytes_to_segment(&pending, &stream_info, frames, chunk_index, timestamp_sec)?;
+                send_to_children!(segment);
+                total_frames += frames as u64;
+                chunk_index += 1;
+            }
+        }
+
+        // Émettre EndOfStream
+        let final_timestamp = total_frames as f64 / stream_info.sample_rate as f64;
+        let eos = AudioSegment::new_end_of_stream(chunk_index, final_timestamp);
+        send_to_children!(eos);
+
+        // Attendre la fin du décodage
+        stream
+            .wait()
+            .await
+            .map_err(|e| AudioError::ProcessingError(format!("Decode task failed: {}", e)))?;
+
+        Ok(())
+    }
+}
+
+// ═════════════════════════════════════════════════════════════════════════════
+// WRAPPER FileSource - Délègue à Node<FileSourceLogic>
+// ═══════════════════════════════════════════════════════════════════════════════
+
 /// 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<FileSourceLogic>` pour séparer
+/// la logique métier (décodage) de la plomberie (spawning, monitoring).
 pub struct FileSource {
-    path: PathBuf,
+    inner: Node<FileSourceLogic>,
-    chunk_frames: usize,
-    child_txs: Vec<mpsc::Sender<Arc<AudioSegment>>>,
-    children: Vec<Box<dyn AudioPipelineNode>>,
 }
 
 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<P: Into<PathBuf>>(path: P, chunk_frames: usize) -> Self {
+        let logic = FileSourceLogic::new(path, chunk_frames);
         Self {
-            path: path.into(),
+            inner: Node::new_source(logic),
-            chunk_frames,
-            child_txs: Vec::new(),
-            children: Vec::new(),
         }
     }
 }
 
+#[async_trait::async_trait]
+impl AudioPipelineNode for FileSource {
+    fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
+        self.inner.get_tx()
+    }
+
+    fn register(&mut self, child: Box<dyn AudioPipelineNode>) {
+        self.inner.register(child)
+    }
+
+    async fn run(
+        self: Box<Self>,
+        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<mpsc::Sender<Arc<AudioSegment>>> {
-        // FileSource est une source, elle n'a pas d'input
-        None
-    }
-
-    fn register(&mut self, child: Box<dyn AudioPipelineNode>) {
-        // Extraire le tx du child avant de le stocker
-        if let Some(tx) = child.get_tx() {
-            self.child_txs.push(tx.clone());
-        }
-        self.children.push(child);
-    }
-
-    async fn run(
-        self: Box<Self>,
-        stop_token: CancellationToken,
-    ) -> Result<(), AudioError> {
-        let Self {
-            path,
-            chunk_frames,
-            child_txs,
-            children,
-        } = *self;
-
-        // 1. Spawner tous les enfants AVANT de commencer à lire
-        let mut child_handles = Vec::new();
-        for child in children {
-            let child_token = stop_token.child_token();
-            let handle = tokio::spawn(async move {
-                child.run(child_token).await
-            });
-            child_handles.push(handle);
-        }
-
-        // 2. Faire le travail de lecture du fichier
-        let work_result: Result<(), AudioError> = async {
-            // Macro helper pour envoyer à tous les enfants
-            macro_rules! send_to_children {
-                ($segment:expr) => {
-                    for tx in &child_txs {
-                        if tx.send($segment.clone()).await.is_err() {
-                            tracing::warn!("FileSource: child died during send");
-                            return Err(AudioError::SendError);
-                        }
-                    }
-                };
-            }
-
-            // Ouvrir le fichier
-            let file = File::open(&path).await.map_err(|e| {
-                AudioError::ProcessingError(format!("Failed to open {:?}: {}", path, e))
-            })?;
-
-            // Décoder le flux audio
-            let mut stream = decode_audio_stream(file)
-                .await
-                .map_err(|e| AudioError::ProcessingError(format!("Decode error: {}", e)))?;
-            let stream_info = stream.info().clone();
-
-            validate_stream(&stream_info)?;
-
-            // Calculer la taille des chunks si non spécifiée (0 = auto)
-            let chunk_frames = if chunk_frames == 0 {
-                let frames =
-                    (stream_info.sample_rate as f64 * DEFAULT_CHUNK_DURATION_MS / 1000.0) as usize;
-                frames.next_power_of_two().max(256)
-            } else {
-                chunk_frames.max(1)
-            };
-
-            // Émettre TopZeroSync
-            let top_zero = AudioSegment::new_top_zero_sync();
-            send_to_children!(top_zero);
-
-            // Extraire et émettre les métadonnées du fichier
-            if let Ok(file_metadata) = AudioFileMetadata::from_file(&path) {
-                let mut metadata = MemoryTrackMetadata::new();
-                if let Some(title) = file_metadata.title {
-                    let _ = metadata.set_title(Some(title)).await;
-                }
-                if let Some(artist) = file_metadata.artist {
-                    let _ = metadata.set_artist(Some(artist)).await;
-                }
-                if let Some(album) = file_metadata.album {
-                    let _ = metadata.set_album(Some(album)).await;
-                }
-                if let Some(year) = file_metadata.year {
-                    let _ = metadata.set_year(Some(year)).await;
-                }
-                if let Some(duration_secs) = file_metadata.duration_secs {
-                    let _ = metadata
-                        .set_duration(Some(Duration::from_secs(duration_secs)))
-                        .await;
-                }
-
-                let track_boundary = AudioSegment::new_track_boundary(
-                    0,
-                    0.0,
-                    Arc::new(tokio::sync::RwLock::new(metadata)),
-                );
-                send_to_children!(track_boundary);
-            }
-
-            // Préparer la lecture des chunks audio
-            let frame_bytes = stream_info.bytes_per_sample() * stream_info.channels as usize;
-            let chunk_byte_len = chunk_frames * frame_bytes;
-            let mut pending = Vec::new();
-            let mut read_buf = vec![0u8; frame_bytes * 512.max(chunk_frames)];
-            let mut chunk_index = 0u64;
-            let mut total_frames = 0u64;
-
-            // Lire et émettre les chunks audio
-            loop {
-                tokio::select! {
-                    _ = stop_token.cancelled() => {
-                        tracing::info!("FileSource: stop requested");
-                        break;
-                    }
-
-                    read_result = stream.read(&mut read_buf) => {
-                        // Remplir le buffer
-                        if pending.len() < chunk_byte_len {
-                            let read = read_result.map_err(|e| {
-                                AudioError::ProcessingError(format!("I/O error while decoding: {}", e))
-                            })?;
-                            if read == 0 && pending.is_empty() {
-                                break;
-                            }
-                            if read > 0 {
-                                pending.extend_from_slice(&read_buf[..read]);
-                            }
-                        }
-
-                        if pending.is_empty() {
-                            break;
-                        }
-
-                        // Extraire un chunk
-                        let frames_in_pending = pending.len() / frame_bytes;
-                        let frames_to_emit = frames_in_pending.min(chunk_frames);
-                        if frames_to_emit == 0 {
-                            break;
-                        }
-                        let take_bytes = frames_to_emit * frame_bytes;
-                        let chunk_bytes = pending.drain(..take_bytes).collect::<Vec<u8>>();
-
-                        // Calculer le timestamp
-                        let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64;
-
-                        // Créer et envoyer le segment audio
-                        let segment = bytes_to_segment(
-                            &chunk_bytes,
-                            &stream_info,
-                            frames_to_emit,
-                            chunk_index,
-                            timestamp_sec,
-                        )?;
-                        send_to_children!(segment);
-
-                        chunk_index += 1;
-                        total_frames += frames_to_emit as u64;
-                    }
-                }
-            }
-
-            // Traiter le reste éventuel (moins qu'un chunk complet)
-            if !pending.is_empty() {
-                let frames = pending.len() / frame_bytes;
-                if frames > 0 {
-                    let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64;
-                    let segment =
-                        bytes_to_segment(&pending, &stream_info, frames, chunk_index, timestamp_sec)?;
-                    send_to_children!(segment);
-                    total_frames += frames as u64;
-                    chunk_index += 1;
-                }
-            }
-
-            // Émettre EndOfStream
-            let final_timestamp = total_frames as f64 / stream_info.sample_rate as f64;
-            let eos = AudioSegment::new_end_of_stream(chunk_index, final_timestamp);
-            send_to_children!(eos);
-
-            // Attendre la fin du décodage
-            stream
-                .wait()
-                .await
-                .map_err(|e| AudioError::ProcessingError(format!("Decode task failed: {}", e)))?;
-
-            Ok(())
-        }.await;
-
-        // 3. Arrêter les enfants qui tournent encore (descendant uniquement)
-        stop_token.cancel();
-
-        // 4. Fermer les channels pour signaler EOF
-        drop(child_txs);
-
-        // 5. Attendre que TOUS les enfants se terminent
-        for handle in child_handles {
-            match handle.await {
-                Ok(Ok(())) => {
-                    // Enfant terminé normalement
-                }
-                Ok(Err(e)) => {
-                    // Enfant en erreur → propager
-                    tracing::error!("FileSource: child error: {}", e);
-                    return Err(e);
-                }
-                Err(e) => {
-                    // Panic dans l'enfant
-                    tracing::error!("FileSource: child panic: {}", e);
-                    return Err(AudioError::ProcessingError(format!("Child panic: {}", e)));
-                }
-            }
-        }
-
-        // 6. Retourner notre propre résultat (montant vers le parent)
-        work_result
-    }
-}
 
 impl TypedAudioNode for FileSource {
     fn input_type(&self) -> Option<TypeRequirement> {

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<Arc<AudioSegment>>,
+// ═══════════════════════════════════════════════════════════════════════════
-    rx: mpsc::Receiver<Arc<AudioSegment>>,
+// FlacFileSinkLogic - Logique métier pure
+// ═══════════════════════════════════════════════════════════════════════════
+
+/// Signal retourné par pump_segments indiquant pourquoi l'encodage s'est arrêté.
+enum StopReason {
+    TrackBoundary(Arc<tokio::sync::RwLock<dyn pmometadata::TrackMetadata>>),
+    EndOfStream,
+    ChannelClosed,
+    Cancelled,
+}
+
+/// Logique pure d'encodage FLAC
+pub struct FlacFileSinkLogic {
     base_path: PathBuf,
     encoder_options: EncoderOptions,
     pcm_buffer_capacity: usize,
 }
 
-impl FlacFileSink {
+impl FlacFileSinkLogic {
-    /// Crée un sink FLAC avec les options par défaut (compression 5, buffer de 16 segments).
+    pub fn new<P: Into<PathBuf>>(
-    ///
-    /// # 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<P: Into<PathBuf>>(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<P: Into<PathBuf>>(
         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<P: Into<PathBuf>>(
-        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<Arc<AudioSegment>>,
-        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<mpsc::Receiver<Arc<AudioSegment>>>,
+        _output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
         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::<Vec<u8>>(pcm_buffer_capacity);
+            let (pcm_tx, pcm_rx) = mpsc::channel::<Vec<u8>>(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<FlacFileSinkLogic>
+// ═══════════════════════════════════════════════════════════════════════════
+
+pub struct FlacFileSink {
+    inner: Node<FlacFileSinkLogic>,
+}
+
+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<P: Into<PathBuf>>(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<P: Into<PathBuf>>(
+        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<P: Into<PathBuf>>(
+        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<tokio::sync::RwLock<dyn pmometadata::TrackMetadata>>),
-    EndOfStream,
-    ChannelClosed,
-    Cancelled,
-}
-
 /// Attend et retourne le premier chunk audio avec les métadonnées du TrackBoundary si présent.
 /// Retourne une erreur si EndOfStream est reçu avant tout audio 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<Vec<u8>
     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<Vec<u8>
             }
         }
         (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<Vec<u8>
 
         // 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<Vec<u8>
             }
         }
         (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<Vec<u8>
 
         // 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<Vec<u8>
             }
         }
         (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<Vec<u8>
             }
         }
         (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<mpsc::Sender<Arc<AudioSegment>>> {
-        Some(self.tx.clone())
+        self.inner.get_tx()
     }
 
     fn register(&mut self, _child: Box<dyn AudioPipelineNode>) {
@@ -540,15 +566,7 @@ impl AudioPipelineNode for FlacFileSink {
         self: Box<Self>,
         stop_token: CancellationToken,
     ) -> Result<(), AudioError> {
-        let FlacFileSink {
+        Box::new(self.inner).run(stop_token).await
-            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
     }
 }
 

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<mpsc::Sender<Arc<AudioSegment>>>,
-    children: Vec<Box<dyn AudioPipelineNode>>,
 }
 
-impl HttpSource {
+impl HttpSourceLogic {
-    /// Crée une nouvelle source HTTP avec calcul automatique de la taille des chunks.
+    pub fn new<S: Into<String>>(url: S, chunk_frames: usize) -> Self {
-    ///
-    /// 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<S: Into<String>>(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<S: Into<String>>(url: S, chunk_frames: usize) -> Self {
         Self {
             url: url.into(),
             chunk_frames,
-            child_txs: Vec::new(),
-            children: Vec::new(),
         }
     }
 
-    async fn run_internal(
+    pub fn get_url(&self) -> String {
-        url: String,
+        self.url.clone()
-        chunk_frames: usize,
+    }
-        child_txs: Vec<mpsc::Sender<Arc<AudioSegment>>>,
+
+    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<mpsc::Receiver<Arc<AudioSegment>>>,
+        output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
         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();
+        send_to_children!(AudioSegment::new_top_zero_sync());
-        for tx in &child_txs {
-            tx.send(top_zero.clone()).await.map_err(|_| AudioError::SendError)?;
-        }
 
         // Émettre TrackBoundary avec les métadonnées HTTP
-        let track_boundary = AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata)));
+        let track_boundary = AudioSegment::new_track_boundary(
-        for tx in &child_txs {
+            0,
-            tx.send(track_boundary.clone()).await.map_err(|_| AudioError::SendError)?;
+            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 {
+            send_to_children!(segment);
-                if tx.send(segment.clone()).await.is_err() {
-                    // Un enfant est mort, arrêter
-                    return Ok(());
-                }
-            }
 
             chunk_index += 1;
             total_frames += frames_to_emit as u64;
@@ -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 {
+                send_to_children!(segment);
-                    let _ = tx.send(segment.clone()).await;
-                }
                 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 {
+        send_to_children!(eos);
-            let _ = tx.send(eos.clone()).await;
-        }
 
         // Attendre la fin du décodage
         stream
@@ -301,6 +280,66 @@ impl HttpSource {
     }
 }
 
+// ═══════════════════════════════════════════════════════════════════════════
+// HttpSource - Wrapper utilisant Node<HttpSourceLogic>
+// ═══════════════════════════════════════════════════════════════════════════
+
+pub struct HttpSource {
+    inner: Node<HttpSourceLogic>,
+}
+
+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<S: Into<String>>(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<S: Into<String>>(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<mpsc::Sender<Arc<AudioSegment>>> {
-        None // HttpSource est une source, pas d'input
+        self.inner.get_tx()
     }
 
     fn register(&mut self, child: Box<dyn AudioPipelineNode>) {
-        if let Some(tx) = child.get_tx() {
+        self.inner.register(child)
-            self.child_txs.push(tx);
-        }
-        self.children.push(child);
     }
 
     async fn run(
         self: Box<Self>,
         stop_token: CancellationToken,
     ) -> Result<(), AudioError> {
-        let HttpSource {
+        Box::new(self.inner).run(stop_token).await
-            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
     }
 }
 
@@ -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.get_url(), "http://example.com/audio.flac");
-        assert_eq!(source.chunk_frames, 0);
+        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.get_url(), "http://example.com/audio.mp3");
-        assert_eq!(source.chunk_frames, 1024);
+        assert_eq!(source.get_chunc_frames(), 1024);
     }
 
     /// Test de téléchargement et décodage d'un fichier FLAC via HTTP

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),
         }
     }
 }

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<Self>,
         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<dyn std::error::Error>> {
+    /// 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<Self>) -> 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<mpsc::Receiver<Arc<AudioSegment>>>,
+///         output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
+///         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<mpsc::Receiver<Arc<AudioSegment>>>,
+        output: Vec<mpsc::Sender<Arc<AudioSegment>>>,
+        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<dyn std::error::Error>> {
+/// 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<Result<(), AudioError>>,
+}
+
+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<AudioError>) {
+        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<AudioError>) -> 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<L: NodeLogic> {
+    /// La logique métier du nœud
+    logic: L,
+
+    /// Receiver pour les données entrantes (None pour les sources)
+    rx: Option<mpsc::Receiver<Arc<AudioSegment>>>,
+
+    /// Sender pour les données entrantes (pour clonage via get_tx)
+    tx: Option<mpsc::Sender<Arc<AudioSegment>>>,
+
+    /// Liste des nœuds enfants
+    children: Vec<Box<dyn AudioPipelineNode>>,
+
+    /// Liste des senders vers les enfants
+    child_txs: Vec<mpsc::Sender<Arc<AudioSegment>>>,
+}
+
+impl<L: NodeLogic> Node<L> {
+    /// 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<L: NodeLogic> AudioPipelineNode for Node<L> {
+    fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> {
+        self.tx.clone()
+    }
+
+    fn register(&mut self, child: Box<dyn AudioPipelineNode>) {
+        if let Some(tx) = child.get_tx() {
+            self.child_txs.push(tx);
+        }
+        self.children.push(child);
+    }
+
+    async fn run(
+        mut self: Box<Self>,
+        stop_token: CancellationToken,
+    ) -> Result<(), AudioError> {
+        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
+    }
 }