zencodec_io/
pump.rs

1//! Sequential stream pump: `std::io::Read` into a resumable decoder.
2
3use how_far::Pulse;
4use zencodec::decode::{DecodeRowSink, Event, IncrementalDecode, OutputInfo, StepError};
5
6use crate::drive::Progressor;
7use crate::stepper::{PumpFailure, ReadCursor};
8use crate::window::check_need;
9
10/// Failure of [`pump_read_to_sink`]: a [`PumpFailure`] whose stage error is
11/// the decoder's [`StepError`].
12pub type PumpError<E> = PumpFailure<StepError<E>>;
13
14/// Pump a sequential stream into `decoder`, writing rows into `sink`.
15///
16/// The pump owns one window over the stream. It reads `read_size` bytes at a
17/// time (more when the decoder's request needs a larger contiguous span, up
18/// to `max_window`), offers the window, and drops what the decoder took. A
19/// decoder decodes straight from the window; rows are emitted as soon as it
20/// has enough. Bytes the decoder skips ahead over are read and discarded; a
21/// request behind the stream position is [`PumpError::NeedsRandomAccess`].
22pub fn pump_read_to_sink<R, D>(
23    mut reader: R,
24    decoder: &mut D,
25    sink: &mut dyn DecodeRowSink,
26    pulse: Option<&dyn Pulse>,
27    read_size: usize,
28    max_window: usize,
29) -> Result<OutputInfo, PumpError<D::Error>>
30where
31    R: std::io::Read,
32    D: IncrementalDecode + ?Sized,
33{
34    let mut progress = Progressor::new(pulse);
35    let mut cursor = ReadCursor::new(read_size, max_window);
36    loop {
37        let input = cursor.offer();
38        let p = decoder.step(input, sink, pulse).map_err(PumpError::Step)?;
39        if p.event == Event::Need {
40            check_need(decoder.requested(), &input, p.consumed).map_err(PumpError::Protocol)?;
41        }
42        cursor.take(p.consumed);
43        progress.observe(decoder, &p.event);
44        match p.event {
45            Event::Done(info) => return Ok(info),
46            Event::Need => cursor.serve(&mut reader, decoder.requested())?,
47            _ => {}
48        }
49    }
50}
zencodec_io/pump.rs

zencodec_io/
pump.rs
