substrate_data/wire/io/
frame_reader.rs

1//! [`FrameReader`]: reads frames into one buffer it keeps.
2
3use std::io::{self, Read};
4
5use super::framing::{body_len, decode_frame, read_len};
6use crate::Schema;
7
8/// Reads frames into a buffer it keeps, so a stream of messages costs no
9/// allocation per frame once the buffer has grown to the biggest. The end
10/// rules are those of [`super::read_frame_bytes`]: a clean end only between
11/// frames; a stream cut inside a length or a frame is an error.
12///
13/// ```
14/// use substrate_data::io::{FrameReader, write_frame};
15///
16/// let mut pipe = Vec::new();
17/// write_frame(&mut pipe, &7_u32)?;
18/// write_frame(&mut pipe, &8_u32)?;
19/// let (mut r, mut frames) = (pipe.as_slice(), FrameReader::new());
20/// assert_eq!(frames.read::<u32>(&mut r)?, Some(7));
21/// assert_eq!(frames.read::<u32>(&mut r)?, Some(8));
22/// assert_eq!(frames.read::<u32>(&mut r)?, None);
23/// # Ok::<(), std::io::Error>(())
24/// ```
25#[derive(Debug, Default)]
26pub struct FrameReader {
27    buf: Vec<u8>,
28}
29
30impl FrameReader {
31    /// A reader with an empty buffer.
32    #[must_use]
33    pub fn new() -> Self {
34        Self::default()
35    }
36
37    /// The next frame's bytes, valid until the next read; `None` where the
38    /// stream ends between frames. The buffer grows as bytes arrive, never
39    /// by a length the peer merely claims.
40    ///
41    /// # Errors
42    ///
43    /// The reader's I/O error; [`io::ErrorKind::UnexpectedEof`] when the
44    /// stream ends inside a length or a frame; [`io::ErrorKind::InvalidData`]
45    /// for a length over [`super::MAX_FRAME`].
46    pub fn read_bytes(&mut self, r: &mut impl Read) -> io::Result<Option<&[u8]>> {
47        let Some(len) = read_len(r)? else {
48            return Ok(None);
49        };
50        let len = body_len(len)?;
51        self.buf.clear();
52        r.take(len as u64).read_to_end(&mut self.buf)?;
53        if self.buf.len() < len {
54            return Err(io::ErrorKind::UnexpectedEof.into());
55        }
56        Ok(Some(&self.buf))
57    }
58
59    /// The next message; `None` where the stream ends between frames.
60    ///
61    /// # Errors
62    ///
63    /// The reader's I/O error, or [`io::ErrorKind::InvalidData`] for a frame
64    /// that isn't a `T`.
65    pub fn read<T: Schema>(&mut self, r: &mut impl Read) -> io::Result<Option<T>> {
66        self.read_bytes(r)?.map(decode_frame).transpose()
67    }
68}
substrate_data/wire/io/frame_reader.rs

substrate_data/wire/io/
frame_reader.rs
