use core::fmt;
use std::collections::TryReserveError;
use std::io::Error;
use std::io::ErrorKind;
use std::io::Result;
use std::io::Seek;
use std::io::SeekFrom;
use std::io::Write;
use std::result::Result as CoreResult;
use qubit_io::BufferedOutput;
use qubit_io::BufferedOutputRecovery;
use qubit_io::ByteOutput;
use qubit_io::Output;
use qubit_io::SeekableByteOutput;
use qubit_utils::SliceRange;
use qubit_utils::allocation_error;
use super::internal::capacity_error_to_invalid_data;
use super::internal::map_encode_value_result;
use super::internal::validate_encode_progress;
use crate::CapacityError;
use crate::Codec;
use crate::TranscodeEncodeErrorOf;
use crate::TranscodeEncoder;
use crate::TranscodeProgress;
use crate::TranscodeStatus;
use crate::Transcoder;
use crate::value::codec_value_lifecycle::encode_complete_value_into_reserved;
use crate::value::codec_value_lifecycle::max_complete_encode_units;
#[must_use = "flush or recover pending output before dropping the adapter"]
pub struct TranscodeEncodeOutput<O>
where
O: Output,
O::Item: Copy + Default,
{
output: BufferedOutput<O>,
}
impl<O> TranscodeEncodeOutput<O>
where
O: Output,
O::Item: Copy + Default,
{
pub fn new(inner: O) -> Self {
Self {
output: BufferedOutput::new(inner),
}
}
pub fn with_capacity(inner: O, capacity: usize) -> Self {
Self {
output: BufferedOutput::with_capacity(inner, capacity),
}
}
pub fn try_with_capacity(inner: O, capacity: usize) -> CoreResult<Self, TryReserveError> {
Ok(Self {
output: BufferedOutput::try_with_capacity(inner, capacity)?,
})
}
#[must_use]
#[inline]
pub const fn inner(&self) -> &O {
self.output.inner()
}
#[must_use]
#[inline]
pub fn spare_capacity(&self) -> usize {
self.output.spare_capacity()
}
#[must_use]
#[inline]
pub fn spare_raw_parts_mut(&mut self) -> (&mut [O::Item], usize, usize) {
self.output.spare_raw_parts_mut()
}
#[inline]
pub unsafe fn advance(&mut self, count: usize) {
unsafe { self.output.advance(count) }
}
pub fn ensure_spare_capacity(&mut self, count: usize) -> Result<()> {
self.output.try_reserve_capacity(count).map_err(allocation_error)?;
self.output.ensure_spare_capacity(count)
}
#[must_use = "the returned output and recovery status must be handled"]
#[inline]
pub fn into_parts(self) -> (O, BufferedOutputRecovery<O::Item>) {
self.output.into_parts()
}
#[inline]
pub fn flush(&mut self) -> Result<()> {
self.output.flush()
}
pub fn write_encoded_with<C, M>(&mut self, codec: &mut C, value: &C::Value, mut map_error: M) -> Result<()>
where
C: Codec<Unit = O::Item>,
M: FnMut(C::EncodeError) -> Error,
{
let max_units = map_encode_value_result(
max_complete_encode_units::<C>().map_err(TranscodeEncodeErrorOf::<C>::from),
&mut map_error,
)?;
self.ensure_spare_capacity(max_units)?;
let (units, output_index, available) = self.output.spare_raw_parts_mut();
debug_assert!(
available >= max_units,
"reserved spare buffer is smaller than codec upper bound",
);
let written = map_encode_value_result(
encode_complete_value_into_reserved(codec, value, units, output_index, max_units),
&mut map_error,
)?;
unsafe {
self.output.advance(written);
}
Ok(())
}
pub fn reset<E, M, Value>(&mut self, encoder: &mut E, map_error: &mut M) -> Result<()>
where
E: Transcoder<Input = Value, Output = O::Item>,
M: FnMut(E::Error) -> Error,
{
let required = encoder.max_reset_output_len().map_err(capacity_error_to_invalid_data)?;
self.ensure_spare_capacity(required)?;
let (units, output_index, available) = self.output.spare_raw_parts_mut();
debug_assert!(
available >= required,
"insufficient reset capacity reserved in spare output buffer",
);
let written = encoder.reset(units, output_index).map_err(&mut *map_error)?;
assert!(written <= required, "reset wrote beyond its bound");
unsafe {
self.output.advance(written);
}
Ok(())
}
pub fn transcode_step<E, M, Value>(
&mut self,
encoder: &mut E,
map_error: &mut M,
input: &[Value],
input_index: usize,
count: usize,
) -> Result<TranscodeProgress>
where
E: TranscodeEncoder<Input = Value, Output = O::Item>,
M: FnMut(E::Error) -> Error,
{
let input_end = SliceRange::checked_range_end(
input.len(),
input_index,
count,
"encode input range exceeds source buffer",
)?;
if count == 0 {
return Ok(TranscodeProgress::complete(0, 0));
}
let required = encoder
.max_transcode_output_len(1)
.map_err(capacity_error_to_invalid_data)?
.max(1);
self.ensure_spare_capacity(required)?;
let (units, output_index, available_output) = self.output.spare_raw_parts_mut();
let progress = encoder
.transcode(&input[..input_end], input_index, units, output_index)
.map_err(&mut *map_error)?;
let progress = validate_encode_progress(progress, input_index, count, output_index, available_output)?;
unsafe {
self.output.advance(progress.written());
}
Ok(progress)
}
pub fn transcode<E, M, Value>(
&mut self,
encoder: &mut E,
map_error: &mut M,
input: &[Value],
input_index: usize,
count: usize,
) -> Result<usize>
where
E: TranscodeEncoder<Input = Value, Output = O::Item>,
M: FnMut(E::Error) -> Error,
{
let input_end = SliceRange::checked_range_end(
input.len(),
input_index,
count,
"encode input range exceeds source buffer",
)?;
if count == 0 {
return Ok(0);
}
let required = encoder
.max_transcode_output_len(count)
.map_err(capacity_error_to_invalid_data)?
.max(1);
self.ensure_spare_capacity(required)?;
let input = &input[..input_end];
let mut read_total = 0;
while read_total < count {
let (units, output_index, available_output) = self.output.spare_raw_parts_mut();
let remaining_input = count - read_total;
let progress = encoder
.transcode(input, input_index + read_total, units, output_index)
.map_err(&mut *map_error)?;
let progress = validate_encode_progress(
progress,
input_index + read_total,
remaining_input,
output_index,
available_output,
)?;
let read = progress.read();
let written = progress.written();
unsafe {
self.output.advance(written);
}
read_total += read;
if progress.is_complete() {
return Ok(read_total);
}
if let TranscodeStatus::NeedOutput { required, .. } = progress.status()
&& (read_total == count || available_output - written < required.get() || (read == 0 && written == 0))
{
return Err(Error::new(
ErrorKind::InvalidData,
"encoder output bound was insufficient",
));
}
}
Ok(read_total)
}
pub fn finish<E, M, Value>(&mut self, encoder: &mut E, map_error: &mut M) -> Result<()>
where
E: Transcoder<Input = Value, Output = O::Item>,
M: FnMut(E::Error) -> Error,
{
self.finish_to_buffer(encoder, map_error)?;
self.output.flush()
}
pub fn finish_to_buffer<E, M, Value>(&mut self, encoder: &mut E, map_error: &mut M) -> Result<()>
where
E: Transcoder<Input = Value, Output = O::Item>,
M: FnMut(E::Error) -> Error,
{
let required = match encoder.max_finish_output_len() {
Ok(required) => required,
Err(error) => return Err(capacity_error_to_invalid_data(error)),
};
let pending = self.output.capacity() - self.output.spare_capacity();
let capacity = pending
.checked_add(required)
.ok_or_else(|| capacity_error_to_invalid_data(CapacityError::OutputLengthOverflow))?;
self.output.try_reserve_capacity(capacity).map_err(allocation_error)?;
let (units, output_index, available) = self.output.spare_raw_parts_mut();
debug_assert!(
available >= required,
"insufficient finish capacity reserved in spare output buffer",
);
let written = encoder.finish(units, output_index).map_err(&mut *map_error)?;
assert!(written <= required, "finish wrote beyond its bound");
unsafe {
self.output.advance(written);
}
Ok(())
}
}
impl<O> TranscodeEncodeOutput<O>
where
O: SeekableByteOutput,
{
#[inline]
pub fn seek(&mut self, position: SeekFrom) -> Result<u64> {
self.output.seek_to(position)
}
}
impl<O> Write for TranscodeEncodeOutput<O>
where
O: ByteOutput,
{
#[inline]
fn write(&mut self, input: &[u8]) -> Result<usize> {
Output::write(&mut self.output, input)
}
fn write_all(&mut self, input: &[u8]) -> Result<()> {
let mut written = 0;
while written < input.len() {
let count = match Output::write(&mut self.output, &input[written..]) {
Ok(count) => count,
Err(error) if error.kind() == ErrorKind::Interrupted => continue,
Err(error) => return Err(error),
};
if count == 0 {
return Err(Error::from(ErrorKind::WriteZero));
}
assert!(
count <= input.len() - written,
"Output::write returned a count beyond the input length",
);
written += count;
}
Ok(())
}
#[inline]
fn flush(&mut self) -> Result<()> {
TranscodeEncodeOutput::flush(self)
}
}
impl<O> Seek for TranscodeEncodeOutput<O>
where
O: SeekableByteOutput,
{
#[inline]
fn seek(&mut self, position: SeekFrom) -> Result<u64> {
self.seek(position)
}
}
impl<O> fmt::Debug for TranscodeEncodeOutput<O>
where
O: Output,
O::Item: Copy + Default,
BufferedOutput<O>: fmt::Debug,
{
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("TranscodeEncodeOutput")
.field("output", &self.output)
.finish()
}
}