karyon_jsonrpc/
codec.rs

1//! JSON codec for JSON-RPC message framing.
2//!
3//! Provides `JsonCodec` which encodes/decodes `serde_json::Value`
4//! messages with a configurable max payload size (default 4MB).
5
6use std::io;
7
8#[cfg(feature = "http3")]
9use bytes::Buf;
10#[cfg(feature = "http")]
11use bytes::Bytes;
12#[cfg(feature = "http")]
13use serde_json::Value;
14
15pub use karyon_net::codec::Codec;
16pub use karyon_net::ByteBuffer;
17
18#[cfg(feature = "ws")]
19use karyon_net::layers::ws::Message as WsMessage;
20
21#[cfg(feature = "http")]
22use crate::error::{Error, Result};
23
24const DEFAULT_MAX_BUFFER_SIZE: usize = 4 * 1024 * 1024; // 4MB
25
26/// Default HTTP `Content-Type`, for the default `JsonCodec`.
27#[cfg(feature = "http")]
28pub(crate) const DEFAULT_CONTENT_TYPE: &str = "application/json";
29
30/// Byte-stream codec used for TCP, TLS, Unix, QUIC, and HTTP.
31pub trait JsonRpcCodec:
32    Codec<ByteBuffer, Message = serde_json::Value, Error = karyon_net::Error>
33    + Clone
34    + Send
35    + Sync
36    + 'static
37{
38}
39
40impl<T> JsonRpcCodec for T where
41    T: Codec<ByteBuffer, Message = serde_json::Value, Error = karyon_net::Error>
42        + Clone
43        + Send
44        + Sync
45        + 'static
46{
47}
48
49/// WebSocket-message codec used for `ws://` and `wss://` endpoints.
50#[cfg(feature = "ws")]
51pub trait JsonRpcWsCodec:
52    Codec<WsMessage, Message = serde_json::Value, Error = karyon_net::Error>
53    + Clone
54    + Send
55    + Sync
56    + 'static
57{
58}
59
60#[cfg(feature = "ws")]
61impl<T> JsonRpcWsCodec for T where
62    T: Codec<WsMessage, Message = serde_json::Value, Error = karyon_net::Error>
63        + Clone
64        + Send
65        + Sync
66        + 'static
67{
68}
69
70/// Default JSON codec with configurable max payload size.
71#[derive(Clone)]
72pub struct JsonCodec {
73    max_size: usize,
74}
75
76impl Default for JsonCodec {
77    fn default() -> Self {
78        Self {
79            max_size: DEFAULT_MAX_BUFFER_SIZE,
80        }
81    }
82}
83
84impl JsonCodec {
85    pub fn new(max_payload_size: usize) -> Self {
86        Self {
87            max_size: max_payload_size,
88        }
89    }
90}
91
92impl Codec<ByteBuffer> for JsonCodec {
93    type Message = serde_json::Value;
94    type Error = karyon_net::Error;
95
96    fn encode(
97        &self,
98        src: &serde_json::Value,
99        dst: &mut ByteBuffer,
100    ) -> std::result::Result<usize, karyon_net::Error> {
101        let msg =
102            serde_json::to_string(src).map_err(|e| karyon_net::Error::IO(io::Error::other(e)))?;
103        let buf = msg.as_bytes();
104
105        if buf.len() > self.max_size {
106            return Err(karyon_net::Error::BufferFull(format!(
107                "Buffer size {} exceeds maximum {}",
108                buf.len(),
109                self.max_size
110            )));
111        }
112
113        dst.extend_from_slice(buf);
114        Ok(buf.len())
115    }
116
117    fn decode(
118        &self,
119        src: &mut ByteBuffer,
120    ) -> std::result::Result<Option<(usize, serde_json::Value)>, karyon_net::Error> {
121        if src.len() > self.max_size {
122            return Err(karyon_net::Error::BufferFull(format!(
123                "Buffer size {} exceeds maximum {}",
124                src.len(),
125                self.max_size
126            )));
127        }
128
129        let de = serde_json::Deserializer::from_slice(src.as_ref());
130        let mut iter = de.into_iter::<serde_json::Value>();
131
132        let item = match iter.next() {
133            Some(Ok(item)) => item,
134            Some(Err(ref e)) if e.is_eof() => return Ok(None),
135            Some(Err(e)) => return Err(karyon_net::Error::IO(io::Error::other(e))),
136            None => return Ok(None),
137        };
138
139        Ok(Some((iter.byte_offset(), item)))
140    }
141}
142
143/// WS codec for JSON values over WebSocket text messages.
144#[cfg(feature = "ws")]
145impl Codec<WsMessage> for JsonCodec {
146    type Message = serde_json::Value;
147    type Error = karyon_net::Error;
148
149    fn encode(
150        &self,
151        src: &serde_json::Value,
152        dst: &mut WsMessage,
153    ) -> std::result::Result<usize, karyon_net::Error> {
154        let json =
155            serde_json::to_string(src).map_err(|e| karyon_net::Error::IO(io::Error::other(e)))?;
156        let len = json.len();
157        *dst = WsMessage::Text(json);
158        Ok(len)
159    }
160
161    fn decode(
162        &self,
163        src: &mut WsMessage,
164    ) -> std::result::Result<Option<(usize, serde_json::Value)>, karyon_net::Error> {
165        match src {
166            WsMessage::Text(text) => {
167                let len = text.len();
168                if len > self.max_size {
169                    return Err(karyon_net::Error::BufferFull(format!(
170                        "Buffer size {} exceeds maximum {}",
171                        len, self.max_size
172                    )));
173                }
174                let val = serde_json::from_str(text)
175                    .map_err(|e| karyon_net::Error::IO(io::Error::other(e)))?;
176                Ok(Some((len, val)))
177            }
178            _ => Ok(None),
179        }
180    }
181}
182
183/// Encode one message with `codec`.
184#[cfg(feature = "http")]
185pub(crate) fn encode_msg<C: JsonRpcCodec>(codec: &C, msg: &Value) -> Result<Bytes> {
186    let mut buf = ByteBuffer::new();
187    codec.encode(msg, &mut buf)?;
188    Ok(Bytes::copy_from_slice(buf.as_ref()))
189}
190
191/// Decode one message from a complete HTTP body.
192#[cfg(feature = "http")]
193pub(crate) fn decode_body<C: JsonRpcCodec>(codec: &C, body: &[u8]) -> Result<Value> {
194    let mut buf = ByteBuffer::from_slice(body);
195    match codec.decode(&mut buf)? {
196        Some((_, msg)) => Ok(msg),
197        None => Err(Error::InvalidMsg("Incomplete message".into())),
198    }
199}
200
201/// Append all bytes of `data` to `buf`.
202#[cfg(feature = "http3")]
203pub(crate) fn extend_buf(buf: &mut ByteBuffer, mut data: impl Buf) {
204    while data.has_remaining() {
205        let chunk = data.chunk();
206        let n = chunk.len();
207        buf.extend_from_slice(chunk);
208        data.advance(n);
209    }
210}
karyon_jsonrpc/codec.rs

karyon_jsonrpc/
codec.rs
