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

qubit_codec/transcode/io/
async_transcode_decode_input.rs
