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

substrate_data/wire/io/
frame_reader.rs
