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

o_sfu_protocol/signaling/
envelope.rs
