From a8a5c2db9f019ee32b99f4a5065672ab3b7fc11b Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Mon, 27 Oct 2025 22:59:06 +0100 Subject: [PATCH] unification des decodeurs --- pmoflac/src/aiff.rs | 44 +++------------------ pmoflac/src/decoder.rs | 73 +++-------------------------------- pmoflac/src/decoder_common.rs | 60 +++++++++++++++++++++++++++- pmoflac/src/mp3.rs | 53 +++---------------------- pmoflac/src/ogg.rs | 58 ++++------------------------ pmoflac/src/opus.rs | 48 ++++------------------- pmoflac/src/wav.rs | 43 +++------------------ 7 files changed, 96 insertions(+), 283 deletions(-) diff --git a/pmoflac/src/aiff.rs b/pmoflac/src/aiff.rs index 189e843a..83f65925 100644 --- a/pmoflac/src/aiff.rs +++ b/pmoflac/src/aiff.rs @@ -5,13 +5,7 @@ //! little-endian interleaved PCM frames compatible with the rest of the //! pipeline. -use std::{ - collections::VecDeque, - fmt, - io::{self, Read}, - pin::Pin, - task::{Context, Poll}, -}; +use std::{collections::VecDeque, fmt, io::Read}; use tokio::{ io::AsyncRead, @@ -21,7 +15,8 @@ use tokio::{ use crate::{ common::ChannelReader, decoder_common::{ - spawn_ingest_task, spawn_writer_task, DecoderError, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE, + spawn_ingest_task, spawn_writer_task, DecodedStream, DecoderError, CHANNEL_CAPACITY, + DUPLEX_BUFFER_SIZE, }, pcm::StreamInfo, stream::ManagedAsyncReader, @@ -142,35 +137,8 @@ impl CommChunk { } } -/// Asynchronous decoded stream wrapper for AIFF data. -pub struct AiffDecodedStream { - info: StreamInfo, - reader: ManagedAsyncReader, -} - -impl AiffDecodedStream { - pub fn info(&self) -> &StreamInfo { - &self.info - } - - pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader) { - (self.info, self.reader) - } - - pub async fn wait(self) -> Result<(), AiffError> { - self.reader.wait().await - } -} - -impl AsyncRead for AiffDecodedStream { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut tokio::io::ReadBuf<'_>, - ) -> Poll> { - Pin::new(&mut self.reader).poll_read(cx, buf) - } -} +/// Async stream alias for decoded AIFF audio. +pub type AiffDecodedStream = DecodedStream; /// Decode an AIFF stream into PCM audio (little-endian interleaved). pub async fn decode_aiff_stream(reader: R) -> Result @@ -362,7 +330,7 @@ where let info = info_rx.await.map_err(|_| AiffError::ChannelClosed)??; let reader = ManagedAsyncReader::new("aiff-decode-writer", pcm_reader, writer_handle); - Ok(AiffDecodedStream { info, reader }) + Ok(DecodedStream::new(info, reader)) } fn parse_extended_f80(bytes: &[u8]) -> Result { diff --git a/pmoflac/src/decoder.rs b/pmoflac/src/decoder.rs index 41a929d0..78c7b36d 100644 --- a/pmoflac/src/decoder.rs +++ b/pmoflac/src/decoder.rs @@ -1,9 +1,3 @@ -use std::{ - io, - pin::Pin, - task::{Context, Poll}, -}; - use tokio::{ io::AsyncRead, sync::{mpsc, oneshot}, @@ -11,72 +5,17 @@ use tokio::{ use crate::{ common::ChannelReader, - decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE}, + decoder_common::{ + spawn_ingest_task, spawn_writer_task, DecodedStream, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE, + }, error::FlacError, pcm::StreamInfo, stream::ManagedAsyncReader, util::interleaved_i32_to_le_bytes, }; -/// An async stream that decodes FLAC audio into PCM samples. -/// -/// This struct implements `AsyncRead`, allowing you to read decoded PCM data -/// as it becomes available. The decoding happens in a background task. -/// -/// # Example -/// -/// ```no_run -/// use pmoflac::decode_flac_stream; -/// use tokio::fs::File; -/// use tokio::io::AsyncReadExt; -/// -/// # #[tokio::main] -/// # async fn main() -> Result<(), Box> { -/// let file = File::open("audio.flac").await?; -/// let mut stream = decode_flac_stream(file).await?; -/// -/// println!("Sample rate: {}", stream.info().sample_rate); -/// -/// let mut pcm = Vec::new(); -/// stream.read_to_end(&mut pcm).await?; -/// stream.wait().await?; -/// # Ok(()) -/// # } -/// ``` -pub struct FlacDecodedStream { - info: StreamInfo, - reader: ManagedAsyncReader, -} - -impl FlacDecodedStream { - /// Returns metadata about the FLAC stream (sample rate, channels, etc.). - pub fn info(&self) -> &StreamInfo { - &self.info - } - - /// Consumes the stream and returns its components. - pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader) { - (self.info, self.reader) - } - - /// Waits for the background decoding task to complete. - /// - /// This should be called after reading all data to ensure proper cleanup - /// and to catch any errors that occurred during decoding. - pub async fn wait(self) -> Result<(), FlacError> { - self.reader.wait().await - } -} - -impl AsyncRead for FlacDecodedStream { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut tokio::io::ReadBuf<'_>, - ) -> Poll> { - Pin::new(&mut self.reader).poll_read(cx, buf) - } -} +/// Async stream alias for decoded FLAC audio. +pub type FlacDecodedStream = DecodedStream; /// Decodes a FLAC stream into PCM audio data. /// @@ -210,5 +149,5 @@ where let info = info_rx.await.map_err(|_| FlacError::ChannelClosed)??; let reader = ManagedAsyncReader::new("flac-decode-writer", pcm_reader, writer_handle); - Ok(FlacDecodedStream { info, reader }) + Ok(DecodedStream::new(info, reader)) } diff --git a/pmoflac/src/decoder_common.rs b/pmoflac/src/decoder_common.rs index c8da6437..34b46b92 100644 --- a/pmoflac/src/decoder_common.rs +++ b/pmoflac/src/decoder_common.rs @@ -3,15 +3,21 @@ //! This module provides shared functionality for the FLAC, MP3, and Ogg/Vorbis decoders, //! reducing code duplication and ensuring consistent behavior across all decoders. -use std::io; +use std::{ + io, + pin::Pin, + task::{Context, Poll}, +}; use bytes::Bytes; use tokio::{ - io::{AsyncRead, AsyncReadExt, AsyncWriteExt, DuplexStream}, + io::{AsyncRead, AsyncReadExt, AsyncWriteExt, DuplexStream, ReadBuf}, sync::mpsc, task::JoinHandle, }; +use crate::{pcm::StreamInfo, stream::ManagedAsyncReader}; + /// Generic error type shared by the streaming decoders. #[derive(thiserror::Error, Debug, Clone)] pub enum DecoderError { @@ -47,6 +53,56 @@ impl From<&str> for DecoderError { } } +/// Generic decoded stream wrapper shared by all decoders. +pub struct DecodedStream +where + E: std::error::Error, +{ + info: StreamInfo, + reader: ManagedAsyncReader, +} + +impl DecodedStream +where + E: std::error::Error, +{ + /// Creates a new decoded stream from metadata and the underlying reader. + pub fn new(info: StreamInfo, reader: ManagedAsyncReader) -> Self { + Self { info, reader } + } + + /// Returns metadata about the decoded audio stream. + pub fn info(&self) -> &StreamInfo { + &self.info + } + + /// Consumes the stream and returns its components. + pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader) { + (self.info, self.reader) + } + + /// Waits for the decoding pipeline to finish. + pub async fn wait(self) -> Result<(), E> + where + E: From, + { + self.reader.wait().await + } +} + +impl AsyncRead for DecodedStream +where + E: std::error::Error, +{ + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.reader).poll_read(cx, buf) + } +} + /// Size of chunks when reading input data. /// /// This size balances between efficient I/O operations and memory usage. diff --git a/pmoflac/src/mp3.rs b/pmoflac/src/mp3.rs index 2488cd73..a54e0a4c 100644 --- a/pmoflac/src/mp3.rs +++ b/pmoflac/src/mp3.rs @@ -87,22 +87,17 @@ //! } //! ``` -use std::{ - io, - pin::Pin, - task::{Context, Poll}, -}; - use minimp3::{Decoder as MiniMp3Decoder, Error as MiniMp3Error}; use tokio::{ - io::{AsyncRead, ReadBuf}, + io::AsyncRead, sync::{mpsc, oneshot}, }; use crate::{ common::ChannelReader, decoder_common::{ - spawn_ingest_task, spawn_writer_task, DecoderError, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE, + spawn_ingest_task, spawn_writer_task, DecodedStream, DecoderError, CHANNEL_CAPACITY, + DUPLEX_BUFFER_SIZE, }, pcm::StreamInfo, stream::ManagedAsyncReader, @@ -111,44 +106,8 @@ use crate::{ /// Errors that can occur while decoding MP3 data. pub type Mp3Error = DecoderError; -/// An async stream that decodes MP3 audio into PCM samples. -/// -/// This struct implements `AsyncRead`, allowing you to read decoded PCM data -/// as it becomes available. The decoding happens in a background task. -pub struct Mp3DecodedStream { - info: StreamInfo, - reader: ManagedAsyncReader, -} - -impl Mp3DecodedStream { - /// Returns metadata about the decoded MP3 stream. - pub fn info(&self) -> &StreamInfo { - &self.info - } - - /// Consumes the stream and returns its components. - pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader) { - (self.info, self.reader) - } - - /// Waits for the background decoding task to complete. - /// - /// This should be called after reading all data to ensure proper cleanup - /// and to catch any errors that occurred during decoding. - pub async fn wait(self) -> Result<(), Mp3Error> { - self.reader.wait().await - } -} - -impl AsyncRead for Mp3DecodedStream { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut ReadBuf<'_>, - ) -> Poll> { - Pin::new(&mut self.reader).poll_read(cx, buf) - } -} +/// Async decoded MP3 stream (type alias over the shared wrapper). +pub type Mp3DecodedStream = DecodedStream; /// Decodes an MP3 stream into PCM audio data (16-bit little-endian interleaved). /// @@ -239,5 +198,5 @@ where let info = info_rx.await.map_err(|_| Mp3Error::ChannelClosed)??; let reader = ManagedAsyncReader::new("mp3-decode-writer", pcm_reader, writer_handle); - Ok(Mp3DecodedStream { info, reader }) + Ok(DecodedStream::new(info, reader)) } diff --git a/pmoflac/src/ogg.rs b/pmoflac/src/ogg.rs index 854230e5..6c3f8b8f 100644 --- a/pmoflac/src/ogg.rs +++ b/pmoflac/src/ogg.rs @@ -101,25 +101,21 @@ //! } //! ``` -use std::{ - io, - pin::Pin, - task::{Context, Poll}, -}; - use lewton::{ audio::{self, read_audio_packet_generic, PreviousWindowRight}, header::{self, read_header_comment, read_header_ident, read_header_setup, CommentHeader}, samples::InterleavedSamples, }; use tokio::{ - io::{AsyncRead, ReadBuf}, + io::AsyncRead, sync::{mpsc, oneshot}, }; use crate::{ common::ChannelReader, - decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE}, + decoder_common::{ + spawn_ingest_task, spawn_writer_task, DecodedStream, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE, + }, ogg_common::{OggContainerError, OggPacketReader, OggReaderOptions}, pcm::StreamInfo, stream::ManagedAsyncReader, @@ -140,48 +136,8 @@ impl From for OggContainerError { } } -/// An async stream that decodes Ogg/Vorbis audio into PCM samples. -/// -/// This struct implements `AsyncRead`, allowing you to read decoded PCM data -/// as it becomes available. The decoding happens in a background task. -pub struct OggDecodedStream { - info: StreamInfo, - reader: ManagedAsyncReader, -} - -impl OggDecodedStream { - /// Returns metadata about the decoded Ogg/Vorbis stream. - /// - /// This includes sample rate, channel count, bits per sample, and block sizes. - pub fn info(&self) -> &StreamInfo { - &self.info - } - - /// Consumes the stream and returns its components. - /// - /// Useful for chaining with encoders without buffering the PCM data. - pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader) { - (self.info, self.reader) - } - - /// Waits for the background decoding task to complete. - /// - /// This should be called after reading all data to ensure proper cleanup - /// and to catch any errors that occurred during decoding. - pub async fn wait(self) -> Result<(), OggError> { - self.reader.wait().await - } -} - -impl AsyncRead for OggDecodedStream { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut ReadBuf<'_>, - ) -> Poll> { - Pin::new(&mut self.reader).poll_read(cx, buf) - } -} +/// Async stream alias for decoded Ogg/Vorbis audio. +pub type OggDecodedStream = DecodedStream; /// Decodes an Ogg/Vorbis stream into PCM audio (16-bit little-endian interleaved). /// @@ -310,5 +266,5 @@ where let info = info_rx.await.map_err(|_| OggError::ChannelClosed)??; let reader = ManagedAsyncReader::new("ogg-decode-writer", pcm_reader, writer_handle); - Ok(OggDecodedStream { info, reader }) + Ok(DecodedStream::new(info, reader)) } diff --git a/pmoflac/src/opus.rs b/pmoflac/src/opus.rs index 640c927c..ee9476e8 100644 --- a/pmoflac/src/opus.rs +++ b/pmoflac/src/opus.rs @@ -4,21 +4,17 @@ //! ingestion/producer pattern established for the other decoders while staying //! 100% streaming (no seeking or buffering entire files). -use std::{ - io, - pin::Pin, - task::{Context, Poll}, -}; - use opus::{Channels, Decoder as OpusDecoder, Error as OpusError}; use tokio::{ - io::{AsyncRead, ReadBuf}, + io::AsyncRead, sync::{mpsc, oneshot}, }; use crate::{ common::ChannelReader, - decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE}, + decoder_common::{ + spawn_ingest_task, spawn_writer_task, DecodedStream, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE, + }, ogg_common::{OggContainerError, OggPacketReader, OggReaderOptions}, pcm::StreamInfo, stream::ManagedAsyncReader, @@ -36,38 +32,8 @@ impl From for OggContainerError { } } -/// An async stream that decodes Ogg/Opus audio into PCM samples. -pub struct OggOpusDecodedStream { - info: StreamInfo, - reader: ManagedAsyncReader, -} - -impl OggOpusDecodedStream { - /// Returns metadata about the decoded stream. - pub fn info(&self) -> &StreamInfo { - &self.info - } - - /// Consumes the stream and returns its components. - pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader) { - (self.info, self.reader) - } - - /// Waits for the decoding pipeline to finish. - pub async fn wait(self) -> Result<(), OggOpusError> { - self.reader.wait().await - } -} - -impl AsyncRead for OggOpusDecodedStream { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut ReadBuf<'_>, - ) -> Poll> { - Pin::new(&mut self.reader).poll_read(cx, buf) - } -} +/// Async stream alias for decoded Ogg/Opus audio. +pub type OggOpusDecodedStream = DecodedStream; /// Decodes an Ogg/Opus stream into PCM audio (16-bit little-endian). pub async fn decode_ogg_opus_stream(reader: R) -> Result @@ -195,7 +161,7 @@ where let info = info_rx.await.map_err(|_| OggOpusError::ChannelClosed)??; let reader = ManagedAsyncReader::new("ogg-opus-writer", pcm_reader, writer_handle); - Ok(OggOpusDecodedStream { info, reader }) + Ok(DecodedStream::new(info, reader)) } /// Parsed OpusHead metadata. diff --git a/pmoflac/src/wav.rs b/pmoflac/src/wav.rs index 541c5a3d..6e595a39 100644 --- a/pmoflac/src/wav.rs +++ b/pmoflac/src/wav.rs @@ -4,12 +4,7 @@ //! header incrementally, validates the format, and then streams `data` chunk //! payload as little-endian PCM frames through the common async pipeline. -use std::{ - fmt, - io::{self, Read}, - pin::Pin, - task::{Context, Poll}, -}; +use std::{fmt, io::Read}; use tokio::{ io::AsyncRead, @@ -19,7 +14,8 @@ use tokio::{ use crate::{ common::ChannelReader, decoder_common::{ - spawn_ingest_task, spawn_writer_task, DecoderError, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE, + spawn_ingest_task, spawn_writer_task, DecodedStream, DecoderError, CHANNEL_CAPACITY, + DUPLEX_BUFFER_SIZE, }, pcm::StreamInfo, stream::ManagedAsyncReader, @@ -145,35 +141,8 @@ impl FmtChunk { } } -/// An async stream that yields PCM decoded from WAV data. -pub struct WavDecodedStream { - info: StreamInfo, - reader: ManagedAsyncReader, -} - -impl WavDecodedStream { - pub fn info(&self) -> &StreamInfo { - &self.info - } - - pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader) { - (self.info, self.reader) - } - - pub async fn wait(self) -> Result<(), WavError> { - self.reader.wait().await - } -} - -impl AsyncRead for WavDecodedStream { - fn poll_read( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut tokio::io::ReadBuf<'_>, - ) -> Poll> { - Pin::new(&mut self.reader).poll_read(cx, buf) - } -} +/// Async stream alias for decoded WAV audio. +pub type WavDecodedStream = DecodedStream; /// Decode a WAV stream into PCM audio. pub async fn decode_wav_stream(reader: R) -> Result @@ -315,5 +284,5 @@ where let info = info_rx.await.map_err(|_| WavError::ChannelClosed)??; let reader = ManagedAsyncReader::new("wav-decode-writer", pcm_reader, writer_handle); - Ok(WavDecodedStream { info, reader }) + Ok(DecodedStream::new(info, reader)) }