qubit_codec/transcode/io/
async_transcode_decode_input.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 asynchronous input driver that decodes units into values.
9
10use core::fmt;
11use std::collections::TryReserveError;
12use std::future::poll_fn;
13use std::io::Error;
14use std::io::ErrorKind;
15use std::io::Result;
16use std::pin::Pin;
17use std::task::Context;
18use std::task::Poll;
19
20use qubit_io::AsyncBufferedInput;
21use qubit_io::AsyncInput;
22use qubit_io::Buffer;
23use qubit_utils::SliceRange;
24use qubit_utils::allocation_error;
25
26use super::async_transcode_decode_step::AsyncTranscodeDecodeStep;
27use super::transcode_progress_validation::validate_decode_progress;
28use crate::CapacityError;
29use crate::TranscodeProgress;
30use crate::Transcoder;
31
32/// Decodes an asynchronous unit stream into values.
33///
34/// The adapter owns unit buffering but not decoder state. Callers supply a
35/// streaming decoder and error mapper for each operation.
36///
37/// # Type Parameters
38///
39/// - `I`: Wrapped asynchronous unit input.
40#[must_use]
41pub struct AsyncTranscodeDecodeInput<I>
42where
43    I: AsyncInput,
44    I::Item: Copy + Default,
45{
46    /// Buffered asynchronous unit input.
47    input: AsyncBufferedInput<I>,
48}
49
50impl<I> AsyncTranscodeDecodeInput<I>
51where
52    I: AsyncInput,
53    I::Item: Copy + Default,
54{
55    /// Creates an adapter with the default unit-buffer capacity.
56    ///
57    /// # Parameters
58    ///
59    /// - `inner`: Unit input read by this adapter.
60    ///
61    /// # Returns
62    ///
63    /// Returns a new buffered asynchronous decoder input.
64    pub fn new(inner: I) -> Self {
65        Self {
66            input: AsyncBufferedInput::new(inner),
67        }
68    }
69
70    /// Creates an adapter with an internal unit buffer of at least capacity.
71    ///
72    /// # Parameters
73    ///
74    /// - `inner`: Unit input read by this adapter.
75    /// - `capacity`: Requested internal unit-buffer capacity.
76    ///
77    /// # Returns
78    ///
79    /// Returns a new buffered asynchronous decoder input.
80    pub fn with_capacity(inner: I, capacity: usize) -> Self {
81        Self {
82            input: AsyncBufferedInput::with_capacity(inner, capacity),
83        }
84    }
85
86    /// Tries to create an adapter with an internal unit buffer of at least
87    /// capacity.
88    ///
89    /// # Errors
90    ///
91    /// Returns the allocation error when the internal buffer cannot be
92    /// allocated.
93    pub fn try_with_capacity(inner: I, capacity: usize) -> std::result::Result<Self, TryReserveError> {
94        Ok(Self {
95            input: AsyncBufferedInput::try_with_capacity(inner, capacity)?,
96        })
97    }
98
99    /// Returns a shared reference to the wrapped asynchronous input.
100    ///
101    /// The input can be physically positioned after units retained in this
102    /// adapter's unread buffer.
103    #[must_use]
104    pub const fn inner(&self) -> &I {
105        self.input.inner()
106    }
107
108    /// Returns a mutable reference to the wrapped asynchronous input.
109    ///
110    /// Direct reads can invalidate the logical stream position represented by
111    /// unread buffered units.
112    #[must_use]
113    pub fn inner_mut(&mut self) -> &mut I {
114        self.input.inner_mut()
115    }
116
117    /// Returns the number of unread units currently buffered.
118    #[must_use]
119    pub const fn unread_len(&self) -> usize {
120        self.input.unread_len()
121    }
122
123    /// Returns the unread buffered unit window.
124    #[must_use]
125    pub fn unread(&self) -> &[I::Item] {
126        self.input.unread()
127    }
128
129    /// Returns the total internal unit-buffer capacity.
130    #[must_use]
131    pub fn capacity(&self) -> usize {
132        self.input.capacity()
133    }
134
135    /// Consumes unread units from the current buffer window.
136    ///
137    /// # Panics
138    ///
139    /// Panics when count exceeds the unread unit count.
140    pub fn consume(&mut self, count: usize) {
141        assert!(count <= self.unread_len(), "cannot consume beyond buffered input",);
142        // SAFETY: The asserted bound proves count fits the unread window.
143        unsafe {
144            self.input.consume(count);
145        }
146    }
147
148    /// Copies unread units into an indexed output range without consuming them.
149    ///
150    /// # Safety
151    ///
152    /// The caller must guarantee that the indexed range fits in output, does
153    /// not overflow, holds no overlapping unread units, and count is no
154    /// greater than the unread unit count.
155    pub unsafe fn copy_unread_to(&self, output: &mut [I::Item], output_index: usize, count: usize) {
156        // SAFETY: The caller upholds the delegated buffer-copy contract.
157        unsafe {
158            self.input.copy_unread_to(output, output_index, count);
159        }
160    }
161
162    /// Consumes this adapter and returns the input plus its unread buffer.
163    #[must_use = "the returned input and unread buffer must be handled"]
164    pub fn into_parts(self) -> (I, Buffer<I::Item>) {
165        self.input.into_parts()
166    }
167
168    /// Runs decoder reset into an indexed output range without I/O.
169    ///
170    /// # Errors
171    ///
172    /// Returns invalid output-range, capacity, or mapped decoder-reset errors.
173    ///
174    /// # Panics
175    ///
176    /// Panics if the decoder writes more values than its reset bound.
177    pub fn reset<D, M, Value>(
178        &self,
179        decoder: &mut D,
180        map_error: &mut M,
181        output: &mut [Value],
182        output_index: usize,
183        count: usize,
184    ) -> Result<usize>
185    where
186        D: Transcoder<Input = I::Item, Output = Value>,
187        M: FnMut(D::Error) -> Error,
188    {
189        let required = decoder.max_reset_output_len().map_err(capacity_to_io_error)?;
190        let output_end = SliceRange::checked_range_end(
191            output.len(),
192            output_index,
193            count,
194            "reset output range exceeds destination buffer",
195        )?;
196        if count < required {
197            return Err(Error::new(
198                ErrorKind::InvalidData,
199                "insufficient output for decoder reset bound",
200            ));
201        }
202        let output = &mut output[..output_end];
203        let written = decoder.reset(output, output_index).map_err(&mut *map_error)?;
204        assert!(written <= required, "reset wrote beyond its bound");
205        Ok(written)
206    }
207
208    /// Finishes a decoder into an indexed output range without I/O.
209    ///
210    /// # Errors
211    ///
212    /// Returns invalid output-range, capacity, or mapped decoder-finish
213    /// errors.
214    ///
215    /// # Panics
216    ///
217    /// Panics if the decoder writes more values than its finish bound.
218    pub fn finish<D, M, Value>(
219        &self,
220        decoder: &mut D,
221        map_error: &mut M,
222        output: &mut [Value],
223        output_index: usize,
224        count: usize,
225    ) -> Result<usize>
226    where
227        D: Transcoder<Input = I::Item, Output = Value>,
228        M: FnMut(D::Error) -> Error,
229    {
230        let required = decoder.max_finish_output_len().map_err(capacity_to_io_error)?;
231        let output_end = SliceRange::checked_range_end(
232            output.len(),
233            output_index,
234            count,
235            "finish output range exceeds destination buffer",
236        )?;
237        if count < required {
238            return Err(Error::new(
239                ErrorKind::InvalidData,
240                "insufficient output for decoder finish bound",
241            ));
242        }
243        let output = &mut output[..output_end];
244        let written = decoder.finish(output, output_index).map_err(&mut *map_error)?;
245        assert!(written <= required, "finish wrote beyond its bound");
246        Ok(written)
247    }
248}
249
250impl<I> AsyncTranscodeDecodeInput<I>
251where
252    I: AsyncInput + Unpin,
253    I::Item: Copy + Default + Unpin,
254{
255    /// Appends at least one unit to the unread buffer.
256    ///
257    /// # Returns
258    ///
259    /// Returns true when data was appended, or false at end of input.
260    ///
261    /// # Errors
262    ///
263    /// Returns input or buffer-management errors while refilling.
264    pub async fn fill_more_async(&mut self) -> Result<bool> {
265        self.input.fill_more_async().await
266    }
267
268    /// Refills until at least count unread units are available.
269    ///
270    /// # Returns
271    ///
272    /// Returns true when count units are available, or false when EOF occurs
273    /// first.
274    ///
275    /// # Errors
276    ///
277    /// Returns allocation, input, or buffer-management errors while refilling.
278    pub async fn fill_until_async(&mut self, count: usize) -> Result<bool> {
279        if count > self.input.capacity() {
280            self.input.try_reserve_capacity(count).map_err(allocation_error)?;
281        }
282        self.input.fill_until_async(count).await
283    }
284
285    /// Polls one cancellation-safe decoder operation.
286    ///
287    /// The operation first refills only when no unread unit is buffered. After
288    /// a decoder call changes either the decoder or the buffered input, it
289    /// immediately returns [`AsyncTranscodeDecodeStep::Progress`] and never
290    /// polls the input again. This makes the returned progress the commit
291    /// boundary for cancellation: resume with the adapter's current state,
292    /// rather than replaying the previous source range.
293    ///
294    /// # Errors
295    ///
296    /// Returns input, allocation, output-range, invalid-progress, or mapped
297    /// decoder errors.
298    pub fn poll_transcode<D, M, Value>(
299        mut self: Pin<&mut Self>,
300        cx: &mut Context<'_>,
301        decoder: &mut D,
302        map_error: &mut M,
303        output: &mut [Value],
304        output_index: usize,
305        count: usize,
306    ) -> Poll<Result<AsyncTranscodeDecodeStep>>
307    where
308        D: Transcoder<Input = I::Item, Output = Value>,
309        M: FnMut(D::Error) -> Error,
310    {
311        let output_end = SliceRange::checked_range_end(
312            output.len(),
313            output_index,
314            count,
315            "decoded output range exceeds destination buffer",
316        )?;
317        if count == 0 {
318            return Poll::Ready(Ok(AsyncTranscodeDecodeStep::Progress(TranscodeProgress::complete(
319                0, 0,
320            ))));
321        }
322        let output = &mut output[..output_end];
323        if self.as_ref().get_ref().unread_len() == 0 {
324            let this = self.as_mut().get_mut();
325            match Pin::new(&mut this.input).poll_fill_more(cx) {
326                Poll::Ready(Ok(true)) => {}
327                Poll::Ready(Ok(false)) => {
328                    return Poll::Ready(Ok(AsyncTranscodeDecodeStep::EndOfInput));
329                }
330                Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
331                Poll::Pending => return Poll::Pending,
332            }
333        }
334        let this = self.as_mut().get_mut();
335        let available_input = this.unread_len();
336        let progress = decoder
337            .transcode(this.unread(), 0, output, output_index)
338            .map_err(&mut *map_error)
339            .and_then(|progress| validate_decode_progress(progress, 0, available_input, output_index, count));
340        let progress = match progress {
341            Ok(progress) => progress,
342            Err(error) => return Poll::Ready(Err(error)),
343        };
344        this.consume(progress.read());
345        Poll::Ready(Ok(AsyncTranscodeDecodeStep::Progress(progress)))
346    }
347
348    /// Performs one decode step after the caller has established EOF.
349    ///
350    /// This method performs no I/O. It passes the current unread buffer to
351    /// [`Transcoder::transcode_eof`], validates progress, and commits consumed
352    /// source units before returning.
353    ///
354    /// # Errors
355    ///
356    /// Returns invalid output-range errors, or decoder errors mapped by
357    /// `map_error`.
358    pub fn transcode_eof_step<D, M, Value>(
359        &mut self,
360        decoder: &mut D,
361        map_error: &mut M,
362        output: &mut [Value],
363        output_index: usize,
364        count: usize,
365    ) -> Result<TranscodeProgress>
366    where
367        D: Transcoder<Input = I::Item, Output = Value>,
368        M: FnMut(D::Error) -> Error,
369    {
370        let output_end = SliceRange::checked_range_end(
371            output.len(),
372            output_index,
373            count,
374            "decoded EOF output range exceeds destination buffer",
375        )?;
376        if count == 0 {
377            return Ok(TranscodeProgress::complete(0, 0));
378        }
379        if self.unread_len() == 0 {
380            return Ok(TranscodeProgress::complete(0, 0));
381        }
382        let available_input = self.unread_len();
383        let progress = decoder
384            .transcode_eof(self.unread(), 0, &mut output[..output_end], output_index)
385            .map_err(&mut *map_error)
386            .and_then(|progress| validate_decode_progress(progress, 0, available_input, output_index, count))?;
387        self.consume(progress.read());
388        Ok(progress)
389    }
390
391    /// Decodes one cancellation-safe progress step into an indexed output
392    /// range.
393    ///
394    /// This is the async wrapper for [`Self::poll_transcode`]. It returns after
395    /// one decoder invocation; callers that require a complete destination
396    /// range must drive successive steps and retain each returned progress.
397    pub async fn transcode_async<D, M, Value>(
398        &mut self,
399        decoder: &mut D,
400        map_error: &mut M,
401        output: &mut [Value],
402        output_index: usize,
403        count: usize,
404    ) -> Result<AsyncTranscodeDecodeStep>
405    where
406        D: Transcoder<Input = I::Item, Output = Value>,
407        M: FnMut(D::Error) -> Error,
408    {
409        poll_fn(|cx| Pin::new(&mut *self).poll_transcode(cx, decoder, map_error, output, output_index, count)).await
410    }
411}
412
413impl<I> AsyncInput for AsyncTranscodeDecodeInput<I>
414where
415    I: AsyncInput,
416    I::Item: Copy + Default,
417{
418    /// Item type read through the retained unit buffer.
419    type Item = I::Item;
420
421    /// Reports that this input retains unread units.
422    fn is_buffered(&self) -> bool {
423        true
424    }
425
426    /// Polls one read through the retained unit buffer.
427    ///
428    /// # Safety
429    ///
430    /// The caller must provide a valid indexed output range.
431    unsafe fn poll_read_unchecked(
432        mut self: Pin<&mut Self>,
433        cx: &mut Context<'_>,
434        output: &mut [Self::Item],
435        index: usize,
436        count: usize,
437    ) -> Poll<Result<usize>> {
438        // SAFETY: Pinning this wrapper pins its buffered input field.
439        let input = unsafe { Pin::new_unchecked(&mut self.as_mut().get_unchecked_mut().input) };
440        // SAFETY: The caller upholds the delegated indexed-read contract.
441        unsafe { input.poll_read_unchecked(cx, output, index, count) }
442    }
443}
444
445impl<I> fmt::Debug for AsyncTranscodeDecodeInput<I>
446where
447    I: AsyncInput,
448    I::Item: Copy + Default,
449    AsyncBufferedInput<I>: fmt::Debug,
450{
451    /// Formats this asynchronous decoder input for debugging.
452    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
453        formatter
454            .debug_struct("AsyncTranscodeDecodeInput")
455            .field("input", &self.input)
456            .finish()
457    }
458}
459
460/// Converts decoder capacity failures into invalid stream data.
461fn capacity_to_io_error(error: CapacityError) -> Error {
462    Error::new(ErrorKind::InvalidData, error)
463}
qubit_codec/transcode/io/async_transcode_decode_input.rs

qubit_codec/transcode/io/
async_transcode_decode_input.rs
