use core::fmt::Debug;
use core::fmt::Formatter;
use core::fmt::Result as FormatResult;
use std::collections::TryReserveError;
use std::io::Error;
use std::io::ErrorKind;
use std::io::Read;
use std::io::Result;
use std::io::Seek;
use std::io::SeekFrom;
use std::result::Result as StdResult;
use qubit_io::Buffer;
use qubit_io::BufferedInput;
use qubit_io::ByteInput;
use qubit_io::Input;
use qubit_io::SeekableByteInput;
use qubit_utils::SliceRange;
use qubit_utils::UncheckedSlice;
use qubit_utils::allocation_error;
use qubit_utils::try_reserve_vec;
use super::internal::CodecDecodeDriver;
use super::internal::capacity_error_to_invalid_data;
use super::internal::validate_decode_progress;
use super::internal::validate_eof_decode_progress;
use crate::Codec;
use crate::DecodeLifecycleOutput;
use crate::DecodeLifecycleProgress;
use crate::TranscodeDomainError;
use crate::TranscodeFailure;
use crate::TranscodeProgress;
use crate::TranscodeStatus;
use crate::Transcoder;
use crate::codec::assert_unit_bounds;
pub struct TranscodeDecodeInput<I>
where
I: Input,
I::Item: Copy + Default,
{
input: BufferedInput<I>,
eof_retry_pending: bool,
}
impl<I> TranscodeDecodeInput<I>
where
I: Input,
I::Item: Copy + Default,
{
#[must_use]
#[inline]
pub fn new(inner: I) -> Self {
Self {
input: BufferedInput::new(inner),
eof_retry_pending: false,
}
}
#[must_use]
#[inline]
pub fn with_capacity(inner: I, capacity: usize) -> Self {
Self {
input: BufferedInput::with_capacity(inner, capacity),
eof_retry_pending: false,
}
}
pub fn try_with_capacity(inner: I, capacity: usize) -> StdResult<Self, TryReserveError> {
Ok(Self {
input: BufferedInput::try_with_capacity(inner, capacity)?,
eof_retry_pending: false,
})
}
#[must_use]
#[inline]
pub const fn inner(&self) -> &I {
self.input.inner()
}
#[must_use]
#[inline]
pub fn unread_len(&self) -> usize {
self.input.unread_len()
}
#[must_use]
#[inline]
pub fn unread(&self) -> &[I::Item] {
self.input.unread()
}
#[must_use]
#[inline]
pub fn capacity(&self) -> usize {
self.input.capacity()
}
pub fn fill_until(&mut self, count: usize) -> Result<bool> {
if count > self.input.capacity() {
self.input.try_reserve_capacity(count).map_err(allocation_error)?;
}
let previous_unread = self.input.unread_len();
let result = self.input.fill_until(count);
if self.input.unread_len() > previous_unread {
self.eof_retry_pending = false;
}
result
}
pub fn transcode_step<D, M, Value>(
&mut self,
decoder: &mut D,
map_error: &mut M,
output: &mut [Value],
output_index: usize,
count: usize,
) -> Result<Option<TranscodeProgress>>
where
D: Transcoder<Input = I::Item, Output = Value>,
M: FnMut(D::Error) -> Error,
{
let output_end = SliceRange::checked_range_end(
output.len(),
output_index,
count,
"decoded output range exceeds destination buffer",
)?;
if count == 0 {
return Ok(Some(TranscodeProgress::complete(0, 0)));
}
if self.unread_len() == 0 && !self.fill_more()? {
return Ok(None);
}
let available_input = self.unread_len();
let progress = decoder
.transcode(self.unread(), 0, &mut output[..output_end], output_index)
.map_err(&mut *map_error)
.and_then(|progress| validate_decode_progress(progress, 0, available_input, output_index, count))?;
self.consume(progress.read());
Ok(Some(progress))
}
pub fn transcode_eof_step<D, M, Value>(
&mut self,
decoder: &mut D,
map_error: &mut M,
output: &mut [Value],
output_index: usize,
count: usize,
) -> Result<TranscodeProgress>
where
D: Transcoder<Input = I::Item, Output = Value>,
M: FnMut(D::Error) -> Error,
{
let output_end = SliceRange::checked_range_end(
output.len(),
output_index,
count,
"decoded EOF output range exceeds destination buffer",
)?;
if count == 0 {
return Ok(TranscodeProgress::complete(0, 0));
}
let available_input = self.unread_len();
let progress = decoder
.transcode_eof(self.unread(), 0, &mut output[..output_end], output_index)
.map_err(&mut *map_error)
.and_then(|progress| validate_eof_decode_progress(progress, 0, available_input, output_index, count))?;
self.consume(progress.read());
self.eof_retry_pending = matches!(progress.status(), TranscodeStatus::NeedOutput { .. });
Ok(progress)
}
#[inline]
pub fn consume(&mut self, count: usize) {
assert!(count <= self.unread_len(), "cannot consume beyond buffered input",);
unsafe {
self.input.consume(count);
}
}
pub unsafe fn copy_unread_to(&self, output: &mut [I::Item], output_index: usize, count: usize) {
let unread = self.unread();
unsafe {
UncheckedSlice::copy_nonoverlapping(unread, 0, output, output_index, count);
}
}
#[must_use = "the returned input and unread buffer must be handled"]
#[inline]
pub fn into_parts(self) -> (I, Buffer<I::Item>) {
self.input.into_parts()
}
pub unsafe fn read_unchecked(
&mut self,
output: &mut [I::Item],
output_index: usize,
count: usize,
) -> Result<usize> {
debug_assert!(
SliceRange::range_fits(output.len(), output_index, count),
"unchecked read output range exceeds destination buffer",
);
if count == 0 {
return Ok(0);
}
self.eof_retry_pending = false;
unsafe { self.input.read_unchecked(output, output_index, count) }
}
pub fn read_decoded_with<C, M>(&mut self, codec: &mut C, map_error: M) -> Result<C::Value>
where
C: Codec<Unit = I::Item>,
M: FnMut(C::DecodeError) -> Error,
{
TranscodeFailure::ensure_no_decode_lifecycle_output::<C>()
.map_err(|error| Error::new(ErrorKind::Unsupported, error))?;
let progress = self.read_decoded_lifecycle_with_scratch_impl(codec, &mut [], &mut [], map_error)?;
let (value, reset_written, finish_written) = progress.into_parts();
debug_assert_eq!(0, reset_written);
debug_assert_eq!(0, finish_written);
Ok(value)
}
pub fn read_decoded_lifecycle_with<C, M>(
&mut self,
codec: &mut C,
map_error: M,
) -> Result<DecodeLifecycleOutput<C::Value>>
where
C: Codec<Unit = I::Item>,
C::Value: Default,
M: FnMut(C::DecodeError) -> Error,
{
assert_unit_bounds::<C>();
let mut reset_output = Vec::new();
try_reserve_vec(&mut reset_output, C::MAX_DECODE_RESET_VALUES).map_err(allocation_error)?;
reset_output.resize_with(C::MAX_DECODE_RESET_VALUES, C::Value::default);
let mut finish_output = Vec::new();
try_reserve_vec(&mut finish_output, C::MAX_DECODE_FINISH_VALUES).map_err(allocation_error)?;
finish_output.resize_with(C::MAX_DECODE_FINISH_VALUES, C::Value::default);
let progress =
self.read_decoded_lifecycle_with_scratch_impl(codec, &mut reset_output, &mut finish_output, map_error)?;
let (value, reset_written, finish_written) = progress.into_parts();
reset_output.truncate(reset_written);
finish_output.truncate(finish_written);
Ok(DecodeLifecycleOutput::new(reset_output, value, finish_output))
}
pub fn read_decoded_lifecycle_with_scratch<C, M>(
&mut self,
codec: &mut C,
reset_output: &mut [C::Value],
finish_output: &mut [C::Value],
map_error: M,
) -> Result<DecodeLifecycleProgress<C::Value>>
where
C: Codec<Unit = I::Item>,
M: FnMut(C::DecodeError) -> Error,
{
self.read_decoded_lifecycle_with_scratch_impl(codec, reset_output, finish_output, map_error)
}
pub fn reset<D, M, Value>(
&self,
decoder: &mut D,
map_error: &mut M,
output: &mut [Value],
output_index: usize,
count: usize,
) -> Result<usize>
where
D: Transcoder<Input = I::Item, Output = Value>,
M: FnMut(D::Error) -> Error,
{
let required = decoder.max_reset_output_len().map_err(capacity_error_to_invalid_data)?;
let output_end = SliceRange::checked_range_end(
output.len(),
output_index,
count,
"reset output range exceeds destination buffer",
)?;
if count < required {
return Err(Error::new(
ErrorKind::InvalidData,
"insufficient output for decoder reset bound",
));
}
let output = &mut output[..output_end];
let written = decoder.reset(output, output_index).map_err(&mut *map_error)?;
assert!(written <= required, "reset wrote beyond its bound");
Ok(written)
}
pub fn transcode<D, M, Value>(
&mut self,
decoder: &mut D,
map_error: &mut M,
output: &mut [Value],
output_index: usize,
count: usize,
) -> Result<usize>
where
D: Transcoder<Input = I::Item, Output = Value>,
M: FnMut(D::Error) -> Error,
{
let output_end = SliceRange::checked_range_end(
output.len(),
output_index,
count,
"decoded output range exceeds destination buffer",
)?;
if count == 0 {
return Ok(0);
}
if self.eof_retry_pending {
return self
.transcode_eof_step(decoder, map_error, output, output_index, count)
.map(|progress| progress.written());
}
let output = &mut output[..output_end];
let mut written_total = 0;
loop {
if self.unread_len() == 0 && !self.fill_more()? {
let remaining_output = count - written_total;
let eof_progress = self.transcode_eof_step(
decoder,
map_error,
output,
output_index + written_total,
remaining_output,
)?;
written_total += eof_progress.written();
return Ok(written_total);
}
let units = self.unread();
let available_input = units.len();
let remaining_output = count - written_total;
let progress = decoder
.transcode(units, 0, output, output_index + written_total)
.map_err(&mut *map_error)?;
let progress = validate_decode_progress(
progress,
0,
available_input,
output_index + written_total,
remaining_output,
)?;
let consumed = progress.read();
let written = progress.written();
self.consume(consumed);
written_total += written;
if written_total == count {
return Ok(written_total);
}
match progress.status() {
TranscodeStatus::Complete => {}
TranscodeStatus::NeedOutput { .. } => {
return Ok(written_total);
}
TranscodeStatus::NeedInput { required, .. } => {
if self.fill_until(required.get())? {
continue;
}
let remaining_output = count - written_total;
let eof_progress = self.transcode_eof_step(
decoder,
map_error,
output,
output_index + written_total,
remaining_output,
)?;
written_total += eof_progress.written();
return Ok(written_total);
}
}
}
}
pub fn finish<D, M, Value>(
&self,
decoder: &mut D,
map_error: &mut M,
output: &mut [Value],
output_index: usize,
count: usize,
) -> Result<usize>
where
D: Transcoder<Input = I::Item, Output = Value>,
M: FnMut(D::Error) -> Error,
{
let required = decoder
.max_finish_output_len()
.map_err(capacity_error_to_invalid_data)?;
let output_end = SliceRange::checked_range_end(
output.len(),
output_index,
count,
"finish output range exceeds destination buffer",
)?;
if count < required {
return Err(Error::new(
ErrorKind::InvalidData,
"insufficient output for decoder finish bound",
));
}
let output = &mut output[..output_end];
let written = decoder.finish(output, output_index).map_err(&mut *map_error)?;
assert!(written <= required, "finish wrote beyond its bound");
Ok(written)
}
}
impl<I> TranscodeDecodeInput<I>
where
I: SeekableByteInput,
{
#[inline]
pub fn seek(&mut self, position: SeekFrom) -> Result<u64> {
let position = self.input.seek_to(position)?;
self.eof_retry_pending = false;
Ok(position)
}
}
impl<I> TranscodeDecodeInput<I>
where
I: Input,
I::Item: Copy + Default,
{
fn read_decoded_lifecycle_with_scratch_impl<C, M>(
&mut self,
codec: &mut C,
reset_output: &mut [C::Value],
finish_output: &mut [C::Value],
mut map_error: M,
) -> Result<DecodeLifecycleProgress<C::Value>>
where
C: Codec<Unit = I::Item>,
M: FnMut(C::DecodeError) -> Error,
{
assert_unit_bounds::<C>();
if reset_output.len() < C::MAX_DECODE_RESET_VALUES {
return Err(Error::new(
ErrorKind::InvalidInput,
"decode reset output is shorter than the codec reset bound",
));
}
if finish_output.len() < C::MAX_DECODE_FINISH_VALUES {
return Err(Error::new(
ErrorKind::InvalidInput,
"decode finish output is shorter than the codec finish bound",
));
}
self.eof_retry_pending = false;
let mut map_domain_error = |error: TranscodeDomainError<C::DecodeError>| {
let contextual = error.map_source(&mut map_error);
Error::new(contextual.source().kind(), contextual)
};
let reset_written = unsafe {
codec.decode_reset(reset_output, 0)
}
.map_err(|source| map_domain_error(TranscodeDomainError::reset(source)))?;
assert!(
reset_written <= C::MAX_DECODE_RESET_VALUES,
"Codec::decode_reset wrote beyond its reset bound",
);
let value = CodecDecodeDriver::new(&mut self.input).read_one(codec, &mut |source| {
map_domain_error(TranscodeDomainError::main(source, 0))
})?;
let finish_written = unsafe {
codec.decode_finish(finish_output, 0)
}
.map_err(|source| map_domain_error(TranscodeDomainError::finish(source)))?;
assert!(
finish_written <= C::MAX_DECODE_FINISH_VALUES,
"Codec::decode_finish wrote beyond its finish bound",
);
Ok(DecodeLifecycleProgress::new(value, reset_written, finish_written))
}
#[inline]
fn fill_more(&mut self) -> Result<bool> {
let result = self.input.fill_more();
if matches!(result, Ok(true)) {
self.eof_retry_pending = false;
}
result
}
}
impl<I> Read for TranscodeDecodeInput<I>
where
I: ByteInput,
{
#[inline]
fn read(&mut self, output: &mut [u8]) -> Result<usize> {
unsafe { self.read_unchecked(output, 0, output.len()) }
}
}
impl<I> Seek for TranscodeDecodeInput<I>
where
I: SeekableByteInput,
{
#[inline]
fn seek(&mut self, position: SeekFrom) -> Result<u64> {
self.seek(position)
}
}
impl<I> Debug for TranscodeDecodeInput<I>
where
I: Input,
I::Item: Copy + Default,
BufferedInput<I>: Debug,
{
fn fmt(&self, formatter: &mut Formatter<'_>) -> FormatResult {
formatter
.debug_struct("TranscodeDecodeInput")
.field("input", &self.input)
.finish()
}
}