From 6d41abc9014ac8aa60d58bce494951ccdbb54625 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Mon, 27 Oct 2025 18:13:50 +0100 Subject: [PATCH] =?UTF-8?q?Ajout=20du=20decodage=20mp3=20=C3=A0=20pmoflac?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 41 +++++ pmoflac/Cargo.toml | 5 +- pmoflac/src/common.rs | 81 +++++++++ pmoflac/src/decoder.rs | 65 +------- pmoflac/src/encoder.rs | 14 +- pmoflac/src/error.rs | 6 + pmoflac/src/lib.rs | 45 ++++- pmoflac/src/mp3.rs | 323 ++++++++++++++++++++++++++++++++++++ pmoflac/src/stream.rs | 41 +++-- pmoflac/tests/mp3_decode.rs | 52 ++++++ 10 files changed, 592 insertions(+), 81 deletions(-) create mode 100644 pmoflac/src/common.rs create mode 100644 pmoflac/src/mp3.rs create mode 100644 pmoflac/tests/mp3_decode.rs diff --git a/Cargo.lock b/Cargo.lock index f6f37c04..729428f5 100644 --- a/Cargo.lock +++ b/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" diff --git a/pmoflac/Cargo.toml b/pmoflac/Cargo.toml index 1742221e..8eb853db 100644 --- a/pmoflac/Cargo.toml +++ b/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"] } diff --git a/pmoflac/src/common.rs b/pmoflac/src/common.rs new file mode 100644 index 00000000..909abad6 --- /dev/null +++ b/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 +where + E: std::error::Error + std::fmt::Display, +{ + rx: mpsc::Receiver>, + current: Bytes, + offset: usize, + finished: bool, +} + +impl ChannelReader +where + E: std::error::Error + std::fmt::Display, +{ + pub fn new(rx: mpsc::Receiver>) -> Self { + Self { + rx, + current: Bytes::new(), + offset: 0, + finished: false, + } + } +} + +impl Read for ChannelReader +where + E: std::error::Error + std::fmt::Display, +{ + fn read(&mut self, buf: &mut [u8]) -> io::Result { + 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); + } + } + } + } +} diff --git a/pmoflac/src/decoder.rs b/pmoflac/src/decoder.rs index e75b7859..1a076752 100644 --- a/pmoflac/src/decoder.rs +++ b/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, } 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) { (self.info, self.reader) } @@ -163,7 +166,7 @@ where let (info_tx, info_rx) = oneshot::channel::>(); let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), FlacError> { - let mut channel_reader = ChannelReader::new(ingest_rx); + let mut channel_reader = ChannelReader::::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>, - current: Bytes, - offset: usize, - finished: bool, -} - -impl ChannelReader { - fn new(rx: mpsc::Receiver>) -> Self { - Self { - rx, - current: Bytes::new(), - offset: 0, - finished: false, - } - } -} - -impl Read for ChannelReader { - fn read(&mut self, buf: &mut [u8]) -> io::Result { - 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); - } - } - } - } -} diff --git a/pmoflac/src/encoder.rs b/pmoflac/src/encoder.rs index e4004e88..7ccd87b5 100644 --- a/pmoflac/src/encoder.rs +++ b/pmoflac/src/encoder.rs @@ -52,7 +52,7 @@ const PCM_FRAMES_PER_CHUNK: usize = 4096; /// ``` pub struct FlacEncodedStream { format: PcmFormat, - reader: ManagedAsyncReader, + reader: ManagedAsyncReader, } 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) { (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(()) /// # } /// ``` diff --git a/pmoflac/src/error.rs b/pmoflac/src/error.rs index 81bd8f91..8c6ddcaa 100644 --- a/pmoflac/src/error.rs +++ b/pmoflac/src/error.rs @@ -25,3 +25,9 @@ impl From for FlacError { FlacError::Decode(err.to_string()) } } + +impl From for FlacError { + fn from(msg: String) -> Self { + FlacError::Decode(msg) + } +} diff --git a/pmoflac/src/lib.rs b/pmoflac/src/lib.rs index 338d51b2..ddffb076 100644 --- a/pmoflac/src/lib.rs +++ b/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. -//! The key feature is **true streaming**: data is processed incrementally without buffering -//! entire files in memory. +//! This library provides streaming FLAC and MP3 decoding, as well as FLAC encoding, +//! with a Tokio-based async API. The key feature is **true streaming**: data is +//! 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> { +//! // 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}; diff --git a/pmoflac/src/mp3.rs b/pmoflac/src/mp3.rs new file mode 100644 index 00000000..52aaa7bd --- /dev/null +++ b/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> { +//! 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> { +//! // 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 for Mp3Error { + fn from(err: io::Error) -> Self { + Mp3Error::Io { + kind: err.kind(), + message: err.to_string(), + } + } +} + +impl From 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, +} + +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) + } +} + +/// 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(reader: R) -> Result +where + R: AsyncRead + Unpin + Send + 'static, +{ + let (ingest_tx, ingest_rx) = mpsc::channel::>(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::, Mp3Error>>(CHANNEL_CAPACITY); + let (pcm_reader, mut pcm_writer) = tokio_io::duplex(256 * 1024); + let (info_tx, info_rx) = oneshot::channel::>(); + + let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), Mp3Error> { + let channel_reader = ChannelReader::::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 }) +} diff --git a/pmoflac/src/stream.rs b/pmoflac/src/stream.rs index b4aba9c8..a5f96df2 100644 --- a/pmoflac/src/stream.rs +++ b/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 +where + E: std::error::Error, +{ inner: Option, - join: Option>>, + join: Option>>, role: &'static str, } -impl ManagedAsyncReader { +impl ManagedAsyncReader +where + E: std::error::Error, +{ pub fn new( role: &'static str, inner: DuplexStream, - join: JoinHandle>, + join: JoinHandle>, ) -> 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, + { match self.join.take() { Some(handle) => match handle.await { Ok(res) => res, - Err(err) => Err(FlacError::TaskJoin { - role: self.role, - details: err.to_string(), - }), + Err(err) => { + let msg = format!("task '{}' failed: {}", self.role, err); + Err(E::from(msg)) + } }, None => Ok(()), } @@ -58,7 +67,10 @@ impl ManagedAsyncReader { } } -impl AsyncRead for ManagedAsyncReader { +impl AsyncRead for ManagedAsyncReader +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 Drop for ManagedAsyncReader +where + E: std::error::Error, +{ fn drop(&mut self) { if let Some(handle) = self.join.take() { handle.abort(); diff --git a/pmoflac/tests/mp3_decode.rs b/pmoflac/tests/mp3_decode.rs new file mode 100644 index 00000000..5524bf4a --- /dev/null +++ b/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> { + 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> { + 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(()) +}