use core::num::NonZeroUsize;
use std::io::Error;
use std::io::ErrorKind;
use std::io::Result;
use qubit_io::BufferedInput;
use qubit_io::Input;
use qubit_utils::allocation_error;
use crate::Codec;
use crate::DecodeFailure;
pub(in crate::transcode::io) struct CodecDecodeDriver<'a, I>
where
I: Input,
I::Item: Copy + Default,
{
input: &'a mut BufferedInput<I>,
}
impl<'a, I> CodecDecodeDriver<'a, I>
where
I: Input,
I::Item: Copy + Default,
{
#[inline]
#[must_use]
pub(in crate::transcode::io) const fn new(input: &'a mut BufferedInput<I>) -> Self {
Self { input }
}
pub(in crate::transcode::io) fn read_one<C, M>(&mut self, codec: &mut C, map_error: &mut M) -> Result<C::Value>
where
C: Codec<Unit = I::Item>,
M: FnMut(C::DecodeError) -> Error,
{
let min_units_per_value = C::MIN_UNITS_PER_VALUE;
let max_units_per_value = C::MAX_DECODE_UNITS_PER_VALUE.max(min_units_per_value);
self.input
.try_reserve_capacity(min_units_per_value)
.map_err(allocation_error)?;
loop {
let (available, end_of_input) = self.prepare_buffered_window(min_units_per_value, max_units_per_value)?;
let units = &self.input.unread()[..available];
debug_assert!(units.len() >= min_units_per_value);
let decode_result = Self::decode_window(codec, units, end_of_input);
match decode_result {
Ok((value, consumed)) => {
return self.accept(value, consumed, available);
}
Err(DecodeFailure::Incomplete { source, required_total }) => {
assert!(
required_total.get() <= C::MAX_DECODE_UNITS_PER_VALUE,
"Codec::decode incomplete required_total exceeded Codec::MAX_DECODE_UNITS_PER_VALUE",
);
if end_of_input {
return self.handle_eof_incomplete::<C, M>(source, required_total, map_error);
}
if !self.refill_after_incomplete(required_total, available)? {
return self.retry_decode_at_eof::<C, M>(codec, map_error);
}
}
Err(DecodeFailure::Invalid { source, consumed }) => {
return self.reject::<C, M>(source, consumed, available, map_error);
}
}
}
}
fn decode_window<C>(
codec: &mut C,
units: &[I::Item],
end_of_input: bool,
) -> core::result::Result<(C::Value, NonZeroUsize), DecodeFailure<C::DecodeError>>
where
C: Codec<Unit = I::Item>,
{
unsafe {
if end_of_input {
codec.decode_eof(units, 0)
} else {
codec.decode(units, 0)
}
}
}
fn retry_decode_at_eof<C, M>(&mut self, codec: &mut C, map_error: &mut M) -> Result<C::Value>
where
C: Codec<Unit = I::Item>,
M: FnMut(C::DecodeError) -> Error,
{
let available = self.input.unread_len();
let units = &self.input.unread()[..available];
let eof_result = unsafe { codec.decode_eof(units, 0) };
match eof_result {
Ok((value, consumed)) => self.accept(value, consumed, available),
Err(DecodeFailure::Invalid { source, consumed }) => {
self.reject::<C, M>(source, consumed, available, map_error)
}
Err(DecodeFailure::Incomplete { source, required_total }) => {
self.handle_eof_incomplete::<C, M>(source, required_total, map_error)
}
}
}
fn prepare_buffered_window(
&mut self,
min_units_per_value: usize,
max_units_per_value: usize,
) -> Result<(usize, bool)> {
let mut end_of_input = false;
let available = self.input.unread_len();
if available < min_units_per_value && !self.input.fill_until(min_units_per_value)? {
end_of_input = true;
}
if self.input.unread_len() < min_units_per_value {
let available = self.input.unread_len();
unsafe {
self.input.consume(available);
}
return Err(Error::new(ErrorKind::UnexpectedEof, "failed to decode complete value"));
}
Ok((self.input.unread_len().min(max_units_per_value), end_of_input))
}
fn accept<Value>(&mut self, value: Value, consumed: NonZeroUsize, available: usize) -> Result<Value> {
if consumed.get() > available {
return Err(Error::new(
ErrorKind::InvalidData,
"codec consumed units exceed unread window",
));
}
unsafe {
self.input.consume(consumed.get());
}
Ok(value)
}
fn refill_after_incomplete(&mut self, required_total: NonZeroUsize, available: usize) -> Result<bool> {
let required_total = required_total.get();
if available >= required_total {
return Err(Error::new(
ErrorKind::InvalidData,
"codec reported incomplete input within available window",
));
}
self.input
.try_reserve_capacity(required_total)
.map_err(allocation_error)?;
self.input.fill_until(required_total)
}
fn handle_eof_incomplete<C, M>(
&mut self,
source: Option<C::DecodeError>,
required_total: NonZeroUsize,
map_error: &mut M,
) -> Result<C::Value>
where
C: Codec<Unit = I::Item>,
M: FnMut(C::DecodeError) -> Error,
{
assert!(
required_total.get() <= C::MAX_DECODE_UNITS_PER_VALUE,
"Codec::decode_eof incomplete required_total exceeded Codec::MAX_DECODE_UNITS_PER_VALUE",
);
let available = self.input.unread_len();
if required_total.get() <= available {
return Err(Error::new(
ErrorKind::InvalidData,
"codec reported incomplete input within available window",
));
}
unsafe {
self.input.consume(available);
}
match source {
Some(source) => Err(map_error(source)),
None => Err(Error::new(ErrorKind::UnexpectedEof, "failed to decode complete value")),
}
}
fn reject<C, M>(
&mut self,
source: C::DecodeError,
consumed: Option<NonZeroUsize>,
available: usize,
map_error: &mut M,
) -> Result<C::Value>
where
C: Codec<Unit = I::Item>,
M: FnMut(C::DecodeError) -> Error,
{
if let Some(consumed) = consumed {
if consumed.get() > available {
return Err(Error::new(
ErrorKind::InvalidData,
"decode error consumed units exceed unread window",
));
}
unsafe {
self.input.consume(consumed.get());
}
}
Err(map_error(source))
}
}