Refactorisation des nœuds audio pour utiliser boxed()

Cette mise à jour refactorise les nœuds audio pour utiliser la méthode `boxed()` lors de l'enregistrement des enfants, améliorant ainsi la cohérence et la lisibilité du code. Les méthodes `make()` sont ajoutées pour faciliter la création d'instances boxées des nœuds, et les exemples sont mis à jour en conséquence.
This commit is contained in:
2026-02-26 20:18:14 +01:00
parent 306e691c61
commit 6f8a80a58a
16 changed files with 180 additions and 92 deletions

View File

@@ -70,8 +70,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let sink = FlacFileSink::new(output_path);
// Construire la chaîne: source → converter → sink
converter.register(Box::new(sink));
source.register(converter);
converter.register(sink.boxed());
source.register(converter.boxed());
// Créer un token d'arrêt pour contrôle manuel si besoin
let stop_token = CancellationToken::new();

View File

@@ -38,7 +38,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let sink = FlacFileSink::new(output_path);
// Enregistrer le sink comme enfant de la source
source.register(Box::new(sink));
source.register(sink.boxed());
// Créer un token d'arrêt pour contrôle manuel si besoin
let stop_token = CancellationToken::new();

View File

@@ -34,7 +34,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let sink = AudioSink::new();
// Connecter la source au sink
source.register(Box::new(sink));
source.register(sink.boxed());
println!("Démarrage de la lecture...");
println!("Appuyez sur Ctrl+C pour arrêter");

View File

@@ -35,22 +35,16 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
println!("Lecture de: {}", file_path);
println!("Sample rate cible: {} Hz", target_sample_rate);
// Créer la source audio
let mut source = FileSource::new(file_path).await?;
// Créer le nœud de resampling
let mut resampler = ResamplingNode::new(target_sample_rate);
// Créer le nœud de conversion vers I24
let mut converter = ToI24Node::new();
// Créer le sink audio avec volume à 80%
let sink = AudioSink::with_volume(0.8);
// Construire le pipeline: Source → Resampler → Converter → Sink
source.register(Box::new(resampler));
resampler.register(Box::new(converter));
converter.register(Box::new(sink));
let sink = AudioSink::new();
let mut converter = ToI24Node::new();
converter.register(sink.boxed());
let mut resampler = ResamplingNode::new(target_sample_rate);
resampler.register(converter.boxed());
let mut source = FileSource::new(file_path);
source.register(resampler.boxed());
println!(
"Pipeline créé: FileSource → Resampling({} Hz) → ToI24 → AudioSink",

View File

@@ -547,19 +547,23 @@ impl AudioSink {
}
}
pub fn make() -> Box<dyn AudioPipelineNode> {
Self::new().boxed()
}
/// Crée un nouveau AudioSink avec une taille de channel personnalisée
pub fn with_channel_size(channel_size: usize) -> Self {
pub fn with_channel_size(channel_size: usize) -> Box<dyn AudioPipelineNode> {
Self {
inner: Node::new_with_input(AudioSinkLogic::new(), channel_size),
}
}.boxed()
}
/// Crée un AudioSink avec null output (pour tests sans carte audio)
/// Consomme les segments audio sans les jouer
pub fn with_null_output() -> Self {
pub fn with_null_output() -> Box<dyn AudioPipelineNode> {
Self {
inner: Node::new_with_input(AudioSinkLogic::with_null_output(), DEFAULT_CHANNEL_SIZE),
}
}.boxed()
}
}

View File

