Ajout d'un decodeur ogg et opus vers pcm

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",

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"

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<R>(reader: R) -> Result<FlacDecodedStream, FlacE
 where
     R: AsyncRead + Unpin + Send + 'static,
 {
-    let (ingest_tx, ingest_rx) = mpsc::channel::<Result<Bytes, FlacError>>(CHANNEL_CAPACITY);
+    let (ingest_tx, ingest_rx) = mpsc::channel(CHANNEL_CAPACITY);
+    spawn_ingest_task(reader, ingest_tx);
 
-    tokio::spawn(async move {
+    let (pcm_tx, pcm_rx) = mpsc::channel(CHANNEL_CAPACITY);
-        let mut reader = tokio::io::BufReader::new(reader);
+    let (pcm_reader, pcm_writer) = tokio::io::duplex(DUPLEX_BUFFER_SIZE);
-        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::<Result<Vec<u8>, FlacError>>(CHANNEL_CAPACITY);
-    let (pcm_reader, mut pcm_writer) = tokio::io::duplex(256 * 1024);
     let (info_tx, info_rx) = oneshot::channel::<Result<StreamInfo, FlacError>>();
 
     let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), FlacError> {
@@ -230,23 +205,7 @@ where
         Ok(())
     });
 
-    let writer_handle = tokio::spawn(async move {
+    let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "flac-decode");
-        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 info = info_rx.await.map_err(|_| FlacError::ChannelClosed)??;
     let reader = ManagedAsyncReader::new("flac-decode-writer", pcm_reader, writer_handle);

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<R, E>(
+    reader: R,
+    ingest_tx: mpsc::Sender<Result<Bytes, E>>,
+) -> JoinHandle<()>
+where
+    R: AsyncRead + Unpin + Send + 'static,
+    E: From<io::Error> + 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<E>(
+    mut pcm_rx: mpsc::Receiver<Result<Vec<u8>, E>>,
+    mut pcm_writer: DuplexStream,
+    blocking_handle: JoinHandle<Result<(), E>>,
+    role: &'static str,
+) -> JoinHandle<Result<(), E>>
+where
+    E: From<io::Error> + From<String> + 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
+            ))),
+        }
+    })
+}

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};

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::{
+    io::{AsyncRead, ReadBuf},
-        self as tokio_io, AsyncRead, AsyncReadExt, AsyncWriteExt, ReadBuf,
-    },
     sync::{mpsc, oneshot},
 };
 
