From a23f340dc91d853c8ab2496a22e471ce62ffab86 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Mon, 27 Oct 2025 23:18:23 +0100 Subject: [PATCH] =?UTF-8?q?Debug=20suite=20=C3=A0=20revue=20de=20code?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pmoflac/src/aiff.rs | 327 ++++++++++++++++++++------------------ pmoflac/src/ogg.rs | 137 +++++++++------- pmoflac/src/ogg_common.rs | 18 ++- pmoflac/src/opus.rs | 193 ++++++++++++---------- pmoflac/src/wav.rs | 217 +++++++++++++------------ 5 files changed, 485 insertions(+), 407 deletions(-) diff --git a/pmoflac/src/aiff.rs b/pmoflac/src/aiff.rs index 83f65925..84c06981 100644 --- a/pmoflac/src/aiff.rs +++ b/pmoflac/src/aiff.rs @@ -153,176 +153,195 @@ where let (info_tx, info_rx) = oneshot::channel::>(); let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), AiffError> { - let channel_reader = ChannelReader::::new(ingest_rx); - let mut aiff_reader = StreamingAiffReader::new(channel_reader); + let mut info_tx = Some(info_tx); - // Parse FORM header - let form_header = aiff_reader.read_exact_vec(12)?; - if &form_header[0..4] != b"FORM" { - return Err(AiffError::Decode("missing FORM header".into())); - } - let form_type = <[u8; 4]>::try_from(&form_header[8..12]).unwrap(); - if form_type != *b"AIFF" && form_type != *b"AIFC" { - return Err(AiffError::Decode( - "unsupported FORM type (expected AIFF/AIFC)".into(), - )); - } + let result: Result<(), AiffError> = (|| { + let channel_reader = ChannelReader::::new(ingest_rx); + let mut aiff_reader = StreamingAiffReader::new(channel_reader); - let mut comm_chunk: Option = None; - let mut stream_info_sent = false; + // Parse FORM header + let form_header = aiff_reader.read_exact_vec(12)?; + if &form_header[0..4] != b"FORM" { + return Err(AiffError::Decode("missing FORM header".into())); + } + let form_type = <[u8; 4]>::try_from(&form_header[8..12]).unwrap(); + if form_type != *b"AIFF" && form_type != *b"AIFC" { + return Err(AiffError::Decode( + "unsupported FORM type (expected AIFF/AIFC)".into(), + )); + } - loop { - let header = match aiff_reader.read_exact_vec(8) { - Ok(bytes) => bytes, - Err(AiffError::Decode(msg)) if msg.contains("unexpected EOF") => break, - Err(err) => return Err(err), - }; - let chunk_id = <[u8; 4]>::try_from(&header[..4]).unwrap(); - let chunk_size = - u32::from_be_bytes([header[4], header[5], header[6], header[7]]) as usize; + let mut comm_chunk: Option = None; + let mut stream_info_sent = false; - let padded_size = if chunk_size % 2 == 0 { - chunk_size - } else { - chunk_size + 1 - }; + loop { + let header = match aiff_reader.read_exact_vec(8) { + Ok(bytes) => bytes, + Err(AiffError::Decode(msg)) if msg.contains("unexpected EOF") => break, + Err(err) => return Err(err), + }; + let chunk_id = <[u8; 4]>::try_from(&header[..4]).unwrap(); + let chunk_size = + u32::from_be_bytes([header[4], header[5], header[6], header[7]]) as usize; - match &chunk_id { - b"COMM" => { - let data = aiff_reader.read_exact_vec(chunk_size)?; - if form_type == *b"AIFF" && data.len() < 18 { - return Err(AiffError::Decode("COMM chunk too small".into())); - } - if data.len() < 18 { - return Err(AiffError::Decode("COMM chunk too small for AIFC".into())); - } - let channels = u16::from_be_bytes([data[0], data[1]]); - let num_frames = u32::from_be_bytes([data[2], data[3], data[4], data[5]]); - let bits_per_sample = u16::from_be_bytes([data[6], data[7]]); - let sample_rate = parse_extended_f80(&data[8..18])?; + let padded_size = if chunk_size % 2 == 0 { + chunk_size + } else { + chunk_size + 1 + }; - let compression = if form_type == *b"AIFC" { - if data.len() < 22 { - return Err(AiffError::Decode( - "AIFC COMM chunk missing compression type".into(), - )); + match &chunk_id { + b"COMM" => { + let data = aiff_reader.read_exact_vec(chunk_size)?; + if form_type == *b"AIFF" && data.len() < 18 { + return Err(AiffError::Decode("COMM chunk too small".into())); } - match &data[18..22] { - b"NONE" => Compression::BigEndianPcm, - b"sowt" => Compression::LittleEndianPcm, - code => { - return Err(AiffError::Decode(format!( - "unsupported AIFC compression type: {}", - String::from_utf8_lossy(code) - ))) + if data.len() < 18 { + return Err(AiffError::Decode("COMM chunk too small for AIFC".into())); + } + let channels = u16::from_be_bytes([data[0], data[1]]); + let num_frames = + u32::from_be_bytes([data[2], data[3], data[4], data[5]]); + let bits_per_sample = u16::from_be_bytes([data[6], data[7]]); + let sample_rate = parse_extended_f80(&data[8..18])?; + + let compression = if form_type == *b"AIFC" { + if data.len() < 22 { + return Err(AiffError::Decode( + "AIFC COMM chunk missing compression type".into(), + )); } - } - } else { - Compression::BigEndianPcm - }; - - let comm = CommChunk { - channels, - num_frames, - bits_per_sample, - sample_rate, - compression, - }; - comm.validate()?; - - comm_chunk = Some(comm); - - if padded_size > chunk_size { - aiff_reader.skip(padded_size - chunk_size)?; - } - } - b"SSND" => { - let comm = comm_chunk.as_ref().ok_or_else(|| { - AiffError::Decode("SSND chunk encountered before COMM".into()) - })?; - - let header = aiff_reader.read_exact_vec(8)?; - let offset = - u32::from_be_bytes([header[0], header[1], header[2], header[3]]) as usize; - let _block_size = - u32::from_be_bytes([header[4], header[5], header[6], header[7]]) as usize; - - if offset > 0 { - aiff_reader.skip(offset)?; - } - - let data_bytes = chunk_size - .checked_sub(8) - .ok_or_else(|| AiffError::Decode("invalid SSND chunk size".into()))?; - let bytes_per_sample = comm.bytes_per_sample(); - - let info = StreamInfo { - sample_rate: comm.sample_rate, - channels: comm.channels as u8, - bits_per_sample: comm.bits_per_sample as u8, - total_samples: Some(comm.num_frames as u64), - max_block_size: 0, - min_block_size: 0, - }; - - if !stream_info_sent { - if info_tx.send(Ok(info.clone())).is_err() { - return Ok(()); - } - stream_info_sent = true; - } - - let mut remaining = data_bytes; - while remaining > 0 { - let mut to_read = remaining.min(8192); - let residue = to_read % bytes_per_sample; - if residue != 0 { - to_read -= residue; - } - if to_read == 0 { - to_read = bytes_per_sample; - } - let mut chunk = aiff_reader.read_exact_vec(to_read)?; - match comm.compression { - Compression::BigEndianPcm => { - chunk = convert_be_pcm(chunk, comm.bits_per_sample)?; + match &data[18..22] { + b"NONE" => Compression::BigEndianPcm, + b"sowt" => Compression::LittleEndianPcm, + code => { + return Err(AiffError::Decode(format!( + "unsupported AIFC compression type: {}", + String::from_utf8_lossy(code) + ))) + } } - Compression::LittleEndianPcm => { - // data already little-endian; no conversion - } - } - if !chunk.is_empty() { - if pcm_tx.blocking_send(Ok(chunk)).is_err() { - return Ok(()); - } - } - remaining = remaining - .checked_sub(to_read) - .ok_or_else(|| AiffError::Decode("SSND chunk underflow".into()))?; - } + } else { + Compression::BigEndianPcm + }; - if padded_size > chunk_size { - aiff_reader.skip(1)?; - } + let comm = CommChunk { + channels, + num_frames, + bits_per_sample, + sample_rate, + compression, + }; + comm.validate()?; - break; - } - _ => { - aiff_reader.skip(chunk_size)?; - if padded_size > chunk_size { - aiff_reader.skip(padded_size - chunk_size)?; + comm_chunk = Some(comm); + + if padded_size > chunk_size { + aiff_reader.skip(padded_size - chunk_size)?; + } + } + b"SSND" => { + let comm = comm_chunk.as_ref().ok_or_else(|| { + AiffError::Decode("SSND chunk encountered before COMM".into()) + })?; + + let header = aiff_reader.read_exact_vec(8)?; + let offset = u32::from_be_bytes([ + header[0], header[1], header[2], header[3], + ]) as usize; + let _block_size = u32::from_be_bytes([ + header[4], header[5], header[6], header[7], + ]) as usize; + + if offset > 0 { + aiff_reader.skip(offset)?; + } + + let data_bytes = chunk_size + .checked_sub(8) + .ok_or_else(|| AiffError::Decode("invalid SSND chunk size".into()))?; + let bytes_per_sample = comm.bytes_per_sample(); + + let info = StreamInfo { + sample_rate: comm.sample_rate, + channels: comm.channels as u8, + bits_per_sample: comm.bits_per_sample as u8, + total_samples: Some(comm.num_frames as u64), + max_block_size: 0, + min_block_size: 0, + }; + + if !stream_info_sent { + if let Some(tx) = info_tx.take() { + if tx.send(Ok(info.clone())).is_err() { + return Ok(()); + } + } + stream_info_sent = true; + } + + let mut remaining = data_bytes; + while remaining > 0 { + let mut to_read = remaining.min(8192); + let residue = to_read % bytes_per_sample; + if residue != 0 { + to_read -= residue; + } + if to_read == 0 { + to_read = bytes_per_sample; + } + let mut chunk = aiff_reader.read_exact_vec(to_read)?; + match comm.compression { + Compression::BigEndianPcm => { + chunk = convert_be_pcm(chunk, comm.bits_per_sample)?; + } + Compression::LittleEndianPcm => { + // data already little-endian; no conversion + } + } + if !chunk.is_empty() { + if pcm_tx.blocking_send(Ok(chunk)).is_err() { + return Ok(()); + } + } + remaining = remaining + .checked_sub(to_read) + .ok_or_else(|| AiffError::Decode("SSND chunk underflow".into()))?; + } + + if padded_size > chunk_size { + aiff_reader.skip(1)?; + } + + break; + } + _ => { + aiff_reader.skip(chunk_size)?; + if padded_size > chunk_size { + aiff_reader.skip(padded_size - chunk_size)?; + } } } } - } - if !stream_info_sent { - return Err(AiffError::Decode( - "no SSND chunk found in AIFF stream".into(), - )); - } + if !stream_info_sent { + return Err(AiffError::Decode( + "no SSND chunk found in AIFF stream".into(), + )); + } - Ok(()) + Ok(()) + })(); + + match result { + Ok(()) => Ok(()), + Err(err) => { + if let Some(tx) = info_tx.take() { + let _ = tx.send(Err(err.clone())); + } + Err(err) + } + } }); let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "aiff-decode"); diff --git a/pmoflac/src/ogg.rs b/pmoflac/src/ogg.rs index 6c3f8b8f..08ec7112 100644 --- a/pmoflac/src/ogg.rs +++ b/pmoflac/src/ogg.rs @@ -184,81 +184,98 @@ where let (info_tx, info_rx) = oneshot::channel::>(); let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), OggError> { - let channel_reader = ChannelReader::::new(ingest_rx); - let mut packet_reader = OggPacketReader::new(channel_reader, OggReaderOptions::default()); + let mut info_tx = Some(info_tx); - // 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()))?; - let ident_hdr = read_header_ident(&ident_packet)?; + let result: Result<(), OggError> = (|| { + let channel_reader = ChannelReader::::new(ingest_rx); + let mut packet_reader = + OggPacketReader::new(channel_reader, OggReaderOptions::default()); - let comment_packet = packet_reader - .next_packet()? - .ok_or_else(|| OggError::Decode("missing Vorbis comment header".into()))?; - let _comment_hdr: CommentHeader = read_header_comment(&comment_packet)?; + // 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()))?; + let ident_hdr = read_header_ident(&ident_packet)?; - 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 comment_packet = packet_reader + .next_packet()? + .ok_or_else(|| OggError::Decode("missing Vorbis comment header".into()))?; + let _comment_hdr: CommentHeader = read_header_comment(&comment_packet)?; - let info = StreamInfo { - sample_rate: ident_hdr.audio_sample_rate, - channels: ident_hdr.audio_channels, - bits_per_sample: 16, - total_samples: None, - max_block_size: 1 << ident_hdr.blocksize_1, - min_block_size: 1 << ident_hdr.blocksize_0, - }; + 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), + )?; - if info_tx.send(Ok(info.clone())).is_err() { - return Ok(()); - } + let info = StreamInfo { + sample_rate: ident_hdr.audio_sample_rate, + channels: ident_hdr.audio_channels, + bits_per_sample: 16, + total_samples: None, + max_block_size: 1 << ident_hdr.blocksize_1, + min_block_size: 1 << ident_hdr.blocksize_0, + }; - // Decode audio packets - let mut pcm_bytes = Vec::new(); - let mut produced_audio = false; - let mut pwr = PreviousWindowRight::new(); - - while let Some(packet) = packet_reader.next_packet()? { - let decoded: InterleavedSamples = - read_audio_packet_generic(&ident_hdr, &setup_hdr, &packet, &mut pwr)?; - - if decoded.samples.is_empty() { - continue; + if let Some(tx) = info_tx.take() { + if tx.send(Ok(info.clone())).is_err() { + return Ok(()); + } } - produced_audio = true; + // Decode audio packets + let mut pcm_bytes = Vec::new(); + let mut produced_audio = false; + let mut pwr = PreviousWindowRight::new(); - // 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()); + while let Some(packet) = packet_reader.next_packet()? { + let decoded: InterleavedSamples = + read_audio_packet_generic(&ident_hdr, &setup_hdr, &packet, &mut pwr)?; + + if decoded.samples.is_empty() { + continue; + } + + 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; + } + + // Pre-allocate for next iteration + pcm_bytes = + Vec::with_capacity(info.max_block_size as usize * info.channels as usize * 2); } - let chunk = std::mem::take(&mut pcm_bytes); - if pcm_tx.blocking_send(Ok(chunk)).is_err() { - break; + if !produced_audio { + let err = OggError::Decode("stream contained no decodable Vorbis packets".into()); + let _ = pcm_tx.blocking_send(Err(err.clone())); + return Err(err); } - // Pre-allocate for next iteration - pcm_bytes = - Vec::with_capacity(info.max_block_size as usize * info.channels as usize * 2); - } + Ok(()) + })(); - if !produced_audio { - let err = OggError::Decode("stream contained no decodable Vorbis packets".into()); - let _ = pcm_tx.blocking_send(Err(err.clone())); - return Err(err); + match result { + Ok(()) => Ok(()), + Err(err) => { + if let Some(tx) = info_tx.take() { + let _ = tx.send(Err(err.clone())); + } + Err(err) + } } - - Ok(()) }); let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "ogg-decode"); diff --git a/pmoflac/src/ogg_common.rs b/pmoflac/src/ogg_common.rs index 331caf28..e236c6d0 100644 --- a/pmoflac/src/ogg_common.rs +++ b/pmoflac/src/ogg_common.rs @@ -82,6 +82,7 @@ pub struct OggPacketReader { finished: bool, stream_serial: Option, sync_buffer: Vec, + sync_bytes_read: usize, synced: bool, options: OggReaderOptions, } @@ -96,6 +97,7 @@ impl OggPacketReader { finished: false, stream_serial: None, sync_buffer: Vec::new(), + sync_bytes_read: 0, synced: !options.find_sync, // If we don't need to find sync, we're already synced options, } @@ -183,14 +185,21 @@ impl OggPacketReader { return Ok(()); } - while self.sync_buffer.len() < self.options.max_sync_search { + while self.sync_bytes_read < self.options.max_sync_search { let mut chunk = [0u8; 1024]; - let n = Read::read(&mut self.reader, &mut chunk).map_err(OggContainerError::from)?; + let remaining = self.options.max_sync_search - self.sync_bytes_read; + let to_read = remaining.min(chunk.len()); + if to_read == 0 { + break; + } + let n = Read::read(&mut self.reader, &mut chunk[..to_read]) + .map_err(OggContainerError::from)?; if n == 0 { return Err(OggContainerError::Decode( "EOF reached while searching for Ogg sync pattern".into(), )); } + self.sync_bytes_read += n; self.sync_buffer.extend_from_slice(&chunk[..n]); // Search for "OggS" pattern @@ -205,9 +214,8 @@ impl OggPacketReader { 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() >= self.options.max_sync_search { + // Keep the buffer compact to avoid unbounded growth. + if self.sync_buffer.len() > 3 { let keep_len = 3.min(self.sync_buffer.len()); self.sync_buffer.drain(..self.sync_buffer.len() - keep_len); } diff --git a/pmoflac/src/opus.rs b/pmoflac/src/opus.rs index ee9476e8..f7a11b55 100644 --- a/pmoflac/src/opus.rs +++ b/pmoflac/src/opus.rs @@ -48,112 +48,129 @@ where let (info_tx, info_rx) = oneshot::channel::>(); let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), OggOpusError> { - let channel_reader = ChannelReader::::new(ingest_rx); - let mut packet_reader = OggPacketReader::new( - channel_reader, - OggReaderOptions { - validate_crc: false, - find_sync: false, - ..OggReaderOptions::default() - }, - ); + let mut info_tx = Some(info_tx); - let header_packet = packet_reader - .next_packet()? - .ok_or_else(|| OggOpusError::Decode("missing OpusHead packet".into()))?; - let header = OpusHead::parse(&header_packet)?; + let result: Result<(), OggOpusError> = (|| { + let channel_reader = ChannelReader::::new(ingest_rx); + let mut packet_reader = OggPacketReader::new( + channel_reader, + OggReaderOptions { + validate_crc: false, + find_sync: false, + ..OggReaderOptions::default() + }, + ); - let tags_packet = packet_reader - .next_packet()? - .ok_or_else(|| OggOpusError::Decode("missing OpusTags packet".into()))?; - let _tags = OpusTags::parse(&tags_packet)?; + let header_packet = packet_reader + .next_packet()? + .ok_or_else(|| OggOpusError::Decode("missing OpusHead packet".into()))?; + let header = OpusHead::parse(&header_packet)?; - let channels_enum = match header.channels { - 1 => Channels::Mono, - 2 => Channels::Stereo, - other => { - return Err(OggOpusError::Decode(format!( - "unsupported channel count: {}", - other - ))) - } - }; + let tags_packet = packet_reader + .next_packet()? + .ok_or_else(|| OggOpusError::Decode("missing OpusTags packet".into()))?; + let _tags = OpusTags::parse(&tags_packet)?; - let mut decoder = OpusDecoder::new(48_000, channels_enum)?; - if header.output_gain != 0 { - decoder.set_gain(i32::from(header.output_gain))?; - } + let channels_enum = match header.channels { + 1 => Channels::Mono, + 2 => Channels::Stereo, + other => { + return Err(OggOpusError::Decode(format!( + "unsupported channel count: {}", + other + ))) + } + }; - 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 mut decoder = OpusDecoder::new(48_000, channels_enum)?; + if header.output_gain != 0 { + decoder.set_gain(i32::from(header.output_gain))?; } - let decoded_frames = decoder.decode( - &packet, - &mut pcm_buffer[..MAX_FRAME_SAMPLES * channels], - false, - )?; - if decoded_frames == 0 { - continue; - } + 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, + }; - 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; + if let Some(tx) = info_tx.take() { + if tx.send(Ok(info.clone())).is_err() { + return Ok(()); } } - let start_index = start_frame * channels; - let end_index = decoded_frames * channels; + 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; - 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()); + 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 = std::mem::take(&mut pcm_bytes); + if pcm_tx.blocking_send(Ok(chunk)).is_err() { + break; + } + pcm_bytes = Vec::with_capacity(MAX_FRAME_SAMPLES * channels * 2); } - if pcm_bytes.is_empty() { - continue; + if !produced_audio { + return Err(OggOpusError::Decode( + "stream contained no decodable Opus packets".into(), + )); } - produced_audio = true; - let chunk = pcm_bytes.clone(); - if pcm_tx.blocking_send(Ok(chunk)).is_err() { - break; + Ok(()) + })(); + + match result { + Ok(()) => Ok(()), + Err(err) => { + if let Some(tx) = info_tx.take() { + let _ = tx.send(Err(err.clone())); + } + Err(err) } } - - 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"); diff --git a/pmoflac/src/wav.rs b/pmoflac/src/wav.rs index 6e595a39..647ecf93 100644 --- a/pmoflac/src/wav.rs +++ b/pmoflac/src/wav.rs @@ -157,126 +157,143 @@ where let (info_tx, info_rx) = oneshot::channel::>(); let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), WavError> { - let channel_reader = ChannelReader::::new(ingest_rx); - let mut wav_reader = StreamingWavReader::new(channel_reader); + let mut info_tx = Some(info_tx); - let riff = wav_reader.read_exact(12)?; - if &riff[0..4] != b"RIFF" { - return Err(WavError::Decode("missing RIFF header".into())); - } - if &riff[8..12] != b"WAVE" { - return Err(WavError::Decode("missing WAVE signature".into())); - } + let result: Result<(), WavError> = (|| { + let channel_reader = ChannelReader::::new(ingest_rx); + let mut wav_reader = StreamingWavReader::new(channel_reader); - let mut fmt_chunk: Option = None; - let mut data_found = false; - - loop { - let mut chunk_header = [0u8; 8]; - match wav_reader.read_exact(8) { - Ok(bytes) => chunk_header.copy_from_slice(bytes), - Err(WavError::Decode(msg)) if msg.contains("unexpected EOF") => break, - Err(err) => return Err(err), + let riff = wav_reader.read_exact(12)?; + if &riff[0..4] != b"RIFF" { + return Err(WavError::Decode("missing RIFF header".into())); + } + if &riff[8..12] != b"WAVE" { + return Err(WavError::Decode("missing WAVE signature".into())); } - let chunk_id = &chunk_header[..4]; - let chunk_size = u32::from_le_bytes([ - chunk_header[4], - chunk_header[5], - chunk_header[6], - chunk_header[7], - ]) as usize; - let padded_size = (chunk_size + 1) & !1; // align to even bytes + let mut fmt_chunk: Option = None; + let mut data_found = false; - match chunk_id { - b"fmt " => { - let bytes = wav_reader.read_exact(chunk_size)?; - if chunk_size < 16 { - return Err(WavError::Decode("fmt chunk too small".into())); - } - let audio_format = u16::from_le_bytes([bytes[0], bytes[1]]); - let channels = u16::from_le_bytes([bytes[2], bytes[3]]); - let sample_rate = u32::from_le_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]); - let bits_per_sample = u16::from_le_bytes([bytes[14], bytes[15]]); - let fmt = FmtChunk { - audio_format, - channels, - sample_rate, - bits_per_sample, - }; - fmt.validate()?; - fmt_chunk = Some(fmt); - if padded_size > chunk_size { - wav_reader.skip(padded_size - chunk_size)?; - } + loop { + let mut chunk_header = [0u8; 8]; + match wav_reader.read_exact(8) { + Ok(bytes) => chunk_header.copy_from_slice(bytes), + Err(WavError::Decode(msg)) if msg.contains("unexpected EOF") => break, + Err(err) => return Err(err), } - b"data" => { - let fmt = fmt_chunk - .as_ref() - .ok_or_else(|| WavError::Decode("data chunk before fmt chunk".into()))?; + let chunk_id = &chunk_header[..4]; + let chunk_size = u32::from_le_bytes([ + chunk_header[4], + chunk_header[5], + chunk_header[6], + chunk_header[7], + ]) as usize; - let info = StreamInfo { - sample_rate: fmt.sample_rate, - channels: fmt.channels as u8, - bits_per_sample: fmt.bits_per_sample as u8, - total_samples: None, - max_block_size: 0, - min_block_size: 0, - }; + let padded_size = (chunk_size + 1) & !1; // align to even bytes - if info_tx.send(Ok(info.clone())).is_err() { - return Ok(()); - } - - let mut remaining = chunk_size; - let bytes_per_frame = fmt.bytes_per_sample() * fmt.channels as usize; - let mut buffer = vec![0u8; 4096]; - - while remaining > 0 { - let to_read = remaining.min(buffer.len()); - let read = wav_reader.reader.read(&mut buffer[..to_read])?; - if read == 0 { - break; + match chunk_id { + b"fmt " => { + let bytes = wav_reader.read_exact(chunk_size)?; + if chunk_size < 16 { + return Err(WavError::Decode("fmt chunk too small".into())); } - remaining -= read; - let aligned = read - (read % bytes_per_frame); - if aligned > 0 { - if pcm_tx - .blocking_send(Ok(buffer[..aligned].to_vec())) - .is_err() - { + let audio_format = u16::from_le_bytes([bytes[0], bytes[1]]); + let channels = u16::from_le_bytes([bytes[2], bytes[3]]); + let sample_rate = + u32::from_le_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]); + let bits_per_sample = u16::from_le_bytes([bytes[14], bytes[15]]); + let fmt = FmtChunk { + audio_format, + channels, + sample_rate, + bits_per_sample, + }; + fmt.validate()?; + fmt_chunk = Some(fmt); + if padded_size > chunk_size { + wav_reader.skip(padded_size - chunk_size)?; + } + } + b"data" => { + let fmt = fmt_chunk.as_ref().ok_or_else(|| { + WavError::Decode("data chunk before fmt chunk".into()) + })?; + + let info = StreamInfo { + sample_rate: fmt.sample_rate, + channels: fmt.channels as u8, + bits_per_sample: fmt.bits_per_sample as u8, + total_samples: None, + max_block_size: 0, + min_block_size: 0, + }; + + if let Some(tx) = info_tx.take() { + if tx.send(Ok(info.clone())).is_err() { return Ok(()); } } - if aligned < read { - return Err(WavError::Decode( - "incomplete frame at end of chunk".into(), - )); + + let mut remaining = chunk_size; + let bytes_per_frame = fmt.bytes_per_sample() * fmt.channels as usize; + let mut buffer = vec![0u8; 4096]; + + while remaining > 0 { + let to_read = remaining.min(buffer.len()); + let read = wav_reader.reader.read(&mut buffer[..to_read])?; + if read == 0 { + break; + } + remaining -= read; + let aligned = read - (read % bytes_per_frame); + if aligned > 0 { + if pcm_tx + .blocking_send(Ok(buffer[..aligned].to_vec())) + .is_err() + { + return Ok(()); + } + } + if aligned < read { + return Err(WavError::Decode( + "incomplete frame at end of chunk".into(), + )); + } } - } - if padded_size > chunk_size { - let mut pad = [0u8; 1]; - wav_reader.reader.read_exact(&mut pad)?; - } + if padded_size > chunk_size { + let mut pad = [0u8; 1]; + wav_reader.reader.read_exact(&mut pad)?; + } - data_found = true; - break; - } - _ => { - wav_reader.skip(chunk_size)?; - if padded_size > chunk_size { - wav_reader.skip(padded_size - chunk_size)?; + data_found = true; + break; + } + _ => { + wav_reader.skip(chunk_size)?; + if padded_size > chunk_size { + wav_reader.skip(padded_size - chunk_size)?; + } } } } - } - if !data_found { - return Err(WavError::Decode("no data chunk found in WAV stream".into())); - } + if !data_found { + return Err(WavError::Decode("no data chunk found in WAV stream".into())); + } - Ok(()) + Ok(()) + })(); + + match result { + Ok(()) => Ok(()), + Err(err) => { + if let Some(tx) = info_tx.take() { + let _ = tx.send(Err(err.clone())); + } + Err(err) + } + } }); let writer_handle = spawn_writer_task(pcm_rx, pcm_writer, blocking_handle, "wav-decode");