diff --git a/.gitignore b/.gitignore index 6f9e14a2..bb8469ee 100644 --- a/.gitignore +++ b/.gitignore @@ -10,6 +10,7 @@ **/*.a **/*.aif **/*.aiff +**/*.flac xxx /dcai/ **/.pmomusic.yml diff --git a/pmoflac/Cargo.lock b/pmoflac/Cargo.lock new file mode 100644 index 00000000..270c2480 --- /dev/null +++ b/pmoflac/Cargo.lock @@ -0,0 +1,272 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "bitflags" +version = "2.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3" + +[[package]] +name = "bytes" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a" + +[[package]] +name = "cc" +version = "1.2.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "739eb0f94557554b3ca9a86d2d37bebd49c5e6d0c1d2bda35ba5bdac830befc2" +dependencies = [ + "find-msvc-tools", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "claxon" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4bfbf56724aa9eca8afa4fcfadeb479e722935bb2a0900c2d37e0cc477af0688" + +[[package]] +name = "cmake" +version = "0.1.54" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7caa3f9de89ddbe2c607f4101924c5abec803763ae9534e4f4d7d8f84aa81f0" +dependencies = [ + "cc", +] + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "fastrand" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" + +[[package]] +name = "find-msvc-tools" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52051878f80a721bb68ebfbc930e07b65ba72f2da88968ea5c06fd6ca3d3a127" + +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasip2", +] + +[[package]] +name = "libc" +version = "0.2.177" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2874a2af47a2325c2001a6e6fad9b16a53b802102b528163885171cf92b15976" + +[[package]] +name = "libflac-sys" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6fc5cbb957a914952ee9b8667e82b984c6dc280087df01497fc5b4776d303582" +dependencies = [ + "cmake", + "libc", +] + +[[package]] +name = "linux-raw-sys" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df1d3c3b53da64cf5760482273a98e575c651a67eec7f77df96b5b642de8f039" + +[[package]] +name = "once_cell" +version = "1.21.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" + +[[package]] +name = "pin-project-lite" +version = "0.2.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b" + +[[package]] +name = "pmoflac" +version = "0.1.0" +dependencies = [ + "bytes", + "claxon", + "libc", + "libflac-sys", + "tempfile", + "thiserror", + "tokio", +] + +[[package]] +name = "proc-macro2" +version = "1.0.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ee95bc4ef87b8d5ba32e8b7714ccc834865276eab0aed5c9958d00ec45f49e8" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.41" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce25767e7b499d1b604768e7cde645d14cc8584231ea6b295e9c9eb22c02e1d1" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + +[[package]] +name = "rustix" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd15f8a2c5551a84d56efdc1cd049089e409ac19a3072d5037a17fd70719ff3e" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys", +] + +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + +[[package]] +name = "syn" +version = "2.0.108" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da58917d35242480a05c2897064da0a80589a2a0476c9a3f2fdc83b53502e917" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tempfile" +version = "3.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d31c77bdf42a745371d260a26ca7163f1e0924b64afa0b688e61b5a9fa02f16" +dependencies = [ + "fastrand", + "getrandom", + "once_cell", + "rustix", + "windows-sys", +] + +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio" +version = "1.48.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff360e02eab121e0bc37a2d3b4d4dc622e6eda3a8e5253d5435ecf5bd4c68408" +dependencies = [ + "bytes", + "pin-project-lite", + "tokio-macros", +] + +[[package]] +name = "tokio-macros" +version = "2.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "unicode-ident" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "462eeb75aeb73aea900253ce739c8e18a67423fadf006037cd3ff27e82748a06" + +[[package]] +name = "wasip2" +version = "1.0.1+wasi-0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0562428422c63773dad2c345a1882263bbf4d65cf3f42e90921f787ef5ad58e7" +dependencies = [ + "wit-bindgen", +] + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "wit-bindgen" +version = "0.46.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f17a85883d4e6d00e8a97c586de764dabcc06133f7f1d55dce5cdc070ad7fe59" diff --git a/pmoflac/Cargo.toml b/pmoflac/Cargo.toml new file mode 100644 index 00000000..ced761fe --- /dev/null +++ b/pmoflac/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "pmoflac" +version = "0.1.0" +edition = "2021" +authors = ["PMOMusic"] +description = "Asynchronous FLAC <-> PCM streaming utilities" +license = "MIT" + +[lib] +name = "pmoflac" +path = "src/lib.rs" + +[workspace] +members = ["."] + +[dependencies] +bytes = "1.6" +claxon = "0.4" +libc = "0.2" +libflac-sys = { version = "0.3.3", default-features = false, features = ["build-flac"] } +thiserror = "1.0" +tokio = { version = "1.37", features = ["rt", "macros", "sync", "io-util"] } + +[dev-dependencies] +tempfile = "3.10" +tokio = { version = "1.37", features = ["rt", "macros", "sync", "io-util"] } diff --git a/pmoflac/src/decoder.rs b/pmoflac/src/decoder.rs new file mode 100644 index 00000000..cb4dce46 --- /dev/null +++ b/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> { + Pin::new(&mut self.reader).poll_read(cx, buf) + } +} + +pub async fn decode_flac_stream(reader: R) -> Result +where + R: AsyncRead + Unpin + Send + 'static, +{ + let (ingest_tx, ingest_rx) = mpsc::channel::>(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::, FlacError>>(CHANNEL_CAPACITY); + let (pcm_reader, mut pcm_writer) = tokio::io::duplex(256 * 1024); + let (info_tx, info_rx) = oneshot::channel::>(); + + 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>, + current: Bytes, + offset: usize, + finished: bool, +} + +impl ChannelReader { + fn new(rx: mpsc::Receiver>) -> Self { + Self { + rx, + current: Bytes::new(), + offset: 0, + finished: false, + } + } +} + +impl Read for ChannelReader { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + 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); + } + } + } + } +} diff --git a/pmoflac/src/encoder.rs b/pmoflac/src/encoder.rs new file mode 100644 index 00000000..16191cbd --- /dev/null +++ b/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> { + 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, + pub block_size: Option, +} + +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( + reader: R, + format: PcmFormat, + options: EncoderOptions, +) -> Result +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::>(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::, FlacError>>(CHANNEL_CAPACITY); + let (init_tx, init_rx) = oneshot::channel::>(); + + 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( + reader: R, + format: PcmFormat, + tx: mpsc::Sender>, +) -> 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>, + tx: mpsc::Sender, FlacError>>, + init_tx: oneshot::Sender>, +) -> 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, FlacError>>, + error: Option, +} + +impl EncoderClientState { + fn new(tx: mpsc::Sender, 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 + } + } +} diff --git a/pmoflac/src/error.rs b/pmoflac/src/error.rs new file mode 100644 index 00000000..81bd8f91 --- /dev/null +++ b/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 for FlacError { + fn from(err: claxon::Error) -> Self { + FlacError::Decode(err.to_string()) + } +} diff --git a/pmoflac/src/lib.rs b/pmoflac/src/lib.rs new file mode 100644 index 00000000..efe06fab --- /dev/null +++ b/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}; diff --git a/pmoflac/src/pcm.rs b/pmoflac/src/pcm.rs new file mode 100644 index 00000000..657efa8a --- /dev/null +++ b/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, + 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, + pub frames: u32, +} + +impl PcmChunk { + pub fn new(data: Vec, 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) +} diff --git a/pmoflac/src/stream.rs b/pmoflac/src/stream.rs new file mode 100644 index 00000000..b4aba9c8 --- /dev/null +++ b/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, + join: Option>>, + role: &'static str, +} + +impl ManagedAsyncReader { + pub fn new( + role: &'static str, + inner: DuplexStream, + join: JoinHandle>, + ) -> 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> { + 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> { + 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(); + } +} diff --git a/pmoflac/src/util.rs b/pmoflac/src/util.rs new file mode 100644 index 00000000..33b7942b --- /dev/null +++ b/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) { + 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, 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) +} diff --git a/pmoflac/tests/flac_streams.rs b/pmoflac/tests/flac_streams.rs new file mode 100644 index 00000000..2fd92eee --- /dev/null +++ b/pmoflac/tests/flac_streams.rs @@ -0,0 +1,112 @@ +use std::{ + io, + path::PathBuf, + pin::Pin, + task::{Context, Poll}, +}; + +use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf}; + +use pmoflac::{decode_flac_stream, encode_flac_stream, EncoderOptions, FlacError, PcmFormat}; + +fn fixture(name: &str) -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("test_data") + .join(name) +} + +struct VecAsyncReader { + data: Vec, + pos: usize, +} + +impl VecAsyncReader { + fn new(data: Vec) -> Self { + Self { data, pos: 0 } + } +} + +impl AsyncRead for VecAsyncReader { + fn poll_read( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + if self.pos >= self.data.len() { + return Poll::Ready(Ok(())); + } + let remaining = &self.data[self.pos..]; + let to_copy = remaining.len().min(buf.remaining()); + if to_copy == 0 { + return Poll::Ready(Ok(())); + } + buf.put_slice(&remaining[..to_copy]); + self.pos += to_copy; + Poll::Ready(Ok(())) + } +} + +#[tokio::test] +async fn decode_stream_info_16_44() -> Result<(), FlacError> { + let bytes = std::fs::read(fixture( + "1abaa2c7fb4302e20ac570e79857b700.32bits-44.1Khz.flac", + ))?; + let mut stream = decode_flac_stream(VecAsyncReader::new(bytes)).await?; + let info = stream.info().clone(); + + assert_eq!(info.sample_rate, 44_100); + assert_eq!(info.channels, 2); + assert_eq!(info.bits_per_sample, 16); + + let mut pcm = Vec::new(); + stream.read_to_end(&mut pcm).await?; + assert!(!pcm.is_empty()); + stream.wait().await?; + + Ok(()) +} + +#[tokio::test] +async fn roundtrip_encode_decode_24_192() -> Result<(), FlacError> { + let bytes = std::fs::read(fixture("Yuri-Korzunov_Movement_24bit-192kHz.flac"))?; + let mut decoder = decode_flac_stream(VecAsyncReader::new(bytes)).await?; + let info = decoder.info().clone(); + + let mut pcm_bytes = Vec::new(); + decoder.read_to_end(&mut pcm_bytes).await?; + decoder.wait().await?; + + assert!(!pcm_bytes.is_empty()); + assert_eq!( + pcm_bytes.len() % (info.bytes_per_sample() * info.channels as usize), + 0 + ); + + let format = PcmFormat { + sample_rate: info.sample_rate, + channels: info.channels, + bits_per_sample: info.bits_per_sample, + }; + let options = EncoderOptions { + total_samples: info.total_samples, + ..Default::default() + }; + + let pcm_reader = VecAsyncReader::new(pcm_bytes.clone()); + let mut encoder_stream = encode_flac_stream(pcm_reader, format, options).await?; + let mut encoded_bytes = Vec::new(); + encoder_stream.read_to_end(&mut encoded_bytes).await?; + encoder_stream.wait().await?; + + assert!(!encoded_bytes.is_empty()); + + let flac_reader = VecAsyncReader::new(encoded_bytes); + let mut decoder_roundtrip = decode_flac_stream(flac_reader).await?; + let mut pcm_roundtrip = Vec::new(); + decoder_roundtrip.read_to_end(&mut pcm_roundtrip).await?; + decoder_roundtrip.wait().await?; + + assert_eq!(pcm_roundtrip, pcm_bytes); + + Ok(()) +}