substrate_data/wire/io/
framing.rs

1//! Frames written and read with their length in front.
2
3use std::io::{self, Read, Write};
4
5use crate::Schema;
6
7/// How a frame's length is written.
8pub type Len = u32;
9/// The longest frame read: anything longer is a broken or hostile peer.
10pub const MAX_FRAME: usize = 256 << 20;
11
12/// Writes `message` as one frame, and flushes.
13///
14/// # Errors
15///
16/// The writer's I/O error.
17pub fn write_frame<T: Schema>(w: &mut impl Write, message: &T) -> io::Result<()> {
18    write_frame_bytes(w, &crate::to_bytes(message))
19}
20
21/// Writes `bytes` (a message's binary form) as one frame, and flushes.
22///
23/// # Errors
24///
25/// The writer's I/O error; [`io::ErrorKind::InvalidInput`] for a frame over [`MAX_FRAME`].
26pub fn write_frame_bytes(w: &mut impl Write, bytes: &[u8]) -> io::Result<()> {
27    let len = frame_len(bytes.len())?;
28    // One write: a reader never sees a length without its body behind it.
29    let mut frame = Vec::with_capacity(size_of::<Len>() + bytes.len());
30    frame.extend_from_slice(&len.to_le_bytes());
31    frame.extend_from_slice(bytes);
32    w.write_all(&frame)?;
33    w.flush()
34}
35
36/// The next frame's bytes; `None` where the stream ends between frames.
37///
38/// # Errors
39///
40/// The reader's I/O error; [`io::ErrorKind::UnexpectedEof`] when the stream
41/// ends inside a length or a frame; [`io::ErrorKind::InvalidData`] for a
42/// length over [`MAX_FRAME`].
43pub fn read_frame_bytes(r: &mut impl Read) -> io::Result<Option<Vec<u8>>> {
44    let Some(len) = read_len(r)? else {
45        return Ok(None);
46    };
47    let len = body_len(len)?;
48    let mut bytes = vec![0; len];
49    r.read_exact(&mut bytes)?;
50    Ok(Some(bytes))
51}
52
53/// The next message; `None` where the stream ends between frames.
54///
55/// # Errors
56///
57/// The reader's I/O error, or [`io::ErrorKind::InvalidData`] for a frame that isn't a `T`.
58pub fn read_frame<T: Schema>(r: &mut impl Read) -> io::Result<Option<T>> {
59    let Some(bytes) = read_frame_bytes(r)? else {
60        return Ok(None);
61    };
62    decode_frame(&bytes).map(Some)
63}
64
65/// A length read off the wire as a body length; [`io::ErrorKind::InvalidData`]
66/// over [`MAX_FRAME`], before anything is allocated for it.
67pub(super) fn body_len(len: Len) -> io::Result<usize> {
68    let len = len as usize;
69    if len > MAX_FRAME {
70        return Err(io::Error::new(io::ErrorKind::InvalidData, "frame too long"));
71    }
72    Ok(len)
73}
74
75/// A frame body's length as written; [`io::ErrorKind::InvalidInput`] over [`MAX_FRAME`].
76pub(super) fn frame_len(len: usize) -> io::Result<Len> {
77    Len::try_from(len)
78        .ok()
79        .filter(|_| len <= MAX_FRAME)
80        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "frame too long"))
81}
82
83/// A frame body read as a `T`; [`io::ErrorKind::InvalidData`] when it isn't one.
84pub(super) fn decode_frame<T: Schema>(bytes: &[u8]) -> io::Result<T> {
85    crate::read_bytes(bytes)
86        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, crate::Errors(e)))
87}
88
89/// A frame's length; `None` only when the stream ends before its first byte
90/// (between frames), an error when it ends inside the length.
91pub(super) fn read_len(r: &mut impl Read) -> io::Result<Option<Len>> {
92    let mut len = [0; size_of::<Len>()];
93    loop {
94        match r.read(&mut len[..1]) {
95            Ok(0) => return Ok(None),
96            Ok(_) => break,
97            Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
98            Err(e) => return Err(e),
99        }
100    }
101    r.read_exact(&mut len[1..])?;
102    Ok(Some(Len::from_le_bytes(len)))
103}
substrate_data/wire/io/framing.rs

substrate_data/wire/io/
framing.rs
