http_streams_core/
json_nl_codec.rs

1//! Decoding JSON Lines.
2//!
3//! The easy direction: `serde_json` escapes newlines inside strings, so a raw `\n` can only
4//! ever be a record separator and a line split is safe.
5
6use crate::error::{StreamError, StreamErrorKind};
7use bytes::BytesMut;
8use serde::Deserialize;
9use std::marker::PhantomData;
10use tokio_util::codec::{Decoder, LinesCodec, LinesCodecError};
11
12/// A [`Decoder`] that yields one deserialised item per line.
13#[derive(Debug)]
14pub struct JsonNewLineCodec<T> {
15    inner: LinesCodec,
16    /// `fn() -> T` rather than `T`: a bare `PhantomData<T>` would make this codec `!Send`
17    /// whenever `T` is, and the crates built on this one promise `Send` streams for item
18    /// types that carry no such bound. The item type is produced, never held, so this is also
19    /// the honest variance.
20    _ph: PhantomData<fn() -> T>,
21}
22
23impl<T> JsonNewLineCodec<T> {
24    /// A codec that rejects any single line longer than `max_length` bytes.
25    pub fn new_with_max_length(max_length: usize) -> Self {
26        Self {
27            inner: LinesCodec::new_with_max_length(max_length),
28            _ph: PhantomData,
29        }
30    }
31}
32
33/// Framing failures are told apart from deserialisation failures, so that a line which blew the
34/// length limit reports as [`MaxLenReachedError`] rather than as a generic codec error.
35///
36/// [`MaxLenReachedError`]: StreamErrorKind::MaxLenReachedError
37fn frame_error(err: LinesCodecError) -> StreamError {
38    match err {
39        LinesCodecError::MaxLineLengthExceeded => StreamError::new(
40            StreamErrorKind::MaxLenReachedError,
41            None,
42            Some("Max line length reached".into()),
43        ),
44        LinesCodecError::Io(err) => StreamError::from(err),
45    }
46}
47
48/// A line that fails to parse is yielded as an error **item**, not as the decoder's error:
49/// the line framed correctly, so the decoder knows exactly where the next one starts and the
50/// stream carries on. Returning the decoder's `Error` here would make one bad line silently
51/// truncate the rest of the body, because `FramedRead` latches its error state.
52fn parse<T>(line: &str) -> Option<Result<T, StreamError>>
53where
54    T: for<'de> Deserialize<'de>,
55{
56    Some(
57        serde_json::from_str(line)
58            .map_err(|err| StreamError::new(StreamErrorKind::CodecError, Some(Box::new(err)), None)),
59    )
60}
61
62impl<T> Decoder for JsonNewLineCodec<T>
63where
64    T: for<'de> Deserialize<'de>,
65{
66    type Item = Result<T, StreamError>;
67    type Error = StreamError;
68
69    fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, StreamError> {
70        match self.inner.decode(buf).map_err(frame_error)? {
71            Some(line) => Ok(parse(&line)),
72            None => Ok(None),
73        }
74    }
75
76    fn decode_eof(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, StreamError> {
77        match self.inner.decode_eof(buf).map_err(frame_error)? {
78            Some(line) => Ok(parse(&line)),
79            None => Ok(None),
80        }
81    }
82}
http_streams_core/json_nl_codec.rs

http_streams_core/
json_nl_codec.rs