@@ -22,6 +22,7 @@ use std::sync::Arc;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
/// Logique de conversion générique
///
/// Cette struct contient la logique pure de conversion d'un type vers un autre.
@@ -112,102 +113,162 @@ where
// ═══════════════════════════════════════════════════════════════════════════
/// Node de conversion vers I16 (16-bit signed integer)
pub struct ToI16Node;
pub struct ToI16Node(Node<ConverterLogic<fn(&AudioChunk) -> AudioChunk>>);
impl ToI16Node {
pub fn new() -> Box<dyn AudioPipelineNode> {
Self::with_channel_size(16)
pub fn new() -> Self {
Self(Node::new_with_input(ConverterLogic::new(|chunk: &AudioChunk| chunk.to_i16()), 16))
}
pub fn make() -> Box<dyn AudioPipelineNode> {
Self::new().boxed()
}
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))
Self(Node::new_with_input(ConverterLogic::new(|chunk: &AudioChunk| chunk.to_i16()), channel_size)).boxed()
}
}
impl Default for ToI16Node {
fn default() -> Self {
Self
Self::new()
}
}
#[async_trait::async_trait]
impl AudioPipelineNode for ToI16Node {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> { self.0.get_tx() }
fn register(&mut self, child: Box<dyn AudioPipelineNode>) { self.0.register(child) }
async fn run(self: Box<Self>, stop_token: CancellationToken) -> Result<(), AudioError> {
Box::new(self.0).run(stop_token).await
}
}
/// Node de conversion vers I24 (24-bit signed integer)
pub struct ToI24Node;
pub struct ToI24Node(Node<ConverterLogic<fn(&AudioChunk) -> AudioChunk>>);
impl ToI24Node {
pub fn new() -> Box<dyn AudioPipelineNode> {
Self::with_channel_size(16)
pub fn new() -> Self {
Self(Node::new_with_input(ConverterLogic::new(|chunk: &AudioChunk| chunk.to_i24()), 16))
}
pub fn make() -> Box<dyn AudioPipelineNode> {
Self::new().boxed()
}
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))
Self(Node::new_with_input(ConverterLogic::new(|chunk: &AudioChunk| chunk.to_i24()), channel_size)).boxed()
}
}
impl Default for ToI24Node {
fn default() -> Self {
Self
Self::new()
}
}
#[async_trait::async_trait]
impl AudioPipelineNode for ToI24Node {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> { self.0.get_tx() }
fn register(&mut self, child: Box<dyn AudioPipelineNode>) { self.0.register(child) }
async fn run(self: Box<Self>, stop_token: CancellationToken) -> Result<(), AudioError> {
Box::new(self.0).run(stop_token).await
}
}
/// Node de conversion vers I32 (32-bit signed integer)
pub struct ToI32Node;
pub struct ToI32Node(Node<ConverterLogic<fn(&AudioChunk) -> AudioChunk>>);
impl ToI32Node {
pub fn new() -> Box<dyn AudioPipelineNode> {
Self::with_channel_size(16)
pub fn new() -> Self {
Self(Node::new_with_input(ConverterLogic::new(|chunk: &AudioChunk| chunk.to_i32()), 16))
}
pub fn make() -> Box<dyn AudioPipelineNode> {
Self::new().boxed()
}
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))
Self(Node::new_with_input(ConverterLogic::new(|chunk: &AudioChunk| chunk.to_i32()), channel_size)).boxed()
}
}
impl Default for ToI32Node {
fn default() -> Self {
Self
Self::new()
}
}
#[async_trait::async_trait]
impl AudioPipelineNode for ToI32Node {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> { self.0.get_tx() }
fn register(&mut self, child: Box<dyn AudioPipelineNode>) { self.0.register(child) }
async fn run(self: Box<Self>, stop_token: CancellationToken) -> Result<(), AudioError> {
Box::new(self.0).run(stop_token).await
}
}
/// Node de conversion vers F32 (32-bit floating point)
pub struct ToF32Node;
pub struct ToF32Node(Node<ConverterLogic<fn(&AudioChunk) -> AudioChunk>>);
impl ToF32Node {
pub fn new() -> Box<dyn AudioPipelineNode> {
Self::with_channel_size(16)
pub fn new() -> Self {
Self(Node::new_with_input(ConverterLogic::new(|chunk: &AudioChunk| chunk.to_f32()), 16))
}
pub fn make() -> Box<dyn AudioPipelineNode> {
Self::new().boxed()
}
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))
Self(Node::new_with_input(ConverterLogic::new(|chunk: &AudioChunk| chunk.to_f32()), channel_size)).boxed()
}
}
impl Default for ToF32Node {
fn default() -> Self {
Self
Self::new()
}
}
#[async_trait::async_trait]
impl AudioPipelineNode for ToF32Node {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> { self.0.get_tx() }
fn register(&mut self, child: Box<dyn AudioPipelineNode>) { self.0.register(child) }
async fn run(self: Box<Self>, stop_token: CancellationToken) -> Result<(), AudioError> {
Box::new(self.0).run(stop_token).await
}
}
/// Node de conversion vers F64 (64-bit floating point)
pub struct ToF64Node;
pub struct ToF64Node(Node<ConverterLogic<fn(&AudioChunk) -> AudioChunk>>);
impl ToF64Node {
pub fn new() -> Box<dyn AudioPipelineNode> {
Self::with_channel_size(16)
pub fn new() -> Self {
Self(Node::new_with_input(ConverterLogic::new(|chunk: &AudioChunk| chunk.to_f64()), 16))
}
pub fn make() -> Box<dyn AudioPipelineNode> {
Self::new().boxed()
}
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))
Self(Node::new_with_input(ConverterLogic::new(|chunk: &AudioChunk| chunk.to_f64()), channel_size)).boxed()
}
}
impl Default for ToF64Node {
fn default() -> Self {
Self
Self::new()
}
}
#[async_trait::async_trait]
impl AudioPipelineNode for ToF64Node {
fn get_tx(&self) -> Option<mpsc::Sender<Arc<AudioSegment>>> { self.0.get_tx() }
fn register(&mut self, child: Box<dyn AudioPipelineNode>) { self.0.register(child) }
async fn run(self: Box<Self>, stop_token: CancellationToken) -> Result<(), AudioError> {
Box::new(self.0).run(stop_token).await
}
}

