scv_protocol/
frame.rs

1//! Newline-delimited framing, without I/O.
2//!
3//! Every SCV connection carries one JSON object per line. Frame limits are a
4//! trust boundary: a peer must not make the other side buffer without bound.
5//! [`FrameDecoder`] is the one implementation of that limit; the reading
6//! loop that feeds it lives with the I/O (`scv_client::read_frame`).
7
8use serde::Serialize;
9
10/// What a decoder does with a line longer than its limit.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum Overflow {
13    /// Report [`Frame::TooLarge`] at once, consuming nothing more. The caller
14    /// is expected to drop the connection.
15    Stop,
16    /// Discard the rest of the line, then report [`Frame::TooLarge`], so the
17    /// connection can carry on with the next line.
18    Skip,
19}
20
21/// One result of decoding.
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub enum Frame {
24    /// A complete line, without its `\n`.
25    Line(Vec<u8>),
26    /// A line over the limit.
27    TooLarge,
28    /// End of input on a line boundary.
29    End,
30    /// End of input in the middle of a line; the bytes read so far.
31    Truncated(Vec<u8>),
32}
33
34/// How much of a buffer [`FrameDecoder::feed`] used, and what it completed.
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct Step {
37    /// Bytes of the buffer that belong to the decoder now; consume them.
38    pub consumed: usize,
39    /// A completed line or an overflow, if any.
40    pub frame: Option<Frame>,
41}
42
43/// Splits a byte stream into bounded lines.
44///
45/// The decoder keeps a partial line between calls, so a read loop that is
46/// cancelled between buffers loses nothing: feed what the reader has, consume
47/// exactly [`Step::consumed`], and call again.
48#[derive(Debug, Clone)]
49pub struct FrameDecoder {
50    /// Most bytes one line may take, counting its `\n`.
51    limit: usize,
52    overflow: Overflow,
53    partial: Vec<u8>,
54    discarding: bool,
55}
56
57impl FrameDecoder {
58    /// A decoder whose lines, counting the `\n`, take at most `limit` bytes.
59    pub fn new(limit: usize, overflow: Overflow) -> Self {
60        Self {
61            limit,
62            overflow,
63            partial: Vec::new(),
64            discarding: false,
65        }
66    }
67
68    /// Change the limit for the lines still to come.
69    pub fn set_limit(&mut self, limit: usize) {
70        self.limit = limit;
71    }
72
73    /// Whether no partial line is buffered or being discarded.
74    #[cfg(test)]
75    pub(crate) fn is_empty(&self) -> bool {
76        self.partial.is_empty() && !self.discarding
77    }
78
79    /// Take bytes from `available` up to and including the next `\n`.
80    pub fn feed(&mut self, available: &[u8]) -> Step {
81        let newline = available.iter().position(|byte| *byte == b'\n');
82        let take = newline.map_or(available.len(), |index| index + 1);
83        if !self.discarding {
84            if self.partial.len().saturating_add(take) > self.limit {
85                match self.overflow {
86                    Overflow::Stop => {
87                        return Step {
88                            consumed: 0,
89                            frame: Some(Frame::TooLarge),
90                        };
91                    }
92                    Overflow::Skip => {
93                        self.discarding = true;
94                        self.partial.clear();
95                    }
96                }
97            } else {
98                self.partial.extend_from_slice(&available[..take]);
99            }
100        }
101        let frame = newline.map(|_| {
102            if std::mem::take(&mut self.discarding) {
103                Frame::TooLarge
104            } else {
105                let mut line = std::mem::take(&mut self.partial);
106                line.pop();
107                Frame::Line(line)
108            }
109        });
110        Step {
111            consumed: take,
112            frame,
113        }
114    }
115
116    /// The input ended: what was left, if anything.
117    pub fn finish(&mut self) -> Frame {
118        if std::mem::take(&mut self.discarding) {
119            Frame::TooLarge
120        } else if self.partial.is_empty() {
121            Frame::End
122        } else {
123            Frame::Truncated(std::mem::take(&mut self.partial))
124        }
125    }
126}
127
128/// Strip a line's trailing `\r` (and `\n`) and check it against `max_bytes`:
129/// the rule for peers that may send CRLF and whose limit excludes the line
130/// ending. `None` means too large.
131pub fn trim_line(mut line: Vec<u8>, max_bytes: usize) -> Option<Vec<u8>> {
132    while matches!(line.last(), Some(b'\n' | b'\r')) {
133        line.pop();
134    }
135    (line.len() <= max_bytes).then_some(line)
136}
137
138/// Encode `message` as one frame: its JSON and a `\n`.
139pub fn encode_frame(message: &impl Serialize) -> serde_json::Result<Vec<u8>> {
140    let mut bytes = serde_json::to_vec(message)?;
141    bytes.push(b'\n');
142    Ok(bytes)
143}
144
145#[cfg(test)]
146mod tests;
scv_protocol/frame.rs

scv_protocol/
frame.rs
