substrate_data/wire/io/
frame_writer.rs

1//! [`FrameWriter`]: sends messages from buffers it keeps.
2
3use std::io::{self, Write};
4
5use super::framing::{Len, frame_len};
6use crate::Schema;
7use crate::binary::Writer;
8use crate::frame::Packer;
9
10/// Sends messages as frames without allocating per message: it keeps the
11/// buffers a message is written into, so once they have grown to the
12/// biggest message, a send only writes. Each frame (its length, then its
13/// body) goes out in one `write_all`, then the writer is flushed.
14///
15/// - [`FrameWriter::send_packed`] sends the packed frame layout
16///   ([`crate::frame`]): allocation-free once warm for messages whose
17///   fields are plain values, strings, lists and records.
18/// - [`FrameWriter::send`] sends the named binary form (`SDAT`): also
19///   allocation-free once warm, except where a field writes through a
20///   [`crate::Stream`], which builds its body apart.
21///
22/// ```
23/// use substrate_data::io::{FrameWriter, read_frame};
24///
25/// let mut sender = FrameWriter::new();
26/// let mut pipe = Vec::new();
27/// sender.send(&mut pipe, &"hello".to_string())?;
28/// sender.send(&mut pipe, &"again".to_string())?;
29/// assert_eq!(read_frame::<String>(&mut pipe.as_slice())?.as_deref(), Some("hello"));
30/// # Ok::<(), std::io::Error>(())
31/// ```
32#[derive(Debug, Default)]
33pub struct FrameWriter {
34    /// The named form's writer: its buffer holds the length, then the body.
35    named: Writer,
36    packer: Packer,
37    /// The packed form's length and body, side by side.
38    packed: Vec<u8>,
39}
40
41impl FrameWriter {
42    /// A writer with nothing grown yet: the first sends allocate.
43    #[must_use]
44    pub fn new() -> Self {
45        Self::default()
46    }
47
48    /// Sends `message` in the named binary form, and flushes.
49    ///
50    /// # Errors
51    ///
52    /// The writer's I/O error; [`io::ErrorKind::InvalidInput`] for a frame
53    /// over [`super::MAX_FRAME`] (nothing is written).
54    pub fn send<T: Schema>(&mut self, w: &mut impl Write, message: &T) -> io::Result<()> {
55        self.named.restart(size_of::<Len>());
56        message.encode(&mut self.named);
57        put_frame(w, self.named.buffer_mut())
58    }
59
60    /// Sends `message` in the packed frame layout, and flushes.
61    ///
62    /// # Errors
63    ///
64    /// The writer's I/O error; [`io::ErrorKind::InvalidInput`] for a frame
65    /// over [`super::MAX_FRAME`] (nothing is written).
66    pub fn send_packed<T: Schema>(&mut self, w: &mut impl Write, message: &T) -> io::Result<()> {
67        let packed = &mut self.packed;
68        packed.clear();
69        packed.extend_from_slice(&[0; size_of::<Len>()]);
70        self.packer
71            .with_frame(message, |frame| packed.extend_from_slice(frame));
72        put_frame(w, packed)
73    }
74}
75
76/// `buf` (room for a length, then a body) with its length filled in, written
77/// in one piece and flushed.
78fn put_frame(w: &mut impl Write, buf: &mut [u8]) -> io::Result<()> {
79    let (len, body) = buf.split_at_mut(size_of::<Len>());
80    len.copy_from_slice(&frame_len(body.len())?.to_le_bytes());
81    w.write_all(buf)?;
82    w.flush()
83}
substrate_data/wire/io/frame_writer.rs

substrate_data/wire/io/
frame_writer.rs
