grpc_webnext/
json_frame.rs

1//! Native-JSON WebSocket frame format for the `+json` codec.
2//!
3//! JSON WebSocket messages are **text** frames carrying a flat object; you read
4//! which field is present to know the frame kind. One WebSocket carries exactly one
5//! stream, so the WS URL *is* the route — frames carry neither `method` nor a stream
6//! id (human-readable):
7//!
8//! ```jsonc
9//! { "metadata": {…}, "timeoutMillis": 5000 } // open (optional; metadata/deadline)
10//! { "message": {…} }                         // data message
11//! { "halfClose": true }                      // client done sending
12//! { "status": { "code": 0, "message": "" } } // terminal (trailer / reset)
13//! ```
14//!
15//! The application `message` is a *native* JSON value (not base64 bytes). Proto
16//! mode uses binary frames (`crate::frame`).
17
18use std::collections::BTreeMap;
19
20use serde::{Deserialize, Serialize};
21use serde_json::Value;
22
23/// Metadata as a JSON object (single value per key).
24pub type JsonMeta = BTreeMap<String, String>;
25
26#[derive(Debug, Serialize, Deserialize)]
27#[serde(rename_all = "camelCase")]
28pub struct JsonStatus {
29    pub code: u32,
30    #[serde(default, skip_serializing_if = "String::is_empty")]
31    pub message: String,
32}
33
34/// One WebSocket text frame. The frame kind is chosen by which of `message` /
35/// `half_close` / `status` (or a bare `metadata` open) is present.
36#[derive(Debug, Default, Serialize, Deserialize)]
37#[serde(rename_all = "camelCase")]
38pub struct JsonFrame {
39    /// Request metadata (on open) or initial response metadata (header) or trailing
40    /// metadata (with `status`).
41    #[serde(default, skip_serializing_if = "Option::is_none")]
42    pub metadata: Option<JsonMeta>,
43    #[serde(default, skip_serializing_if = "Option::is_none")]
44    pub timeout_millis: Option<u32>,
45    /// A data message (native JSON). On an open frame, the optional first message.
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub message: Option<Value>,
48    /// Set → the client is done sending.
49    #[serde(default, skip_serializing_if = "Option::is_none")]
50    pub half_close: Option<bool>,
51    /// Set → terminal status (trailer, or a reset/cancel).
52    #[serde(default, skip_serializing_if = "Option::is_none")]
53    pub status: Option<JsonStatus>,
54}
55
56/// Parse a WebSocket text frame.
57pub fn decode_json_frame(text: &str) -> Result<JsonFrame, serde_json::Error> {
58    serde_json::from_str(text)
59}
60
61/// Serialize a `JsonFrame` to a WebSocket text payload.
62pub fn encode_json_frame(frame: &JsonFrame) -> String {
63    serde_json::to_string(frame).expect("JsonFrame serializes")
64}
65
66// --- Conversions to/from the internal protobuf `Frame` (server-side only) -----
67
68use crate::pb::{frame::Kind, metadatum, Frame, HalfClose, Message, Metadatum, Reset, Subscribe};
69
70fn meta_vec_to_json(items: &[Metadatum]) -> Option<JsonMeta> {
71    let map: JsonMeta = items
72        .iter()
73        .filter_map(|m| match &m.value {
74            Some(metadatum::Value::AsciiValue(s)) => Some((m.key.clone(), s.clone())),
75            _ => None, // binary metadata is omitted from JSON frames
76        })
77        .collect();
78    (!map.is_empty()).then_some(map)
79}
80
81fn json_to_meta_vec(meta: &Option<JsonMeta>) -> Vec<Metadatum> {
82    meta.iter()
83        .flatten()
84        .map(|(k, v)| Metadatum {
85            key: k.clone(),
86            value: Some(metadatum::Value::AsciiValue(v.clone())),
87        })
88        .collect()
89}
90
91fn to_bytes(v: &Value) -> bytes::Bytes {
92    serde_json::to_vec(v).unwrap_or_default().into()
93}
94
95/// Build a `Subscribe` from the open frame, taking the method from the WS route
96/// (the frame itself carries no `method`).
97pub fn json_open_to_subscribe(f: JsonFrame, method: String) -> Subscribe {
98    Subscribe {
99        method,
100        headers: json_to_meta_vec(&f.metadata),
101        timeout_millis: f.timeout_millis.unwrap_or(0),
102        initial_payload: f.message.as_ref().map(to_bytes).unwrap_or_default(),
103    }
104}
105
106/// Convert a post-open client `JsonFrame` (WS text) into the internal `Frame`. The
107/// frame kind is chosen by which field is present.
108pub fn json_frame_to_proto(f: JsonFrame) -> Frame {
109    let kind = if let Some(status) = f.status {
110        // A terminal status from the client is a cancel/reset.
111        Kind::Reset(Reset { status_code: status.code, status_message: status.message })
112    } else if f.half_close == Some(true) {
113        Kind::HalfClose(HalfClose {})
114    } else if let Some(message) = f.message {
115        Kind::Message(Message { payload: to_bytes(&message) })
116    } else {
117        // Bare frame is treated as half-close.
118        Kind::HalfClose(HalfClose {})
119    };
120    Frame { kind: Some(kind) }
121}
122
123/// Convert an internal server `Frame` into a `JsonFrame` for a WS text message.
124/// Returns `None` for kinds with no JSON form (client-only kinds).
125pub fn proto_frame_to_json(frame: &Frame) -> Option<JsonFrame> {
126    let from_bytes = |b: &[u8]| serde_json::from_slice::<Value>(b).unwrap_or(Value::Null);
127    Some(match frame.kind.as_ref()? {
128        Kind::Message(m) => JsonFrame {
129            message: Some(from_bytes(&m.payload)),
130            ..Default::default()
131        },
132        Kind::Header(h) => JsonFrame {
133            metadata: Some(meta_vec_to_json(&h.headers).unwrap_or_default()),
134            ..Default::default()
135        },
136        Kind::Trailer(t) => JsonFrame {
137            status: Some(JsonStatus { code: t.status_code, message: t.status_message.clone() }),
138            metadata: meta_vec_to_json(&t.trailers),
139            ..Default::default()
140        },
141        Kind::Reset(r) => JsonFrame {
142            status: Some(JsonStatus { code: r.status_code, message: r.status_message.clone() }),
143            ..Default::default()
144        },
145        _ => return None,
146    })
147}
grpc_webnext/json_frame.rs

grpc_webnext/
json_frame.rs
