zencodec_io/
controlled_stepper.rs

1//! Expose mandatory suspensions through the existing byte cursors.
2
3use zencodec::ImageInfo;
4use zencodec::decode::{
5    BoundaryAction, BoundaryId, ControlledIncrementalDecode, ControlledProgress, DecodeBoundary,
6    DecodeRowSink, Event, Fused, IncrementalDecode, Input, InputRequest, StepError,
7};
8use zenpixels::PixelDescriptor;
9
10use crate::{ByteStepper, Step, Stepped};
11
12/// A controlled decoder's observable progress. A suspension is never completion.
13#[derive(Debug)]
14#[non_exhaustive]
15pub enum ControlledEvent {
16    /// Ordinary metadata, rows or completed output.
17    Decode(Event),
18    /// A required decision before affected native or sink work can continue.
19    Suspended(DecodeBoundary),
20}
21
22/// Drive a controlled decoder with [`crate::SliceCursor`], [`crate::ReadAtCursor`]
23/// or an async cursor, returning mandatory boundaries to the adapter.
24///
25/// Use this when a transcode adapter must prepare conversion, metadata and encode
26/// together while the decoder is suspended. Ordinary callers can instead use
27/// [`crate::PolicyDecoder`] to resolve each boundary automatically.
28///
29/// Repeated cursor calls at a suspension return the same boundary without native
30/// progress or reads. Only a matching [`Self::resume`] can continue. A native or
31/// sink failure, cancellation or unwind drops the native session, so retaining
32/// this adapter does not retain its managed buffers after failure.
33pub struct ControlledStepper<'a, D, S> {
34    decoder: Option<Fused<D>>,
35    sink: &'a mut S,
36}
37
38impl<'a, D: ControlledIncrementalDecode, S: DecodeRowSink> ControlledStepper<'a, D, S> {
39    /// Bind a decoder to its destination; construction performs no decoding.
40    pub fn new(decoder: D, sink: &'a mut S) -> Self {
41        Self {
42            decoder: Some(Fused::new(decoder)),
43            sink,
44        }
45    }
46
47    /// Inspect the adapter's destination without releasing the native session.
48    pub fn sink(&self) -> &S {
49        self.sink
50    }
51
52    /// Configure a destination while the native session is suspended, for
53    /// example by installing the selected conversion and encoder together.
54    /// This does not resume the decoder or approve an interpretation.
55    pub fn sink_mut(&mut self) -> &mut S {
56        self.sink
57    }
58
59    /// Facts currently available from the held native session.
60    pub fn info(&self) -> Option<&ImageInfo> {
61        self.decoder.as_ref().and_then(IncrementalDecode::info)
62    }
63
64    /// Proposed output layout and interpretation, if known.
65    pub fn output_descriptor(&self) -> Option<PixelDescriptor> {
66        self.decoder
67            .as_ref()
68            .and_then(IncrementalDecode::output_descriptor)
69    }
70
71    /// Apply a decision to the exact held boundary. This resumes the protocol;
72    /// the cursor's next call performs subsequent native work.
73    /// `None` adds no per-call control and preserves native retained controls.
74    pub fn resume(
75        &mut self,
76        boundary: BoundaryId,
77        action: BoundaryAction,
78        pulse: Option<&dyn zencodec::Pulse>,
79    ) -> Result<(), StepError<D::Error>> {
80        let mut decoder = self.decoder.take().ok_or(StepError::Finished)?;
81        decoder.resume(boundary, action, pulse)?;
82        self.decoder = Some(decoder);
83        Ok(())
84    }
85}
86
87impl<D: ControlledIncrementalDecode, S: DecodeRowSink> ByteStepper for ControlledStepper<'_, D, S> {
88    type Output = ControlledEvent;
89    type Error = StepError<D::Error>;
90
91    fn step(
92        &mut self,
93        input: Input<'_>,
94        pulse: Option<&dyn zencodec::Pulse>,
95    ) -> Result<Stepped<Self::Output>, Self::Error> {
96        // Taking ownership also makes a caught native/sink unwind terminal and
97        // releases the session before control returns to its caller.
98        let mut decoder = self.decoder.take().ok_or(StepError::Finished)?;
99        let (consumed, step) = match decoder.step_controlled(input, self.sink, pulse)? {
100            ControlledProgress::Suspended { consumed, boundary } => {
101                (consumed, Step::Output(ControlledEvent::Suspended(boundary)))
102            }
103            ControlledProgress::Decode(progress) => {
104                let step = match progress.event {
105                    Event::Need => Step::Need,
106                    event @ Event::Done(_) => Step::Finished(ControlledEvent::Decode(event)),
107                    event => Step::Output(ControlledEvent::Decode(event)),
108                };
109                (progress.consumed, step)
110            }
111        };
112        self.decoder = Some(decoder);
113        Ok(Stepped { consumed, step })
114    }
115
116    fn requested(&self) -> &[InputRequest] {
117        self.decoder
118            .as_ref()
119            .map_or(&[], IncrementalDecode::requested)
120    }
121}
zencodec_io/controlled_stepper.rs

zencodec_io/
controlled_stepper.rs
