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

unification des decodeurs

Browse Source
This commit is contained in:
Eric Coissac
2025-10-27 22:59:06 +01:00
parent e15eaf9b53
commit 4b642d69e5
7 changed files with 96 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
pmoflac/src/aiff.rs
View File

@@ -5,13 +5,7 @@
//! little-endian interleaved PCM frames compatible with the rest of the
//! pipeline.
use std::{
collections::VecDeque,
fmt,
io::{self, Read},
pin::Pin,
task::{Context, Poll},
};
use std::{collections::VecDeque, fmt, io::Read};
use tokio::{
io::AsyncRead,
@@ -21,7 +15,8 @@ use tokio::{
use crate::{
common::ChannelReader,
decoder_common::{
spawn_ingest_task, spawn_writer_task, DecoderError, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE,
spawn_ingest_task, spawn_writer_task, DecodedStream, DecoderError, CHANNEL_CAPACITY,
DUPLEX_BUFFER_SIZE,
},
pcm::StreamInfo,
stream::ManagedAsyncReader,
@@ -142,35 +137,8 @@ impl CommChunk {
}
}
/// Asynchronous decoded stream wrapper for AIFF data.
pub struct AiffDecodedStream {
info: StreamInfo,
reader: ManagedAsyncReader<AiffError>,
}
impl AiffDecodedStream {
pub fn info(&self) -> &StreamInfo {
&self.info
}
pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader<AiffError>) {
(self.info, self.reader)
}
pub async fn wait(self) -> Result<(), AiffError> {
self.reader.wait().await
}
}
impl AsyncRead for AiffDecodedStream {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.reader).poll_read(cx, buf)
}
}
/// Async stream alias for decoded AIFF audio.
pub type AiffDecodedStream = DecodedStream<AiffError>;
/// Decode an AIFF stream into PCM audio (little-endian interleaved).
pub async fn decode_aiff_stream<R>(reader: R) -> Result<AiffDecodedStream, AiffError>
@@ -362,7 +330,7 @@ where
let info = info_rx.await.map_err(|_| AiffError::ChannelClosed)??;
let reader = ManagedAsyncReader::new("aiff-decode-writer", pcm_reader, writer_handle);
Ok(AiffDecodedStream { info, reader })
Ok(DecodedStream::new(info, reader))
}
fn parse_extended_f80(bytes: &[u8]) -> Result<u32, AiffError> {

73
pmoflac/src/decoder.rs
View File

@@ -1,9 +1,3 @@
use std::{
io,
pin::Pin,
task::{Context, Poll},
};
use tokio::{
io::AsyncRead,
sync::{mpsc, oneshot},
@@ -11,72 +5,17 @@ use tokio::{
use crate::{
common::ChannelReader,
decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE},
decoder_common::{
spawn_ingest_task, spawn_writer_task, DecodedStream, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE,
},
error::FlacError,
pcm::StreamInfo,
stream::ManagedAsyncReader,
util::interleaved_i32_to_le_bytes,
};
/// An async stream that decodes FLAC 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.
///
/// # Example
///
/// ```no_run
/// use pmoflac::decode_flac_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.flac").await?;
/// let mut stream = decode_flac_stream(file).await?;
///
/// println!("Sample rate: {}", stream.info().sample_rate);
///
/// let mut pcm = Vec::new();
/// stream.read_to_end(&mut pcm).await?;
/// stream.wait().await?;
/// # Ok(())
/// # }
/// ```
pub struct FlacDecodedStream {
info: StreamInfo,
reader: ManagedAsyncReader<FlacError>,
}
impl FlacDecodedStream {
/// Returns metadata about the FLAC stream (sample rate, channels, etc.).
pub fn info(&self) -> &StreamInfo {
&self.info
}
/// Consumes the stream and returns its components.
pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader<FlacError>) {
(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<(), FlacError> {
self.reader.wait().await
}
}
impl AsyncRead for FlacDecodedStream {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.reader).poll_read(cx, buf)
}
}
/// Async stream alias for decoded FLAC audio.
pub type FlacDecodedStream = DecodedStream<FlacError>;
/// Decodes a FLAC stream into PCM audio data.
///
@@ -210,5 +149,5 @@ where
let info = info_rx.await.map_err(|_| FlacError::ChannelClosed)??;
let reader = ManagedAsyncReader::new("flac-decode-writer", pcm_reader, writer_handle);
Ok(FlacDecodedStream { info, reader })
Ok(DecodedStream::new(info, reader))
}

60
pmoflac/src/decoder_common.rs
View File

