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

qubit_codec/transcode/io/
async_transcode_decode_input.rs
