Ajout du decodage mp3 à pmoflac

Cargo.lock

@@ -2135,6 +2135,15 @@ dependencies = [
  "imgref",
 ]
 
+[[package]]
+name = "mach2"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d640282b302c0bb0a2a8e0233ead9035e3bed871f0b7e81fe4a1ec829765db44"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "matchers"
 version = "0.2.0"
@@ -2204,6 +2213,26 @@ version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
 
+[[package]]
+name = "minimp3"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9a3ed9d34ed1a9190336a2b165bf09ac447693dfd9a61684597aaae2ee12df53"
+dependencies = [
+ "minimp3-sys",
+ "slice-ring-buffer",
+ "thiserror 1.0.69",
+]
+
+[[package]]
+name = "minimp3-sys"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e21c73734c69dc95696c9ed8926a2b393171d98b3f5f5935686a26a487ab9b90"
+dependencies = [
+ "cc",
+]
+
 [[package]]
 name = "miniz_oxide"
 version = "0.8.9"
@@ -2735,6 +2764,7 @@ dependencies = [
  "claxon",
  "libc",
  "libflac-sys",
+ "minimp3",
  "tempfile",
  "thiserror 1.0.69",
  "tokio",
@@ -3734,6 +3764,17 @@ version = "0.4.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7a2ae44ef20feb57a68b23d846850f861394c2e02dc425a50098ae8c90267589"
 
+[[package]]
+name = "slice-ring-buffer"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "84ae312bda09b2368f79f985fdb4df4a0b5cbc75546b511303972d195f8c27d6"
+dependencies = [
+ "libc",
+ "mach2",
+ "winapi 0.3.9",
+]
+
 [[package]]
 name = "smallvec"
 version = "1.15.1"

pmoflac/Cargo.toml

@@ -16,8 +16,9 @@ claxon = "0.4"
 libc = "0.2"
 libflac-sys = { version = "0.3.3", default-features = false, features = ["build-flac"] }
 thiserror = "1.0"
-tokio = { version = "1.37", features = ["rt", "macros", "sync", "io-util"] }
+tokio = { version = "1.37", features = ["rt", "rt-multi-thread", "macros", "sync", "io-util", "fs"] }
+minimp3 = "0.5"
 
 [dev-dependencies]
 tempfile = "3.10"
-tokio = { version = "1.37", features = ["rt", "macros", "sync", "io-util", "time"] }
+tokio = { version = "1.37", features = ["rt", "rt-multi-thread", "macros", "sync", "io-util", "time", "fs"] }

pmoflac/src/common.rs

@@ -0,0 +1,81 @@
+//! Common utilities shared between decoders.
+//!
+//! This module contains structures and adapters used by both FLAC and MP3 decoders
+//! to bridge async channels with synchronous I/O requirements.
+
+use std::io::{self, Read};
+
+use bytes::Bytes;
+use tokio::sync::mpsc;
+
+/// Internal adapter that bridges async channel reading to sync `std::io::Read`.
+///
+/// Many decoder libraries (like minimp3, claxon) require a synchronous `Read`
+/// implementation, but our data arrives via an async channel. This adapter uses
+/// `blocking_recv` to bridge the gap, buffering chunks as they arrive.
+///
+/// This is generic over the error type to support both FLAC and MP3 decoders.
+pub(crate) struct ChannelReader<E>
+where
+    E: std::error::Error + std::fmt::Display,
+{
+    rx: mpsc::Receiver<Result<Bytes, E>>,
+    current: Bytes,
+    offset: usize,
+    finished: bool,
+}
+
+impl<E> ChannelReader<E>
+where
+    E: std::error::Error + std::fmt::Display,
+{
+    pub fn new(rx: mpsc::Receiver<Result<Bytes, E>>) -> Self {
+        Self {
+            rx,
+            current: Bytes::new(),
+            offset: 0,
+            finished: false,
+        }
+    }
+}
+
+impl<E> Read for ChannelReader<E>
+where
+    E: std::error::Error + std::fmt::Display,
+{
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        loop {
+            // If we have data in the current buffer, copy it out
+            if self.offset < self.current.len() {
+                let n = buf.len().min(self.current.len() - self.offset);
+                buf[..n].copy_from_slice(&self.current[self.offset..self.offset + n]);
+                self.offset += n;
+                return Ok(n);
+            }
+
+            // If we're finished, return EOF
+            if self.finished {
+                return Ok(0);
+            }
+
+            // Try to receive the next chunk from the channel
+            match self.rx.blocking_recv() {
+                Some(Ok(bytes)) => {
+                    if bytes.is_empty() {
+                        continue;
+                    }
+                    self.current = bytes;
+                    self.offset = 0;
+                }
+                Some(Err(err)) => {
+                    self.finished = true;
+                    return Err(io::Error::new(io::ErrorKind::Other, err.to_string()));
+                }
+                None => {
+                    self.finished = true;
+                    return Ok(0);
+                }
+            }
+        }
+    }
+}

