rskit_codec/framing/
value.rs

1//! Typed value framing: encode/decode one [`Codec`] payload per frame.
2
3use std::io::{Read, Write};
4
5use rskit_errors::{AppError, AppResult};
6use serde::Serialize;
7use serde::de::DeserializeOwned;
8
9use crate::codec::{Codec, decode, encode};
10
11use super::frame::{read_frame, write_frame};
12
13/// Encode `value` with `codec` and write it as one length-delimited frame.
14///
15/// # Errors
16///
17/// Returns a typed [`AppError`] (cause preserved) if encoding or the frame
18/// write fails, or the payload exceeds `max_bytes`.
19pub fn write_value<W, T>(
20    writer: &mut W,
21    codec: &dyn Codec,
22    value: &T,
23    max_bytes: usize,
24) -> AppResult<()>
25where
26    W: Write,
27    T: Serialize + ?Sized,
28{
29    let text = encode(codec, value)?;
30    write_frame(writer, text.as_bytes(), max_bytes)
31}
32
33/// Read one frame and decode it into `T` through `codec`.
34///
35/// Returns `Ok(None)` on a clean end-of-stream between frames.
36///
37/// # Errors
38///
39/// Returns a typed [`AppError`] (cause preserved) on a transport failure or a
40/// payload that does not decode into `T`.
41pub fn read_value<R, T>(reader: &mut R, codec: &dyn Codec, max_bytes: usize) -> AppResult<Option<T>>
42where
43    R: Read,
44    T: DeserializeOwned,
45{
46    let Some(payload) = read_frame(reader, max_bytes)? else {
47        return Ok(None);
48    };
49    decode_value(codec, &payload).map(Some)
50}
51
52/// Decode an already-read frame `payload` into `T` through `codec`.
53///
54/// Split from [`read_value`] so a caller that must inspect one frame as more
55/// than one shape can decode the same bytes without re-reading the stream.
56///
57/// # Errors
58///
59/// Returns a typed [`AppError`] (cause preserved) if the payload is not valid
60/// UTF-8 or does not decode into `T`.
61pub fn decode_value<T>(codec: &dyn Codec, payload: &[u8]) -> AppResult<T>
62where
63    T: DeserializeOwned,
64{
65    let text = std::str::from_utf8(payload).map_err(|error| {
66        AppError::invalid_input("frame", "frame payload is not valid UTF-8").with_cause(error)
67    })?;
68    decode(codec, text)
69}
70
71#[cfg(test)]
72mod tests {
73    use super::super::frame::DEFAULT_MAX_FRAME_BYTES;
74    use super::{read_value, write_value};
75    use crate::JsonCodec;
76
77    #[test]
78    fn value_round_trips_as_one_frame() {
79        let mut buffer = Vec::new();
80        let sent = vec!["a".to_string(), "b".to_string()];
81        write_value(
82            &mut buffer,
83            &JsonCodec::default(),
84            &sent,
85            DEFAULT_MAX_FRAME_BYTES,
86        )
87        .expect("write");
88        let mut cursor = std::io::Cursor::new(buffer);
89        let got: Vec<String> =
90            read_value(&mut cursor, &JsonCodec::default(), DEFAULT_MAX_FRAME_BYTES)
91                .expect("read")
92                .expect("present");
93        assert_eq!(got, sent);
94    }
95
96    #[test]
97    fn non_utf8_payload_is_invalid_input() {
98        let mut buffer = Vec::new();
99        super::write_frame(&mut buffer, &[0xff, 0xfe], DEFAULT_MAX_FRAME_BYTES).expect("write");
100        let mut cursor = std::io::Cursor::new(buffer);
101        let error =
102            read_value::<_, String>(&mut cursor, &JsonCodec::default(), DEFAULT_MAX_FRAME_BYTES)
103                .unwrap_err();
104        assert_eq!(error.code(), rskit_errors::ErrorCode::InvalidInput);
105    }
106}
rskit_codec/framing/value.rs

rskit_codec/framing/
value.rs
