sim_lib_net_core/
ndjson.rs

1//! Incremental newline-delimited JSON decoding.
2
3use crate::NetError;
4use crate::line::{DEFAULT_MAX_LINE_BYTES, 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/// The default constructor uses [`DEFAULT_MAX_LINE_BYTES`] to guard against an
13/// unbounded buffered line on an adversarial stream; exceeding it fails closed
14/// with [`NetError::LineTooLong`]. [`unbounded`](NdjsonDecoder::unbounded) is
15/// the explicit opt-out for trusted in-memory inputs.
16#[derive(Debug)]
17pub struct NdjsonDecoder {
18    lines: LineDecoder,
19    max_line_bytes: Option<usize>,
20}
21
22impl Default for NdjsonDecoder {
23    fn default() -> Self {
24        Self::new()
25    }
26}
27
28impl NdjsonDecoder {
29    /// Create a decoder using [`DEFAULT_MAX_LINE_BYTES`].
30    pub fn new() -> Self {
31        Self::with_max_line_bytes(DEFAULT_MAX_LINE_BYTES)
32    }
33
34    /// Create a decoder with no per-line size limit.
35    ///
36    /// Use this only for trusted, already-bounded inputs.
37    pub fn unbounded() -> Self {
38        Self {
39            lines: LineDecoder::unbounded(),
40            max_line_bytes: None,
41        }
42    }
43
44    /// Create a decoder that rejects any line (or buffered partial) exceeding
45    /// `max_line_bytes`.
46    pub fn with_max_line_bytes(max_line_bytes: usize) -> Self {
47        Self {
48            lines: LineDecoder::with_max_line_bytes(max_line_bytes),
49            max_line_bytes: Some(max_line_bytes),
50        }
51    }
52
53    /// Append bytes and return completed JSON record strings, skipping blank
54    /// lines. Fails closed if any completed line or the buffered remainder
55    /// exceeds the configured limit.
56    pub fn push(&mut self, bytes: &[u8]) -> Result<Vec<String>, NetError> {
57        let mut records = Vec::new();
58        for line in self.lines.push_checked(bytes)? {
59            self.check_limit(line.len())?;
60            if line.is_empty() {
61                continue;
62            }
63            records.push(String::from_utf8_lossy(&line).into_owned());
64        }
65        Ok(records)
66    }
67
68    /// Take any buffered remainder as a final record. Returns `None` when the
69    /// buffer is empty or holds only whitespace.
70    pub fn flush(&mut self) -> Option<String> {
71        self.flush_checked()
72            .expect("ndjson line cap exceeded; use push to handle oversized input")
73    }
74
75    /// Take any buffered remainder as a final record, checking the configured
76    /// line cap first.
77    pub fn flush_checked(&mut self) -> Result<Option<String>, NetError> {
78        let Some(line) = self.lines.flush_checked()? else {
79            return Ok(None);
80        };
81        if line.is_empty() {
82            return Ok(None);
83        }
84        self.check_limit(line.len())?;
85        Ok(Some(String::from_utf8_lossy(&line).into_owned()))
86    }
87
88    fn check_limit(&self, len: usize) -> Result<(), NetError> {
89        if let Some(limit) = self.max_line_bytes
90            && len > limit
91        {
92            return Err(NetError::LineTooLong { max: limit });
93        }
94        Ok(())
95    }
96}
sim_lib_net_core/ndjson.rs

sim_lib_net_core/
ndjson.rs
