zencodec_io/
read_at_reader.rs

1//! Synchronous parser bridge over an existing positional source.
2
3use alloc::boxed::Box;
4use std::io::{self, Read, Seek, SeekFrom};
5use zencodec::source::ReadAt;
6
7use crate::StdSourceError;
8
9/// Adapt a positional source to [`Read`] and [`Seek`] with an independent cursor.
10///
11/// This is for synchronous parsers that already require standard I/O. It adds
12/// no buffering, prefetch, retry or asynchronous suspension. Wrap it in
13/// [`std::io::BufReader`] only when its extra reads fit the transport's plan.
14/// Batched range requests remain available through [`ReadAt::read_many_at`];
15/// a parser making individual `Read` calls cannot express such a batch here.
16///
17/// End-relative seeks call [`ReadAt::len`], which can consume a sequential source
18/// to discover its length. Other seeks and [`Seek::stream_position`] do no I/O.
19/// Seeking past EOF is allowed. A failed read or seek leaves the logical cursor
20/// unchanged, but may have changed the source's state or part of the read buffer.
21/// Whether retry is valid remains a source contract.
22///
23/// Native `std::io::Error` values (including [`StdSourceError::Io`]) are returned
24/// unchanged. Other errors retain their concrete value as an I/O payload, with
25/// an underlying I/O kind when present. [`zencodec::find_cause`] inspects these
26/// payloads as well as ordinary error chains.
27#[derive(Debug)]
28pub struct ReadAtReader<S> {
29    source: S,
30    position: u64,
31}
32
33impl<S> ReadAtReader<S> {
34    /// Take ownership of a source, starting the logical cursor at zero.
35    pub const fn new(source: S) -> Self {
36        Self {
37            source,
38            position: 0,
39        }
40    }
41
42    /// Recover the source. The bridge retains no buffered bytes.
43    pub fn into_inner(self) -> S {
44        self.source
45    }
46}
47
48fn into_io(error: impl core::error::Error + Send + Sync + 'static) -> io::Error {
49    let error: Box<dyn core::error::Error + Send + Sync> = Box::new(error);
50    let error = match error.downcast::<io::Error>() {
51        Ok(native) => return *native,
52        Err(other) => other,
53    };
54    let error = match error.downcast::<StdSourceError>() {
55        Ok(source) => match *source {
56            StdSourceError::Io(native) => return native,
57            other => return io::Error::other(other),
58        },
59        Err(other) => other,
60    };
61    let kind = zencodec::find_cause::<io::Error>(error.as_ref())
62        .map_or(io::ErrorKind::Other, io::Error::kind);
63    io::Error::new(kind, error)
64}
65
66impl<S: ReadAt> Read for ReadAtReader<S> {
67    fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
68        if out.is_empty() {
69            return Ok(0);
70        }
71        let n = self.source.read_at(self.position, out).map_err(into_io)?;
72        if n > out.len() {
73            return Err(io::Error::new(
74                io::ErrorKind::InvalidData,
75                "positional source returned a count larger than its buffer",
76            ));
77        }
78        self.position = self.position.checked_add(n as u64).ok_or_else(|| {
79            io::Error::new(io::ErrorKind::InvalidData, "source position overflow")
80        })?;
81        Ok(n)
82    }
83}
84
85impl<S: ReadAt> Seek for ReadAtReader<S> {
86    fn seek(&mut self, to: SeekFrom) -> io::Result<u64> {
87        let position = match to {
88            SeekFrom::Start(n) => Some(n),
89            SeekFrom::Current(n) => self.position.checked_add_signed(n),
90            SeekFrom::End(n) => self.source.len().map_err(into_io)?.checked_add_signed(n),
91        }
92        .ok_or_else(|| {
93            io::Error::new(io::ErrorKind::InvalidInput, "seek outside u64 offset space")
94        })?;
95        self.position = position;
96        Ok(position)
97    }
98
99    fn stream_position(&mut self) -> io::Result<u64> {
100        Ok(self.position)
101    }
102}
zencodec_io/read_at_reader.rs

zencodec_io/
read_at_reader.rs
