o_sfu_protocol/signaling/
wire_catalog.rs

1use o_sfu_model::RecordingOptions;
2use serde::{Serialize, de::DeserializeOwned};
3use serde_json::Value;
4
5use super::{
6    AuthPayload, ClientBroadcastPayload, Envelope, PeerInfoPayload, PeerLeftPayload,
7    RecordingActionResult, RequestId, ServerBroadcastPayload, SessionDescriptionPayload,
8    SourceDescriptor, StreamIntentPayload, SubscribePayload, TrackBinding, WelcomePayload,
9};
10use crate::shared::{RecordingStateUpdate, UserInfo};
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13enum WireTag {
14    Auth,
15    Broadcast,
16    Info,
17    Offer,
18    PeerInfo,
19    PeerJoined,
20    PeerLeft,
21    Publish,
22    RecordingChange,
23    Renegotiate,
24    StartRecording,
25    StopRecording,
26    Subscribe,
27    Sources,
28    Tracks,
29    Unpublish,
30    Welcome,
31}
32
33impl WireTag {
34    const fn as_str(self) -> &'static str {
35        match self {
36            Self::Auth => "auth",
37            Self::Broadcast => "broadcast",
38            Self::Info => "info",
39            Self::Offer => "offer",
40            Self::PeerInfo => "peerinfo",
41            Self::PeerJoined => "peerjoined",
42            Self::PeerLeft => "peerleft",
43            Self::Publish => "publish",
44            Self::RecordingChange => "recordingchange",
45            Self::Renegotiate => "renegotiate",
46            Self::StartRecording => "startrecording",
47            Self::StopRecording => "stoprecording",
48            Self::Subscribe => "subscribe",
49            Self::Sources => "sources",
50            Self::Tracks => "tracks",
51            Self::Unpublish => "unpublish",
52            Self::Welcome => "welcome",
53        }
54    }
55}
56
57type EntryDecode<T> = fn(WireTag, Option<Value>) -> Result<T, EnvelopeDecodeError>;
58
59#[derive(Clone, Copy)]
60struct EnvelopeEntry<T> {
61    tag: WireTag,
62    decode: EntryDecode<T>,
63}
64
65impl<T> EnvelopeEntry<T> {
66    const fn new(tag: WireTag, decode: EntryDecode<T>) -> Self {
67        Self { tag, decode }
68    }
69
70    fn decode(&self, payload: Option<Value>) -> Result<T, EnvelopeDecodeError> {
71        (self.decode)(self.tag, payload)
72    }
73}
74
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub enum EnvelopeDecodeError {
77    UnknownTag(String),
78    InvalidPayload(String),
79    UnexpectedPayload(String),
80}
81
82#[derive(Debug, Clone, PartialEq, Eq)]
83pub enum ClientMessage {
84    Auth(AuthPayload),
85    Publish(StreamIntentPayload),
86    Unpublish(StreamIntentPayload),
87    Subscribe(SubscribePayload),
88    Info(UserInfo),
89    Broadcast(ClientBroadcastPayload),
90}
91
92impl ClientMessage {
93    const ENTRIES: &'static [EnvelopeEntry<Self>] = &[
94        EnvelopeEntry::new(WireTag::Auth, |tag, payload| {
95            decode_payload(tag, payload, Self::Auth)
96        }),
97        EnvelopeEntry::new(WireTag::Publish, |tag, payload| {
98            decode_payload(tag, payload, Self::Publish)
99        }),
100        EnvelopeEntry::new(WireTag::Unpublish, |tag, payload| {
101            decode_payload(tag, payload, Self::Unpublish)
102        }),
103        EnvelopeEntry::new(WireTag::Subscribe, |tag, payload| {
104            decode_payload(tag, payload, Self::Subscribe)
105        }),
106        EnvelopeEntry::new(WireTag::Info, |tag, payload| {
107            decode_payload(tag, payload, Self::Info)
108        }),
109        EnvelopeEntry::new(WireTag::Broadcast, |tag, payload| {
110            decode_payload(tag, payload, Self::Broadcast)
111        }),
112    ];
113
114    pub(crate) fn into_envelope(self) -> Result<Envelope, serde_json::Error> {
115        match self {
116            Self::Auth(payload) => encode_message(WireTag::Auth, payload),
117            Self::Publish(payload) => encode_message(WireTag::Publish, payload),
118            Self::Unpublish(payload) => encode_message(WireTag::Unpublish, payload),
119            Self::Subscribe(payload) => encode_message(WireTag::Subscribe, payload),
120            Self::Info(payload) => encode_message(WireTag::Info, payload),
121            Self::Broadcast(payload) => encode_message(WireTag::Broadcast, payload),
122        }
123    }
124
125    pub(crate) fn decode(tag: &str, payload: Option<Value>) -> Result<Self, EnvelopeDecodeError> {
126        decode_entry(tag, payload, Self::ENTRIES)
127    }
128}
129
130#[derive(Debug, Clone, PartialEq, Eq)]
131pub enum ClientRequest {
132    StartRecording(RecordingOptions),
133    StopRecording,
134}
135
136impl ClientRequest {
137    const ENTRIES: &'static [EnvelopeEntry<Self>] = &[
138        EnvelopeEntry::new(WireTag::StartRecording, |tag, payload| {
139            decode_payload(tag, payload, Self::StartRecording)
140        }),
141        EnvelopeEntry::new(WireTag::StopRecording, |tag, payload| {
142            decode_empty(tag, payload.as_ref(), Self::StopRecording)
143        }),
144    ];
145
146    pub(crate) fn into_envelope(
147        self,
148        request_id: RequestId,
149    ) -> Result<Envelope, serde_json::Error> {
150        match self {
151            Self::StartRecording(payload) => {
152                encode_request(WireTag::StartRecording, request_id, payload)
153            }
154            Self::StopRecording => Ok(Envelope::request(
155                WireTag::StopRecording.as_str(),
156                request_id,
157                None,
158            )),
159        }
160    }
161
162    pub(crate) fn decode(tag: &str, payload: Option<Value>) -> Result<Self, EnvelopeDecodeError> {
163        decode_entry(tag, payload, Self::ENTRIES)
164    }
165}
166
167#[derive(Debug, Clone, PartialEq, Eq)]
168pub enum ServerRequest {
169    Offer(SessionDescriptionPayload),
170    Renegotiate(SessionDescriptionPayload),
171}
172
173impl ServerRequest {
174    const ENTRIES: &'static [EnvelopeEntry<Self>] = &[
175        EnvelopeEntry::new(WireTag::Offer, |tag, payload| {
176            decode_payload(tag, payload, Self::Offer)
177        }),
178        EnvelopeEntry::new(WireTag::Renegotiate, |tag, payload| {
179            decode_payload(tag, payload, Self::Renegotiate)
180        }),
181    ];
182
183    /// encode one server-authored request into the Odoo wire envelope
184    ///
185    /// # Errors
186    ///
187    /// returns an error when the typed payload cannot be serialized into the
188    /// JSON envelope payload
189    pub fn into_envelope(self, request_id: RequestId) -> Result<Envelope, serde_json::Error> {
190        match self {
191            Self::Offer(payload) => encode_request(WireTag::Offer, request_id, payload),
192            Self::Renegotiate(payload) => encode_request(WireTag::Renegotiate, request_id, payload),
193        }
194    }
195
196    pub(crate) fn decode(tag: &str, payload: Option<Value>) -> Result<Self, EnvelopeDecodeError> {
197        decode_entry(tag, payload, Self::ENTRIES)
198    }
199}
200
201#[derive(Debug, Clone, PartialEq, Eq)]
202pub enum ClientResponse {
203    Offer(SessionDescriptionPayload),
204    Renegotiate(SessionDescriptionPayload),
205}
206
207impl ClientResponse {
208    const ENTRIES: &'static [EnvelopeEntry<Self>] = &[
209        EnvelopeEntry::new(WireTag::Offer, |tag, payload| {
210            decode_payload(tag, payload, Self::Offer)
211        }),
212        EnvelopeEntry::new(WireTag::Renegotiate, |tag, payload| {
213            decode_payload(tag, payload, Self::Renegotiate)
214        }),
215    ];
216
217    pub(crate) fn into_envelope(
218        self,
219        response_to: RequestId,
220    ) -> Result<Envelope, serde_json::Error> {
221        match self {
222            Self::Offer(payload) => encode_response(WireTag::Offer, response_to, payload),
223            Self::Renegotiate(payload) => {
224                encode_response(WireTag::Renegotiate, response_to, payload)
225            }
226        }
227    }
228
229    pub(crate) fn decode(tag: &str, payload: Option<Value>) -> Result<Self, EnvelopeDecodeError> {
230        decode_entry(tag, payload, Self::ENTRIES)
231    }
232}
233
234#[derive(Debug, Clone, PartialEq, Eq)]
235pub enum ServerMessage {
236    Welcome(WelcomePayload),
237    Tracks(Vec<TrackBinding>),
238    Sources(Vec<SourceDescriptor>),
239    PeerInfo(PeerInfoPayload),
240    PeerJoined(PeerInfoPayload),
241    PeerLeft(PeerLeftPayload),
242    Broadcast(ServerBroadcastPayload),
243    RecordingChange(RecordingStateUpdate),
244}
245
246impl ServerMessage {
247    const ENTRIES: &'static [EnvelopeEntry<Self>] = &[
248        EnvelopeEntry::new(WireTag::Welcome, |tag, payload| {
249            decode_payload(tag, payload, Self::Welcome)
250        }),
251        EnvelopeEntry::new(WireTag::Tracks, |tag, payload| {
252            decode_payload(tag, payload, Self::Tracks)
253        }),
254        EnvelopeEntry::new(WireTag::Sources, |tag, payload| {
255            decode_payload(tag, payload, Self::Sources)
256        }),
257        EnvelopeEntry::new(WireTag::PeerInfo, |tag, payload| {
258            decode_payload(tag, payload, Self::PeerInfo)
259        }),
260        EnvelopeEntry::new(WireTag::PeerJoined, |tag, payload| {
261            decode_payload(tag, payload, Self::PeerJoined)
262        }),
263        EnvelopeEntry::new(WireTag::PeerLeft, |tag, payload| {
264            decode_payload(tag, payload, Self::PeerLeft)
265        }),
266        EnvelopeEntry::new(WireTag::Broadcast, |tag, payload| {
267            decode_payload(tag, payload, Self::Broadcast)
268        }),
269        EnvelopeEntry::new(WireTag::RecordingChange, |tag, payload| {
270            decode_payload(tag, payload, Self::RecordingChange)
271        }),
272    ];
273
274    /// encode one server message into the Odoo wire envelope
275    ///
276    /// # Errors
277    ///
278    /// returns an error when the typed payload cannot be serialized into the
279    /// JSON envelope payload
280    pub fn into_envelope(self) -> Result<Envelope, serde_json::Error> {
281        match self {
282            Self::Welcome(payload) => encode_message(WireTag::Welcome, payload),
283            Self::Tracks(payload) => encode_message(WireTag::Tracks, payload),
284            Self::Sources(payload) => encode_message(WireTag::Sources, payload),
285            Self::PeerInfo(payload) => encode_message(WireTag::PeerInfo, payload),
286            Self::PeerJoined(payload) => encode_message(WireTag::PeerJoined, payload),
287            Self::PeerLeft(payload) => encode_message(WireTag::PeerLeft, payload),
288            Self::Broadcast(payload) => encode_message(WireTag::Broadcast, payload),
289            Self::RecordingChange(payload) => encode_message(WireTag::RecordingChange, payload),
290        }
291    }
292
293    pub(crate) fn decode(tag: &str, payload: Option<Value>) -> Result<Self, EnvelopeDecodeError> {
294        decode_entry(tag, payload, Self::ENTRIES)
295    }
296}
297
298#[derive(Debug, Clone, PartialEq, Eq)]
299pub enum ServerResponse {
300    StartRecording(RecordingActionResult),
301    StopRecording(RecordingActionResult),
302}
303
304impl ServerResponse {
305    const ENTRIES: &'static [EnvelopeEntry<Self>] = &[
306        EnvelopeEntry::new(WireTag::StartRecording, |tag, payload| {
307            decode_payload(tag, payload, Self::StartRecording)
308        }),
309        EnvelopeEntry::new(WireTag::StopRecording, |tag, payload| {
310            decode_payload(tag, payload, Self::StopRecording)
311        }),
312    ];
313
314    /// encode one server response into the Odoo wire envelope
315    ///
316    /// # Errors
317    ///
318    /// returns an error when the typed payload cannot be serialized into the
319    /// JSON envelope payload
320    pub fn into_envelope(self, response_to: RequestId) -> Result<Envelope, serde_json::Error> {
321        match self {
322            Self::StartRecording(payload) => {
323                encode_response(WireTag::StartRecording, response_to, payload)
324            }
325            Self::StopRecording(payload) => {
326                encode_response(WireTag::StopRecording, response_to, payload)
327            }
328        }
329    }
330
331    pub(crate) fn decode(tag: &str, payload: Option<Value>) -> Result<Self, EnvelopeDecodeError> {
332        decode_entry(tag, payload, Self::ENTRIES)
333    }
334}
335
336fn encode_message<T: Serialize>(tag: WireTag, payload: T) -> Result<Envelope, serde_json::Error> {
337    Ok(Envelope::message(
338        tag.as_str(),
339        Some(serde_json::to_value(payload)?),
340    ))
341}
342
343fn encode_request<T: Serialize>(
344    tag: WireTag,
345    request_id: RequestId,
346    payload: T,
347) -> Result<Envelope, serde_json::Error> {
348    Ok(Envelope::request(
349        tag.as_str(),
350        request_id,
351        Some(serde_json::to_value(payload)?),
352    ))
353}
354
355fn encode_response<T: Serialize>(
356    tag: WireTag,
357    response_to: RequestId,
358    payload: T,
359) -> Result<Envelope, serde_json::Error> {
360    Ok(Envelope::response(
361        tag.as_str(),
362        response_to,
363        Some(serde_json::to_value(payload)?),
364    ))
365}
366
367fn decode_entry<T>(
368    tag: &str,
369    payload: Option<Value>,
370    entries: &[EnvelopeEntry<T>],
371) -> Result<T, EnvelopeDecodeError> {
372    entries
373        .iter()
374        .find(|entry| entry.tag.as_str() == tag)
375        .ok_or_else(|| unknown_tag(tag))?
376        .decode(payload)
377}
378
379fn decode_payload<T, P>(
380    tag: WireTag,
381    payload: Option<Value>,
382    build: fn(P) -> T,
383) -> Result<T, EnvelopeDecodeError>
384where
385    P: DeserializeOwned,
386{
387    parse_payload(tag.as_str(), payload).map(build)
388}
389
390fn decode_empty<T>(
391    tag: WireTag,
392    payload: Option<&Value>,
393    value: T,
394) -> Result<T, EnvelopeDecodeError> {
395    ensure_empty_payload(tag.as_str(), payload)?;
396    Ok(value)
397}
398
399fn unknown_tag(tag: &str) -> EnvelopeDecodeError {
400    EnvelopeDecodeError::UnknownTag(tag.to_owned())
401}
402
403fn parse_payload<T: DeserializeOwned>(
404    tag: &str,
405    payload: Option<Value>,
406) -> Result<T, EnvelopeDecodeError> {
407    serde_json::from_value(
408        payload.ok_or_else(|| EnvelopeDecodeError::InvalidPayload(tag.to_owned()))?,
409    )
410    .map_err(|_error| EnvelopeDecodeError::InvalidPayload(tag.to_owned()))
411}
412
413fn ensure_empty_payload(tag: &str, payload: Option<&Value>) -> Result<(), EnvelopeDecodeError> {
414    if payload.is_some() {
415        return Err(EnvelopeDecodeError::UnexpectedPayload(tag.to_owned()));
416    }
417    Ok(())
418}
o_sfu_protocol/signaling/wire_catalog.rs

o_sfu_protocol/signaling/
wire_catalog.rs
