Crate pmoflac

pmoflac/src/decoder.rs

@@ -0,0 +1,219 @@
+use std::{
+    io::{self, Read},
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use bytes::Bytes;
+use tokio::{
+    io::{AsyncRead, AsyncReadExt, AsyncWriteExt},
+    sync::{mpsc, oneshot},
+};
+
+use crate::{
+    error::FlacError, pcm::StreamInfo, stream::ManagedAsyncReader,
+    util::interleaved_i32_to_le_bytes,
+};
+
+const INGEST_CHUNK_SIZE: usize = 32 * 1024;
+const CHANNEL_CAPACITY: usize = 8;
+
+pub struct FlacDecodedStream {
+    info: StreamInfo,
+    reader: ManagedAsyncReader,
+}
+
+impl FlacDecodedStream {
+    pub fn info(&self) -> &StreamInfo {
+        &self.info
+    }
+
+    pub fn into_parts(self) -> (StreamInfo, ManagedAsyncReader) {
+        (self.info, self.reader)
+    }
+
+    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)
+    }
+}
+
+pub async fn decode_flac_stream<R>(reader: R) -> Result<FlacDecodedStream, FlacError>
+where
+    R: AsyncRead + Unpin + Send + 'static,
+{
+    let (ingest_tx, ingest_rx) = mpsc::channel::<Result<Bytes, FlacError>>(CHANNEL_CAPACITY);
+
+    tokio::spawn(async move {
+        let mut reader = tokio::io::BufReader::new(reader);
+        let mut buf = vec![0u8; INGEST_CHUNK_SIZE];
+        loop {
+            match reader.read(&mut buf).await {
+                Ok(0) => break,
+                Ok(n) => {
+                    let chunk = Bytes::copy_from_slice(&buf[..n]);
+                    if ingest_tx.send(Ok(chunk)).await.is_err() {
+                        break;
+                    }
+                }
+                Err(err) => {
+                    let _ = ingest_tx.send(Err(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> {
+        let mut channel_reader = ChannelReader::new(ingest_rx);
+        let mut flac_reader = match claxon::FlacReader::new(&mut channel_reader) {
+            Ok(reader) => reader,
+            Err(err) => {
+                let msg = err.to_string();
+                let _ = info_tx.send(Err(FlacError::Decode(msg.clone())));
+                return Err(FlacError::Decode(msg));
+            }
+        };
+
+        let flac_info = flac_reader.streaminfo();
+        let info = StreamInfo {
+            sample_rate: flac_info.sample_rate,
+            channels: flac_info.channels as u8,
+            bits_per_sample: flac_info.bits_per_sample as u8,
+            total_samples: flac_info.samples,
+            max_block_size: flac_info.max_block_size,
+            min_block_size: flac_info.min_block_size,
+        };
+        if info_tx.send(Ok(info.clone())).is_err() {
+            return Ok(());
+        }
+
+        let mut blocks = flac_reader.blocks();
+        let mut buffer = Vec::new();
+        let mut interleaved = Vec::new();
+        let mut pcm_bytes = Vec::new();
+        loop {
+            match blocks.read_next_or_eof(buffer) {
+                Ok(Some(block)) => {
+                    let frames = block.duration() as usize;
+                    let channels = block.channels() as usize;
+
+                    interleaved.clear();
+                    interleaved.reserve(frames * channels);
+                    for frame_idx in 0..frames {
+                        for channel_idx in 0..channels {
+                            interleaved.push(block.sample(channel_idx as u32, frame_idx as u32));
+                        }
+                    }
+
+                    pcm_bytes.clear();
+                    pcm_bytes.reserve(frames * channels * info.bytes_per_sample());
+                    interleaved_i32_to_le_bytes(&interleaved, info.bits_per_sample, &mut pcm_bytes);
+                    let chunk = std::mem::take(&mut pcm_bytes);
+                    if pcm_tx.blocking_send(Ok(chunk)).is_err() {
+                        break;
+                    }
+
+                    pcm_bytes = Vec::with_capacity(frames * channels * info.bytes_per_sample());
+                    buffer = block.into_buffer();
+                }
+                Ok(None) => break,
+                Err(err) => {
+                    let msg = err.to_string();
+                    let _ = pcm_tx.blocking_send(Err(FlacError::Decode(msg.clone())));
+                    return Err(FlacError::Decode(msg));
+                }
+            }
+        }
+
+        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 info = info_rx.await.map_err(|_| FlacError::ChannelClosed)??;
+    let reader = ManagedAsyncReader::new("flac-decode-writer", pcm_reader, writer_handle);
+
+    Ok(FlacDecodedStream { info, reader })
+}
+
+struct ChannelReader {
+    rx: mpsc::Receiver<Result<Bytes, FlacError>>,
+    current: Bytes,
+    offset: usize,
+    finished: bool,
+}
+
+impl ChannelReader {
+    fn new(rx: mpsc::Receiver<Result<Bytes, FlacError>>) -> Self {
+        Self {
+            rx,
+            current: Bytes::new(),
+            offset: 0,
+            finished: false,
+        }
+    }
+}
+
+impl Read for ChannelReader {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        loop {
+            if self.offset < self.current.len() {
+                let n = std::cmp::min(buf.len(), self.current.len() - self.offset);
+                buf[..n].copy_from_slice(&self.current[self.offset..self.offset + n]);
+                self.offset += n;
+                return Ok(n);
+            }
+            if self.finished {
+                return Ok(0);
+            }
+
+            match self.rx.blocking_recv() {
+                Some(Ok(bytes)) => {
+                    if bytes.is_empty() {
+                        continue;
+                    }
+                    self.current = bytes;
+                    self.offset = 0;
+                }
+                Some(Err(err)) => {
+                    self.finished = true;
+                    return Err(io::Error::new(io::ErrorKind::Other, err.to_string()));
+                }
+                None => {
+                    self.finished = true;
+                    return Ok(0);
+                }
+            }
+        }
+    }
+}

pmoflac/src/encoder.rs

@@ -0,0 +1,357 @@
+use std::{
+    ffi::c_void,
+    io,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use tokio::{
+    io::{AsyncRead, AsyncReadExt, AsyncWriteExt},
+    sync::{mpsc, oneshot},
+};
+
+use crate::{
+    error::FlacError,
+    pcm::{PcmChunk, PcmFormat},
+    stream::ManagedAsyncReader,
+    util::le_bytes_to_interleaved_i32,
+};
+
+const CHANNEL_CAPACITY: usize = 8;
+const PCM_FRAMES_PER_CHUNK: usize = 4096;
+
+pub struct FlacEncodedStream {
+    format: PcmFormat,
+    reader: ManagedAsyncReader,
+}
+
+impl FlacEncodedStream {
+    pub fn format(&self) -> PcmFormat {
+        self.format
+    }
+
+    pub fn into_parts(self) -> (PcmFormat, ManagedAsyncReader) {
+        (self.format, self.reader)
+    }
+
+    pub async fn wait(self) -> Result<(), FlacError> {
+        self.reader.wait().await
+    }
+}
+
+impl tokio::io::AsyncRead for FlacEncodedStream {
+    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)
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct EncoderOptions {
+    pub compression_level: u32,
+    pub verify: bool,
+    pub total_samples: Option<u64>,
+    pub block_size: Option<u32>,
+}
+
+impl Default for EncoderOptions {
+    fn default() -> Self {
+        Self {
+            compression_level: 5,
+            verify: false,
+            total_samples: None,
+            block_size: None,
+        }
+    }
+}
+
+pub async fn encode_flac_stream<R>(
+    reader: R,
+    format: PcmFormat,
+    options: EncoderOptions,
+) -> Result<FlacEncodedStream, FlacError>
+where
+    R: AsyncRead + Unpin + Send + 'static,
+{
+    format
+        .validate()
+        .map_err(|msg| FlacError::Unsupported(format!("invalid PCM format: {msg}")))?;
+
+    if options.compression_level > 12 {
+        return Err(FlacError::Unsupported(
+            "compression level must be in 0..=12".into(),
+        ));
+    }
+
+    let (pcm_tx, pcm_rx) = mpsc::channel::<Result<PcmChunk, FlacError>>(CHANNEL_CAPACITY);
+    let format_for_reader = format;
+    tokio::spawn(async move {
+        let _ = feed_pcm_chunks(reader, format_for_reader, pcm_tx).await;
+    });
+
+    let (flac_reader, mut flac_writer) = tokio::io::duplex(256 * 1024);
+    let (flac_tx, mut flac_rx) = mpsc::channel::<Result<Vec<u8>, FlacError>>(CHANNEL_CAPACITY);
+    let (init_tx, init_rx) = oneshot::channel::<Result<(), FlacError>>();
+
+    let format_for_encoder = format;
+    let options_for_encoder = options;
+    let blocking_handle = tokio::task::spawn_blocking(move || {
+        run_encoder(
+            format_for_encoder,
+            options_for_encoder,
+            pcm_rx,
+            flac_tx,
+            init_tx,
+        )
+    });
+
+    let writer_handle = tokio::spawn(async move {
+        while let Some(chunk) = flac_rx.recv().await {
+            let bytes = chunk?;
+            if bytes.is_empty() {
+                continue;
+            }
+            flac_writer.write_all(&bytes).await?;
+        }
+        flac_writer.shutdown().await?;
+        match blocking_handle.await {
+            Ok(res) => res,
+            Err(err) => Err(FlacError::TaskJoin {
+                role: "flac-encode",
+                details: err.to_string(),
+            }),
+        }
+    });
+
+    init_rx.await.map_err(|_| FlacError::ChannelClosed)??;
+
+    let reader = ManagedAsyncReader::new("flac-encode-writer", flac_reader, writer_handle);
+    Ok(FlacEncodedStream { format, reader })
+}
+
+async fn feed_pcm_chunks<R>(
+    reader: R,
+    format: PcmFormat,
+    tx: mpsc::Sender<Result<PcmChunk, FlacError>>,
+) -> Result<(), FlacError>
+where
+    R: AsyncRead + Unpin,
+{
+    let bytes_per_frame = format.bytes_per_sample() * format.channels as usize;
+    let chunk_bytes = PCM_FRAMES_PER_CHUNK * bytes_per_frame;
+    let mut pending = Vec::with_capacity(chunk_bytes * 2);
+    let mut reader = tokio::io::BufReader::new(reader);
+
+    loop {
+        while pending.len() >= chunk_bytes {
+            let samples =
+                le_bytes_to_interleaved_i32(&pending[..chunk_bytes], format.bits_per_sample)
+                    .map_err(|msg| FlacError::Encode(msg))?;
+            pending.drain(..chunk_bytes);
+            let frames = (samples.len() / format.channels as usize) as u32;
+            let chunk = PcmChunk::new(samples, frames, format.channels);
+            if tx.send(Ok(chunk)).await.is_err() {
+                return Ok(());
+            }
+        }
+
+        let read = reader.read_buf(&mut pending).await?;
+        if read == 0 {
+            break;
+        }
+    }
+
+    if !pending.is_empty() {
+        if pending.len() % bytes_per_frame != 0 {
+            let msg = "PCM stream ended with a partial frame (incomplete sample data)".to_string();
+            let _ = tx.send(Err(FlacError::Encode(msg.clone()))).await;
+            return Err(FlacError::Encode(msg));
+        }
+        let samples = le_bytes_to_interleaved_i32(&pending, format.bits_per_sample)
+            .map_err(|msg| FlacError::Encode(msg))?;
+        let frames = (samples.len() / format.channels as usize) as u32;
+        let chunk = PcmChunk::new(samples, frames, format.channels);
+        let _ = tx.send(Ok(chunk)).await;
+    }
+
+    Ok(())
+}
+
+fn run_encoder(
+    format: PcmFormat,
+    options: EncoderOptions,
+    mut rx: mpsc::Receiver<Result<PcmChunk, FlacError>>,
+    tx: mpsc::Sender<Result<Vec<u8>, FlacError>>,
+    init_tx: oneshot::Sender<Result<(), FlacError>>,
+) -> Result<(), FlacError> {
+    use libflac_sys::*;
+
+    unsafe {
+        let encoder = FLAC__stream_encoder_new();
+        if encoder.is_null() {
+            let _ = init_tx.send(Err(FlacError::LibFlacInit(
+                "FLAC__stream_encoder_new returned null".into(),
+            )));
+            return Err(FlacError::LibFlacInit(
+                "FLAC__stream_encoder_new returned null".into(),
+            ));
+        }
+
+        let _encoder_guard = EncoderHandle { ptr: encoder };
+
+        let mut state = EncoderClientState::new(tx);
+
+        let ensure = |ok: FLAC__bool, msg: &str| {
+            if ok == 0 {
+                Err(FlacError::LibFlacInit(msg.into()))
+            } else {
+                Ok(())
+            }
+        };
+
+        ensure(
+            FLAC__stream_encoder_set_channels(encoder, format.channels as u32),
+            "set_channels failed",
+        )?;
+        ensure(
+            FLAC__stream_encoder_set_bits_per_sample(encoder, format.bits_per_sample as u32),
+            "set_bits_per_sample failed",
+        )?;
+        ensure(
+            FLAC__stream_encoder_set_sample_rate(encoder, format.sample_rate),
+            "set_sample_rate failed",
+        )?;
+        ensure(
+            FLAC__stream_encoder_set_compression_level(encoder, options.compression_level),
+            "set_compression_level failed",
+        )?;
+        ensure(
+            FLAC__stream_encoder_set_streamable_subset(encoder, 1),
+            "set_streamable_subset failed",
+        )?;
+        ensure(
+            FLAC__stream_encoder_set_verify(encoder, if options.verify { 1 } else { 0 }),
+            "set_verify failed",
+        )?;
+        if let Some(total) = options.total_samples {
+            ensure(
+                FLAC__stream_encoder_set_total_samples_estimate(encoder, total),
+                "set_total_samples_estimate failed",
+            )?;
+        }
+        if let Some(block_size) = options.block_size {
+            ensure(
+                FLAC__stream_encoder_set_blocksize(encoder, block_size),
+                "set_blocksize failed",
+            )?;
+        }
+
+        let init_status = FLAC__stream_encoder_init_stream(
+            encoder,
+            Some(write_callback),
+            None,
+            None,
+            None,
+            &mut state as *mut EncoderClientState as *mut c_void,
+        );
+        if init_status != libflac_sys::FLAC__STREAM_ENCODER_INIT_STATUS_OK {
+            let msg = format!("init_stream failed: status {init_status}");
+            let _ = init_tx.send(Err(FlacError::LibFlacInit(msg.clone())));
+            return Err(FlacError::LibFlacInit(msg));
+        }
+
+        let _ = init_tx.send(Ok(()));
+
+        while let Some(chunk_result) = rx.blocking_recv() {
+            let chunk = match chunk_result {
+                Ok(chunk) => chunk,
+                Err(err) => {
+                    FLAC__stream_encoder_finish(encoder);
+                    return Err(err);
+                }
+            };
+            if chunk.frames == 0 {
+                continue;
+            }
+
+            let success = FLAC__stream_encoder_process_interleaved(
+                encoder,
+                chunk.data.as_ptr(),
+                chunk.frames,
+            );
+            if success == 0 {
+                if let Some(err) = state.error.take() {
+                    FLAC__stream_encoder_finish(encoder);
+                    return Err(err);
+                }
+                FLAC__stream_encoder_finish(encoder);
+                return Err(FlacError::Encode("libFLAC reported encode failure".into()));
+            }
+        }
+
+        let finish_ok = FLAC__stream_encoder_finish(encoder);
+        if finish_ok == 0 {
+            if let Some(err) = state.error.take() {
+                return Err(err);
+            }
+            return Err(FlacError::Encode(
+                "libFLAC failed to finalize stream".into(),
+            ));
+        }
+
+        if let Some(err) = state.error.take() {
+            return Err(err);
+        }
+    }
+
+    Ok(())
+}
+
+struct EncoderClientState {
+    tx: mpsc::Sender<Result<Vec<u8>, FlacError>>,
+    error: Option<FlacError>,
+}
+
+impl EncoderClientState {
+    fn new(tx: mpsc::Sender<Result<Vec<u8>, FlacError>>) -> Self {
+        Self { tx, error: None }
+    }
+}
+
+struct EncoderHandle {
+    ptr: *mut libflac_sys::FLAC__StreamEncoder,
+}
+
+impl Drop for EncoderHandle {
+    fn drop(&mut self) {
+        unsafe {
+            if !self.ptr.is_null() {
+                libflac_sys::FLAC__stream_encoder_delete(self.ptr);
+            }
+        }
+    }
+}
+
+unsafe extern "C" fn write_callback(
+    _encoder: *const libflac_sys::FLAC__StreamEncoder,
+    buffer: *const libflac_sys::FLAC__byte,
+    bytes: usize,
+    _samples: u32,
+    _current_frame: u32,
+    client_data: *mut c_void,
+) -> libflac_sys::FLAC__StreamEncoderWriteStatus {
+    let state = &mut *(client_data as *mut EncoderClientState);
+    let slice = std::slice::from_raw_parts(buffer, bytes);
+    match state.tx.blocking_send(Ok(slice.to_vec())) {
+        Ok(_) => libflac_sys::FLAC__STREAM_ENCODER_WRITE_STATUS_OK,
+        Err(_) => {
+            state.error = Some(FlacError::LibFlacWrite(
+                "failed to send encoded data (receiver dropped)".into(),
+            ));
+            libflac_sys::FLAC__STREAM_ENCODER_WRITE_STATUS_FATAL_ERROR
+        }
+    }
+}

pmoflac/src/error.rs

@@ -0,0 +1,27 @@
+use std::io;
+
+#[derive(thiserror::Error, Debug)]
+pub enum FlacError {
+    #[error("I/O error: {0}")]
+    Io(#[from] io::Error),
+    #[error("FLAC decode error: {0}")]
+    Decode(String),
+    #[error("FLAC encode error: {0}")]
+    Encode(String),
+    #[error("libFLAC initialization failed: {0}")]
+    LibFlacInit(String),
+    #[error("libFLAC write callback failed: {0}")]
+    LibFlacWrite(String),
+    #[error("internal channel closed unexpectedly")]
+    ChannelClosed,
+    #[error("unsupported configuration: {0}")]
+    Unsupported(String),
+    #[error("{role} task failed: {details}")]
+    TaskJoin { role: &'static str, details: String },
+}
+
+impl From<claxon::Error> for FlacError {
+    fn from(err: claxon::Error) -> Self {
+        FlacError::Decode(err.to_string())
+    }
+}

pmoflac/src/lib.rs

@@ -0,0 +1,11 @@
+pub mod decoder;
+pub mod encoder;
+pub mod error;
+mod pcm;
+mod stream;
+mod util;
+
+pub use decoder::{decode_flac_stream, FlacDecodedStream};
+pub use encoder::{encode_flac_stream, EncoderOptions, FlacEncodedStream};
+pub use error::FlacError;
+pub use pcm::{PcmFormat, StreamInfo};

pmoflac/src/pcm.rs

@@ -0,0 +1,65 @@
+use std::cmp;
+
+/// Describes the properties of a FLAC/PCM stream.
+#[derive(Debug, Clone)]
+pub struct StreamInfo {
+    pub sample_rate: u32,
+    pub channels: u8,
+    pub bits_per_sample: u8,
+    pub total_samples: Option<u64>,
+    pub max_block_size: u16,
+    pub min_block_size: u16,
+}
+
+impl StreamInfo {
+    pub fn bytes_per_sample(&self) -> usize {
+        bytes_per_sample(self.bits_per_sample)
+    }
+}
+
+/// Basic PCM format used when encoding to FLAC.
+#[derive(Debug, Clone, Copy)]
+pub struct PcmFormat {
+    pub sample_rate: u32,
+    pub channels: u8,
+    pub bits_per_sample: u8,
+}
+
+impl PcmFormat {
+    pub fn validate(&self) -> Result<(), String> {
+        if self.channels == 0 {
+            return Err("channel count must be greater than 0".into());
+        }
+        if self.channels > 8 {
+            return Err("channel count greater than 8 is unsupported".into());
+        }
+        if self.sample_rate == 0 {
+            return Err("sample rate must be greater than 0".into());
+        }
+        if self.bits_per_sample == 0 || self.bits_per_sample > 32 {
+            return Err("bits per sample must be in 1..=32".into());
+        }
+        Ok(())
+    }
+
+    pub fn bytes_per_sample(&self) -> usize {
+        bytes_per_sample(self.bits_per_sample)
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct PcmChunk {
+    pub data: Vec<i32>,
+    pub frames: u32,
+}
+
+impl PcmChunk {
+    pub fn new(data: Vec<i32>, frames: u32, channels: u8) -> Self {
+        debug_assert_eq!(data.len(), frames as usize * channels as usize);
+        Self { data, frames }
+    }
+}
+
+pub(crate) fn bytes_per_sample(bits_per_sample: u8) -> usize {
+    cmp::max(1, ((bits_per_sample as usize) + 7) / 8)
+}

pmoflac/src/stream.rs

@@ -0,0 +1,78 @@
+use std::{
+    io,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use tokio::{
+    io::{AsyncRead, DuplexStream, ReadBuf},
+    task::JoinHandle,
+};
+
+use crate::error::FlacError;
+
+/// Async reader that is backed by a spawned task writing into it.
+pub struct ManagedAsyncReader {
+    inner: Option<DuplexStream>,
+    join: Option<JoinHandle<Result<(), FlacError>>>,
+    role: &'static str,
+}
+
+impl ManagedAsyncReader {
+    pub fn new(
+        role: &'static str,
+        inner: DuplexStream,
+        join: JoinHandle<Result<(), FlacError>>,
+    ) -> Self {
+        Self {
+            inner: Some(inner),
+            join: Some(join),
+            role,
+        }
+    }
+
+    /// Waits for the producer task to finish.
+    pub async fn wait(mut self) -> Result<(), FlacError> {
+        match self.join.take() {
+            Some(handle) => match handle.await {
+                Ok(res) => res,
+                Err(err) => Err(FlacError::TaskJoin {
+                    role: self.role,
+                    details: err.to_string(),
+                }),
+            },
+            None => Ok(()),
+        }
+    }
+
+    fn poll_read_inner(
+        &mut self,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        let inner = self
+            .inner
+            .as_mut()
+            .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "reader dropped"))?;
+        Pin::new(inner).poll_read(cx, buf)
+    }
+}
+
+impl AsyncRead for ManagedAsyncReader {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        self.poll_read_inner(cx, buf)
+    }
+}
+
+impl Drop for ManagedAsyncReader {
+    fn drop(&mut self) {
+        if let Some(handle) = self.join.take() {
+            handle.abort();
+        }
+        self.inner.take();
+    }
+}

pmoflac/src/util.rs

@@ -0,0 +1,47 @@
+use crate::pcm::bytes_per_sample;
+
+pub fn interleaved_i32_to_le_bytes(samples: &[i32], bits_per_sample: u8, out: &mut Vec<u8>) {
+    let bytes_per = bytes_per_sample(bits_per_sample);
+    out.clear();
+    out.reserve(samples.len() * bytes_per);
+
+    for &sample in samples {
+        let mut value = sample;
+        if bits_per_sample < 32 {
+            let shift = 32 - bits_per_sample as u32;
+            value = (value << shift) >> shift;
+        }
+        for i in 0..bytes_per {
+            out.push(((value >> (i * 8)) & 0xFF) as u8);
+        }
+    }
+}
+
+pub fn le_bytes_to_interleaved_i32(bytes: &[u8], bits_per_sample: u8) -> Result<Vec<i32>, String> {
+    let bytes_per = bytes_per_sample(bits_per_sample);
+    if bytes.len() % bytes_per != 0 {
+        return Err(format!(
+            "PCM byte stream length {} is not aligned to {} bytes/sample",
+            bytes.len(),
+            bytes_per
+        ));
+    }
+
+    let mut samples = Vec::with_capacity(bytes.len() / bytes_per);
+    let shift = 32 - (bits_per_sample as u32);
+
+    let mut idx = 0;
+    while idx < bytes.len() {
+        let mut value = 0i32;
+        for i in 0..bytes_per {
+            value |= (bytes[idx + i] as i32) << (8 * i);
+        }
+        if bits_per_sample < 32 {
+            value = (value << shift) >> shift;
+        }
+        samples.push(value);
+        idx += bytes_per;
+    }
+
+    Ok(samples)
+}