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

Récupération de l'erreur git cleaning

Browse Source
This commit is contained in:
Eric Coissac
2025-11-04 20:34:44 +01:00
parent 83d7520840
commit 88349e7797
48 changed files with 1754 additions and 1029 deletions
Download Patch File Download Diff File Expand all files Collapse all files

513
pmoaudio/src/nodes/converter_nodes.rs Normal file → Executable file
View File

@@ -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},
type_constraints::{SampleType, TypeRequirement},
AudioPipelineNode, AudioSegment,
nodes::AudioError,
pipeline::{Node, NodeLogic},
AudioChunk, AudioPipelineNode, AudioSegment,
};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
// Macro pour générer les converter nodes avec la nouvelle architecture AudioPipelineNode
macro_rules! converter_node {
($node_name:ident, $convert_method:ident, $output_type:expr, $doc:expr) => {
#[doc = $doc]
pub struct $node_name {
tx: mpsc::Sender<Arc<AudioSegment>>,
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()
}
}
};
/// Logique de conversion générique
///
/// Cette struct contient la logique pure de conversion d'un type vers un autre.
/// Elle reçoit des segments, convertit les chunks audio, et relay les syncmarkers.
pub struct ConverterLogic<F> {
convert_fn: F,
}
// Générer les 5 converter nodes
converter_node!(
ToI16Node,
to_i16,
SampleType::I16,
"Node de conversion vers I16 (16-bit signed integer)"
);
converter_node!(
ToI24Node,
to_i24,
SampleType::I24,
"Node de conversion vers I24 (24-bit signed integer)"
);
converter_node!(
ToI32Node,
to_i32,
SampleType::I32,
"Node de conversion vers I32 (32-bit signed integer)"
);
converter_node!(
ToF32Node,
to_f32,
SampleType::F32,
"Node de conversion vers F32 (32-bit floating point)"
);
converter_node!(
ToF64Node,
to_f64,
SampleType::F64,
"Node de conversion vers F64 (64-bit floating point)"
);
impl<F> ConverterLogic<F>
where
F: Fn(&AudioChunk) -> AudioChunk + Send + 'static,
{
pub fn new(convert_fn: F) -> Self {
Self { convert_fn }
}
}
#[async_trait::async_trait]
impl<F> NodeLogic for ConverterLogic<F>
where
F: Fn(&AudioChunk) -> AudioChunk + Send + 'static,
{
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("Converter must have input");
tracing::debug!("ConverterLogic::process started, {} children", output.len());
loop {
let segment = tokio::select! {
_ = stop_token.cancelled() => {
tracing::debug!("ConverterLogic cancelled");
break;
}
result = rx.recv() => {
match result {
Some(seg) => seg,
None => {
tracing::debug!("ConverterLogic received EOF");
break; // EOF
}
}
}
};
// Convertir si c'est un chunk audio, sinon passer tel quel
let output_segment = if segment.is_audio_chunk() {
if let Some(chunk) = segment.as_chunk() {
let converted_chunk = (self.convert_fn)(chunk);
// Debug: afficher le type du chunk converti (seulement pour le premier)
if segment.order == 0 {
let chunk_type = match &converted_chunk {
crate::AudioChunk::I16(_) => "I16",
crate::AudioChunk::I24(_) => "I24",
crate::AudioChunk::I32(_) => "I32",
crate::AudioChunk::F32(_) => "F32",
crate::AudioChunk::F64(_) => "F64",
};
tracing::debug!("ConverterLogic: converted chunk type = {}", chunk_type);
}
Arc::new(AudioSegment {
order: segment.order,
timestamp_sec: segment.timestamp_sec,
segment: crate::_AudioSegment::Chunk(Arc::new(converted_chunk)),
})
} else {
segment
}
} else {
segment
};
// Envoyer à tous les enfants
for tx in &output {
tx.send(output_segment.clone())
.await
.map_err(|_| AudioError::ChildDied)?;
}
}
Ok(())
}
}
// ═══════════════════════════════════════════════════════════════════════════
// Converters spécifiques - Fonctions factory simplifiées
// ═══════════════════════════════════════════════════════════════════════════
/// Node de conversion vers I16 (16-bit signed integer)
pub struct ToI16Node;
impl ToI16Node {
pub fn new() -> Box<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() {
let node = ToF32Node::new();
async fn test_converter_logic() {
// Test unitaire de la logique pure
let mut logic = ConverterLogic::new(|chunk: &AudioChunk| chunk.to_f32());
// Vérifier les types d'entrée/sortie
assert_eq!(
node.input_type().unwrap().get_accepted_types().len(),
5,
"Should accept all 5 types"
);
assert_eq!(
node.output_type()
.unwrap()
.get_accepted_types()
.first()
.copied(),
Some(SampleType::F32),
"Should output F32 only"
);
}
// Nœud de test simple qui collecte les segments
struct TestCollectorNode {
input_tx: mpsc::Sender<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 (input_tx, input_rx) = mpsc::channel(10);
let (output_tx, mut output_rx) = mpsc::channel(10);
let stop_token = CancellationToken::new();
let handle = tokio::spawn(async move { Box::new(node).run(stop_token).await });
// Créer et envoyer un chunk I32
let stereo = vec![[1_000_000i32 << 16, -500_000i32 << 16]; 100];
let chunk_data = AudioChunkData::new(stereo.clone(), 48_000, 0.0);
let chunk = AudioChunk::I32(chunk_data);
// Créer un chunk de test
let test_chunk = AudioChunk::I16(AudioChunkData::new(
vec![[100, 200], [300, 400]],
48000,
0.0,
));
// Créer le segment directement
let segment = Arc::new(AudioSegment {
order: 0,
timestamp_sec: 0.0,
segment: crate::_AudioSegment::Chunk(Arc::new(chunk)),
segment: crate::_AudioSegment::Chunk(Arc::new(test_chunk)),
});
tx.send(segment).await.unwrap();
drop(tx);
// Envoyer le segment
input_tx.send(segment).await.unwrap();
drop(input_tx); // EOF
// Recevoir le chunk converti
let result = out_rx.recv().await.unwrap();
// Lancer le traitement
tokio::spawn(async move {
logic
.process(Some(input_rx), vec![output_tx], stop_token)
.await
.unwrap();
});
// Vérifier le résultat
let result = output_rx.recv().await.unwrap();
assert!(result.is_audio_chunk());
if let Some(converted) = result.as_chunk() {
assert_eq!(converted.type_name(), "i16");
assert_eq!(converted.len(), 100);
// Vérifier la conversion (downsampling de I32 vers I16)
if let AudioChunk::I16(data) = &**converted {
for (orig, converted_frame) in stereo.iter().zip(data.frames().iter()) {
let expected_l = (orig[0] >> 16) as i16;
let expected_r = (orig[1] >> 16) as i16;
assert_eq!(converted_frame[0], expected_l);
assert_eq!(converted_frame[1], expected_r);
}
}
if let Some(chunk) = result.as_chunk() {
assert!(matches!(chunk.as_ref(), AudioChunk::F32(_)));
} else {
panic!("Expected audio chunk");
}
handle.await.unwrap().unwrap();
}
#[tokio::test]
async fn test_syncmarkers_passthrough() {
let mut node = ToI16Node::new();
let (out_tx, mut out_rx) = mpsc::channel(16);
let collector = TestCollectorNode::new(out_tx);
node.register(Box::new(collector));
let tx = node.get_tx().unwrap();
let stop_token = CancellationToken::new();
tokio::spawn(async move {
Box::new(node).run(stop_token).await.unwrap();
});
// Envoyer un syncmarker
let top_zero = AudioSegment::new_top_zero_sync();
tx.send(top_zero.clone()).await.unwrap();
drop(tx);
// Recevoir le syncmarker
let result = out_rx.recv().await.unwrap();
assert!(!result.is_audio_chunk());
assert!(result.as_sync_marker().is_some());
}
}

452
pmoaudio/src/nodes/file_source.rs Normal file → Executable file
View File

@@ -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,
chunk_frames: usize,
child_txs: Vec<mpsc::Sender<Arc<AudioSegment>>>,
children: Vec<Box<dyn AudioPipelineNode>>,
inner: Node<FileSourceLogic>,
}
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(),
chunk_frames,
child_txs: Vec::new(),
children: Vec::new(),
inner: Node::new_source(logic),
}
}
}
#[async_trait::async_trait]
impl AudioPipelineNode for FileSource {
fn get_tx(&self) -> Option<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> {

196
pmoaudio/src/nodes/flac_file_sink.rs Normal file → Executable file
View File

@@ -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 {
/// 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>>(
impl FlacFileSinkLogic {
pub fn new<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 {
tx: _tx,
rx,
base_path,
encoder_options,
pcm_buffer_capacity,
} = *self;
Self::run_internal(rx, base_path, encoder_options, pcm_buffer_capacity, stop_token).await
Box::new(self.inner).run(stop_token).await
}
}

234
pmoaudio/src/nodes/http_source.rs Normal file → Executable file
View File

@@ -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 {
/// 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 {
impl HttpSourceLogic {
pub fn new<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(
url: String,
chunk_frames: usize,
child_txs: Vec<mpsc::Sender<Arc<AudioSegment>>>,
pub fn get_url(&self) -> String {
self.url.clone()
}
pub fn get_chunc_frames(&self) -> usize {
self.chunk_frames
}
}
#[async_trait::async_trait]
impl NodeLogic for HttpSourceLogic {
async fn process(
&mut self,
_input: Option<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();
for tx in &child_txs {
tx.send(top_zero.clone()).await.map_err(|_| AudioError::SendError)?;
}
send_to_children!(AudioSegment::new_top_zero_sync());
// Émettre TrackBoundary avec les métadonnées HTTP
let track_boundary = AudioSegment::new_track_boundary(0, 0.0, Arc::new(tokio::sync::RwLock::new(metadata)));
for tx in &child_txs {
tx.send(track_boundary.clone()).await.map_err(|_| AudioError::SendError)?;
}
let track_boundary = AudioSegment::new_track_boundary(
0,
0.0,
Arc::new(tokio::sync::RwLock::new(metadata)),
);
send_to_children!(track_boundary);
// Préparer la lecture des chunks audio
let frame_bytes = stream_info.bytes_per_sample() * stream_info.channels as usize;
@@ -258,12 +246,7 @@ impl HttpSource {
timestamp_sec,
)?;
for tx in &child_txs {
if tx.send(segment.clone()).await.is_err() {
// Un enfant est mort, arrêter
return Ok(());
}
}
send_to_children!(segment);
chunk_index += 1;
total_frames += frames_to_emit as u64;
@@ -276,9 +259,7 @@ impl HttpSource {
let timestamp_sec = total_frames as f64 / stream_info.sample_rate as f64;
let segment =
bytes_to_segment(&pending, &stream_info, frames, chunk_index, timestamp_sec)?;
for tx in &child_txs {
let _ = tx.send(segment.clone()).await;
}
send_to_children!(segment);
total_frames += frames as u64;
chunk_index += 1;
}
@@ -287,9 +268,7 @@ impl HttpSource {
// Émettre EndOfStream
let final_timestamp = total_frames as f64 / stream_info.sample_rate as f64;
let eos = AudioSegment::new_end_of_stream(chunk_index, final_timestamp);
for tx in &child_txs {
let _ = tx.send(eos.clone()).await;
}
send_to_children!(eos);
// Attendre la fin du décodage
stream
@@ -301,6 +280,66 @@ impl HttpSource {
}
}
// ═══════════════════════════════════════════════════════════════════════════
// HttpSource - Wrapper utilisant Node<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.child_txs.push(tx);
}
self.children.push(child);
self.inner.register(child)
}
async fn run(
self: Box<Self>,
stop_token: CancellationToken,
) -> Result<(), AudioError> {
let HttpSource {
url,
chunk_frames,
child_txs,
children,
} = *self;
// Spawner tous les enfants
let mut child_handles = Vec::new();
for child in children {
let child_token = stop_token.child_token();
let handle = tokio::spawn(async move {
child.run(child_token).await
});
child_handles.push(handle);
}
// Lancer la logique interne
let work_result = Self::run_internal(url, chunk_frames, child_txs, stop_token.clone()).await;
// Attendre que tous les enfants se terminent
for handle in child_handles {
match handle.await {
Ok(Ok(())) => {},
Ok(Err(e)) => return Err(e),
Err(e) => {
return Err(AudioError::ProcessingError(format!(
"Child task panicked: {}",
e
)))
}
}
}
work_result
Box::new(self.inner).run(stop_token).await
}
}
@@ -598,16 +600,16 @@ mod tests {
#[test]
fn test_http_source_creation() {
let source = HttpSource::new("http://example.com/audio.flac");
assert_eq!(source.url, "http://example.com/audio.flac");
assert_eq!(source.chunk_frames, 0);
assert_eq!(source.get_url(), "http://example.com/audio.flac");
assert_eq!(source.get_chunc_frames(), 0);
}
/// Test de création avec taille de chunk personnalisée
#[test]
fn test_http_source_with_chunk_size() {
let source = HttpSource::with_chunk_size("http://example.com/audio.mp3", 1024);
assert_eq!(source.url, "http://example.com/audio.mp3");
assert_eq!(source.chunk_frames, 1024);
assert_eq!(source.get_url(), "http://example.com/audio.mp3");
assert_eq!(source.get_chunc_frames(), 1024);
}
/// Test de téléchargement et décodage d'un fichier FLAC via HTTP

9
pmoaudio/src/nodes/mod.rs Normal file → Executable file
View File

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