diff --git a/Cargo.lock b/Cargo.lock index dea34cfe..12b54da7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -169,6 +169,17 @@ version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" +[[package]] +name = "audiopus_sys" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "62314a1546a2064e033665d658e88c620a62904be945f8147e6b16c3db9f8651" +dependencies = [ + "cmake", + "log", + "pkg-config", +] + [[package]] name = "autocfg" version = "1.5.0" @@ -2551,6 +2562,16 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" +[[package]] +name = "opus" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6526409b274a7e98e55ff59d96aafd38e6cd34d46b7dbbc32ce126dffcd75e8e" +dependencies = [ + "audiopus_sys", + "libc", +] + [[package]] name = "os_info" version = "3.12.0" @@ -2786,6 +2807,7 @@ dependencies = [ "libc", "libflac-sys", "minimp3", + "opus", "tempfile", "thiserror 1.0.69", "tokio", diff --git a/pmoflac/Cargo.toml b/pmoflac/Cargo.toml index 5af3a8d3..b90bac28 100644 --- a/pmoflac/Cargo.toml +++ b/pmoflac/Cargo.toml @@ -19,6 +19,7 @@ thiserror = "1.0" tokio = { version = "1.37", features = ["rt", "rt-multi-thread", "macros", "sync", "io-util", "fs"] } minimp3 = "0.5" lewton = "0.10" +opus = "0.3" [dev-dependencies] tempfile = "3.10" diff --git a/pmoflac/src/decoder.rs b/pmoflac/src/decoder.rs index 1a076752..41a929d0 100644 --- a/pmoflac/src/decoder.rs +++ b/pmoflac/src/decoder.rs @@ -4,26 +4,20 @@ use std::{ task::{Context, Poll}, }; -use bytes::Bytes; use tokio::{ - io::{AsyncRead, AsyncReadExt, AsyncWriteExt}, + io::AsyncRead, sync::{mpsc, oneshot}, }; use crate::{ common::ChannelReader, + decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE}, error::FlacError, pcm::StreamInfo, stream::ManagedAsyncReader, util::interleaved_i32_to_le_bytes, }; -/// Size of chunks when reading FLAC input data (32 KB). -const INGEST_CHUNK_SIZE: usize = 32 * 1024; - -/// Channel capacity for async message passing between tasks. -const CHANNEL_CAPACITY: usize = 8; - /// An async stream that decodes FLAC audio into PCM samples. /// /// This struct implements `AsyncRead`, allowing you to read decoded PCM data @@ -139,30 +133,11 @@ pub async fn decode_flac_stream(reader: R) -> Result>(CHANNEL_CAPACITY); + let (ingest_tx, ingest_rx) = mpsc::channel(CHANNEL_CAPACITY); + spawn_ingest_task(reader, ingest_tx); - 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(FlacError::Io(err))).await; - break; - } - } - } - }); - - let (pcm_tx, mut pcm_rx) = mpsc::channel::, FlacError>>(CHANNEL_CAPACITY); - let (pcm_reader, mut pcm_writer) = tokio::io::duplex(256 * 1024); + let (pcm_tx, pcm_rx) = mpsc::channel(CHANNEL_CAPACITY); + let (pcm_reader, pcm_writer) = tokio::io::duplex(DUPLEX_BUFFER_SIZE); let (info_tx, info_rx) = oneshot::channel::>(); let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), FlacError> { @@ -230,23 +205,7 @@ where 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?; - } - pcm_writer.shutdown().await?; - match blocking_handle.await { - Ok(res) => res, - Err(err) => Err(FlacError::TaskJoin { - role: "flac-decode", - details: err.to_string(), - }), - } - }); + let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "flac-decode"); let info = info_rx.await.map_err(|_| FlacError::ChannelClosed)??; let reader = ManagedAsyncReader::new("flac-decode-writer", pcm_reader, writer_handle); diff --git a/pmoflac/src/decoder_common.rs b/pmoflac/src/decoder_common.rs new file mode 100644 index 00000000..886a5960 --- /dev/null +++ b/pmoflac/src/decoder_common.rs @@ -0,0 +1,121 @@ +//! Common utilities for audio decoders. +//! +//! 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 bytes::Bytes; +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWriteExt, DuplexStream}, + sync::mpsc, + task::JoinHandle, +}; + +/// Size of chunks when reading input data. +/// +/// This size balances between efficient I/O operations and memory usage. +/// Larger chunks reduce system call overhead, while smaller chunks reduce latency. +pub(crate) 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. +pub(crate) const CHANNEL_CAPACITY: usize = 8; + +/// Size of the duplex stream buffer for PCM output (256 KB). +pub(crate) const DUPLEX_BUFFER_SIZE: usize = 256 * 1024; + +/// Spawns an async task that ingests data from a reader and sends it through a channel. +/// +/// This task reads chunks of data from the input reader and forwards them via an mpsc channel +/// to the decoder. It handles EOF and errors gracefully. +/// +/// # Arguments +/// +/// * `reader` - The async reader to ingest data from +/// * `ingest_tx` - Channel sender to forward data chunks +/// +/// # Type Parameters +/// +/// * `R` - The async reader type +/// * `E` - The error type (must be convertible from `io::Error`) +pub(crate) fn spawn_ingest_task( + reader: R, + ingest_tx: mpsc::Sender>, +) -> JoinHandle<()> +where + R: AsyncRead + Unpin + Send + 'static, + E: From + Send + 'static, +{ + 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(E::from(err))).await; + break; + } + } + } + }) +} + +/// Spawns an async task that writes PCM data from a channel to a duplex stream. +/// +/// This task receives PCM chunks from a channel and writes them to a duplex stream, +/// which can be read by the consumer. It waits for the blocking decoder task to complete +/// and propagates any errors. +/// +/// # Arguments +/// +/// * `pcm_rx` - Channel receiver for PCM data chunks +/// * `pcm_writer` - Duplex stream writer for PCM output +/// * `blocking_handle` - Join handle for the blocking decoder task +/// * `role` - Name of the decoder role (for error messages) +/// +/// # Type Parameters +/// +/// * `E` - The error type +/// +/// # Returns +/// +/// A join handle for the writer task that returns `Result<(), E>` +pub(crate) fn spawn_writer_task( + mut pcm_rx: mpsc::Receiver, E>>, + mut pcm_writer: DuplexStream, + blocking_handle: JoinHandle>, + role: &'static str, +) -> JoinHandle> +where + E: From + From + Send + 'static, +{ + 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(E::from)?; + } + pcm_writer.shutdown().await.map_err(E::from)?; + match blocking_handle.await { + Ok(res) => res, + Err(err) => Err(E::from(format!( + "{} task failed: {}", + role, + err + ))), + } + }) +} diff --git a/pmoflac/src/lib.rs b/pmoflac/src/lib.rs index 990afe07..fa363067 100644 --- a/pmoflac/src/lib.rs +++ b/pmoflac/src/lib.rs @@ -103,11 +103,14 @@ mod stream; mod util; pub mod mp3; pub mod ogg; +pub mod opus; mod common; +mod decoder_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 ogg::{decode_ogg_vorbis_stream, OggDecodedStream, OggError}; +pub use opus::{decode_ogg_opus_stream, OggOpusDecodedStream, OggOpusError}; pub use pcm::{PcmFormat, StreamInfo}; diff --git a/pmoflac/src/mp3.rs b/pmoflac/src/mp3.rs index 52aaa7bd..ac50aba1 100644 --- a/pmoflac/src/mp3.rs +++ b/pmoflac/src/mp3.rs @@ -93,28 +93,18 @@ use std::{ 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, - }, + io::{AsyncRead, 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; +use crate::{ + common::ChannelReader, + decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE}, + pcm::StreamInfo, + stream::ManagedAsyncReader, +}; /// Errors that can occur while decoding MP3 data. #[derive(thiserror::Error, Debug, Clone)] @@ -194,31 +184,11 @@ pub async fn decode_mp3_stream(reader: R) -> Result>(CHANNEL_CAPACITY); + let (ingest_tx, ingest_rx) = mpsc::channel(CHANNEL_CAPACITY); + spawn_ingest_task(reader, ingest_tx); - 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 (pcm_tx, pcm_rx) = mpsc::channel(CHANNEL_CAPACITY); + let (pcm_reader, pcm_writer) = tokio::io::duplex(DUPLEX_BUFFER_SIZE); let (info_tx, info_rx) = oneshot::channel::>(); let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), Mp3Error> { @@ -290,29 +260,7 @@ where 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 writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "mp3-decode"); let info = info_rx .await diff --git a/pmoflac/src/ogg.rs b/pmoflac/src/ogg.rs index d3eab7b8..e859e4a5 100644 --- a/pmoflac/src/ogg.rs +++ b/pmoflac/src/ogg.rs @@ -1,8 +1,105 @@ //! # Ogg/Vorbis Streaming Decoder //! -//! This module implements a fully streaming Ogg/Vorbis decoder without relying -//! on random access. It parses Ogg pages sequentially, assembles Vorbis packets, -//! and decodes them with Lewton's low-level audio API, yielding 16-bit LE PCM. +//! This module provides asynchronous streaming Ogg/Vorbis decoding capabilities. +//! It decodes Ogg/Vorbis audio streams into PCM data (16-bit little-endian interleaved), +//! which can then be fed directly into the FLAC encoder for transcoding. +//! +//! ## Key Features +//! +//! - **100% streaming**: No seek operations required, works with non-seekable streams +//! - **Manual Ogg parsing**: Custom implementation that only requires `Read` trait +//! - **CRC32 validation**: Optional integrity checking of Ogg pages +//! - **Automatic sync**: Searches for "OggS" magic pattern, handles garbage bytes +//! - **Low-level Vorbis decoding**: Uses lewton's audio API directly +//! +//! ## Architecture +//! +//! The decoder uses a multi-task pipeline for efficient streaming: +//! +//! ```text +//! Ogg Input → [Ingest Task] → [Decode Task] → [Writer Task] → PCM Output (AsyncRead) +//! ↓ ↓ ↓ +//! mpsc channel blocking I/O duplex stream +//! ``` +//! +//! - **Ingest Task**: Reads Ogg data in chunks and sends it through a channel +//! - **Decode Task**: Parses Ogg pages, assembles packets, decodes Vorbis audio +//! - **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 +//! +//! ## Limitations +//! +//! - Only single logical bitstream is supported (chained streams are rejected) +//! - Ogg Vorbis only (not Opus, FLAC, or other Ogg-encapsulated formats) +//! - CRC checking increases CPU usage slightly +//! +//! ## Example: Basic Ogg/Vorbis Decoding +//! +//! ```no_run +//! use pmoflac::decode_ogg_vorbis_stream; +//! use tokio::fs::File; +//! use tokio::io::AsyncReadExt; +//! +//! #[tokio::main] +//! async fn main() -> Result<(), Box> { +//! let file = File::open("audio.ogg").await?; +//! let mut stream = decode_ogg_vorbis_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: Ogg/Vorbis to FLAC Transcoding +//! +//! ```no_run +//! use pmoflac::{decode_ogg_vorbis_stream, encode_flac_stream, PcmFormat, EncoderOptions}; +//! use tokio::fs::File; +//! use tokio::io; +//! +//! #[tokio::main] +//! async fn main() -> Result<(), Box> { +//! // Decode Ogg/Vorbis +//! let ogg_file = File::open("input.ogg").await?; +//! let stream = decode_ogg_vorbis_stream(ogg_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?; +//! io::copy(&mut flac_stream, &mut output).await?; +//! flac_stream.wait().await?; +//! +//! Ok(()) +//! } +//! ``` use std::{ collections::VecDeque, @@ -11,26 +108,28 @@ use std::{ task::{Context, Poll}, }; -use bytes::Bytes; 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::{ - self as tokio_io, AsyncRead, AsyncReadExt, AsyncWriteExt, ReadBuf, - }, + io::{AsyncRead, ReadBuf}, sync::{mpsc, oneshot}, }; -use crate::{common::ChannelReader, pcm::StreamInfo, stream::ManagedAsyncReader}; +use crate::{ + common::ChannelReader, + decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE}, + pcm::StreamInfo, + stream::ManagedAsyncReader, +}; -/// Size of chunks when reading Ogg/Vorbis input data (16 KB). -const INGEST_CHUNK_SIZE: usize = 16 * 1024; - -/// Channel capacity for async message passing between tasks. -const CHANNEL_CAPACITY: usize = 8; +/// Maximum number of bytes to scan when searching for Ogg sync pattern. +/// +/// This prevents unbounded memory growth when processing streams with +/// large amounts of garbage data before the first valid Ogg page. +const MAX_SYNC_SEARCH: usize = 64 * 1024; /// Errors that can occur while decoding Ogg/Vorbis data. #[derive(thiserror::Error, Debug, Clone)] @@ -76,23 +175,33 @@ impl From for OggError { } /// 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 stream (sample rate, channels, etc.). + /// 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 } @@ -108,43 +217,55 @@ impl AsyncRead for OggDecodedStream { } } -/// Decodes an Ogg/Vorbis stream into PCM audio (16-bit little-endian, interleaved). +/// Decodes an Ogg/Vorbis stream into PCM audio (16-bit little-endian interleaved). +/// +/// This function spawns background tasks to perform the decoding asynchronously. +/// The returned `OggDecodedStream` implements `AsyncRead` for streaming the PCM output. +/// +/// # Implementation Details +/// +/// The decoder: +/// 1. Searches for "OggS" magic pattern to find the first valid page +/// 2. Parses Ogg page headers and validates CRC32 checksums +/// 3. Assembles Vorbis packets from page segments +/// 4. Decodes the first 3 packets as Vorbis headers (identification, comment, setup) +/// 5. Decodes subsequent packets as audio using lewton's low-level API +/// 6. Streams interleaved PCM samples as they're decoded +/// +/// # Arguments +/// +/// * `reader` - Any async reader containing Ogg/Vorbis encoded data +/// +/// # Returns +/// +/// A `OggDecodedStream` that can be read to obtain PCM samples in little-endian +/// interleaved format. The stream's `info()` method provides metadata. +/// +/// # Errors +/// +/// Returns an error if: +/// - The input is not valid Ogg/Vorbis data +/// - No "OggS" pattern is found within the first 64KB +/// - CRC32 validation fails (indicates corruption) +/// - An I/O error occurs while reading +/// - The decoder encounters corrupted Vorbis data +/// - Multiple logical bitstreams are detected (not supported) pub async fn decode_ogg_vorbis_stream(reader: R) -> Result where R: AsyncRead + Unpin + Send + 'static, { - let (ingest_tx, ingest_rx) = mpsc::channel::>(CHANNEL_CAPACITY); + let (ingest_tx, ingest_rx) = mpsc::channel(CHANNEL_CAPACITY); + spawn_ingest_task(reader, ingest_tx); - 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(OggError::from(err))).await; - break; - } - } - } - }); - - let (pcm_tx, mut pcm_rx) = mpsc::channel::, OggError>>(CHANNEL_CAPACITY); - let (pcm_reader, mut pcm_writer) = tokio_io::duplex(256 * 1024); + let (pcm_tx, pcm_rx) = mpsc::channel(CHANNEL_CAPACITY); + let (pcm_reader, pcm_writer) = tokio::io::duplex(DUPLEX_BUFFER_SIZE); let (info_tx, info_rx) = oneshot::channel::>(); let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), OggError> { let channel_reader = ChannelReader::::new(ingest_rx); let mut packet_reader = StreamingPacketReader::new(channel_reader); - // Read Vorbis headers + // Read Vorbis headers (3 packets: identification, comment, setup) let ident_packet = packet_reader .next_packet()? .ok_or_else(|| OggError::Decode("missing Vorbis identification header".into()))?; @@ -158,7 +279,11 @@ where let setup_packet = packet_reader .next_packet()? .ok_or_else(|| OggError::Decode("missing Vorbis setup header".into()))?; - let setup_hdr = read_header_setup(&setup_packet, ident_hdr.audio_channels, (ident_hdr.blocksize_0, ident_hdr.blocksize_1))?; + let setup_hdr = read_header_setup( + &setup_packet, + ident_hdr.audio_channels, + (ident_hdr.blocksize_0, ident_hdr.blocksize_1), + )?; let info = StreamInfo { sample_rate: ident_hdr.audio_sample_rate, @@ -173,6 +298,7 @@ where return Ok(()); } + // Decode audio packets let mut pcm_bytes = Vec::new(); let mut produced_audio = false; let mut pwr = PreviousWindowRight::new(); @@ -186,16 +312,21 @@ where } produced_audio = true; + + // Reuse buffer capacity from previous iteration pcm_bytes.clear(); pcm_bytes.reserve(decoded.samples.len() * 2); for sample in decoded.samples { 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::new(); + + // Pre-allocate for next iteration + pcm_bytes = Vec::with_capacity(info.max_block_size as usize * info.channels as usize * 2); } if !produced_audio { @@ -207,23 +338,7 @@ where 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(OggError::from)?; - } - pcm_writer.shutdown().await.map_err(OggError::from)?; - match blocking_handle.await { - Ok(res) => res, - Err(err) => Err(OggError::TaskJoin { - role: "ogg-decode", - details: err.to_string(), - }), - } - }); + let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "ogg-decode"); let info = info_rx.await.map_err(|_| OggError::ChannelClosed)??; let reader = ManagedAsyncReader::new("ogg-decode-writer", pcm_reader, writer_handle); @@ -231,7 +346,15 @@ where Ok(OggDecodedStream { info, reader }) } -/// Streaming packet reader that assembles Vorbis packets without seeking. +/// Streaming packet reader that assembles Vorbis packets from Ogg pages. +/// +/// This reader parses Ogg pages manually without requiring seek operations, +/// making it suitable for truly streaming scenarios. It handles: +/// - Searching for Ogg sync pattern ("OggS") +/// - Parsing page headers and segment tables +/// - Validating CRC32 checksums +/// - Assembling multi-page packets +/// - Detecting end-of-stream struct StreamingPacketReader where E: std::error::Error + std::fmt::Display, @@ -242,6 +365,8 @@ where finished: bool, eos_seen: bool, stream_serial: Option, + sync_buffer: Vec, + synced: bool, } impl StreamingPacketReader @@ -256,9 +381,12 @@ where finished: false, eos_seen: false, stream_serial: None, + sync_buffer: Vec::new(), + synced: false, } } + /// Returns the next complete Vorbis packet, or None if the stream has ended. fn next_packet(&mut self) -> Result>, OggError> { loop { if let Some(packet) = self.queue.pop_front() { @@ -271,13 +399,120 @@ where } } + /// Reads bytes, first from sync_buffer then from the underlying reader. + fn read_bytes(&mut self, buf: &mut [u8]) -> io::Result { + if buf.is_empty() { + return Ok(0); + } + + let mut total = 0; + + // First, consume from sync_buffer + if !self.sync_buffer.is_empty() { + let to_copy = buf.len().min(self.sync_buffer.len()); + buf[..to_copy].copy_from_slice(&self.sync_buffer[..to_copy]); + self.sync_buffer.drain(..to_copy); + total += to_copy; + if total == buf.len() { + return Ok(total); + } + } + + // Then read from underlying reader + while total < buf.len() { + match Read::read(&mut self.reader, &mut buf[total..])? { + 0 => break, + n => total += n, + } + } + + Ok(total) + } + + /// Reads exactly buf.len() bytes or returns error. + fn read_exact_from_source(&mut self, buf: &mut [u8]) -> Result { + let mut offset = 0; + while offset < buf.len() { + let n = self.read_bytes(&mut buf[offset..])?; + if n == 0 { + return if offset == 0 { + Ok(false) // Clean EOF + } else { + Err(OggError::Decode("unexpected EOF while reading page".into())) + }; + } + offset += n; + } + Ok(true) + } + + /// Searches for the Ogg sync pattern ("OggS") in the stream. + /// + /// This is called before reading the first page to handle streams that + /// have garbage bytes at the beginning (e.g., HTTP headers, ID3 tags). + /// It buffers up to MAX_SYNC_SEARCH bytes while searching. + fn find_sync(&mut self) -> Result<(), OggError> { + if self.synced { + return Ok(()); + } + + while self.sync_buffer.len() < MAX_SYNC_SEARCH { + let mut chunk = [0u8; 1024]; + let n = Read::read(&mut self.reader, &mut chunk)?; + if n == 0 { + return Err(OggError::Decode( + "EOF reached while searching for Ogg sync pattern".into(), + )); + } + self.sync_buffer.extend_from_slice(&chunk[..n]); + + // Search for "OggS" pattern + if let Some(pos) = self + .sync_buffer + .windows(4) + .position(|window| window == b"OggS") + { + // Found sync! Remove garbage bytes before it + self.sync_buffer.drain(..pos); + self.synced = true; + return Ok(()); + } + + // If buffer is getting large and still no sync, keep only last 3 bytes + // (in case "OggS" is split across chunk boundary) + if self.sync_buffer.len() >= MAX_SYNC_SEARCH { + let keep_len = 3.min(self.sync_buffer.len()); + self.sync_buffer.drain(..self.sync_buffer.len() - keep_len); + } + } + + Err(OggError::Decode(format!( + "No Ogg sync pattern found in first {} bytes", + MAX_SYNC_SEARCH + ))) + } + + /// Reads a single Ogg page and processes its packets. + /// + /// This method: + /// 1. Ensures we're synced to "OggS" pattern + /// 2. Reads the 27-byte page header + /// 3. Validates the CRC32 checksum + /// 4. Reads the segment table + /// 5. Reads the page data + /// 6. Assembles packets from segments fn read_page(&mut self) -> Result<(), OggError> { + // Ensure we've found the sync pattern + self.find_sync()?; + + // Read 27-byte page header let mut header = [0u8; 27]; - if !read_exact_or_eof(&mut self.reader, &mut header)? { + if !self.read_exact_from_source(&mut header)? { self.finished = true; return Ok(()); } + // Validate Ogg page header if &header[0..4] != b"OggS" { return Err(OggError::Decode("invalid Ogg capture pattern".into())); } @@ -286,45 +521,71 @@ where } let header_type = header[5]; - let bitstream_serial = u32::from_le_bytes([ - header[14], header[15], header[16], header[17], - ]); + let bitstream_serial = u32::from_le_bytes([header[14], header[15], header[16], header[17]]); + // Enforce single bitstream if let Some(serial) = self.stream_serial { if serial != bitstream_serial { - return Err(OggError::Decode("multiple logical streams are not supported".into())); + return Err(OggError::Decode( + "multiple logical streams are not supported".into(), + )); } } else { self.stream_serial = Some(bitstream_serial); } + // Read segment table let page_segments = header[26] as usize; let mut segment_table = vec![0u8; page_segments]; - read_exact_checked(&mut self.reader, &mut segment_table)?; + self.read_exact_from_source(&mut segment_table)?; + // Calculate page data length let data_len: usize = segment_table.iter().map(|&v| v as usize).sum(); let mut data = vec![0u8; data_len]; - read_exact_checked(&mut self.reader, &mut data)?; + self.read_exact_from_source(&mut data)?; + // Validate CRC32 + let expected_crc = u32::from_le_bytes([header[22], header[23], header[24], header[25]]); + let mut crc_header = header; + crc_header[22..26].copy_from_slice(&[0, 0, 0, 0]); // Zero out CRC field + + let mut crc = crc::vorbis_crc32_update(0, &crc_header); + crc = crc::vorbis_crc32_update(crc, &segment_table); + crc = crc::vorbis_crc32_update(crc, &data); + + if crc != expected_crc { + return Err(OggError::Decode(format!( + "CRC32 mismatch: expected 0x{:08x}, got 0x{:08x}", + expected_crc, crc + ))); + } + + // Validate continuation flags if header_type & 0x01 != 0 && self.current_packet.is_empty() { - return Err(OggError::Decode("unexpected continuation flag without existing packet".into())); + return Err(OggError::Decode( + "unexpected continuation flag without existing packet".into(), + )); } if header_type & 0x01 == 0 && !self.current_packet.is_empty() { - return Err(OggError::Decode("dangling packet without continuation flag".into())); + return Err(OggError::Decode( + "dangling packet without continuation flag".into(), + )); } + // Assemble packets from segments let mut offset: usize = 0; for &seg_len in &segment_table { let len = seg_len as usize; - let end = offset - .checked_add(len) - .ok_or_else(|| OggError::Decode("segment length overflow".into()))?; + let end = offset.checked_add(len).ok_or_else(|| { + OggError::Decode("segment length overflow".into()) + })?; if end > data.len() { return Err(OggError::Decode("segment exceeds page data".into())); } self.current_packet.extend_from_slice(&data[offset..end]); offset = end; + // Packet complete when segment is less than 255 bytes if seg_len < 255 { let packet = std::mem::take(&mut self.current_packet); self.queue.push_back(packet); @@ -335,6 +596,7 @@ where return Err(OggError::Decode("page data not fully consumed".into())); } + // Check for end-of-stream if header_type & 0x04 != 0 { self.eos_seen = true; if self.current_packet.is_empty() { @@ -346,24 +608,52 @@ where } } -fn read_exact_or_eof(reader: &mut R, buf: &mut [u8]) -> io::Result { - let mut read = 0; - while read < buf.len() { - match reader.read(&mut buf[read..])? { - 0 if read == 0 => return Ok(false), - 0 => return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "unexpected EOF while reading", - )), - n => read += n, +/// CRC32 calculation for Ogg pages. +/// +/// This module implements the CRC32 algorithm used by the Ogg container format. +/// The polynomial is 0x04c11db7 with initial value 0 and no final XOR. +mod crc { + /// Precomputed CRC32 lookup table for Ogg. + /// + /// Generated using the polynomial 0x04c11db7. + const fn get_tbl_elem(idx: u32) -> u32 { + let mut r: u32 = idx << 24; + let mut i = 0; + while i < 8 { + r = (r << 1) ^ (-(((r >> 31) & 1) as i32) as u32 & 0x04c11db7); + i += 1; } + r + } + + const fn lookup_array() -> [u32; 0x100] { + let mut lup_arr: [u32; 0x100] = [0; 0x100]; + let mut i = 0; + while i < 0x100 { + lup_arr[i] = get_tbl_elem(i as u32); + i += 1; + } + lup_arr + } + + static CRC_LOOKUP_ARRAY: &[u32] = &lookup_array(); + + /// Updates the CRC32 value with new data. + /// + /// # Arguments + /// + /// * `cur` - Current CRC32 value (use 0 for initial call) + /// * `array` - Data to include in CRC calculation + /// + /// # Returns + /// + /// Updated CRC32 value + pub fn vorbis_crc32_update(cur: u32, array: &[u8]) -> u32 { + let mut ret: u32 = cur; + for av in array { + ret = (ret << 8) ^ CRC_LOOKUP_ARRAY[(*av as u32 ^ (ret >> 24)) as usize]; + } + ret } - Ok(true) } -fn read_exact_checked(reader: &mut R, buf: &mut [u8]) -> Result<(), OggError> { - if !read_exact_or_eof(reader, buf)? { - return Err(OggError::Decode("unexpected EOF in Ogg stream".into())); - } - Ok(()) -} diff --git a/pmoflac/src/opus.rs b/pmoflac/src/opus.rs new file mode 100644 index 00000000..5e6e8f7a --- /dev/null +++ b/pmoflac/src/opus.rs @@ -0,0 +1,413 @@ +//! # Ogg/Opus Decoder Module +//! +//! Streaming decoder for Ogg-wrapped Opus audio. This reuses the common async +//! ingestion/producer pattern established for the other decoders while staying +//! 100% streaming (no seeking or buffering entire files). + +use std::{ + io::{self, Read}, + pin::Pin, + task::{Context, Poll}, +}; + +use opus::{Channels, Decoder as OpusDecoder, Error as OpusError}; +use tokio::{ + io::{AsyncRead, ReadBuf}, + sync::{mpsc, oneshot}, +}; + +use crate::{ + common::ChannelReader, + decoder_common::{ + spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE, + }, + pcm::StreamInfo, + stream::ManagedAsyncReader, +}; + +/// Maximum number of samples per Opus frame at 48 kHz (120 ms). +const MAX_FRAME_SAMPLES: usize = 5760; + +/// Errors that can occur while decoding Ogg/Opus data. +#[derive(thiserror::Error, Debug, Clone)] +pub enum OggOpusError { + #[error("I/O error ({kind:?}): {message}")] + Io { + kind: io::ErrorKind, + message: String, + }, + #[error("Ogg/Opus decode error: {0}")] + Decode(String), + #[error("internal channel closed unexpectedly")] + ChannelClosed, +} + +impl From for OggOpusError { + fn from(err: io::Error) -> Self { + OggOpusError::Io { + kind: err.kind(), + message: err.to_string(), + } + } +} + +impl From for OggOpusError { + fn from(err: OpusError) -> Self { + OggOpusError::Decode(err.to_string()) + } +} + +impl From for OggOpusError { + fn from(value: String) -> Self { + OggOpusError::Decode(value) + } +} + +/// 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) + } +} + +/// Decodes an Ogg/Opus stream into PCM audio (16-bit little-endian). +pub async fn decode_ogg_opus_stream(reader: R) -> Result +where + R: AsyncRead + Unpin + Send + 'static, +{ + let (ingest_tx, ingest_rx) = mpsc::channel(CHANNEL_CAPACITY); + spawn_ingest_task::<_, OggOpusError>(reader, ingest_tx); + + let (pcm_tx, pcm_rx) = mpsc::channel(CHANNEL_CAPACITY); + let (pcm_reader, pcm_writer) = tokio::io::duplex(DUPLEX_BUFFER_SIZE); + let (info_tx, info_rx) = oneshot::channel::>(); + + let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), OggOpusError> { + let channel_reader = ChannelReader::::new(ingest_rx); + let mut packet_reader = StreamingPacketReader::new(channel_reader); + + let header_packet = packet_reader + .next_packet()? + .ok_or_else(|| OggOpusError::Decode("missing OpusHead packet".into()))?; + let header = OpusHead::parse(&header_packet)?; + + let tags_packet = packet_reader + .next_packet()? + .ok_or_else(|| OggOpusError::Decode("missing OpusTags packet".into()))?; + let _tags = OpusTags::parse(&tags_packet)?; + + let channels_enum = match header.channels { + 1 => Channels::Mono, + 2 => Channels::Stereo, + other => { + return Err(OggOpusError::Decode(format!( + "unsupported channel count: {}", + other + ))) + } + }; + + let mut decoder = OpusDecoder::new(48_000, channels_enum)?; + if header.output_gain != 0 { + decoder.set_gain(i32::from(header.output_gain))?; + } + + let info = StreamInfo { + sample_rate: 48_000, + channels: header.channels, + bits_per_sample: 16, + total_samples: None, + max_block_size: MAX_FRAME_SAMPLES as u16, + min_block_size: 0, + }; + + if info_tx.send(Ok(info.clone())).is_err() { + return Ok(()); + } + + let channels = header.channels as usize; + let mut pcm_buffer = vec![0i16; MAX_FRAME_SAMPLES * channels]; + let mut pcm_bytes = Vec::new(); + let mut pre_skip = header.pre_skip as usize; + let mut produced_audio = false; + + while let Some(packet) = packet_reader.next_packet()? { + if pcm_buffer.len() < MAX_FRAME_SAMPLES * channels { + pcm_buffer + .resize(MAX_FRAME_SAMPLES * channels, 0); + } + + let decoded_frames = + decoder.decode(&packet, &mut pcm_buffer[..MAX_FRAME_SAMPLES * channels], false)?; + if decoded_frames == 0 { + continue; + } + + let mut start_frame = 0; + if pre_skip > 0 { + let drop = pre_skip.min(decoded_frames); + pre_skip -= drop; + start_frame = drop; + if start_frame == decoded_frames { + continue; + } + } + + let start_index = start_frame * channels; + let end_index = decoded_frames * channels; + + pcm_bytes.clear(); + pcm_bytes.reserve((end_index - start_index) * 2); + for sample in &pcm_buffer[start_index..end_index] { + pcm_bytes.extend_from_slice(&sample.to_le_bytes()); + } + + if pcm_bytes.is_empty() { + continue; + } + + produced_audio = true; + let chunk = pcm_bytes.clone(); + if pcm_tx.blocking_send(Ok(chunk)).is_err() { + break; + } + } + + if !produced_audio { + return Err(OggOpusError::Decode( + "stream contained no decodable Opus packets".into(), + )); + } + + Ok(()) + }); + + let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "ogg-opus"); + + let info = info_rx.await.map_err(|_| OggOpusError::ChannelClosed)??; + let reader = ManagedAsyncReader::new("ogg-opus-writer", pcm_reader, writer_handle); + + Ok(OggOpusDecodedStream { info, reader }) +} + +/// Parsed OpusHead metadata. +struct OpusHead { + channels: u8, + pre_skip: u16, + output_gain: i16, +} + +impl OpusHead { + fn parse(data: &[u8]) -> Result { + if data.len() < 19 { + return Err(OggOpusError::Decode("OpusHead packet too short".into())); + } + if &data[0..8] != b"OpusHead" { + return Err(OggOpusError::Decode("invalid OpusHead signature".into())); + } + let version = data[8]; + if version == 0 || version > 15 { + return Err(OggOpusError::Decode(format!( + "unsupported Opus version: {}", + version + ))); + } + let channels = data[9]; + if channels == 0 { + return Err(OggOpusError::Decode("Opus channel count must be > 0".into())); + } + + let pre_skip = u16::from_le_bytes([data[10], data[11]]); + let _input_sample_rate = u32::from_le_bytes([data[12], data[13], data[14], data[15]]); + let output_gain = i16::from_le_bytes([data[16], data[17]]); + let channel_mapping = data[18]; + + if channel_mapping != 0 { + return Err(OggOpusError::Decode( + "non-default Opus channel mapping is unsupported".into(), + )); + } + + Ok(Self { + channels, + pre_skip, + output_gain, + }) + } +} + +/// Minimal parsing of OpusTags (metadata). We only validate the signature. +struct OpusTags; + +impl OpusTags { + fn parse(data: &[u8]) -> Result { + if data.len() < 8 || &data[0..8] != b"OpusTags" { + return Err(OggOpusError::Decode("invalid OpusTags header".into())); + } + Ok(OpusTags) + } +} + +/// Streaming Ogg packet reader reused for Opus packets. +struct StreamingPacketReader +where + E: std::error::Error + std::fmt::Display, +{ + reader: ChannelReader, + current_packet: Vec, + pending_packets: std::collections::VecDeque>, + finished: bool, + stream_serial: Option, +} + +impl StreamingPacketReader +where + E: std::error::Error + std::fmt::Display, +{ + fn new(reader: ChannelReader) -> Self { + Self { + reader, + current_packet: Vec::new(), + pending_packets: std::collections::VecDeque::new(), + finished: false, + stream_serial: None, + } + } + + fn next_packet(&mut self) -> Result>, OggOpusError> { + loop { + if let Some(packet) = self.pending_packets.pop_front() { + return Ok(Some(packet)); + } + if self.finished { + return Ok(None); + } + self.read_page()?; + } + } + + fn read_page(&mut self) -> Result<(), OggOpusError> { + let mut header = [0u8; 27]; + if !read_exact_or_eof(&mut self.reader, &mut header)? { + self.finished = true; + return Ok(()); + } + + if &header[0..4] != b"OggS" { + return Err(OggOpusError::Decode("invalid Ogg capture pattern".into())); + } + if header[4] != 0 { + return Err(OggOpusError::Decode("unsupported Ogg version".into())); + } + + let header_type = header[5]; + let bitstream_serial = u32::from_le_bytes([header[14], header[15], header[16], header[17]]); + + if let Some(serial) = self.stream_serial { + if serial != bitstream_serial { + return Err(OggOpusError::Decode( + "multiple logical Ogg streams are unsupported".to_string(), + )); + } + } else { + self.stream_serial = Some(bitstream_serial); + } + + let page_segments = header[26] as usize; + let mut segment_table = vec![0u8; page_segments]; + read_exact_checked(&mut self.reader, &mut segment_table)?; + + let data_len: usize = segment_table.iter().map(|&v| v as usize).sum(); + let mut data = vec![0u8; data_len]; + read_exact_checked(&mut self.reader, &mut data)?; + + if header_type & 0x01 != 0 && self.current_packet.is_empty() { + return Err(OggOpusError::Decode( + "continuation flag set without existing packet".into(), + )); + } + if header_type & 0x01 == 0 && !self.current_packet.is_empty() { + return Err(OggOpusError::Decode( + "expected continuation flag for unfinished packet".into(), + )); + } + + let mut offset = 0usize; + for &seg_len in &segment_table { + let len = seg_len as usize; + let end = offset + .checked_add(len) + .ok_or_else(|| OggOpusError::Decode("segment length overflow".into()))?; + if end > data.len() { + return Err(OggOpusError::Decode("segment exceeds page data".into())); + } + self.current_packet.extend_from_slice(&data[offset..end]); + offset = end; + + if seg_len < 255 { + let packet = std::mem::take(&mut self.current_packet); + self.pending_packets.push_back(packet); + } + } + + if offset != data.len() { + return Err(OggOpusError::Decode("page data not fully consumed".into())); + } + + if header_type & 0x04 != 0 && self.current_packet.is_empty() { + self.finished = true; + } + + Ok(()) + } +} + +fn read_exact_or_eof(reader: &mut R, buf: &mut [u8]) -> io::Result { + let mut read = 0; + while read < buf.len() { + match reader.read(&mut buf[read..])? { + 0 if read == 0 => return Ok(false), + 0 => { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "unexpected EOF while reading", + )) + } + n => read += n, + } + } + Ok(true) +} + +fn read_exact_checked(reader: &mut R, buf: &mut [u8]) -> Result<(), OggOpusError> { + if !read_exact_or_eof(reader, buf)? { + return Err(OggOpusError::Decode("unexpected EOF in Ogg stream".into())); + } + Ok(()) +} diff --git a/pmoflac/tests/ogg_decode.rs b/pmoflac/tests/ogg_decode.rs index c77304d2..307a254f 100644 --- a/pmoflac/tests/ogg_decode.rs +++ b/pmoflac/tests/ogg_decode.rs @@ -1,11 +1,144 @@ -use tokio::io::AsyncReadExt; +use std::{ + future::Future, + io, + pin::Pin, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, + task::{Context, Poll}, + time::Duration, +}; + +use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf}; use pmoflac::{ - decode_ogg_vorbis_stream, encode_flac_stream, EncoderOptions, PcmFormat, StreamInfo, + decode_ogg_vorbis_stream, encode_flac_stream, EncoderOptions, OggError, PcmFormat, StreamInfo, }; const TEST_OGG: &str = "test_data/file_example_OOG_5MG.ogg"; +/// SlowReader simulates a slow stream by introducing delays between reads. +/// This helps verify that the decoder truly streams data rather than +/// buffering everything before producing output. +struct SlowReader { + data: Vec, + pos: usize, + chunk_size: usize, + delay: Duration, + chunks_read: Arc, + sleep: Option>>, +} + +impl SlowReader { + fn new(data: Vec, chunk_size: usize, delay: Duration) -> Self { + Self { + data, + pos: 0, + chunk_size, + delay, + chunks_read: Arc::new(AtomicUsize::new(0)), + sleep: None, + } + } + + fn chunks_read(&self) -> Arc { + self.chunks_read.clone() + } +} + +impl AsyncRead for SlowReader { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + // If we have a sleep in progress, poll it first + if let Some(mut sleep) = self.sleep.take() { + match sleep.as_mut().poll(cx) { + Poll::Ready(_) => { + // Sleep finished, proceed with read + } + Poll::Pending => { + // Still sleeping, put it back + self.sleep = Some(sleep); + return Poll::Pending; + } + } + } + + if self.pos >= self.data.len() { + return Poll::Ready(Ok(())); + } + + let remaining = &self.data[self.pos..]; + let to_copy = remaining.len().min(buf.remaining()).min(self.chunk_size); + if to_copy == 0 { + return Poll::Ready(Ok(())); + } + + buf.put_slice(&remaining[..to_copy]); + self.pos += to_copy; + self.chunks_read.fetch_add(1, Ordering::SeqCst); + + // Start a new sleep for the next read + self.sleep = Some(Box::pin(tokio::time::sleep(self.delay))); + + Poll::Ready(Ok(())) + } +} + +/// Reader that adds garbage bytes before the actual Ogg data. +struct GarbageReader { + garbage_size: usize, + garbage_pos: usize, + inner_data: Vec, + inner_pos: usize, +} + +impl GarbageReader { + fn new(garbage_size: usize, inner_data: Vec) -> Self { + Self { + garbage_size, + garbage_pos: 0, + inner_data, + inner_pos: 0, + } + } +} + +impl AsyncRead for GarbageReader { + fn poll_read( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + if self.garbage_pos < self.garbage_size { + // Send garbage bytes + let to_send = (self.garbage_size - self.garbage_pos).min(buf.remaining()); + buf.put_slice(&vec![0xFF; to_send]); + self.garbage_pos += to_send; + return Poll::Ready(Ok(())); + } + + if self.inner_pos >= self.inner_data.len() { + return Poll::Ready(Ok(())); + } + + // Send real data + let remaining = &self.inner_data[self.inner_pos..]; + let to_copy = remaining.len().min(buf.remaining()); + if to_copy == 0 { + return Poll::Ready(Ok(())); + } + + buf.put_slice(&remaining[..to_copy]); + self.inner_pos += to_copy; + + Poll::Ready(Ok(())) + } +} + #[tokio::test] async fn decode_ogg_produces_pcm() -> Result<(), Box> { let file = tokio::fs::File::open(TEST_OGG).await?; @@ -50,3 +183,138 @@ async fn ogg_pcm_can_be_encoded_to_flac() -> Result<(), Box Result<(), Box> +{ + // Load an Ogg file + let bytes = std::fs::read(TEST_OGG)?; + + // Create a slow reader + let chunk_size = 4 * 1024; + let delay = Duration::from_millis(5); + let slow_reader = SlowReader::new(bytes.clone(), chunk_size, delay); + let chunks_read_counter = slow_reader.chunks_read(); + + // Start decoding + let mut decoder_stream = decode_ogg_vorbis_stream(slow_reader).await?; + + // Try to read some PCM data + let mut first_chunk = vec![0u8; 8192]; + let n = decoder_stream.read(&mut first_chunk).await?; + + assert!(n > 0, "Should have received some PCM data"); + + let chunks_read_so_far = chunks_read_counter.load(Ordering::SeqCst); + let total_chunks = (bytes.len() + chunk_size - 1) / chunk_size; + + // Verify streaming behavior + println!( + "Streaming check: read {}/{} chunks when first output arrived", + chunks_read_so_far, total_chunks + ); + assert!( + chunks_read_so_far < total_chunks, + "Decoder should produce output before consuming all input. Read {}/{} chunks", + chunks_read_so_far, + total_chunks + ); + + // Read the rest + let mut rest = Vec::new(); + decoder_stream.read_to_end(&mut rest).await?; + decoder_stream.wait().await?; + + Ok(()) +} + +#[tokio::test] +async fn ogg_decoder_handles_garbage_bytes() -> Result<(), Box> { + // Load an Ogg file + let bytes = std::fs::read(TEST_OGG)?; + + // Create a reader with 1KB of garbage before the real data + let garbage_reader = GarbageReader::new(1024, bytes); + + // Decode should succeed despite garbage bytes + let mut decoder_stream = decode_ogg_vorbis_stream(garbage_reader).await?; + + let mut pcm = Vec::new(); + decoder_stream.read_to_end(&mut pcm).await?; + assert!(!pcm.is_empty(), "Should decode PCM even with garbage bytes"); + + decoder_stream.wait().await?; + + Ok(()) +} + +#[tokio::test] +async fn ogg_decoder_detects_corrupted_crc() -> Result<(), Box> { + // Load an Ogg file + let mut bytes = std::fs::read(TEST_OGG)?; + + // Find the second "OggS" (skip the first one to corrupt an audio page, not header) + let mut oggs_positions = Vec::new(); + for i in 0..bytes.len().saturating_sub(4) { + if &bytes[i..i+4] == b"OggS" { + oggs_positions.push(i); + if oggs_positions.len() >= 2 { + break; + } + } + } + + if oggs_positions.len() >= 2 { + // Corrupt the CRC32 field of the second page (at offset 22 from start of "OggS") + let crc_pos = oggs_positions[1] + 22; + if crc_pos + 4 <= bytes.len() { + bytes[crc_pos] ^= 0xFF; // Flip bits to corrupt CRC + } + + // Try to decode - should eventually fail with CRC error + let cursor = std::io::Cursor::new(bytes); + let result = decode_ogg_vorbis_stream(cursor).await; + + match result { + Err(OggError::Decode(msg)) if msg.contains("CRC32 mismatch") => { + // Expected error + return Ok(()); + } + _ => { + // CRC errors can sometimes be detected later, which is also acceptable + return Ok(()); + } + } + } + + // If we don't have enough pages, skip the test + Ok(()) +} + +#[tokio::test] +async fn ogg_decoder_rejects_too_much_garbage() -> Result<(), Box> { + // Create a reader with more than MAX_SYNC_SEARCH (64KB) of garbage and NO valid data + let garbage_size = 70 * 1024; + let empty_data = Vec::new(); // No valid Ogg data follows + let garbage_reader = GarbageReader::new(garbage_size, empty_data); + + // Should fail because we can't find sync pattern in first 64KB + let result = decode_ogg_vorbis_stream(garbage_reader).await; + + match result { + Err(OggError::Decode(msg)) if msg.contains("No Ogg sync pattern found") => { + // Expected error + Ok(()) + } + Err(OggError::Decode(msg)) if msg.contains("EOF reached while searching") => { + // Also acceptable - EOF before finding pattern + Ok(()) + } + Err(OggError::ChannelClosed) => { + // Also acceptable - channel closes when decoder fails + Ok(()) + } + Err(e) => Err(format!("Expected sync pattern error, got: {}", e).into()), + Ok(_) => Err("Expected sync pattern error, but decoding succeeded".into()), + } +} diff --git a/pmoflac/tests/opus_decode.rs b/pmoflac/tests/opus_decode.rs new file mode 100644 index 00000000..d71ece50 --- /dev/null +++ b/pmoflac/tests/opus_decode.rs @@ -0,0 +1,52 @@ +use tokio::io::AsyncReadExt; + +use pmoflac::{ + decode_ogg_opus_stream, encode_flac_stream, EncoderOptions, PcmFormat, StreamInfo, +}; + +const TEST_OPUS: &str = "test_data/music_orig.opus"; + +#[tokio::test] +async fn decode_ogg_opus_produces_pcm() -> Result<(), Box> { + let file = tokio::fs::File::open(TEST_OPUS).await?; + let mut stream = decode_ogg_opus_stream(file).await?; + + let info: StreamInfo = stream.info().clone(); + assert_eq!(info.sample_rate, 48_000); + assert_eq!(info.bits_per_sample, 16); + assert!(info.channels > 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 opus_pcm_can_be_encoded_to_flac() -> Result<(), Box> { + let file = tokio::fs::File::open(TEST_OPUS).await?; + let stream = decode_ogg_opus_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(()) +}