o_sfu_protocol/signaling/
envelope.rs

1use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
2use serde_json::Value;
3
4pub type EnvelopeBatch = Vec<Envelope>;
5
6pub const MAX_ENVELOPE_BATCH_LEN: usize = 64;
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
9pub enum EnvelopeBatchDecodeError {
10    #[error("invalid JSON envelope batch")]
11    InvalidJson,
12    #[error("envelope batch contains {actual} entries, exceeding the limit of {limit}")]
13    BatchTooLarge { actual: usize, limit: usize },
14    #[error("invalid envelope routing metadata")]
15    InvalidRoutingMetadata,
16}
17
18#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
19#[serde(transparent)]
20pub struct RequestId(String);
21
22impl RequestId {
23    #[must_use]
24    pub fn new(value: impl Into<String>) -> Self {
25        Self(value.into())
26    }
27
28    #[must_use]
29    pub fn as_str(&self) -> &str {
30        self.0.as_str()
31    }
32}
33
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub(super) enum EnvelopeRoute {
36    Message,
37    Request(RequestId),
38    Response(RequestId),
39}
40
41impl EnvelopeRoute {
42    fn from_wire(request_id: Option<RequestId>, response_to: Option<RequestId>) -> Option<Self> {
43        match (request_id, response_to) {
44            (None, None) => Some(Self::Message),
45            (Some(request_id), None) => Some(Self::Request(request_id)),
46            (None, Some(response_to)) => Some(Self::Response(response_to)),
47            (Some(_), Some(_)) => None,
48        }
49    }
50}
51
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub struct Envelope {
54    tag: String,
55    payload: Option<Value>,
56    route: EnvelopeRoute,
57}
58
59#[derive(Deserialize)]
60struct WireEnvelope {
61    #[serde(rename = "t")]
62    tag: String,
63    #[serde(rename = "p")]
64    payload: Option<Value>,
65    #[serde(rename = "q")]
66    request_id: Option<RequestId>,
67    #[serde(rename = "r")]
68    response_to: Option<RequestId>,
69}
70
71#[derive(Serialize)]
72struct WireEnvelopeRef<'a> {
73    #[serde(rename = "t")]
74    tag: &'a str,
75    #[serde(rename = "p", skip_serializing_if = "Option::is_none")]
76    payload: Option<&'a Value>,
77    #[serde(rename = "q", skip_serializing_if = "Option::is_none")]
78    request_id: Option<&'a RequestId>,
79    #[serde(rename = "r", skip_serializing_if = "Option::is_none")]
80    response_to: Option<&'a RequestId>,
81}
82
83impl Envelope {
84    #[must_use]
85    pub fn message(tag: &str, payload: Option<Value>) -> Self {
86        Self {
87            tag: tag.to_owned(),
88            payload,
89            route: EnvelopeRoute::Message,
90        }
91    }
92
93    #[must_use]
94    pub fn request(tag: &str, request_id: RequestId, payload: Option<Value>) -> Self {
95        Self {
96            tag: tag.to_owned(),
97            payload,
98            route: EnvelopeRoute::Request(request_id),
99        }
100    }
101
102    #[must_use]
103    pub fn response(tag: &str, response_to: RequestId, payload: Option<Value>) -> Self {
104        Self {
105            tag: tag.to_owned(),
106            payload,
107            route: EnvelopeRoute::Response(response_to),
108        }
109    }
110
111    pub(super) fn into_parts(self) -> (String, Option<Value>, EnvelopeRoute) {
112        (self.tag, self.payload, self.route)
113    }
114}
115
116impl WireEnvelope {
117    fn into_envelope(self) -> Option<Envelope> {
118        let route = EnvelopeRoute::from_wire(self.request_id, self.response_to)?;
119
120        Some(Envelope {
121            tag: self.tag,
122            payload: self.payload,
123            route,
124        })
125    }
126}
127
128/// Decode a websocket envelope batch while preserving route validation errors
129/// and checking a caller-provided batch limit before route conversion.
130///
131/// # Errors
132///
133/// Returns `InvalidJson` when the payload cannot be decoded as the envelope
134/// wire shape. Returns `BatchTooLarge` when the decoded batch exceeds `limit`.
135/// Returns `InvalidRoutingMetadata` when an envelope contains both a request id
136/// and response id.
137pub fn decode_envelope_batch(
138    payload: &str,
139    limit: usize,
140) -> Result<EnvelopeBatch, EnvelopeBatchDecodeError> {
141    let wire_batch = serde_json::from_str::<Vec<WireEnvelope>>(payload)
142        .map_err(|_error| EnvelopeBatchDecodeError::InvalidJson)?;
143    if wire_batch.len() > limit {
144        return Err(EnvelopeBatchDecodeError::BatchTooLarge {
145            actual: wire_batch.len(),
146            limit,
147        });
148    }
149
150    wire_batch
151        .into_iter()
152        .map(WireEnvelope::into_envelope)
153        .collect::<Option<EnvelopeBatch>>()
154        .ok_or(EnvelopeBatchDecodeError::InvalidRoutingMetadata)
155}
156
157impl Serialize for Envelope {
158    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
159    where
160        S: Serializer,
161    {
162        let (request_id, response_to) = match &self.route {
163            EnvelopeRoute::Message => (None, None),
164            EnvelopeRoute::Request(request_id) => (Some(request_id), None),
165            EnvelopeRoute::Response(response_to) => (None, Some(response_to)),
166        };
167
168        WireEnvelopeRef {
169            tag: self.tag.as_str(),
170            payload: self.payload.as_ref(),
171            request_id,
172            response_to,
173        }
174        .serialize(serializer)
175    }
176}
177
178impl<'de> Deserialize<'de> for Envelope {
179    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
180    where
181        D: Deserializer<'de>,
182    {
183        WireEnvelope::deserialize(deserializer)?
184            .into_envelope()
185            .ok_or_else(|| de::Error::custom("envelope cannot be both request and response"))
186    }
187}
o_sfu_protocol/signaling/envelope.rs

o_sfu_protocol/signaling/
envelope.rs
