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

qubit_codec/transcode/io/
async_transcode_decode_input.rs