View File

@@ -224,18 +224,21 @@ impl FileSource {
///
/// * `path` - chemin du fichier audio à lire
pub fn new<P: Into<PathBuf>>(path: P) -> Self {
Self::with_chunk_size(path, 0) // 0 = auto-calculer
let logic = FileSourceLogic::new(path, 0);
Self { inner: Node::new_source(logic) }
}
pub fn make<P: Into<PathBuf>>(path: P) -> Box<dyn AudioPipelineNode> {
Self::new(path).boxed()
}
/// Crée une nouvelle source de fichier avec une taille de chunk spécifique.
///
/// * `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 {
pub fn with_chunk_size<P: Into<PathBuf>>(path: P, chunk_frames: usize) -> Box<dyn AudioPipelineNode> {
let logic = FileSourceLogic::new(path, chunk_frames);
Self {
inner: Node::new_source(logic),
}
Self { inner: Node::new_source(logic) }.boxed()
}
}

View File

@@ -305,7 +305,12 @@ impl FlacFileSink {
/// * `base_path` - Chemin de base pour les fichiers FLAC. Si des TrackBoundary sont reçus,
/// des fichiers seront créés avec des suffixes (_01, _02, etc.)
pub fn new<P: Into<PathBuf>>(base_path: P) -> Self {
Self::with_channel_size(base_path, DEFAULT_CHANNEL_SIZE)
let logic = FlacFileSinkLogic::new(base_path, EncoderOptions::default(), 8);
Self { inner: Node::new_with_input(logic, DEFAULT_CHANNEL_SIZE) }
}
pub fn make<P: Into<PathBuf>>(base_path: P) -> Box<dyn AudioPipelineNode> {
Self::new(base_path).boxed()
}
/// Crée un sink FLAC avec une taille de buffer MPSC personnalisée.
@@ -314,8 +319,9 @@ impl FlacFileSink {
///
/// * `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())
pub fn with_channel_size<P: Into<PathBuf>>(base_path: P, channel_size: usize) -> Box<dyn AudioPipelineNode> {
let logic = FlacFileSinkLogic::new(base_path, EncoderOptions::default(), 8);
Self { inner: Node::new_with_input(logic, channel_size) }.boxed()
}
/// Crée un sink FLAC avec une configuration complète.
@@ -329,11 +335,11 @@ impl FlacFileSink {
base_path: P,
channel_size: usize,
encoder_options: EncoderOptions,
) -> Self {
) -> Box<dyn AudioPipelineNode> {
let logic = FlacFileSinkLogic::new(base_path, encoder_options, 8);
Self {
inner: Node::new_with_input(logic, channel_size),
}
}.boxed()
}
}