-use crate::{common::ChannelReader, pcm::StreamInfo, stream::ManagedAsyncReader};
+use crate::{
-
+    common::ChannelReader,
-/// Size of chunks when reading MP3 input data (16 KB).
+    decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE},
-///
+    pcm::StreamInfo,
-/// This size balances between efficient I/O operations and memory usage.
+    stream::ManagedAsyncReader,
-/// 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)]
@@ -194,31 +184,11 @@ pub async fn decode_mp3_stream<R>(reader: R) -> Result<Mp3DecodedStream, Mp3Erro
 where
     R: AsyncRead + Unpin + Send + 'static,
 {
-    let (ingest_tx, ingest_rx) = mpsc::channel::<Result<Bytes, Mp3Error>>(CHANNEL_CAPACITY);
+    let (ingest_tx, ingest_rx) = mpsc::channel(CHANNEL_CAPACITY);
+    spawn_ingest_task(reader, ingest_tx);
 
-    tokio::spawn(async move {
+    let (pcm_tx, pcm_rx) = mpsc::channel(CHANNEL_CAPACITY);
-        let mut reader = tokio_io::BufReader::new(reader);
+    let (pcm_reader, pcm_writer) = tokio::io::duplex(DUPLEX_BUFFER_SIZE);
-        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> {
@@ -290,29 +260,7 @@ where
         Ok(())
     });
 
-    let writer_handle = tokio::spawn(async move {
+    let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "mp3-decode");
-        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

pmoflac/src/ogg.rs

@@ -1,8 +1,105 @@
 //! # Ogg/Vorbis Streaming Decoder
 //!
-//! This module implements a fully streaming Ogg/Vorbis decoder without relying
+//! This module provides asynchronous streaming Ogg/Vorbis decoding capabilities.
-//! on random access. It parses Ogg pages sequentially, assembles Vorbis packets,
+//! It decodes Ogg/Vorbis audio streams into PCM data (16-bit little-endian interleaved),
-//! and decodes them with Lewton's low-level audio API, yielding 16-bit LE PCM.
+//! 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<dyn std::error::Error>> {
+//!     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<dyn std::error::Error>> {
+//!     // 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::{
+    io::{AsyncRead, ReadBuf},
-        self as tokio_io, AsyncRead, AsyncReadExt, AsyncWriteExt, 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).
+/// Maximum number of bytes to scan when searching for Ogg sync pattern.
-const INGEST_CHUNK_SIZE: usize = 16 * 1024;
+///
-
+/// This prevents unbounded memory growth when processing streams with
-/// Channel capacity for async message passing between tasks.
+/// large amounts of garbage data before the first valid Ogg page.
-const CHANNEL_CAPACITY: usize = 8;
+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<String> 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<OggError>,
 }
 
 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<OggError>) {
         (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<R>(reader: R) -> Result<OggDecodedStream, OggError>
 where
     R: AsyncRead + Unpin + Send + 'static,
 {
-    let (ingest_tx, ingest_rx) = mpsc::channel::<Result<Bytes, OggError>>(CHANNEL_CAPACITY);
+    let (ingest_tx, ingest_rx) = mpsc::channel(CHANNEL_CAPACITY);
+    spawn_ingest_task(reader, ingest_tx);
 
-    tokio::spawn(async move {
+    let (pcm_tx, pcm_rx) = mpsc::channel(CHANNEL_CAPACITY);
-        let mut reader = tokio_io::BufReader::new(reader);
+    let (pcm_reader, pcm_writer) = tokio::io::duplex(DUPLEX_BUFFER_SIZE);
-        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::<Result<Vec<u8>, OggError>>(CHANNEL_CAPACITY);
-    let (pcm_reader, mut pcm_writer) = tokio_io::duplex(256 * 1024);
     let (info_tx, info_rx) = oneshot::channel::<Result<StreamInfo, OggError>>();
 
     let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), OggError> {
         let channel_reader = ChannelReader::<OggError>::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 {
+    let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "ogg-decode");
-        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 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<E>
 where
     E: std::error::Error + std::fmt::Display,
@@ -242,6 +365,8 @@ where
     finished: bool,
     eos_seen: bool,
     stream_serial: Option<u32>,
+    sync_buffer: Vec<u8>,
+    synced: bool,
 }
 
 impl<E> StreamingPacketReader<E>
@@ -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<Option<Vec<u8>>, 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<usize> {
+        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<bool, OggError> {
+        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([
+        let bitstream_serial = u32::from_le_bytes([header[14], header[15], header[16], header[17]]);
-            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
+            let end = offset.checked_add(len).ok_or_else(|| {
-                .checked_add(len)
+                OggError::Decode("segment length overflow".into())
-                .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<R: Read>(reader: &mut R, buf: &mut [u8]) -> io::Result<bool> {
+/// CRC32 calculation for Ogg pages.
-    let mut read = 0;
+///
-    while read < buf.len() {
+/// This module implements the CRC32 algorithm used by the Ogg container format.
-        match reader.read(&mut buf[read..])? {
+/// The polynomial is 0x04c11db7 with initial value 0 and no final XOR.
-            0 if read == 0 => return Ok(false),
+mod crc {
-            0 => return Err(io::Error::new(
+    /// Precomputed CRC32 lookup table for Ogg.
-                io::ErrorKind::UnexpectedEof,
+    ///
-                "unexpected EOF while reading",
+    /// Generated using the polynomial 0x04c11db7.
-            )),
+    const fn get_tbl_elem(idx: u32) -> u32 {
-            n => read += n,
+        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<R: Read>(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(())
-}

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<io::Error> for OggOpusError {
+    fn from(err: io::Error) -> Self {
+        OggOpusError::Io {
+            kind: err.kind(),
+            message: err.to_string(),
+        }
+    }
+}
+
+impl From<OpusError> for OggOpusError {
+    fn from(err: OpusError) -> Self {
+        OggOpusError::Decode(err.to_string())
+    }
+}
+
+impl From<String> 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<OggOpusError>,
+}
+
+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<OggOpusError>) {
+        (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<io::Result<()>> {
+        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<R>(reader: R) -> Result<OggOpusDecodedStream, OggOpusError>
+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::<Result<StreamInfo, OggOpusError>>();
+
+    let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), OggOpusError> {
+        let channel_reader = ChannelReader::<OggOpusError>::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<Self, OggOpusError> {
+        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<Self, OggOpusError> {
+        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<E>
+where
+    E: std::error::Error + std::fmt::Display,
+{
+    reader: ChannelReader<E>,
+    current_packet: Vec<u8>,
+    pending_packets: std::collections::VecDeque<Vec<u8>>,
+    finished: bool,
+    stream_serial: Option<u32>,
+}
+
+impl<E> StreamingPacketReader<E>
+where
+    E: std::error::Error + std::fmt::Display,
+{
+    fn new(reader: ChannelReader<E>) -> Self {
+        Self {
+            reader,
+            current_packet: Vec::new(),
+            pending_packets: std::collections::VecDeque::new(),
+            finished: false,
+            stream_serial: None,
+        }
+    }
+
+    fn next_packet(&mut self) -> Result<Option<Vec<u8>>, 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<R: Read>(reader: &mut R, buf: &mut [u8]) -> io::Result<bool> {
+    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<R: Read>(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(())
+}

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<u8>,
+    pos: usize,
+    chunk_size: usize,
+    delay: Duration,
+    chunks_read: Arc<AtomicUsize>,
+    sleep: Option<Pin<Box<tokio::time::Sleep>>>,
+}
+
+impl SlowReader {
+    fn new(data: Vec<u8>, 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<AtomicUsize> {
+        self.chunks_read.clone()
+    }
+}
+
+impl AsyncRead for SlowReader {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        // 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<u8>,
+    inner_pos: usize,
+}
+
+impl GarbageReader {
+    fn new(garbage_size: usize, inner_data: Vec<u8>) -> 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<io::Result<()>> {
+        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<dyn std::error::Error>> {
     let file = tokio::fs::File::open(TEST_OGG).await?;
@@ -50,3 +183,138 @@ async fn ogg_pcm_can_be_encoded_to_flac() -> Result<(), Box<dyn std::error::Erro
 
     Ok(())
 }
+
+#[tokio::test]
+async fn ogg_decoder_streams_without_buffering_all_input() -> Result<(), Box<dyn std::error::Error>>
+{
+    // 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<dyn std::error::Error>> {
+    // 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<dyn std::error::Error>> {
+    // 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<dyn std::error::Error>> {
+    // 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()),
+    }
+}

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<dyn std::error::Error>> {
+    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<dyn std::error::Error>> {
+    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(())
+}