eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Ajout d'un decodeur ogg et opus vers pcm

Browse Source
This commit is contained in:
Eric Coissac
2025-10-27 19:25:51 +01:00
parent 6b52b6ffee
commit b7661b5767
10 changed files with 1280 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
pmoflac/src/decoder.rs
View File

@@ -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 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::<Result<Vec<u8>, 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::<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 {
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);

121
pmoflac/src/decoder_common.rs Normal file
View File

@@ -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
))),
}
})
}

3
pmoflac/src/lib.rs
View File

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

76
pmoflac/src/mp3.rs
View File

@@ -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<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 mut reader = tokio_io::BufReader::new(reader);
let mut buf = vec![0u8; INGEST_CHUNK_SIZE];
loop {
match reader.read(&mut buf).await {
Ok(0) => break,
Ok(n) => {
let chunk = Bytes::copy_from_slice(&buf[..n]);
if ingest_tx.send(Ok(chunk)).await.is_err() {
break;
}
}
Err(err) => {
let _ = ingest_tx.send(Err(Mp3Error::from(err))).await;
break;
}
}
}
});
let (pcm_tx, mut pcm_rx) = mpsc::channel::<Result<Vec<u8>, Mp3Error>>(CHANNEL_CAPACITY);
let (pcm_reader, mut pcm_writer) = tokio_io::duplex(256 * 1024);
let (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, Mp3Error>>();
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

468
pmoflac/src/ogg.rs
View File

@@ -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<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::{
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<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 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::<Result<Vec<u8>, 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::<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 {
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<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([
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<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,
/// 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<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(())
}

413
pmoflac/src/opus.rs Normal file
View File

@@ -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(())
}