View File

@@ -297,7 +297,12 @@ impl 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)
let logic = HttpSourceLogic::new(url.into(), 0);
Self { inner: Node::new_source(logic) }
}
pub fn make<S: Into<String>>(url: S) -> Box<dyn AudioPipelineNode> {
Self::new(url).boxed()
}
/// Crée une nouvelle source HTTP avec une taille de chunk spécifique.
@@ -315,11 +320,9 @@ impl 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 {
pub fn with_chunk_size<S: Into<String>>(url: S, chunk_frames: usize) -> Box<dyn AudioPipelineNode> {
let logic = HttpSourceLogic::new(url.into(), chunk_frames);
Self {
inner: Node::new_source(logic),
}
Self { inner: Node::new_source(logic) }.boxed()
}
pub fn get_url(&self) -> String {

View File

@@ -333,8 +333,13 @@ impl ResamplingNode {
/// Crée un nouveau node de resampling
///
/// * `target_sample_rate` - Sample rate de sortie en Hz (ex: 48000)
pub fn new(target_sample_rate: u32) -> Box<dyn AudioPipelineNode> {
Self::with_channel_size(target_sample_rate, 16)
pub fn new(target_sample_rate: u32) -> Self {
let logic = ResamplingLogic::new(target_sample_rate);
Self { inner: Node::new_with_input(logic, 16) }
}
pub fn make(target_sample_rate: u32) -> Box<dyn AudioPipelineNode> {
Self::new(target_sample_rate).boxed()
}
/// Crée un nouveau node de resampling avec taille de canal personnalisée
@@ -346,9 +351,7 @@ impl ResamplingNode {
channel_size: usize,
) -> Box<dyn AudioPipelineNode> {
let logic = ResamplingLogic::new(target_sample_rate);
Box::new(Self {
inner: Node::new_with_input(logic, channel_size),
})
Self { inner: Node::new_with_input(logic, channel_size) }.boxed()
}
}

View File

@@ -305,7 +305,12 @@ impl TimerBufferNode {
/// let buffer = TimerBufferNode::new(3.0);
/// ```
pub fn new(capacity_sec: f64) -> Self {
Self::with_channel_size(capacity_sec, DEFAULT_CHANNEL_SIZE)
let logic = TimerBufferNodeLogic::new(capacity_sec);
Self { inner: Node::new_with_input(logic, DEFAULT_CHANNEL_SIZE) }
}
pub fn make(capacity_sec: f64) -> Box<dyn AudioPipelineNode> {
Self::new(capacity_sec).boxed()
}
/// Crée un TimerBufferNode avec une taille de buffer MPSC personnalisée
@@ -314,11 +319,9 @@ impl TimerBufferNode {
///
/// * `capacity_sec` - Capacité du buffer en secondes
/// * `channel_size` - Taille du buffer MPSC (nombre de segments en attente)
pub fn with_channel_size(capacity_sec: f64, channel_size: usize) -> Self {
pub fn with_channel_size(capacity_sec: f64, channel_size: usize) -> Box<dyn AudioPipelineNode> {
let logic = TimerBufferNodeLogic::new(capacity_sec);
Self {
inner: Node::new_with_input(logic, channel_size),
}
Self { inner: Node::new_with_input(logic, channel_size) }.boxed()
}
}

View File

@@ -308,7 +308,12 @@ impl TimerNode {
/// let timer = TimerNode::new(3.0);
/// ```
pub fn new(max_lead_time_sec: f64) -> Self {
Self::with_channel_size(max_lead_time_sec, DEFAULT_CHANNEL_SIZE)
let logic = TimerNodeLogic::new(max_lead_time_sec);
Self { inner: Node::new_with_input(logic, DEFAULT_CHANNEL_SIZE) }
}
pub fn make(max_lead_time_sec: f64) -> Box<dyn AudioPipelineNode> {
Self::new(max_lead_time_sec).boxed()
}
/// Crée un TimerNode avec une taille de buffer MPSC personnalisée
@@ -317,11 +322,9 @@ impl TimerNode {
///
/// * `max_lead_time_sec` - Avance maximale en secondes
/// * `channel_size` - Taille du buffer MPSC (nombre de segments en attente)
pub fn with_channel_size(max_lead_time_sec: f64, channel_size: usize) -> Self {
pub fn with_channel_size(max_lead_time_sec: f64, channel_size: usize) -> Box<dyn AudioPipelineNode> {
let logic = TimerNodeLogic::new(max_lead_time_sec);
Self {
inner: Node::new_with_input(logic, channel_size),
}
Self { inner: Node::new_with_input(logic, channel_size) }.boxed()
}
}

View File

@@ -82,6 +82,14 @@ pub trait AudioPipelineNode: Send + 'static {
/// Le parent extrait le tx via `child.get_tx()` avant de stocker le child.
fn register(&mut self, child: Box<dyn AudioPipelineNode>);
/// Encapsule ce nœud dans un `Box<dyn AudioPipelineNode>` pour l'utiliser dans un pipeline.
fn boxed(self) -> Box<dyn AudioPipelineNode>
where
Self: Sized + 'static,
{
Box::new(self)
}
/// Lance le nœud et tous ses enfants
///
/// # Arguments

View File

@@ -117,7 +117,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let sink = FlacFileSink::new(&base_path);
// Construire la chaîne: source → sink
source.register(Box::new(sink));
source.register(sink.boxed());
// Créer un token d'arrêt
let stop_token = CancellationToken::new();

View File

@@ -217,15 +217,15 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let audio_sink = if use_null_audio {
AudioSink::with_null_output()
} else {
AudioSink::new()
AudioSink::make()
};
tracing::debug!("AudioSink created");
// Connecter timer → audio (AVANT de mettre timer dans une Box)
timer.register(Box::new(audio_sink));
timer.register(audio_sink);
// Connecter playlist → timer
playlist_source.register(Box::new(timer));
playlist_source.register(timer.boxed());
tracing::info!("Playback pipeline connected: PlaylistSource → TimerNode → AudioSink");
// ═══════════════════════════════════════════════════════════════════════════

View File

@@ -83,21 +83,21 @@ impl InstancePipeline {
// Nœud de suivi de position : lit le timestamp des chunks sortant du buffer
let (mut position_tracker, position_handle) = PositionTrackerNode::new();
position_tracker.register(Box::new(sink));
position_tracker.register(sink.boxed());
// Nœud de pacing : régule le débit pour éviter les rafales et pertes de segments
// 2s de buffer absorbe les irrégularités de la source réseau
let mut timer_buffer = TimerBufferNode::new(2.0);
timer_buffer.register(Box::new(position_tracker));
timer_buffer.register(position_tracker.boxed());
// Nœud de conversion de profondeur : tout type entier → I24
// Placé avant le buffer pour réduire la mémoire utilisée
let mut to_i24 = ToI24Node::new();
to_i24.register(Box::new(timer_buffer));
to_i24.register(timer_buffer.boxed());
// Nœud de rééchantillonnage : n'importe quel sample rate → 96 kHz
let mut resampler = ResamplingNode::new(DIRECT_OGG_FLAC_SAMPLE_RATE);
resampler.register(to_i24);
resampler.register(to_i24.boxed());
// Le tx d'entrée du resampler est le point d'entrée du pipeline
let segment_tx = resampler.get_tx().expect("ResamplingNode doit avoir un sender");
@@ -110,7 +110,7 @@ impl InstancePipeline {
// Lancer la chaîne resampler → to_i24 → sink en background
let sink_stop = stop_token.clone();
tokio::spawn(async move {
if let Err(e) = resampler.run(sink_stop).await {
if let Err(e) = resampler.boxed().run(sink_stop).await {
warn!("Audio pipeline error: {:?}", e);
}
debug!("Sink task terminated");