zencodec_io/
encoder_sink.rs

1//! Transcode without an intermediate image: a sink that hands the decoder
2//! its own strip buffer and pushes each finished strip into an encoder.
3
4use alloc::boxed::Box;
5use alloc::vec::Vec;
6use core::fmt;
7
8use zencodec::decode::{DecodeRowSink, SinkError};
9use zencodec::encode::{EncodeOutput, Encoder, OutputTiming};
10use zenpixels::{PixelDescriptor, PixelSlice, PixelSliceMut};
11
12/// Failure of [`EncoderSink::finish`].
13#[derive(Debug)]
14#[non_exhaustive]
15pub enum EncoderSinkError<E> {
16    /// The sink failed before the encoder could finish.
17    Sink(SinkError),
18    /// The encoder failed.
19    Encoder(E),
20}
21
22impl<E: fmt::Display> fmt::Display for EncoderSinkError<E> {
23    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
24        match self {
25            Self::Sink(e) => write!(f, "encoder sink: {e}"),
26            Self::Encoder(e) => write!(f, "encoder: {e}"),
27        }
28    }
29}
30
31impl<E: core::error::Error + 'static> core::error::Error for EncoderSinkError<E> {
32    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
33        match self {
34            Self::Sink(e) => Some(e.as_ref()),
35            Self::Encoder(e) => Some(e),
36        }
37    }
38}
39
40/// [`DecodeRowSink`] that feeds an [`Encoder`].
41///
42/// `begin` declares the image to the encoder ([`Encoder::begin`]); an
43/// encoder that answers with a streaming [`OutputTiming`] writes its header
44/// at once and makes bytes available as rows arrive.
45/// [`output_timing`](Self::output_timing) reports that answer.
46/// `provide_next_buffer` returns the sink's own strip buffer, so the
47/// decoder writes straight into the bytes the encoder reads; the strip is
48/// pushed on the next `provide_next_buffer` or on `finish`. No pixel is
49/// copied by this handoff; the encoder may copy or retain its own input/state.
50///
51/// Drain encoded bytes between decode steps with
52/// [`pending_output`](Self::pending_output) and
53/// [`consume_output`](Self::consume_output) to keep output memory bounded.
54/// That bounds queued encoded bytes only. Awaiting a slow writer can keep
55/// much larger raw/intermediate allocations alive; output timing alone does
56/// not establish when those allocations are released.
57/// Encoder errors reach the decoder's caller as the sink error, typed, so
58/// they can be told apart from malformed input.
59pub struct EncoderSink<E> {
60    encoder: Option<E>,
61    strip: Vec<u8>,
62    pending: Option<(u32, u32, PixelDescriptor)>,
63    rows_pushed: u64,
64    timing: Option<OutputTiming>,
65}
66
67impl<E: Encoder> EncoderSink<E>
68where
69    E::Error: Send + Sync,
70{
71    /// Wrap `encoder`; rows go to its `push_rows`.
72    pub fn new(encoder: E) -> Self {
73        Self {
74            encoder: Some(encoder),
75            strip: Vec::new(),
76            pending: None,
77            rows_pushed: 0,
78            timing: None,
79        }
80    }
81
82    /// What the encoder's [`begin`](Encoder::begin) reported: whether its
83    /// bytes stream and how many rows it holds back. `None` until the
84    /// decoder has declared the image.
85    ///
86    /// The sink pushes each strip when the decoder asks for the next one,
87    /// so one strip more than the encoder's
88    /// [`lag_rows`](OutputTiming::lag_rows) can wait here.
89    pub fn output_timing(&self) -> Option<OutputTiming> {
90        self.timing
91    }
92
93    /// Rows pushed into the encoder so far.
94    pub fn rows_pushed(&self) -> u64 {
95        self.rows_pushed
96    }
97
98    /// The encoder.
99    pub fn encoder(&self) -> &E {
100        self.encoder.as_ref().expect("present until finish")
101    }
102
103    /// Encoded bytes ready to send; see [`Encoder::pending_output`].
104    pub fn pending_output(&self) -> &[u8] {
105        self.encoder().pending_output()
106    }
107
108    /// Mark the first `n` pending bytes as sent.
109    pub fn consume_output(&mut self, n: usize) {
110        if let Some(enc) = self.encoder.as_mut() {
111            enc.consume_output(n);
112        }
113    }
114
115    /// Append every pending byte to `into` and consume it; see
116    /// [`Encoder::take_output`].
117    pub fn take_output(&mut self, into: &mut Vec<u8>) {
118        if let Some(enc) = self.encoder.as_mut() {
119            enc.take_output(into);
120        }
121    }
122
123    /// Write every pending byte to `w`. Returns how many were written.
124    #[cfg(feature = "std")]
125    pub fn write_pending(&mut self, w: &mut impl std::io::Write) -> std::io::Result<usize> {
126        let n = self.pending_output().len();
127        if n > 0 {
128            w.write_all(self.pending_output())?;
129            self.consume_output(n);
130        }
131        Ok(n)
132    }
133
134    fn push_pending(&mut self) -> Result<(), SinkError> {
135        let Some((rows, width, desc)) = self.pending.take() else {
136            return Ok(());
137        };
138        let stride = width as usize * desc.bytes_per_pixel();
139        let slice = PixelSlice::new(
140            &self.strip[..stride * rows as usize],
141            width,
142            rows,
143            stride,
144            desc,
145        )
146        .map_err(|e| SinkError::from(alloc::format!("EncoderSink: {e}")))?;
147        let enc = self
148            .encoder
149            .as_mut()
150            .ok_or_else(|| SinkError::from("EncoderSink: encoder already finished"))?;
151        enc.push_rows(slice).map_err(|e| Box::new(e) as SinkError)?;
152        self.rows_pushed += rows as u64;
153        Ok(())
154    }
155
156    /// Push any pending strip and finish the encoder. Returns the bytes not
157    /// already taken through [`consume_output`](Self::consume_output).
158    pub fn finish(mut self) -> Result<EncodeOutput, EncoderSinkError<E::Error>> {
159        self.push_pending().map_err(EncoderSinkError::Sink)?;
160        let enc = self.encoder.take().expect("encoder present until finish");
161        enc.finish().map_err(EncoderSinkError::Encoder)
162    }
163}
164
165impl<E: Encoder> fmt::Debug for EncoderSink<E> {
166    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
167        f.debug_struct("EncoderSink")
168            .field("rows_pushed", &self.rows_pushed)
169            .field("pending", &self.pending.is_some())
170            .finish_non_exhaustive()
171    }
172}
173
174impl<E: Encoder> DecodeRowSink for EncoderSink<E>
175where
176    E::Error: Send + Sync,
177{
178    fn begin(
179        &mut self,
180        width: u32,
181        height: u32,
182        descriptor: PixelDescriptor,
183    ) -> Result<(), SinkError> {
184        let enc = self
185            .encoder
186            .as_mut()
187            .ok_or_else(|| SinkError::from("EncoderSink: encoder already finished"))?;
188        let timing = enc
189            .begin(width, height, descriptor)
190            .map_err(|e| Box::new(e) as SinkError)?;
191        self.timing = Some(timing);
192        Ok(())
193    }
194
195    fn provide_next_buffer(
196        &mut self,
197        _y: u32,
198        height: u32,
199        width: u32,
200        descriptor: PixelDescriptor,
201    ) -> Result<PixelSliceMut<'_>, SinkError> {
202        self.push_pending()?;
203        let stride = width as usize * descriptor.bytes_per_pixel();
204        let len = stride
205            .checked_mul(height as usize)
206            .ok_or_else(|| SinkError::from("EncoderSink: strip too large"))?;
207        if self.strip.len() < len {
208            self.strip.resize(len, 0);
209        }
210        self.pending = Some((height, width, descriptor));
211        PixelSliceMut::new(&mut self.strip[..len], width, height, stride, descriptor)
212            .map_err(|e| alloc::format!("EncoderSink: {e}").into())
213    }
214
215    fn finish(&mut self) -> Result<(), SinkError> {
216        self.push_pending()
217    }
218}
zencodec_io/encoder_sink.rs

zencodec_io/
encoder_sink.rs
