qubit_codec/transcode/io/transcode_encode_output.rs
1// =============================================================================
2// Copyright (c) 2026 Haixing Hu.
3//
4// SPDX-License-Identifier: Apache-2.0
5//
6// Licensed under the Apache License, Version 2.0.
7// =============================================================================
8//! Buffered output driver that encodes values into units.
9
10use core::fmt;
11use std::collections::TryReserveError;
12use std::io::Error;
13use std::io::ErrorKind;
14use std::io::Result;
15use std::io::Seek;
16use std::io::SeekFrom;
17use std::io::Write;
18use std::result::Result as CoreResult;
19
20use qubit_io::BufferedOutput;
21use qubit_io::BufferedOutputRecovery;
22use qubit_io::ByteOutput;
23use qubit_io::Output;
24use qubit_io::SeekableByteOutput;
25use qubit_utils::SliceRange;
26use qubit_utils::allocation_error;
27
28use super::internal::capacity_error_to_invalid_data;
29use super::internal::map_encode_value_result;
30use super::internal::validate_encode_progress;
31use crate::CapacityError;
32use crate::Codec;
33use crate::TranscodeEncodeErrorOf;
34use crate::TranscodeEncoder;
35use crate::TranscodeProgress;
36use crate::TranscodeStatus;
37use crate::Transcoder;
38use crate::value::codec_value_lifecycle::encode_complete_value_into_reserved;
39use crate::value::codec_value_lifecycle::max_complete_encode_units;
40
41/// Encodes an [`Output`] value stream into an [`Output`] unit stream.
42///
43/// This type owns only the unit-level [`qubit_io::BufferedOutput`]. Callers
44/// pass a [`crate::Codec`] and error mapper to each encode operation, which
45/// lets one buffered output drive different encoders without nesting buffers or
46/// storing codec-specific state in the buffer owner.
47///
48/// [`Self::flush`] only drains already buffered units. State-aware streaming
49/// encoders can use [`Self::reset`], [`Self::transcode`], and [`Self::finish`]
50/// explicitly.
51///
52/// Pending units should be flushed or recovered before this adapter is
53/// dropped. Its wrapped buffer only makes a best-effort drain during drop,
54/// skips that drain after an inner I/O panic, and ignores any output error.
55///
56/// # Type Parameters
57///
58/// * `O` - Wrapped unit output.
59///
60/// # Examples
61///
62/// ```
63/// use core::convert::Infallible;
64/// use core::num::NonZeroUsize;
65/// use std::io::Cursor;
66/// use qubit_codec::{Codec, CodecTranscodeEncoder, DecodeFailure, TranscodeEncodeOutput};
67/// use qubit_codec::{TranscodeEncoder, TranscodeStatus};
68///
69/// struct ByteCodec;
70/// impl Codec for ByteCodec {
71/// type Value = u8;
72/// type Unit = u8;
73/// type DecodeError = Infallible;
74/// type EncodeError = Infallible;
75/// const MIN_UNITS_PER_VALUE: usize = 1;
76/// const MAX_DECODE_UNITS_PER_VALUE: usize = 1;
77/// const MAX_ENCODE_UNITS_PER_VALUE: usize = 1;
78/// unsafe fn decode(&mut self, input: &[u8], index: usize)
79/// -> Result<(u8, NonZeroUsize), DecodeFailure<Infallible>> {
80/// Ok((input[index], NonZeroUsize::MIN))
81/// }
82/// unsafe fn encode(&mut self, value: &u8, output: &mut [u8], index: usize)
83/// -> Result<usize, Infallible> {
84/// output[index] = *value;
85/// Ok(1)
86/// }
87/// }
88///
89/// let mut encoder = CodecTranscodeEncoder::new(ByteCodec);
90/// let mut output = TranscodeEncodeOutput::with_capacity(Cursor::new(Vec::new()), 8);
91/// let mut map_error = |_| std::io::Error::other("encode error");
92/// output.reset(&mut encoder, &mut map_error)?;
93/// let values = [1_u8, 2, 3];
94/// let mut offset = 0;
95/// while offset < values.len() {
96/// let progress = output.transcode_step(
97/// &mut encoder, &mut map_error, &values, offset, values.len() - offset,
98/// )?;
99/// offset += progress.read();
100/// if let TranscodeStatus::NeedOutput { required, .. } = progress.status() {
101/// output.ensure_spare_capacity(required.get())?;
102/// }
103/// }
104/// output.finish(&mut encoder, &mut map_error)?;
105/// assert_eq!(output.inner().get_ref(), &[1, 2, 3]);
106/// # Ok::<(), std::io::Error>(())
107/// ```
108///
109/// `finish` commits final units, drains the buffered output, and flushes the
110/// wrapped sink. Use `finish_to_buffer` followed by `flush` when those steps
111/// need to be controlled separately.
112#[must_use = "flush or recover pending output before dropping the adapter"]
113pub struct TranscodeEncodeOutput<O>
114where
115 O: Output,
116 O::Item: Copy + Default,
117{
118 /// Owns the wrapped unit sink and pending encoded units awaiting delivery.
119 output: BufferedOutput<O>,
120}
121
122impl<O> TranscodeEncodeOutput<O>
123where
124 O: Output,
125 O::Item: Copy + Default,
126{
127 /// Creates an encoder output with the default unit buffer capacity.
128 ///
129 /// # Parameters
130 ///
131 /// * `inner` - Unit output written by this adapter.
132 ///
133 /// # Returns
134 ///
135 /// A new buffered encoder output.
136 ///
137 /// # Panics
138 ///
139 /// Panics if initializing the default buffer requires
140 /// `O::Item::default()` or `O::Item::clone()` and either operation panics,
141 /// or if the default backing length exceeds `Vec`'s supported capacity.
142 pub fn new(inner: O) -> Self {
143 Self {
144 output: BufferedOutput::new(inner),
145 }
146 }
147
148 /// Creates an encoder output with a unit buffer of at least `capacity`.
149 ///
150 /// # Parameters
151 ///
152 /// * `inner` - Unit output written by this adapter.
153 /// * `capacity` - Requested internal unit buffer capacity.
154 ///
155 /// # Returns
156 ///
157 /// A new buffered encoder output.
158 ///
159 /// # Panics
160 ///
161 /// Panics if initializing the buffer requires `O::Item::default()` or
162 /// `O::Item::clone()` and either operation panics, or if the requested
163 /// backing length exceeds `Vec`'s supported capacity.
164 pub fn with_capacity(inner: O, capacity: usize) -> Self {
165 Self {
166 output: BufferedOutput::with_capacity(inner, capacity),
167 }
168 }
169
170 /// Creates an encoder output with a unit buffer of at least `capacity`.
171 ///
172 /// # Parameters
173 ///
174 /// * `inner` - Unit output written by this adapter.
175 /// * `capacity` - Requested internal unit buffer capacity.
176 ///
177 /// # Returns
178 ///
179 /// A buffered adapter owning `inner` on success.
180 ///
181 /// # Errors
182 ///
183 /// Returns an allocation error when the requested buffer cannot be
184 /// allocated.
185 ///
186 /// # Panics
187 ///
188 /// Panics if initializing the buffer requires `O::Item::default()` or
189 /// `O::Item::clone()` and either operation panics. Allocation failures are
190 /// returned as errors; panics from element initialization are not caught.
191 pub fn try_with_capacity(inner: O, capacity: usize) -> CoreResult<Self, TryReserveError> {
192 Ok(Self {
193 output: BufferedOutput::try_with_capacity(inner, capacity)?,
194 })
195 }
196
197 /// Returns a shared reference to the wrapped unit output.
198 ///
199 /// Pending units remain in this adapter's internal buffer and are not
200 /// visible through the returned output until [`Self::flush`] succeeds.
201 ///
202 /// # Returns
203 ///
204 /// A shared reference to the wrapped unit output.
205 #[must_use]
206 #[inline]
207 pub const fn inner(&self) -> &O {
208 self.output.inner()
209 }
210
211 /// Returns the available capacity of the spare output buffer.
212 ///
213 /// # Returns
214 ///
215 /// The number of output units that can still be appended without flushing.
216 #[must_use]
217 #[inline]
218 pub fn spare_capacity(&self) -> usize {
219 self.output.spare_capacity()
220 }
221
222 /// Returns raw spare-buffer parts for the internal output buffer.
223 ///
224 /// # Returns
225 ///
226 /// The full backing storage, the spare start index, and the spare unit
227 /// count.
228 #[must_use]
229 #[inline]
230 pub fn spare_raw_parts_mut(&mut self) -> (&mut [O::Item], usize, usize) {
231 self.output.spare_raw_parts_mut()
232 }
233
234 /// Marks `count` units from [`Self::spare_raw_parts_mut`] as written.
235 ///
236 /// # Parameters
237 ///
238 /// * `count` - Initialized units to commit from the spare window.
239 ///
240 /// # Safety
241 ///
242 /// The caller must guarantee that `count <= Self::spare_capacity()` and
243 /// that the corresponding units in the returned spare slice have been
244 /// initialized.
245 #[inline]
246 pub unsafe fn advance(&mut self, count: usize) {
247 // SAFETY: The caller guarantees `count` and initialization invariants.
248 unsafe { self.output.advance(count) }
249 }
250
251 /// Ensures that at least `count` spare units are available.
252 ///
253 /// # Parameters
254 ///
255 /// * `count` - Number of spare units required.
256 ///
257 /// # Errors
258 ///
259 /// Returns allocation errors mapped to [`ErrorKind::OutOfMemory`], or I/O
260 /// errors from the wrapped output while delivering pending units.
261 ///
262 /// # Panics
263 ///
264 /// Panics if growing the buffer requires `O::Item::default()` or
265 /// `O::Item::clone()` and either operation panics.
266 pub fn ensure_spare_capacity(&mut self, count: usize) -> Result<()> {
267 self.output.try_reserve_capacity(count).map_err(allocation_error)?;
268 self.output.ensure_spare_capacity(count)
269 }
270
271 /// Consumes this adapter without flushing the wrapped output.
272 ///
273 /// This method does not call [`Self::flush`] and performs no I/O. `Pending`
274 /// units were not reported as written and may be retried. `Indeterminate`
275 /// means an inner write panicked and may have produced an external prefix;
276 /// do not replay it automatically.
277 ///
278 /// # Returns
279 ///
280 /// The wrapped output and pending or indeterminate units.
281 #[must_use = "the returned output and recovery status must be handled"]
282 #[inline]
283 pub fn into_parts(self) -> (O, BufferedOutputRecovery<O::Item>) {
284 self.output.into_parts()
285 }
286
287 /// Flushes buffered units without finishing any encoder stream.
288 ///
289 /// # Errors
290 ///
291 /// Returns errors from the wrapped output while flushing pending units.
292 #[inline]
293 pub fn flush(&mut self) -> Result<()> {
294 self.output.flush()
295 }
296
297 /// Encodes one codec value into this buffered unit output.
298 ///
299 /// The method grows the persistent internal buffer when necessary, then
300 /// writes the complete encoded value into its spare window.
301 ///
302 /// # Type Parameters
303 ///
304 /// * `C` - Codec producing the wrapped output unit type.
305 /// * `M` - Mapper from codec errors into I/O errors.
306 ///
307 /// # Parameters
308 ///
309 /// * `codec` - Codec used for this single-value encode.
310 /// * `value` - Value to encode.
311 /// * `map_error` - Mapper for codec-domain encode errors.
312 ///
313 /// # Errors
314 ///
315 /// Returns I/O errors from the wrapped output, `InvalidInput` when the
316 /// codec output bound overflows or the value is outside the codec domain,
317 /// or the error returned by `map_error` for codec encode, reset, and finish
318 /// failures. A mapped codec error is wrapped in
319 /// [`crate::TranscodeDomainError`] so its reset, main, or finish phase
320 /// remains available through the returned I/O error's source chain. The
321 /// mapped error itself is retained as that wrapper's source, including any
322 /// custom payload or OS error code.
323 ///
324 /// Capacity planning fails before the codec runs. Preparing the spare
325 /// buffer can first deliver older pending units to the wrapped output; it
326 /// is not a transaction boundary for the whole call. If reset, encode, or
327 /// finish fails, the codec may already have changed state, but units
328 /// written into this call's spare window are not committed to the
329 /// adapter. Older pending units remain subject to the wrapped output's
330 /// buffering contract, and units delivered while preparing capacity
331 /// cannot be rolled back. A successful call commits the actual written
332 /// count to pending output but does not flush the sink. Do not replay
333 /// the value after an error unless the codec provides a recovery
334 /// protocol; reset does not undo state changes or output already
335 /// delivered.
336 ///
337 /// Callers that previously downcast the outer I/O error directly to their
338 /// mapped payload should first downcast it to
339 /// `TranscodeDomainError<std::io::Error>`, then inspect its `source()`.
340 /// For mapped OS errors, `raw_os_error()` is available on that inner
341 /// source.
342 ///
343 /// # Panics
344 ///
345 /// Panics when the codec violates its declared reset, value-width, or
346 /// finish bounds, or when `encode` writes a length different from
347 /// `encode_len` in the same reset state. Also panics if growing the buffer
348 /// requires `O::Item::default()` or `O::Item::clone()` and either
349 /// operation panics.
350 pub fn write_encoded_with<C, M>(&mut self, codec: &mut C, value: &C::Value, mut map_error: M) -> Result<()>
351 where
352 C: Codec<Unit = O::Item>,
353 M: FnMut(C::EncodeError) -> Error,
354 {
355 let max_units = map_encode_value_result(
356 max_complete_encode_units::<C>().map_err(TranscodeEncodeErrorOf::<C>::from),
357 &mut map_error,
358 )?;
359 self.ensure_spare_capacity(max_units)?;
360 let (units, output_index, available) = self.output.spare_raw_parts_mut();
361 debug_assert!(
362 available >= max_units,
363 "reserved spare buffer is smaller than codec upper bound",
364 );
365 let written = map_encode_value_result(
366 encode_complete_value_into_reserved(codec, value, units, output_index, max_units),
367 &mut map_error,
368 )?;
369 // SAFETY: The spare buffer has the conservative complete lifecycle
370 // capacity, and the helper reports how many initialized units it
371 // actually wrote.
372 unsafe {
373 self.output.advance(written);
374 }
375 Ok(())
376 }
377
378 /// Runs encoder reset and buffers any stream-prefix units.
379 ///
380 /// This method reserves enough persistent buffer capacity for all reset
381 /// output before calling `encoder`. Preparing that space may deliver
382 /// previously buffered units to the wrapped output and can return an I/O
383 /// error before `encoder.reset` runs. This delivery does not call the
384 /// wrapped output's `flush` method.
385 ///
386 /// # Type Parameters
387 ///
388 /// * `E` - Transcoder producing the wrapped output unit type.
389 /// * `M` - Mapper from transcoder errors into I/O errors.
390 /// * `Value` - Source value type accepted by the transcoder.
391 ///
392 /// # Parameters
393 ///
394 /// * `encoder` - Transcoder whose stream prefix is being collected.
395 /// * `map_error` - Mapper for reset failures.
396 ///
397 /// # Errors
398 ///
399 /// Returns capacity errors, allocation errors, or reset errors mapped by
400 /// `map_error`.
401 ///
402 /// # Panics
403 ///
404 /// Panics when `encoder` writes more than its declared reset bound.
405 /// Also panics if growing the buffer requires `O::Item::default()` or
406 /// `O::Item::clone()` and either operation panics.
407 pub fn reset<E, M, Value>(&mut self, encoder: &mut E, map_error: &mut M) -> Result<()>
408 where
409 E: Transcoder<Input = Value, Output = O::Item>,
410 M: FnMut(E::Error) -> Error,
411 {
412 let required = encoder.max_reset_output_len().map_err(capacity_error_to_invalid_data)?;
413 self.ensure_spare_capacity(required)?;
414 let (units, output_index, available) = self.output.spare_raw_parts_mut();
415 debug_assert!(
416 available >= required,
417 "insufficient reset capacity reserved in spare output buffer",
418 );
419 let written = encoder.reset(units, output_index).map_err(&mut *map_error)?;
420 assert!(written <= required, "reset wrote beyond its bound");
421 // SAFETY: The encoder reported initialized units within the spare
422 // range reserved above.
423 unsafe {
424 self.output.advance(written);
425 }
426 Ok(())
427 }
428
429 /// Encodes at most one step of the selected input range.
430 ///
431 /// Unlike [`Self::transcode`], this method does not reserve output space
432 /// for the entire range and calls the encoder exactly once. The returned
433 /// progress is relative to this call's input and output windows. Advance
434 /// the input index by [`TranscodeProgress::read`]; if the status is
435 /// `NeedOutput`, provide the requested spare capacity before retrying.
436 /// Written units have already been committed to this adapter's pending
437 /// buffer and must not be advanced again. A zero count validates the range
438 /// and returns complete progress without delivering pending units or
439 /// calling the encoder.
440 ///
441 /// # Type Parameters
442 ///
443 /// * `E` - Encoder producing the wrapped output unit type.
444 /// * `M` - Mapper from encoder errors into I/O errors.
445 /// * `Value` - Source value type accepted by the encoder.
446 ///
447 /// # Parameters
448 ///
449 /// * `encoder` - Encoder that processes one progress step.
450 /// * `map_error` - Maps encoder errors to I/O errors.
451 /// * `input` - Source values.
452 /// * `input_index` - Start index inside `input`.
453 /// * `count` - Number of values available from `input_index`.
454 ///
455 /// # Returns
456 ///
457 /// Progress for the single encoder call, including consumed input,
458 /// committed output, and whether more output capacity is needed.
459 ///
460 /// # Errors
461 ///
462 /// Returns invalid input ranges, capacity, encoder, or output errors. An
463 /// invalid encoder progress report is returned as `ErrorKind::InvalidData`.
464 /// An error does not provide general transactional rollback or complete
465 /// progress. The encoder may already have advanced, but units written by
466 /// its failing call are not committed to this adapter's pending output.
467 /// Do not replay the original input range without an encoder-specific
468 /// recovery protocol. Earlier pending units retain the wrapped output's
469 /// recovery contract; reset starts a new logical stream and does not undo
470 /// output already delivered.
471 ///
472 /// # Panics
473 ///
474 /// Panics if growing the buffer requires `O::Item::default()` or
475 /// `O::Item::clone()` and either operation panics.
476 ///
477 /// # Examples
478 ///
479 /// ```
480 /// use std::io::Cursor;
481 ///
482 /// # use core::convert::Infallible;
483 /// # use core::num::NonZeroUsize;
484 /// # use qubit_codec::Codec;
485 /// # use qubit_codec::CodecTranscodeEncoder;
486 /// # use qubit_codec::DecodeFailure;
487 /// use qubit_codec::TranscodeEncodeOutput;
488 /// use qubit_codec::TranscodeEncoder;
489 /// use qubit_codec::TranscodeStatus;
490 ///
491 /// # struct ByteCodec;
492 /// # impl Codec for ByteCodec {
493 /// # type Value = u8;
494 /// # type Unit = u8;
495 /// # type DecodeError = Infallible;
496 /// # type EncodeError = Infallible;
497 /// # const MIN_UNITS_PER_VALUE: usize = 1;
498 /// # const MAX_ENCODE_UNITS_PER_VALUE: usize = 1;
499 /// # const MAX_DECODE_UNITS_PER_VALUE: usize = 1;
500 /// # unsafe fn decode(
501 /// # &mut self,
502 /// # input: &[u8],
503 /// # index: usize,
504 /// # ) -> Result<(u8, NonZeroUsize), DecodeFailure<Infallible>> {
505 /// # Ok((input[index], NonZeroUsize::MIN))
506 /// # }
507 /// # unsafe fn encode(
508 /// # &mut self,
509 /// # value: &u8,
510 /// # output: &mut [u8],
511 /// # index: usize,
512 /// # ) -> Result<usize, Infallible> {
513 /// # output[index] = *value;
514 /// # Ok(1)
515 /// # }
516 /// # }
517 /// fn encode<E>(encoder: &mut E, values: &[E::Input]) -> std::io::Result<Vec<u8>>
518 /// where
519 /// E: TranscodeEncoder<Output = u8>,
520 /// {
521 /// let mut output = TranscodeEncodeOutput::with_capacity(Cursor::new(Vec::new()), 8);
522 /// let mut map_error = |_| std::io::Error::other("encode error");
523 /// output.reset(encoder, &mut map_error)?;
524 /// let mut offset = 0;
525 /// while offset < values.len() {
526 /// let progress = output.transcode_step(
527 /// encoder,
528 /// &mut map_error,
529 /// values,
530 /// offset,
531 /// values.len() - offset,
532 /// )?;
533 /// offset += progress.read();
534 /// if let TranscodeStatus::NeedOutput { required, .. } = progress.status() {
535 /// output.ensure_spare_capacity(required.get())?;
536 /// }
537 /// }
538 /// output.finish(encoder, &mut map_error)?;
539 /// let (cursor, _) = output.into_parts();
540 /// Ok(cursor.into_inner())
541 /// }
542 ///
543 /// let mut encoder = CodecTranscodeEncoder::new(ByteCodec);
544 /// assert_eq!(encode(&mut encoder, &[1, 2, 3]).unwrap(), vec![1, 2, 3]);
545 /// let mut empty_encoder = CodecTranscodeEncoder::new(ByteCodec);
546 /// assert_eq!(encode(&mut empty_encoder, &[]).unwrap(), Vec::<u8>::new());
547 /// ```
548 pub fn transcode_step<E, M, Value>(
549 &mut self,
550 encoder: &mut E,
551 map_error: &mut M,
552 input: &[Value],
553 input_index: usize,
554 count: usize,
555 ) -> Result<TranscodeProgress>
556 where
557 E: TranscodeEncoder<Input = Value, Output = O::Item>,
558 M: FnMut(E::Error) -> Error,
559 {
560 let input_end = SliceRange::checked_range_end(
561 input.len(),
562 input_index,
563 count,
564 "encode input range exceeds source buffer",
565 )?;
566 if count == 0 {
567 return Ok(TranscodeProgress::complete(0, 0));
568 }
569 let required = encoder
570 .max_transcode_output_len(1)
571 .map_err(capacity_error_to_invalid_data)?
572 .max(1);
573 self.ensure_spare_capacity(required)?;
574 let (units, output_index, available_output) = self.output.spare_raw_parts_mut();
575 let progress = encoder
576 .transcode(&input[..input_end], input_index, units, output_index)
577 .map_err(&mut *map_error)?;
578 let progress = validate_encode_progress(progress, input_index, count, output_index, available_output)?;
579 // SAFETY: Validated progress fits the spare output window.
580 unsafe {
581 self.output.advance(progress.written());
582 }
583 Ok(progress)
584 }
585
586 /// Encodes values from an indexed input range using a streaming
587 /// [`Transcoder`].
588 ///
589 /// This streaming path reserves the encoder's conservative output bound
590 /// for the whole batch before calling it. Preparing that space may deliver
591 /// previously buffered units to the wrapped output and can return an I/O
592 /// error before consuming source values. Split large batches into smaller
593 /// calls when you need to limit the memory required for one operation.
594 ///
595 /// # Type Parameters
596 ///
597 /// * `E` - Encoder producing the wrapped output unit type.
598 /// * `M` - Mapper from encoder errors into I/O errors.
599 /// * `Value` - Source value type accepted by the encoder.
600 ///
601 /// # Parameters
602 ///
603 /// * `encoder` - Streaming encoder used for this operation.
604 /// * `map_error` - Function mapping transcode errors into I/O errors.
605 /// * `input` - Source values.
606 /// * `input_index` - Start index inside `input`.
607 /// * `count` - Maximum number of values to encode.
608 ///
609 /// # Returns
610 ///
611 /// The number of source values consumed.
612 ///
613 /// # Errors
614 ///
615 /// Returns invalid input ranges, capacity, transcode, or output errors. An
616 /// output-bound contract violation is reported as `ErrorKind::InvalidData`.
617 /// An error does not provide general transactional rollback or complete
618 /// progress. The encoder may already have advanced, while units written by
619 /// its failing call are not committed to this adapter's pending output.
620 /// Do not replay the original input range unless the concrete encoder
621 /// provides an explicit recovery protocol. Previously committed pending
622 /// output remains subject to the wrapped output's recovery contract, and
623 /// reset starts a new logical stream without undoing output already
624 /// delivered.
625 ///
626 /// # Panics
627 ///
628 /// Panics if growing the buffer requires `O::Item::default()` or
629 /// `O::Item::clone()` and either operation panics.
630 ///
631 /// # Examples
632 ///
633 /// A value that only implements [`Transcoder`] is not an encoder and cannot
634 /// be passed to this method.
635 ///
636 /// ```compile_fail
637 /// use std::io::Error;
638 /// use std::io::sink;
639 ///
640 /// use qubit_codec::CapacityError;
641 /// use qubit_codec::TranscodeEncodeError;
642 /// use qubit_codec::TranscodeEncodeOutput;
643 /// use qubit_codec::TranscodeProgress;
644 /// use qubit_codec::Transcoder;
645 ///
646 /// struct GenericTranscoder;
647 ///
648 /// impl Transcoder for GenericTranscoder {
649 /// type Input = u8;
650 /// type Output = u8;
651 /// type Error = TranscodeEncodeError<(), u8>;
652 ///
653 /// fn max_transcode_output_len(
654 /// &self,
655 /// input_len: usize,
656 /// ) -> Result<usize, CapacityError> {
657 /// Ok(input_len)
658 /// }
659 ///
660 /// fn reset(
661 /// &mut self,
662 /// _output: &mut [u8],
663 /// _output_index: usize,
664 /// ) -> Result<usize, Self::Error> {
665 /// Ok(0)
666 /// }
667 ///
668 /// fn transcode(
669 /// &mut self,
670 /// input: &[u8],
671 /// input_index: usize,
672 /// _output: &mut [u8],
673 /// _output_index: usize,
674 /// ) -> Result<TranscodeProgress, Self::Error> {
675 /// Ok(TranscodeProgress::complete(input.len() - input_index, 0))
676 /// }
677 ///
678 /// fn finish(
679 /// &mut self,
680 /// _output: &mut [u8],
681 /// _output_index: usize,
682 /// ) -> Result<usize, Self::Error> {
683 /// Ok(0)
684 /// }
685 /// }
686 ///
687 /// let mut output = TranscodeEncodeOutput::with_capacity(sink(), 1);
688 /// let mut transcoder = GenericTranscoder;
689 /// let mut map_error = |_| Error::other("transcode error");
690 /// let _ = output.transcode(
691 /// &mut transcoder,
692 /// &mut map_error,
693 /// &[0_u8],
694 /// 0,
695 /// 1,
696 /// );
697 /// ```
698 pub fn transcode<E, M, Value>(
699 &mut self,
700 encoder: &mut E,
701 map_error: &mut M,
702 input: &[Value],
703 input_index: usize,
704 count: usize,
705 ) -> Result<usize>
706 where
707 E: TranscodeEncoder<Input = Value, Output = O::Item>,
708 M: FnMut(E::Error) -> Error,
709 {
710 let input_end = SliceRange::checked_range_end(
711 input.len(),
712 input_index,
713 count,
714 "encode input range exceeds source buffer",
715 )?;
716 if count == 0 {
717 return Ok(0);
718 }
719 let required = encoder
720 .max_transcode_output_len(count)
721 .map_err(capacity_error_to_invalid_data)?
722 .max(1);
723 self.ensure_spare_capacity(required)?;
724 let input = &input[..input_end];
725 let mut read_total = 0;
726 while read_total < count {
727 let (units, output_index, available_output) = self.output.spare_raw_parts_mut();
728 let remaining_input = count - read_total;
729 let progress = encoder
730 .transcode(input, input_index + read_total, units, output_index)
731 .map_err(&mut *map_error)?;
732 let progress = validate_encode_progress(
733 progress,
734 input_index + read_total,
735 remaining_input,
736 output_index,
737 available_output,
738 )?;
739 let read = progress.read();
740 let written = progress.written();
741 // SAFETY: The progress bounds check above proved that the encoder
742 // initialized no more than the available spare output window.
743 unsafe {
744 self.output.advance(written);
745 }
746 read_total += read;
747 if progress.is_complete() {
748 return Ok(read_total);
749 }
750 if let TranscodeStatus::NeedOutput { required, .. } = progress.status()
751 && (read_total == count || available_output - written < required.get() || (read == 0 && written == 0))
752 {
753 return Err(Error::new(
754 ErrorKind::InvalidData,
755 "encoder output bound was insufficient",
756 ));
757 }
758 }
759 Ok(read_total)
760 }
761
762 /// Finishes the encoder and flushes the wrapped unit output.
763 ///
764 /// This method writes final units directly into the internal spare buffer,
765 /// growing it when [`Transcoder::max_finish_output_len`] exceeds the
766 /// current capacity.
767 ///
768 /// # Type Parameters
769 ///
770 /// * `E` - Transcoder producing the wrapped output unit type.
771 /// * `M` - Mapper from transcoder errors into I/O errors.
772 /// * `Value` - Source value type accepted by the transcoder.
773 ///
774 /// # Parameters
775 ///
776 /// * `encoder` - Encoder whose final units are being collected.
777 /// * `map_error` - Function mapping transcode errors into I/O errors.
778 ///
779 /// # Errors
780 ///
781 /// Returns capacity, transcode finalization, or wrapped output flush
782 /// errors.
783 ///
784 /// # Panics
785 ///
786 /// Panics when the encoder exceeds its declared finish-output bound.
787 /// Also panics if growing the buffer requires `O::Item::default()` or
788 /// `O::Item::clone()` and either operation panics.
789 pub fn finish<E, M, Value>(&mut self, encoder: &mut E, map_error: &mut M) -> Result<()>
790 where
791 E: Transcoder<Input = Value, Output = O::Item>,
792 M: FnMut(E::Error) -> Error,
793 {
794 self.finish_to_buffer(encoder, map_error)?;
795 self.output.flush()
796 }
797
798 /// Finishes the encoder while retaining final units in the output buffer.
799 ///
800 /// This method separates encoder finalization from output delivery. It is
801 /// useful when a caller must record successful finalization before a later
802 /// flush can fail. Call [`Self::flush`] to deliver the retained units.
803 ///
804 /// # Type Parameters
805 ///
806 /// * `E` - Transcoder producing the wrapped output unit type.
807 /// * `M` - Mapper from transcoder errors into I/O errors.
808 /// * `Value` - Source value type accepted by the transcoder.
809 ///
810 /// # Parameters
811 ///
812 /// * `encoder` - Encoder whose final units are being collected.
813 /// * `map_error` - Function mapping transcode errors into I/O errors.
814 ///
815 /// # Errors
816 ///
817 /// Returns capacity planning, allocation, or transcoder finalization
818 /// errors. It does not perform output I/O; use [`Self::finish`] to deliver
819 /// the buffered final units and flush the wrapped output.
820 ///
821 /// # Panics
822 ///
823 /// Panics when `encoder` writes more units than its declared finish bound.
824 /// Also panics if growing the buffer requires `O::Item::default()` or
825 /// `O::Item::clone()` and either operation panics.
826 pub fn finish_to_buffer<E, M, Value>(&mut self, encoder: &mut E, map_error: &mut M) -> Result<()>
827 where
828 E: Transcoder<Input = Value, Output = O::Item>,
829 M: FnMut(E::Error) -> Error,
830 {
831 let required = match encoder.max_finish_output_len() {
832 Ok(required) => required,
833 Err(error) => return Err(capacity_error_to_invalid_data(error)),
834 };
835 let pending = self.output.capacity() - self.output.spare_capacity();
836 let capacity = pending
837 .checked_add(required)
838 .ok_or_else(|| capacity_error_to_invalid_data(CapacityError::OutputLengthOverflow))?;
839 self.output.try_reserve_capacity(capacity).map_err(allocation_error)?;
840 let (units, output_index, available) = self.output.spare_raw_parts_mut();
841 debug_assert!(
842 available >= required,
843 "insufficient finish capacity reserved in spare output buffer",
844 );
845 let written = encoder.finish(units, output_index).map_err(&mut *map_error)?;
846 assert!(written <= required, "finish wrote beyond its bound");
847 // SAFETY: The encoder reported initialized units within the spare
848 // range that was reserved above.
849 unsafe {
850 self.output.advance(written);
851 }
852 Ok(())
853 }
854}
855
856impl<O> TranscodeEncodeOutput<O>
857where
858 O: SeekableByteOutput,
859{
860 /// Flushes pending bytes, then seeks the wrapped byte output.
861 ///
862 /// # Parameters
863 ///
864 /// * `position` - Target seek position.
865 ///
866 /// # Returns
867 ///
868 /// The new stream position reported by the wrapped output.
869 ///
870 /// # Errors
871 ///
872 /// Returns flush or seek errors from the wrapped output.
873 #[inline]
874 pub fn seek(&mut self, position: SeekFrom) -> Result<u64> {
875 self.output.seek_to(position)
876 }
877}
878
879impl<O> Write for TranscodeEncodeOutput<O>
880where
881 O: ByteOutput,
882{
883 /// Writes raw bytes through the internal buffer without encoding them.
884 ///
885 /// # Parameters
886 ///
887 /// * `input` - Bytes to append to the buffered unit output.
888 ///
889 /// # Returns
890 ///
891 /// The number of bytes accepted, which may be shorter than `input`.
892 ///
893 /// # Errors
894 ///
895 /// Propagates wrapped-output errors when buffer delivery is needed.
896 #[inline]
897 fn write(&mut self, input: &[u8]) -> Result<usize> {
898 Output::write(&mut self.output, input)
899 }
900
901 /// Writes all raw bytes through the internal buffer without encoding them.
902 ///
903 /// # Parameters
904 ///
905 /// * `input` - Bytes that must all be accepted before success.
906 ///
907 /// # Errors
908 ///
909 /// Propagates output errors and returns `WriteZero` if output stops
910 /// progressing.
911 ///
912 /// # Panics
913 ///
914 /// Panics when the wrapped output reports more bytes than supplied.
915 fn write_all(&mut self, input: &[u8]) -> Result<()> {
916 let mut written = 0;
917 while written < input.len() {
918 let count = match Output::write(&mut self.output, &input[written..]) {
919 Ok(count) => count,
920 Err(error) if error.kind() == ErrorKind::Interrupted => continue,
921 Err(error) => return Err(error),
922 };
923 if count == 0 {
924 return Err(Error::from(ErrorKind::WriteZero));
925 }
926 assert!(
927 count <= input.len() - written,
928 "Output::write returned a count beyond the input length",
929 );
930 written += count;
931 }
932 Ok(())
933 }
934
935 /// Flushes buffered bytes to the wrapped output without finishing an
936 /// encoder.
937 ///
938 /// # Errors
939 ///
940 /// Propagates wrapped-output flush errors.
941 #[inline]
942 fn flush(&mut self) -> Result<()> {
943 TranscodeEncodeOutput::flush(self)
944 }
945}
946
947impl<O> Seek for TranscodeEncodeOutput<O>
948where
949 O: SeekableByteOutput,
950{
951 /// Flushes pending bytes, then seeks the wrapped byte output.
952 ///
953 /// # Parameters
954 ///
955 /// * `position` - Target position relative to the selected seek origin.
956 ///
957 /// # Returns
958 ///
959 /// The new absolute byte position.
960 ///
961 /// # Errors
962 ///
963 /// Propagates flush or seek errors from the wrapped output.
964 #[inline]
965 fn seek(&mut self, position: SeekFrom) -> Result<u64> {
966 self.seek(position)
967 }
968}
969
970impl<O> fmt::Debug for TranscodeEncodeOutput<O>
971where
972 O: Output,
973 O::Item: Copy + Default,
974 BufferedOutput<O>: fmt::Debug,
975{
976 /// Formats the owned buffer and underlying output for diagnostics.
977 ///
978 /// # Parameters
979 ///
980 /// * `formatter` - Destination and formatting options.
981 ///
982 /// # Errors
983 ///
984 /// Returns a formatting error if the destination rejects output.
985 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
986 formatter
987 .debug_struct("TranscodeEncodeOutput")
988 .field("output", &self.output)
989 .finish()
990 }
991}