pmoflac/src/decoder.rs

@@ -1,5 +1,5 @@
 use std::{
-    io::{self, Read},
+    io,
     pin::Pin,
     task::{Context, Poll},
 };
@@ -11,7 +11,10 @@ use tokio::{
 };
 
 use crate::{
-    error::FlacError, pcm::StreamInfo, stream::ManagedAsyncReader,
+    common::ChannelReader,
+    error::FlacError,
+    pcm::StreamInfo,
+    stream::ManagedAsyncReader,
     util::interleaved_i32_to_le_bytes,
 };
 
@@ -48,7 +51,7 @@ const CHANNEL_CAPACITY: usize = 8;
 /// ```
 pub struct FlacDecodedStream {
     info: StreamInfo,
-    reader: ManagedAsyncReader,
+    reader: ManagedAsyncReader<FlacError>,
 }
 
 impl FlacDecodedStream {
@@ -58,7 +61,7 @@ impl FlacDecodedStream {
     }
 
     /// Consumes the stream and returns its components.
-    pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader) {
+    pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader<FlacError>) {
         (self.info, self.reader)
     }
 
@@ -163,7 +166,7 @@ where
     let (info_tx, info_rx) = oneshot::channel::<Result<StreamInfo, FlacError>>();
 
     let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), FlacError> {
-        let mut channel_reader = ChannelReader::new(ingest_rx);
+        let mut channel_reader = ChannelReader::<FlacError>::new(ingest_rx);
         let mut flac_reader = match claxon::FlacReader::new(&mut channel_reader) {
             Ok(reader) => reader,
             Err(err) => {
@@ -250,55 +253,3 @@ where
 
     Ok(FlacDecodedStream { info, reader })
 }
-
-struct ChannelReader {
-    rx: mpsc::Receiver<Result<Bytes, FlacError>>,
-    current: Bytes,
-    offset: usize,
-    finished: bool,
-}
-
-impl ChannelReader {
-    fn new(rx: mpsc::Receiver<Result<Bytes, FlacError>>) -> Self {
-        Self {
-            rx,
-            current: Bytes::new(),
-            offset: 0,
-            finished: false,
-        }
-    }
-}
-
-impl Read for ChannelReader {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        loop {
-            if self.offset < self.current.len() {
-                let n = std::cmp::min(buf.len(), self.current.len() - self.offset);
-                buf[..n].copy_from_slice(&self.current[self.offset..self.offset + n]);
-                self.offset += n;
-                return Ok(n);
-            }
-            if self.finished {
-                return Ok(0);
-            }
-
-            match self.rx.blocking_recv() {
-                Some(Ok(bytes)) => {
-                    if bytes.is_empty() {
-                        continue;
-                    }
-                    self.current = bytes;
-                    self.offset = 0;
-                }
-                Some(Err(err)) => {
-                    self.finished = true;
-                    return Err(io::Error::new(io::ErrorKind::Other, err.to_string()));
-                }
-                None => {
-                    self.finished = true;
-                    return Ok(0);
-                }
-            }
-        }
-    }
-}

pmoflac/src/encoder.rs

@@ -52,7 +52,7 @@ const PCM_FRAMES_PER_CHUNK: usize = 4096;
 /// ```
 pub struct FlacEncodedStream {
     format: PcmFormat,
-    reader: ManagedAsyncReader,
+    reader: ManagedAsyncReader<FlacError>,
 }
 
 impl FlacEncodedStream {
@@ -62,7 +62,7 @@ impl FlacEncodedStream {
     }
 
     /// Consumes the stream and returns its components.
-    pub fn into_parts(self) -> (PcmFormat, ManagedAsyncReader) {
+    pub fn into_parts(self) -> (PcmFormat, ManagedAsyncReader<FlacError>) {
         (self.format, self.reader)
     }
 
@@ -166,7 +166,7 @@ impl Default for EncoderOptions {
 /// // Generate 1 second of silence at 44.1kHz stereo 16-bit
 /// let sample_rate = 44_100u32;
 /// let channels = 2u8;
-/// let pcm_data = vec![0u8; sample_rate as usize * channels as usize * 2];
+/// let pcm_len = sample_rate as usize * channels as usize * 2;
 ///
 /// let format = PcmFormat {
 ///     sample_rate,
@@ -180,13 +180,17 @@ impl Default for EncoderOptions {
 ///     ..Default::default()
 /// };
 ///
-/// let mut stream = encode_flac_stream(&pcm_data[..], format, options).await?;
+/// let mut stream = encode_flac_stream(
+///     tokio::io::repeat(0).take(pcm_len as u64),
+///     format,
+///     options,
+/// ).await?;
 /// let mut flac_data = Vec::new();
 /// stream.read_to_end(&mut flac_data).await?;
 /// stream.wait().await?;
 ///
 /// println!("Encoded {} bytes of PCM to {} bytes of FLAC",
-///     pcm_data.len(), flac_data.len());
+///     pcm_len, flac_data.len());
 /// # Ok(())
 /// # }
 /// ```

pmoflac/src/error.rs

@@ -25,3 +25,9 @@ impl From<claxon::Error> for FlacError {
         FlacError::Decode(err.to_string())
     }
 }
+
+impl From<String> for FlacError {
+    fn from(msg: String) -> Self {
+        FlacError::Decode(msg)
+    }
+}

pmoflac/src/lib.rs

@@ -1,17 +1,20 @@
 //! # pmoflac
 //!
-//! Asynchronous FLAC encoding and decoding library for Rust.
+//! Asynchronous audio encoding and decoding library for Rust.
 //!
-//! This library provides streaming FLAC encoding and decoding with a Tokio-based async API.
+//! This library provides streaming FLAC and MP3 decoding, as well as FLAC encoding,
-//! The key feature is **true streaming**: data is processed incrementally without buffering
+//! with a Tokio-based async API. The key feature is **true streaming**: data is
-//! entire files in memory.
+//! processed incrementally without buffering entire files in memory.
 //!
 //! ## Features
 //!
+//! - **MP3 decoding**: Stream MP3 files to PCM data
+//! - **FLAC encoding/decoding**: Bidirectional FLAC ↔ PCM conversion
 //! - **Async streaming API**: Built on Tokio's `AsyncRead` trait
 //! - **Low memory footprint**: Processes data in chunks, not entire files
 //! - **Zero-copy where possible**: Efficient buffer management
 //! - **Thread-safe**: Uses channels for inter-task communication
+//! - **Composable**: Chain decoders and encoders (e.g., MP3 → PCM → FLAC)
 //!
 //! ## Example: Decode FLAC to PCM
 //!
@@ -60,6 +63,37 @@
 //!     Ok(())
 //! }
 //! ```
+//!
+//! ## Example: Transcode MP3 to FLAC
+//!
+//! ```no_run
+//! use pmoflac::{decode_mp3_stream, encode_flac_stream, PcmFormat, EncoderOptions};
+//! use tokio::fs::File;
+//! use tokio::io;
+//!
+//! #[tokio::main]
+//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
+//!     // Decode MP3 to PCM stream
+//!     let mp3_file = File::open("input.mp3").await?;
+//!     let mp3_stream = decode_mp3_stream(mp3_file).await?;
+//!     let (info, pcm_reader) = mp3_stream.into_parts();
+//!
+//!     // Encode PCM stream to FLAC
+//!     let format = PcmFormat {
+//!         sample_rate: info.sample_rate,
+//!         channels: info.channels,
+//!         bits_per_sample: info.bits_per_sample,
+//!     };
+//!     let mut flac_stream = encode_flac_stream(pcm_reader, format, EncoderOptions::default()).await?;
+//!
+//!     // Write to output file
+//!     let mut output = File::create("output.flac").await?;
+//!     io::copy(&mut flac_stream, &mut output).await?;
+//!     flac_stream.wait().await?;
+//!
+//!     Ok(())
+//! }
+//! ```
 
 pub mod decoder;
 pub mod encoder;
@@ -67,8 +101,11 @@ pub mod error;
 mod pcm;
 mod stream;
 mod util;
+pub mod mp3;
+mod common;
 
 pub use decoder::{decode_flac_stream, FlacDecodedStream};
 pub use encoder::{encode_flac_stream, EncoderOptions, FlacEncodedStream};
 pub use error::FlacError;
+pub use mp3::{decode_mp3_stream, Mp3DecodedStream, Mp3Error};
 pub use pcm::{PcmFormat, StreamInfo};

pmoflac/src/mp3.rs

@@ -0,0 +1,323 @@
+//! # MP3 Decoder Module
+//!
+//! This module provides asynchronous streaming MP3 decoding capabilities.
+//! It decodes MP3 audio streams into PCM data (16-bit little-endian interleaved),
+//! which can then be fed directly into the FLAC encoder for transcoding.
+//!
+//! ## Architecture
+//!
+//! The decoder uses a multi-task pipeline for efficient streaming:
+//!
+//! ```text
+//! MP3 Input → [Ingest Task] → [Decode Task] → [Writer Task] → PCM Output (AsyncRead)
+//!                  ↓              ↓                ↓
+//!              mpsc channel   blocking I/O    duplex stream
+//! ```
+//!
+//! - **Ingest Task**: Reads MP3 data in chunks and sends it through a channel
+//! - **Decode Task**: Runs in a blocking thread, decodes MP3 frames using minimp3
+//! - **Writer Task**: Writes decoded PCM data to a duplex stream
+//!
+//! This architecture ensures:
+//! - True streaming with minimal memory footprint
+//! - Non-blocking async I/O for the consumer
+//! - Proper backpressure through bounded channels
+//!
+//! ## Example: Basic MP3 Decoding
+//!
+//! ```no_run
+//! use pmoflac::decode_mp3_stream;
+//! use tokio::fs::File;
+//! use tokio::io::AsyncReadExt;
+//!
+//! #[tokio::main]
+//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
+//!     let file = File::open("audio.mp3").await?;
+//!     let mut stream = decode_mp3_stream(file).await?;
+//!
+//!     // Get stream information
+//!     let info = stream.info();
+//!     println!("Sample rate: {} Hz", info.sample_rate);
+//!     println!("Channels: {}", info.channels);
+//!     println!("Bits per sample: {}", info.bits_per_sample);
+//!
+//!     // Read PCM data
+//!     let mut pcm_buffer = Vec::new();
+//!     stream.read_to_end(&mut pcm_buffer).await?;
+//!
+//!     // Wait for decoding to complete
+//!     stream.wait().await?;
+//!
+//!     Ok(())
+//! }
+//! ```
+//!
+//! ## Example: MP3 to FLAC Transcoding
+//!
+//! ```no_run
+//! use pmoflac::{decode_mp3_stream, encode_flac_stream, PcmFormat, EncoderOptions};
+//! use tokio::fs::File;
+//! use tokio::io::AsyncReadExt;
+//!
+//! #[tokio::main]
+//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
+//!     // Decode MP3
+//!     let mp3_file = File::open("input.mp3").await?;
+//!     let stream = decode_mp3_stream(mp3_file).await?;
+//!     let (info, pcm_reader) = stream.into_parts();
+//!
+//!     // Encode to FLAC
+//!     let format = PcmFormat {
+//!         sample_rate: info.sample_rate,
+//!         channels: info.channels,
+//!         bits_per_sample: info.bits_per_sample,
+//!     };
+//!     let mut flac_stream = encode_flac_stream(
+//!         pcm_reader,
+//!         format,
+//!         EncoderOptions::default()
+//!     ).await?;
+//!
+//!     // Write FLAC output
+//!     let mut output = File::create("output.flac").await?;
+//!     tokio::io::copy(&mut flac_stream, &mut output).await?;
+//!     flac_stream.wait().await?;
+//!
+//!     Ok(())
+//! }
+//! ```
+
+use std::{
+    io,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use bytes::Bytes;
+use minimp3::{Decoder as MiniMp3Decoder, Error as MiniMp3Error};
+use tokio::{
+    io::{
+        self as tokio_io, AsyncRead, AsyncReadExt, AsyncWriteExt, ReadBuf,
+    },
+    sync::{mpsc, oneshot},
+};
+
+use crate::{common::ChannelReader, pcm::StreamInfo, stream::ManagedAsyncReader};
+
+/// Size of chunks when reading MP3 input data (16 KB).
+///
+/// This size balances between efficient I/O operations and memory usage.
+/// Larger chunks reduce system call overhead, while smaller chunks reduce latency.
+const INGEST_CHUNK_SIZE: usize = 16 * 1024;
+
+/// Channel capacity for async message passing between tasks.
+///
+/// This bounded capacity provides backpressure: if the decoder can't keep up,
+/// the ingest task will wait before reading more data.
+const CHANNEL_CAPACITY: usize = 8;
+
+/// Errors that can occur while decoding MP3 data.
+#[derive(thiserror::Error, Debug, Clone)]
+pub enum Mp3Error {
+    #[error("I/O error ({kind:?}): {message}")]
+    Io {
+        kind: io::ErrorKind,
+        message: String,
+    },
+    #[error("MP3 decode error: {0}")]
+    Decode(String),
+    #[error("internal channel closed unexpectedly")]
+    ChannelClosed,
+    #[error("{role} task failed: {details}")]
+    TaskJoin { role: &'static str, details: String },
+}
+
+impl From<io::Error> for Mp3Error {
+    fn from(err: io::Error) -> Self {
+        Mp3Error::Io {
+            kind: err.kind(),
+            message: err.to_string(),
+        }
+    }
+}
+
+impl From<String> for Mp3Error {
+    fn from(msg: String) -> Self {
+        Mp3Error::Decode(msg)
+    }
+}
+
+/// 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<Mp3Error>,
+}
+
+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<Mp3Error>) {
+        (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<io::Result<()>> {
+        Pin::new(&mut self.reader).poll_read(cx, buf)
+    }
+}
+
+/// Decodes an MP3 stream into PCM audio data (16-bit little-endian interleaved).
+///
+/// This function spawns background tasks to perform the decoding asynchronously.
+/// The returned `Mp3DecodedStream` implements `AsyncRead` for streaming the PCM output.
+pub async fn decode_mp3_stream<R>(reader: R) -> Result<Mp3DecodedStream, Mp3Error>
+where
+    R: AsyncRead + Unpin + Send + 'static,
+{
+    let (ingest_tx, ingest_rx) = mpsc::channel::<Result<Bytes, Mp3Error>>(CHANNEL_CAPACITY);
+
+    tokio::spawn(async move {
+        let mut reader = tokio_io::BufReader::new(reader);
+        let mut buf = vec![0u8; INGEST_CHUNK_SIZE];
+
+        loop {
+            match reader.read(&mut buf).await {
+                Ok(0) => break,
+                Ok(n) => {
+                    let chunk = Bytes::copy_from_slice(&buf[..n]);
+                    if ingest_tx.send(Ok(chunk)).await.is_err() {
+                        break;
+                    }
+                }
+                Err(err) => {
+                    let _ = ingest_tx.send(Err(Mp3Error::from(err))).await;
+                    break;
+                }
+            }
+        }
+    });
+
+    let (pcm_tx, mut pcm_rx) = mpsc::channel::<Result<Vec<u8>, Mp3Error>>(CHANNEL_CAPACITY);
+    let (pcm_reader, mut pcm_writer) = tokio_io::duplex(256 * 1024);
+    let (info_tx, info_rx) = oneshot::channel::<Result<StreamInfo, Mp3Error>>();
+
+    let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), Mp3Error> {
+        let channel_reader = ChannelReader::<Mp3Error>::new(ingest_rx);
+        let mut decoder = MiniMp3Decoder::new(channel_reader);
+        let mut info_tx = Some(info_tx);
+        let mut pcm_bytes = Vec::new();
+
+        loop {
+            match decoder.next_frame() {
+                Ok(frame) => {
+                    if frame.channels == 0 {
+                        let err = Mp3Error::Decode("MP3 frame reported zero channels".into());
+                        if let Some(tx) = info_tx.take() {
+                            let _ = tx.send(Err(err.clone()));
+                        }
+                        return Err(err);
+                    }
+
+                    if let Some(tx) = info_tx.take() {
+                        let info = StreamInfo {
+                            sample_rate: frame.sample_rate as u32,
+                            channels: frame.channels as u8,
+                            bits_per_sample: 16,
+                            total_samples: None,
+                            max_block_size: 0,
+                            min_block_size: 0,
+                        };
+
+                        if tx.send(Ok(info.clone())).is_err() {
+                            // Consumer dropped; we can stop decoding early.
+                            return Ok(());
+                        }
+                    }
+
+                    pcm_bytes.clear();
+                    pcm_bytes.reserve(frame.data.len() * 2);
+                    for sample in &frame.data {
+                        pcm_bytes.extend_from_slice(&sample.to_le_bytes());
+                    }
+
+                    let chunk = std::mem::take(&mut pcm_bytes);
+                    if pcm_tx.blocking_send(Ok(chunk)).is_err() {
+                        break;
+                    }
+                    pcm_bytes = Vec::with_capacity(frame.data.len() * 2);
+                }
+                Err(MiniMp3Error::Eof) => break,
+                Err(MiniMp3Error::InsufficientData) | Err(MiniMp3Error::SkippedData) => {
+                    // Decoder needs more data; continue ingesting.
+                    continue;
+                }
+                Err(MiniMp3Error::Io(err)) => {
+                    let err = Mp3Error::from(err);
+                    if let Some(tx) = info_tx.take() {
+                        let _ = tx.send(Err(err.clone()));
+                    }
+                    return Err(err);
+                }
+            }
+        }
+
+        if let Some(tx) = info_tx.take() {
+            let err = Mp3Error::Decode("stream contained no decodable MP3 frames".into());
+            let _ = tx.send(Err(err.clone()));
+            return Err(err);
+        }
+
+        Ok(())
+    });
+
+    let writer_handle = tokio::spawn(async move {
+        while let Some(chunk_result) = pcm_rx.recv().await {
+            let chunk = chunk_result?;
+            if chunk.is_empty() {
+                continue;
+            }
+            pcm_writer
+                .write_all(&chunk)
+                .await
+                .map_err(Mp3Error::from)?;
+        }
+        pcm_writer
+            .shutdown()
+            .await
+            .map_err(Mp3Error::from)?;
+        match blocking_handle.await {
+            Ok(res) => res,
+            Err(err) => Err(Mp3Error::TaskJoin {
+                role: "mp3-decode",
+                details: err.to_string(),
+            }),
+        }
+    });
+
+    let info = info_rx
+        .await
+        .map_err(|_| Mp3Error::ChannelClosed)??;
+    let reader = ManagedAsyncReader::new("mp3-decode-writer", pcm_reader, writer_handle);
+
+    Ok(Mp3DecodedStream { info, reader })
+}

pmoflac/src/stream.rs

@@ -9,20 +9,26 @@ use tokio::{
     task::JoinHandle,
 };
 
-use crate::error::FlacError;
-
 /// Async reader that is backed by a spawned task writing into it.
-pub struct ManagedAsyncReader {
+///
+/// This is generic over the error type to support both FLAC and MP3 decoders.
+pub struct ManagedAsyncReader<E>
+where
+    E: std::error::Error,
+{
     inner: Option<DuplexStream>,
-    join: Option<JoinHandle<Result<(), FlacError>>>,
+    join: Option<JoinHandle<Result<(), E>>>,
     role: &'static str,
 }
 
-impl ManagedAsyncReader {
+impl<E> ManagedAsyncReader<E>
+where
+    E: std::error::Error,
+{
     pub fn new(
         role: &'static str,
         inner: DuplexStream,
-        join: JoinHandle<Result<(), FlacError>>,
+        join: JoinHandle<Result<(), E>>,
     ) -> Self {
         Self {
             inner: Some(inner),
@@ -32,14 +38,17 @@ impl ManagedAsyncReader {
     }
 
     /// Waits for the producer task to finish.
-    pub async fn wait(mut self) -> Result<(), FlacError> {
+    pub async fn wait(mut self) -> Result<(), E>
+    where
+        E: From<String>,
+    {
         match self.join.take() {
             Some(handle) => match handle.await {
                 Ok(res) => res,
-                Err(err) => Err(FlacError::TaskJoin {
+                Err(err) => {
-                    role: self.role,
+                    let msg = format!("task '{}' failed: {}", self.role, err);
-                    details: err.to_string(),
+                    Err(E::from(msg))
-                }),
+                }
             },
             None => Ok(()),
         }
@@ -58,7 +67,10 @@ impl ManagedAsyncReader {
     }
 }
 
-impl AsyncRead for ManagedAsyncReader {
+impl<E> AsyncRead for ManagedAsyncReader<E>
+where
+    E: std::error::Error,
+{
     fn poll_read(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
@@ -68,7 +80,10 @@ impl AsyncRead for ManagedAsyncReader {
     }
 }
 
-impl Drop for ManagedAsyncReader {
+impl<E> Drop for ManagedAsyncReader<E>
+where
+    E: std::error::Error,
+{
     fn drop(&mut self) {
         if let Some(handle) = self.join.take() {
             handle.abort();

pmoflac/tests/mp3_decode.rs

@@ -0,0 +1,52 @@
+use tokio::io::AsyncReadExt;
+
+use pmoflac::{
+    decode_mp3_stream, encode_flac_stream, EncoderOptions, PcmFormat, StreamInfo,
+};
+
+const TEST_MP3: &str = "test_data/file_example_MP3_5MG.mp3";
+
+#[tokio::test]
+async fn decode_mp3_produces_pcm() -> Result<(), Box<dyn std::error::Error>> {
+    let file = tokio::fs::File::open(TEST_MP3).await?;
+    let mut stream = decode_mp3_stream(file).await?;
+
+    let info: StreamInfo = stream.info().clone();
+    assert_eq!(info.bits_per_sample, 16);
+    assert!(info.channels > 0);
+    assert!(info.sample_rate > 0);
+
+    let mut pcm = Vec::new();
+    stream.read_to_end(&mut pcm).await?;
+    assert!(!pcm.is_empty());
+    let frame_width = info.channels as usize * info.bytes_per_sample();
+    assert_eq!(pcm.len() % frame_width, 0, "PCM data should align on frame");
+
+    stream.wait().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn mp3_pcm_can_be_encoded_to_flac() -> Result<(), Box<dyn std::error::Error>> {
+    let file = tokio::fs::File::open(TEST_MP3).await?;
+    let stream = decode_mp3_stream(file).await?;
+    let (info, reader) = stream.into_parts();
+
+    let format = PcmFormat {
+        sample_rate: info.sample_rate,
+        channels: info.channels,
+        bits_per_sample: info.bits_per_sample,
+    };
+
+    let mut flac = encode_flac_stream(reader, format, EncoderOptions::default()).await?;
+    let mut encoded = Vec::new();
+    flac.read_to_end(&mut encoded).await?;
+    assert!(!encoded.is_empty());
+    assert!(
+        encoded.starts_with(b"fLaC"),
+        "Encoded data should start with FLAC marker"
+    );
+    flac.wait().await?;
+
+    Ok(())
+}