@@ -3,15 +3,21 @@
//! 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 std::{
io,
pin::Pin,
task::{Context, Poll},
};
use bytes::Bytes;
use tokio::{
io::{AsyncRead, AsyncReadExt, AsyncWriteExt, DuplexStream},
io::{AsyncRead, AsyncReadExt, AsyncWriteExt, DuplexStream, ReadBuf},
sync::mpsc,
task::JoinHandle,
};
use crate::{pcm::StreamInfo, stream::ManagedAsyncReader};
/// Generic error type shared by the streaming decoders.
#[derive(thiserror::Error, Debug, Clone)]
pub enum DecoderError {
@@ -47,6 +53,56 @@ impl From<&str> for DecoderError {
}
}
/// Generic decoded stream wrapper shared by all decoders.
pub struct DecodedStream<E>
where
E: std::error::Error,
{
info: StreamInfo,
reader: ManagedAsyncReader<E>,
}
impl<E> DecodedStream<E>
where
E: std::error::Error,
{
/// Creates a new decoded stream from metadata and the underlying reader.
pub fn new(info: StreamInfo, reader: ManagedAsyncReader<E>) -> Self {
Self { info, reader }
}
/// Returns metadata about the decoded audio stream.
pub fn info(&self) -> &StreamInfo {
&self.info
}
/// Consumes the stream and returns its components.
pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader<E>) {
(self.info, self.reader)
}
/// Waits for the decoding pipeline to finish.
pub async fn wait(self) -> Result<(), E>
where
E: From<String>,
{
self.reader.wait().await
}
}
impl<E> AsyncRead for DecodedStream<E>
where
E: std::error::Error,
{
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)
}
}
/// Size of chunks when reading input data.
///
/// This size balances between efficient I/O operations and memory usage.

53
pmoflac/src/mp3.rs
View File

@@ -87,22 +87,17 @@
//! }
//! ```
use std::{
io,
pin::Pin,
task::{Context, Poll},
};
use minimp3::{Decoder as MiniMp3Decoder, Error as MiniMp3Error};
use tokio::{
io::{AsyncRead, ReadBuf},
io::AsyncRead,
sync::{mpsc, oneshot},
};
use crate::{
common::ChannelReader,
decoder_common::{
spawn_ingest_task, spawn_writer_task, DecoderError, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE,
spawn_ingest_task, spawn_writer_task, DecodedStream, DecoderError, CHANNEL_CAPACITY,
DUPLEX_BUFFER_SIZE,
},
pcm::StreamInfo,
stream::ManagedAsyncReader,
@@ -111,44 +106,8 @@ use crate::{
/// Errors that can occur while decoding MP3 data.
pub type Mp3Error = DecoderError;
/// An async stream that decodes MP3 audio into PCM samples.
///
/// This struct implements `AsyncRead`, allowing you to read decoded PCM data
/// as it becomes available. The decoding happens in a background task.
pub struct Mp3DecodedStream {
info: StreamInfo,
reader: ManagedAsyncReader<Mp3Error>,
}
impl Mp3DecodedStream {
/// Returns metadata about the decoded MP3 stream.
pub fn info(&self) -> &StreamInfo {
&self.info
}
/// Consumes the stream and returns its components.
pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader<Mp3Error>) {
(self.info, self.reader)
}
/// Waits for the background decoding task to complete.
///
/// This should be called after reading all data to ensure proper cleanup
/// and to catch any errors that occurred during decoding.
pub async fn wait(self) -> Result<(), Mp3Error> {
self.reader.wait().await
}
}
impl AsyncRead for Mp3DecodedStream {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.reader).poll_read(cx, buf)
}
}
/// Async decoded MP3 stream (type alias over the shared wrapper).
pub type Mp3DecodedStream = DecodedStream<Mp3Error>;
/// Decodes an MP3 stream into PCM audio data (16-bit little-endian interleaved).
///
@@ -239,5 +198,5 @@ where
let info = info_rx.await.map_err(|_| Mp3Error::ChannelClosed)??;
let reader = ManagedAsyncReader::new("mp3-decode-writer", pcm_reader, writer_handle);
Ok(Mp3DecodedStream { info, reader })
Ok(DecodedStream::new(info, reader))
}

58
pmoflac/src/ogg.rs
View File

