moqtap_codec/draft16/
fields.rs

1use crate::draft16::message::ControlMessage;
2use crate::fields::{FieldMap as Map, FieldValue as Value};
3use crate::kvp::{KeyValuePair, KvpValue};
4use crate::types::*;
5use crate::varint::VarInt;
6
7fn vi(v: u64) -> Value {
8    Value::Uint(v)
9}
10
11fn ns_to_json(ns: &TrackNamespace) -> Value {
12    Value::Array(
13        ns.0.iter().map(|e| Value::Text(String::from_utf8_lossy(e).into_owned())).collect(),
14    )
15}
16
17fn d16_setup_param_name(key: u64) -> Option<&'static str> {
18    match key {
19        0x01 => Some("path"),
20        0x02 => Some("max_request_id"),
21        0x03 => Some("authorization_token"),
22        0x04 => Some("max_auth_token_cache_size"),
23        0x05 => Some("authority"),
24        0x07 => Some("moqt_implementation"),
25        _ => None,
26    }
27}
28
29fn d16_msg_param_name(key: u64) -> Option<&'static str> {
30    match key {
31        0x02 => Some("delivery_timeout"),
32        0x03 => Some("authorization_token"),
33        0x04 => Some("max_cache_duration"),
34        0x08 => Some("expires"),
35        0x09 => Some("largest_object"),
36        0x0e => Some("publisher_priority"),
37        0x10 => Some("forward"),
38        0x20 => Some("subscriber_priority"),
39        0x21 => Some("subscription_filter"),
40        0x22 => Some("group_order"),
41        0x30 => Some("dynamic_groups"),
42        0x32 => Some("new_group_request"),
43        _ => None,
44    }
45}
46
47fn auth_token_to_json_d16(bytes: &[u8]) -> Value {
48    let mut buf = bytes;
49    let alias_type = match VarInt::decode(&mut buf) {
50        Ok(v) => v,
51        Err(_) => return Value::Bytes(bytes.to_vec()),
52    };
53    let at = alias_type.into_inner();
54    let mut o = Map::new();
55    o.insert("alias_type".into(), vi(at));
56    match at {
57        0 | 2 => {
58            if let Ok(ta) = VarInt::decode(&mut buf) {
59                o.insert("token_alias".into(), vi(ta.into_inner()));
60            }
61        }
62        1 => {
63            if let Ok(ta) = VarInt::decode(&mut buf) {
64                o.insert("token_alias".into(), vi(ta.into_inner()));
65            }
66            if let Ok(tt) = VarInt::decode(&mut buf) {
67                o.insert("token_type".into(), vi(tt.into_inner()));
68            }
69            o.insert("token_value".into(), Value::Bytes(buf.to_vec()));
70        }
71        _ => {
72            if let Ok(tt) = VarInt::decode(&mut buf) {
73                o.insert("token_type".into(), vi(tt.into_inner()));
74            }
75            o.insert("token_value".into(), Value::Bytes(buf.to_vec()));
76        }
77    }
78    Value::Map(o)
79}
80
81fn decode_subscription_filter(bytes: &[u8]) -> Value {
82    let mut buf = bytes;
83    let filter_type = VarInt::decode(&mut buf).unwrap().into_inner();
84    let mut obj = Map::new();
85    obj.insert("filter_type".into(), vi(filter_type));
86    match filter_type {
87        3 => {
88            let start_group = VarInt::decode(&mut buf).unwrap().into_inner();
89            let start_object = VarInt::decode(&mut buf).unwrap().into_inner();
90            obj.insert("start_group".into(), vi(start_group));
91            obj.insert("start_object".into(), vi(start_object));
92        }
93        4 => {
94            let start_group = VarInt::decode(&mut buf).unwrap().into_inner();
95            let start_object = VarInt::decode(&mut buf).unwrap().into_inner();
96            let end_group = VarInt::decode(&mut buf).unwrap().into_inner();
97            obj.insert("start_group".into(), vi(start_group));
98            obj.insert("start_object".into(), vi(start_object));
99            obj.insert("end_group".into(), vi(end_group));
100        }
101        _ => {}
102    }
103    Value::Map(obj)
104}
105
106/// Render a draft-16 LARGEST_OBJECT (0x09) parameter value: a Group and an
107/// Object, as two varints.
108///
109/// # Nothing has checked that the value is two varints
110///
111/// Drafts 17 and later give 0x09 a `Location` encoding: their decoders read the
112/// two varints and re-serialise them into the stored value, so what reaches
113/// their extractor is two varints by construction. Draft-16 has no such table.
114/// 0x09 is an odd Type, so `KeyValuePair::decode` keeps whatever
115/// length-prefixed bytes arrived, and `decode_parameters_in` checks duplicates,
116/// authorization tokens, varint value ranges and subscription filters — none of
117/// which looks at 0x09. `KNOWN_MESSAGE_PARAMETERS` admits it and
118/// `check_parameter_scope` permits it on SUBSCRIBE_OK, so a short SUBSCRIBE_OK
119/// carrying `0x09` with an empty value reaches here.
120///
121/// An empty value fails the first read and a single `0x00` fails the *second*,
122/// which is the nastier of the two: the first varint decodes cleanly and the
123/// value looks well formed right up to the point where it is not.
124///
125/// # What a value it cannot read renders as
126///
127/// The raw bytes, as `fields::params` and this file's own
128/// `auth_token_to_json_d16` do. Field extraction runs on a message that has
129/// already decoded, so it has no refusal to give: what a peer sent is what
130/// there is to show.
131fn decode_largest_object(bytes: &[u8]) -> Value {
132    let mut buf = bytes;
133    let Ok(group) = VarInt::decode(&mut buf) else {
134        return Value::Bytes(bytes.to_vec());
135    };
136    let Ok(object) = VarInt::decode(&mut buf) else {
137        return Value::Bytes(bytes.to_vec());
138    };
139    let mut obj = Map::new();
140    obj.insert("group".into(), vi(group.into_inner()));
141    obj.insert("object".into(), vi(object.into_inner()));
142    Value::Map(obj)
143}
144
145fn kvp_to_json_d16_inner(
146    params: &[KeyValuePair],
147    name_fn: fn(u64) -> Option<&'static str>,
148) -> Value {
149    let mut obj = Map::new();
150    let mut unknown = Vec::new();
151
152    for p in params {
153        let key = p.key.into_inner();
154        if let Some(name) = name_fn(key) {
155            match (&p.value, key) {
156                (KvpValue::Bytes(b), 0x21) => {
157                    obj.insert(name.to_string(), decode_subscription_filter(b));
158                }
159                (KvpValue::Bytes(b), 0x09) => {
160                    obj.insert(name.to_string(), decode_largest_object(b));
161                }
162                (KvpValue::Bytes(b), _) if name == "authorization_token" => {
163                    obj.insert(name.to_string(), auth_token_to_json_d16(b));
164                }
165                (KvpValue::Varint(v), _) => {
166                    obj.insert(name.to_string(), vi(v.into_inner()));
167                }
168                (KvpValue::Bytes(b), _) => {
169                    obj.insert(
170                        name.to_string(),
171                        Value::Text(String::from_utf8_lossy(b).into_owned()),
172                    );
173                }
174            }
175        } else {
176            let mut entry = Map::new();
177            entry.insert("id".to_string(), Value::Text(format!("0x{:x}", key)));
178            match &p.value {
179                KvpValue::Varint(v) => {
180                    entry.insert("length".to_string(), vi(v.into_inner()));
181                }
182                KvpValue::Bytes(b) => {
183                    entry.insert("length".to_string(), vi(b.len() as u64));
184                    entry.insert("raw_hex".to_string(), Value::Bytes(b.to_vec()));
185                }
186            }
187            unknown.push(Value::Map(entry));
188        }
189    }
190
191    if !unknown.is_empty() {
192        obj.insert("unknown".to_string(), Value::Array(unknown));
193    }
194
195    Value::Map(obj)
196}
197
198fn kvp_to_json_d16(params: &[KeyValuePair]) -> Value {
199    kvp_to_json_d16_inner(params, d16_msg_param_name)
200}
201
202fn kvp_to_json_d16_setup(params: &[KeyValuePair]) -> Value {
203    kvp_to_json_d16_inner(params, d16_setup_param_name)
204}
205
206/// Track-extension parameter names. `default_publisher_priority` shares key
207/// 0x0e with the message-level `publisher_priority`, so a separate table is
208/// used when rendering `track_extensions` blocks.
209fn d16_track_ext_name(key: u64) -> Option<&'static str> {
210    match key {
211        0x02 => Some("delivery_timeout"),
212        0x04 => Some("max_cache_duration"),
213        0x0e => Some("default_publisher_priority"),
214        _ => None,
215    }
216}
217
218fn kvp_to_json_d16_track_ext(params: &[KeyValuePair]) -> Value {
219    kvp_to_json_d16_inner(params, d16_track_ext_name)
220}
221
222/// This draft's field names for a decoded control message.
223///
224/// Keys are the names this draft gives its fields, in the order it defines
225/// them. An optional field the message did not carry is absent rather than
226/// zero.
227pub fn message_fields(msg: &ControlMessage) -> Map {
228    let obj = match msg {
229        ControlMessage::ClientSetup(m) => {
230            let mut o = Map::new();
231            o.insert("parameters".into(), kvp_to_json_d16_setup(&m.parameters));
232            o
233        }
234        ControlMessage::ServerSetup(m) => {
235            let mut o = Map::new();
236            o.insert("parameters".into(), kvp_to_json_d16_setup(&m.parameters));
237            o
238        }
239        ControlMessage::GoAway(m) => {
240            let mut o = Map::new();
241            o.insert(
242                "new_session_uri".into(),
243                Value::Text(String::from_utf8_lossy(&m.new_session_uri).into_owned()),
244            );
245            o
246        }
247        ControlMessage::MaxRequestId(m) => {
248            let mut o = Map::new();
249            o.insert("max_request_id".into(), vi(m.request_id.into_inner()));
250            o
251        }
252        ControlMessage::RequestsBlocked(m) => {
253            let mut o = Map::new();
254            o.insert("maximum_request_id".into(), vi(m.maximum_request_id.into_inner()));
255            o
256        }
257        ControlMessage::RequestOk(m) => {
258            let mut o = Map::new();
259            o.insert("request_id".into(), vi(m.request_id.into_inner()));
260            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
261            o
262        }
263        ControlMessage::RequestError(m) => {
264            let mut o = Map::new();
265            o.insert("request_id".into(), vi(m.request_id.into_inner()));
266            o.insert("error_code".into(), vi(m.error_code.into_inner()));
267            o.insert("retry_interval".into(), vi(m.retry_interval.into_inner()));
268            o.insert(
269                "reason_phrase".into(),
270                Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
271            );
272            o
273        }
274        ControlMessage::Subscribe(m) => {
275            let mut o = Map::new();
276            o.insert("request_id".into(), vi(m.request_id.into_inner()));
277            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
278            o.insert(
279                "track_name".into(),
280                Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
281            );
282            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
283            o
284        }
285        ControlMessage::SubscribeOk(m) => {
286            let mut o = Map::new();
287            o.insert("request_id".into(), vi(m.request_id.into_inner()));
288            o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
289            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
290            if !m.track_extensions.is_empty() {
291                o.insert("track_extensions".into(), kvp_to_json_d16_track_ext(&m.track_extensions));
292            }
293            o
294        }
295        ControlMessage::RequestUpdate(m) => {
296            let mut o = Map::new();
297            o.insert("request_id".into(), vi(m.request_id.into_inner()));
298            o.insert("existing_request_id".into(), vi(m.existing_request_id.into_inner()));
299            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
300            o
301        }
302        ControlMessage::Unsubscribe(m) => {
303            let mut o = Map::new();
304            o.insert("request_id".into(), vi(m.request_id.into_inner()));
305            o
306        }
307        ControlMessage::Publish(m) => {
308            let mut o = Map::new();
309            o.insert("request_id".into(), vi(m.request_id.into_inner()));
310            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
311            o.insert(
312                "track_name".into(),
313                Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
314            );
315            o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
316            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
317            if !m.track_extensions.is_empty() {
318                o.insert("track_extensions".into(), kvp_to_json_d16_track_ext(&m.track_extensions));
319            }
320            o
321        }
322        ControlMessage::PublishOk(m) => {
323            let mut o = Map::new();
324            o.insert("request_id".into(), vi(m.request_id.into_inner()));
325            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
326            o
327        }
328        ControlMessage::PublishDone(m) => {
329            let mut o = Map::new();
330            o.insert("request_id".into(), vi(m.request_id.into_inner()));
331            o.insert("status_code".into(), vi(m.status_code.into_inner()));
332            o.insert("stream_count".into(), vi(m.stream_count.into_inner()));
333            o.insert(
334                "reason_phrase".into(),
335                Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
336            );
337            o
338        }
339        ControlMessage::PublishNamespace(m) => {
340            let mut o = Map::new();
341            o.insert("request_id".into(), vi(m.request_id.into_inner()));
342            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
343            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
344            o
345        }
346        ControlMessage::PublishNamespaceDone(m) => {
347            let mut o = Map::new();
348            o.insert("request_id".into(), vi(m.request_id.into_inner()));
349            o
350        }
351        ControlMessage::PublishNamespaceCancel(m) => {
352            let mut o = Map::new();
353            o.insert("request_id".into(), vi(m.request_id.into_inner()));
354            o.insert("error_code".into(), vi(m.error_code.into_inner()));
355            o.insert(
356                "reason_phrase".into(),
357                Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
358            );
359            o
360        }
361        ControlMessage::Namespace(m) => {
362            let mut o = Map::new();
363            o.insert("namespace_suffix".into(), ns_to_json(&m.namespace_suffix));
364            o
365        }
366        ControlMessage::NamespaceDone(m) => {
367            let mut o = Map::new();
368            o.insert("namespace_suffix".into(), ns_to_json(&m.namespace_suffix));
369            o
370        }
371        ControlMessage::SubscribeNamespace(m) => {
372            let mut o = Map::new();
373            o.insert("request_id".into(), vi(m.request_id.into_inner()));
374            o.insert("namespace_prefix".into(), ns_to_json(&m.namespace_prefix));
375            o.insert("subscribe_options".into(), vi(m.subscribe_options.into_inner()));
376            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
377            o
378        }
379        ControlMessage::TrackStatus(m) => {
380            let mut o = Map::new();
381            o.insert("request_id".into(), vi(m.request_id.into_inner()));
382            o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
383            o.insert(
384                "track_name".into(),
385                Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
386            );
387            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
388            o
389        }
390        ControlMessage::Fetch(m) => {
391            let mut o = Map::new();
392            o.insert("request_id".into(), vi(m.request_id.into_inner()));
393            o.insert("fetch_type".into(), vi(m.fetch_type as u64));
394            match &m.fetch_payload {
395                crate::draft16::message::FetchPayload::Standalone {
396                    track_namespace,
397                    track_name,
398                    start_group,
399                    start_object,
400                    end_group,
401                    end_object,
402                } => {
403                    o.insert("track_namespace".into(), ns_to_json(track_namespace));
404                    o.insert(
405                        "track_name".into(),
406                        Value::Text(String::from_utf8_lossy(track_name).into_owned()),
407                    );
408                    o.insert("start_group".into(), vi(start_group.into_inner()));
409                    o.insert("start_object".into(), vi(start_object.into_inner()));
410                    o.insert("end_group".into(), vi(end_group.into_inner()));
411                    o.insert("end_object".into(), vi(end_object.into_inner()));
412                }
413                crate::draft16::message::FetchPayload::Joining {
414                    joining_request_id,
415                    joining_start,
416                } => {
417                    o.insert("joining_request_id".into(), vi(joining_request_id.into_inner()));
418                    o.insert("joining_start".into(), vi(joining_start.into_inner()));
419                }
420            }
421            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
422            o
423        }
424        ControlMessage::FetchOk(m) => {
425            let mut o = Map::new();
426            o.insert("request_id".into(), vi(m.request_id.into_inner()));
427            o.insert("end_of_track".into(), vi(m.end_of_track as u64));
428            o.insert("end_group".into(), vi(m.end_group.into_inner()));
429            o.insert("end_object".into(), vi(m.end_object.into_inner()));
430            o.insert("parameters".into(), kvp_to_json_d16(&m.parameters));
431            if !m.track_extensions.is_empty() {
432                o.insert("track_extensions".into(), kvp_to_json_d16_track_ext(&m.track_extensions));
433            }
434            o
435        }
436        ControlMessage::FetchCancel(m) => {
437            let mut o = Map::new();
438            o.insert("request_id".into(), vi(m.request_id.into_inner()));
439            o
440        }
441    };
442    obj
443}
moqtap_codec/draft16/fields.rs

moqtap_codec/draft16/
fields.rs
