sim_lib_net_core/
ndjson.rs

1//! Incremental newline-delimited JSON decoding.
2
3use crate::NetError;
4use crate::line::LineDecoder;
5
6/// Incremental newline-delimited JSON decoder.
7///
8/// Each complete line is returned verbatim as one record string (parsing is the
9/// caller's job). A partial final line is buffered until a later push completes
10/// it or [`flush`](NdjsonDecoder::flush) is called. Blank lines are skipped.
11///
12/// An optional per-line size limit guards against an unbounded buffered line on
13/// an adversarial stream; exceeding it fails closed with
14/// [`NetError::OversizeBody`].
15#[derive(Debug, Default)]
16pub struct NdjsonDecoder {
17    lines: LineDecoder,
18    max_line_bytes: Option<usize>,
19}
20
21impl NdjsonDecoder {
22    /// Create a decoder with no per-line size limit.
23    pub fn new() -> Self {
24        Self::default()
25    }
26
27    /// Create a decoder that rejects any line (or buffered partial) exceeding
28    /// `max_line_bytes`.
29    pub fn with_max_line_bytes(max_line_bytes: usize) -> Self {
30        Self {
31            lines: LineDecoder::new(),
32            max_line_bytes: Some(max_line_bytes),
33        }
34    }
35
36    /// Append bytes and return completed JSON record strings, skipping blank
37    /// lines. Fails closed if any completed line or the buffered remainder
38    /// exceeds the configured limit.
39    pub fn push(&mut self, bytes: &[u8]) -> Result<Vec<String>, NetError> {
40        let mut records = Vec::new();
41        for line in self.lines.push(bytes) {
42            self.check_limit(line.len())?;
43            if line.is_empty() {
44                continue;
45            }
46            records.push(String::from_utf8_lossy(&line).into_owned());
47        }
48        // A still-buffered partial line can also blow the limit before its
49        // newline ever arrives.
50        self.check_limit(self.lines.buffered_len())?;
51        Ok(records)
52    }
53
54    /// Take any buffered remainder as a final record. Returns `None` when the
55    /// buffer is empty or holds only whitespace.
56    pub fn flush(&mut self) -> Option<String> {
57        let line = self.lines.flush()?;
58        if line.is_empty() {
59            return None;
60        }
61        Some(String::from_utf8_lossy(&line).into_owned())
62    }
63
64    fn check_limit(&self, len: usize) -> Result<(), NetError> {
65        if let Some(limit) = self.max_line_bytes
66            && len > limit
67        {
68            return Err(NetError::OversizeBody(limit));
69        }
70        Ok(())
71    }
72}
sim_lib_net_core/ndjson.rs

sim_lib_net_core/
ndjson.rs