@@ -101,25 +101,21 @@
//! }
//! ```
use std::{
io,
pin::Pin,
task::{Context, Poll},
};
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::{AsyncRead, ReadBuf},
io::AsyncRead,
sync::{mpsc, oneshot},
};
use crate::{
common::ChannelReader,
decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE},
decoder_common::{
spawn_ingest_task, spawn_writer_task, DecodedStream, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE,
},
ogg_common::{OggContainerError, OggPacketReader, OggReaderOptions},
pcm::StreamInfo,
stream::ManagedAsyncReader,
@@ -140,48 +136,8 @@ impl From<audio::AudioReadError> for OggContainerError {
}
}
/// 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 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
}
}
impl AsyncRead for OggDecodedStream {
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)
}
}
/// Async stream alias for decoded Ogg/Vorbis audio.
pub type OggDecodedStream = DecodedStream<OggError>;
/// Decodes an Ogg/Vorbis stream into PCM audio (16-bit little-endian interleaved).
///
@@ -310,5 +266,5 @@ where
let info = info_rx.await.map_err(|_| OggError::ChannelClosed)??;
let reader = ManagedAsyncReader::new("ogg-decode-writer", pcm_reader, writer_handle);
Ok(OggDecodedStream { info, reader })
Ok(DecodedStream::new(info, reader))
}

48
pmoflac/src/opus.rs
View File

@@ -4,21 +4,17 @@
//! ingestion/producer pattern established for the other decoders while staying
//! 100% streaming (no seeking or buffering entire files).
use std::{
io,
pin::Pin,
task::{Context, Poll},
};
use opus::{Channels, Decoder as OpusDecoder, Error as OpusError};
use tokio::{
io::{AsyncRead, ReadBuf},
io::AsyncRead,
sync::{mpsc, oneshot},
};
use crate::{
common::ChannelReader,
decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE},
decoder_common::{
spawn_ingest_task, spawn_writer_task, DecodedStream, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE,
},
ogg_common::{OggContainerError, OggPacketReader, OggReaderOptions},
pcm::StreamInfo,
stream::ManagedAsyncReader,
@@ -36,38 +32,8 @@ impl From<OpusError> for OggContainerError {
}
}
/// 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)
}
}
/// Async stream alias for decoded Ogg/Opus audio.
pub type OggOpusDecodedStream = DecodedStream<OggOpusError>;
/// 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>
@@ -195,7 +161,7 @@ where
let info = info_rx.await.map_err(|_| OggOpusError::ChannelClosed)??;
let reader = ManagedAsyncReader::new("ogg-opus-writer", pcm_reader, writer_handle);
Ok(OggOpusDecodedStream { info, reader })
Ok(DecodedStream::new(info, reader))
}
/// Parsed OpusHead metadata.

43
pmoflac/src/wav.rs
View File

@@ -4,12 +4,7 @@
//! header incrementally, validates the format, and then streams `data` chunk
//! payload as little-endian PCM frames through the common async pipeline.
use std::{
fmt,
io::{self, Read},
pin::Pin,
task::{Context, Poll},
};
use std::{fmt, io::Read};
use tokio::{
io::AsyncRead,
@@ -19,7 +14,8 @@ use tokio::{
use crate::{
common::ChannelReader,
decoder_common::{
spawn_ingest_task, spawn_writer_task, DecoderError, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE,
spawn_ingest_task, spawn_writer_task, DecodedStream, DecoderError, CHANNEL_CAPACITY,
DUPLEX_BUFFER_SIZE,
},
pcm::StreamInfo,
stream::ManagedAsyncReader,
@@ -145,35 +141,8 @@ impl FmtChunk {
}
}
/// An async stream that yields PCM decoded from WAV data.
pub struct WavDecodedStream {
info: StreamInfo,
reader: ManagedAsyncReader<WavError>,
}
impl WavDecodedStream {
pub fn info(&self) -> &StreamInfo {
&self.info
}
pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader<WavError>) {
(self.info, self.reader)
}
pub async fn wait(self) -> Result<(), WavError> {
self.reader.wait().await
}
}
impl AsyncRead for WavDecodedStream {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.reader).poll_read(cx, buf)
}
}
/// Async stream alias for decoded WAV audio.
pub type WavDecodedStream = DecodedStream<WavError>;
/// Decode a WAV stream into PCM audio.
pub async fn decode_wav_stream<R>(reader: R) -> Result<WavDecodedStream, WavError>
@@ -315,5 +284,5 @@ where
let info = info_rx.await.map_err(|_| WavError::ChannelClosed)??;
let reader = ManagedAsyncReader::new("wav-decode-writer", pcm_reader, writer_handle);
Ok(WavDecodedStream { info, reader })
Ok(DecodedStream::new(info, reader))
}