moqtap_codec/draft12/
message.rs

1//! Draft-12 control message encoding and decoding.
2//!
3//! Key changes from draft-11:
4//! - Subscribe: `track_alias` removed (moved to SubscribeOk)
5//! - SubscribeOk: `track_alias` added (after request_id)
6//! - SubscribeError: trailing `track_alias` removed
7//! - New messages: Publish (0x1D), PublishOk (0x1E), PublishError (0x1F)
8//! - Same message type IDs for all other messages
9//! - Same framing: type_id(vi) + payload_length(16) + payload
10//! - Same even/odd KVP encoding
11
12use crate::auth_token::{AuthorizationToken, AUTH_TOKEN_PARAMETER};
13use crate::error::{
14    CodecError, MAX_FULL_TRACK_NAME_LENGTH, MAX_GOAWAY_URI_LENGTH, MAX_MESSAGE_LENGTH,
15    MAX_REASON_PHRASE_LENGTH,
16};
17use crate::kvp::{KeyValuePair, KvpValue};
18use crate::types::{self, *};
19pub use crate::types::{check_group_range, check_location_range, check_open_ended_group_range};
20use crate::varint::VarInt;
21use bytes::{Buf, BufMut};
22
23/// Control message type IDs (draft-12).
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25#[repr(u64)]
26pub enum MessageType {
27    SubscribeUpdate = 0x02,
28    Subscribe = 0x03,
29    SubscribeOk = 0x04,
30    SubscribeError = 0x05,
31    Announce = 0x06,
32    AnnounceOk = 0x07,
33    AnnounceError = 0x08,
34    Unannounce = 0x09,
35    Unsubscribe = 0x0A,
36    SubscribeDone = 0x0B,
37    AnnounceCancel = 0x0C,
38    TrackStatusRequest = 0x0D,
39    TrackStatus = 0x0E,
40    GoAway = 0x10,
41    SubscribeAnnounces = 0x11,
42    SubscribeAnnouncesOk = 0x12,
43    SubscribeAnnouncesError = 0x13,
44    UnsubscribeAnnounces = 0x14,
45    MaxRequestId = 0x15,
46    Fetch = 0x16,
47    FetchCancel = 0x17,
48    FetchOk = 0x18,
49    FetchError = 0x19,
50    RequestsBlocked = 0x1A,
51    Publish = 0x1D,
52    PublishOk = 0x1E,
53    PublishError = 0x1F,
54    ClientSetup = 0x20,
55    ServerSetup = 0x21,
56}
57
58impl MessageType {
59    pub fn from_id(id: u64) -> Option<Self> {
60        match id {
61            0x02 => Some(MessageType::SubscribeUpdate),
62            0x03 => Some(MessageType::Subscribe),
63            0x04 => Some(MessageType::SubscribeOk),
64            0x05 => Some(MessageType::SubscribeError),
65            0x06 => Some(MessageType::Announce),
66            0x07 => Some(MessageType::AnnounceOk),
67            0x08 => Some(MessageType::AnnounceError),
68            0x09 => Some(MessageType::Unannounce),
69            0x0A => Some(MessageType::Unsubscribe),
70            0x0B => Some(MessageType::SubscribeDone),
71            0x0C => Some(MessageType::AnnounceCancel),
72            0x0D => Some(MessageType::TrackStatusRequest),
73            0x0E => Some(MessageType::TrackStatus),
74            0x10 => Some(MessageType::GoAway),
75            0x11 => Some(MessageType::SubscribeAnnounces),
76            0x12 => Some(MessageType::SubscribeAnnouncesOk),
77            0x13 => Some(MessageType::SubscribeAnnouncesError),
78            0x14 => Some(MessageType::UnsubscribeAnnounces),
79            0x15 => Some(MessageType::MaxRequestId),
80            0x16 => Some(MessageType::Fetch),
81            0x17 => Some(MessageType::FetchCancel),
82            0x18 => Some(MessageType::FetchOk),
83            0x19 => Some(MessageType::FetchError),
84            0x1A => Some(MessageType::RequestsBlocked),
85            0x1D => Some(MessageType::Publish),
86            0x1E => Some(MessageType::PublishOk),
87            0x1F => Some(MessageType::PublishError),
88            0x20 => Some(MessageType::ClientSetup),
89            0x21 => Some(MessageType::ServerSetup),
90            _ => None,
91        }
92    }
93
94    pub fn id(&self) -> u64 {
95        *self as u64
96    }
97}
98
99// ============================================================
100// Session Lifecycle Messages
101// ============================================================
102
103#[derive(Debug, Clone, PartialEq, Eq)]
104pub struct ClientSetup {
105    pub supported_versions: Vec<VarInt>,
106    pub parameters: Vec<KeyValuePair>,
107}
108
109#[derive(Debug, Clone, PartialEq, Eq)]
110pub struct ServerSetup {
111    pub selected_version: VarInt,
112    pub parameters: Vec<KeyValuePair>,
113}
114
115#[derive(Debug, Clone, PartialEq, Eq)]
116pub struct GoAway {
117    pub new_session_uri: Vec<u8>,
118}
119
120#[derive(Debug, Clone, PartialEq, Eq)]
121pub struct MaxRequestId {
122    pub request_id: VarInt,
123}
124
125#[derive(Debug, Clone, PartialEq, Eq)]
126pub struct RequestsBlocked {
127    pub maximum_request_id: VarInt,
128}
129
130// ============================================================
131// Subscribe Messages
132// ============================================================
133
134/// SUBSCRIBE message (type 0x03). Draft-12: no track_alias (moved to SubscribeOk).
135#[derive(Debug, Clone, PartialEq, Eq)]
136pub struct Subscribe {
137    pub request_id: VarInt,
138    pub track_namespace: TrackNamespace,
139    pub track_name: Vec<u8>,
140    pub subscriber_priority: u8,
141    pub group_order: GroupOrder,
142    pub forward: Forward,
143    pub filter_type: VarInt,
144    pub start_group: Option<VarInt>,
145    pub start_object: Option<VarInt>,
146    pub end_group: Option<VarInt>,
147    pub parameters: Vec<KeyValuePair>,
148}
149
150/// SUBSCRIBE_OK message (type 0x04). Draft-12: gains track_alias.
151#[derive(Debug, Clone, PartialEq, Eq)]
152pub struct SubscribeOk {
153    pub request_id: VarInt,
154    pub track_alias: VarInt,
155    pub expires: VarInt,
156    pub group_order: GroupOrder,
157    pub content_exists: ContentExists,
158    pub largest_location: Option<Location>,
159    pub parameters: Vec<KeyValuePair>,
160}
161
162/// SUBSCRIBE_ERROR message (type 0x05). Draft-12: no trailing track_alias.
163#[derive(Debug, Clone, PartialEq, Eq)]
164pub struct SubscribeError {
165    pub request_id: VarInt,
166    pub error_code: VarInt,
167    pub reason_phrase: Vec<u8>,
168}
169
170#[derive(Debug, Clone, PartialEq, Eq)]
171pub struct SubscribeUpdate {
172    pub request_id: VarInt,
173    pub start_group: VarInt,
174    pub start_object: VarInt,
175    pub end_group: VarInt,
176    pub subscriber_priority: u8,
177    pub forward: Forward,
178    pub parameters: Vec<KeyValuePair>,
179}
180
181#[derive(Debug, Clone, PartialEq, Eq)]
182pub struct SubscribeDone {
183    pub request_id: VarInt,
184    pub status_code: VarInt,
185    pub stream_count: VarInt,
186    pub reason_phrase: Vec<u8>,
187}
188
189#[derive(Debug, Clone, PartialEq, Eq)]
190pub struct Unsubscribe {
191    pub request_id: VarInt,
192}
193
194// ============================================================
195// Announce Messages
196// ============================================================
197
198#[derive(Debug, Clone, PartialEq, Eq)]
199pub struct Announce {
200    pub request_id: VarInt,
201    pub track_namespace: TrackNamespace,
202    pub parameters: Vec<KeyValuePair>,
203}
204
205#[derive(Debug, Clone, PartialEq, Eq)]
206pub struct AnnounceOk {
207    pub request_id: VarInt,
208}
209
210#[derive(Debug, Clone, PartialEq, Eq)]
211pub struct AnnounceError {
212    pub request_id: VarInt,
213    pub error_code: VarInt,
214    pub reason_phrase: Vec<u8>,
215}
216
217#[derive(Debug, Clone, PartialEq, Eq)]
218pub struct AnnounceCancel {
219    pub track_namespace: TrackNamespace,
220    pub error_code: VarInt,
221    pub reason_phrase: Vec<u8>,
222}
223
224#[derive(Debug, Clone, PartialEq, Eq)]
225pub struct Unannounce {
226    pub track_namespace: TrackNamespace,
227}
228
229// ============================================================
230// Subscribe Announces Messages
231// ============================================================
232
233#[derive(Debug, Clone, PartialEq, Eq)]
234pub struct SubscribeAnnounces {
235    pub request_id: VarInt,
236    pub track_namespace_prefix: TrackNamespace,
237    pub parameters: Vec<KeyValuePair>,
238}
239
240#[derive(Debug, Clone, PartialEq, Eq)]
241pub struct SubscribeAnnouncesOk {
242    pub request_id: VarInt,
243}
244
245#[derive(Debug, Clone, PartialEq, Eq)]
246pub struct SubscribeAnnouncesError {
247    pub request_id: VarInt,
248    pub error_code: VarInt,
249    pub reason_phrase: Vec<u8>,
250}
251
252#[derive(Debug, Clone, PartialEq, Eq)]
253pub struct UnsubscribeAnnounces {
254    pub track_namespace_prefix: TrackNamespace,
255}
256
257// ============================================================
258// Track Status Messages
259// ============================================================
260
261#[derive(Debug, Clone, PartialEq, Eq)]
262pub struct TrackStatusRequest {
263    pub request_id: VarInt,
264    pub track_namespace: TrackNamespace,
265    pub track_name: Vec<u8>,
266    pub parameters: Vec<KeyValuePair>,
267}
268
269#[derive(Debug, Clone, PartialEq, Eq)]
270pub struct TrackStatus {
271    pub request_id: VarInt,
272    pub status_code: VarInt,
273    pub largest_location: Location,
274    pub parameters: Vec<KeyValuePair>,
275}
276
277// ============================================================
278// Fetch Messages
279// ============================================================
280
281#[derive(Debug, Clone, Copy, PartialEq, Eq)]
282#[repr(u64)]
283pub enum FetchType {
284    Standalone = 1,
285    RelativeJoining = 2,
286    AbsoluteJoining = 3,
287}
288
289impl FetchType {
290    pub fn from_u64(v: u64) -> Option<Self> {
291        match v {
292            1 => Some(FetchType::Standalone),
293            2 => Some(FetchType::RelativeJoining),
294            3 => Some(FetchType::AbsoluteJoining),
295            _ => None,
296        }
297    }
298}
299
300#[derive(Debug, Clone, PartialEq, Eq)]
301pub struct Fetch {
302    pub request_id: VarInt,
303    pub subscriber_priority: u8,
304    pub group_order: GroupOrder,
305    pub fetch_type: FetchType,
306    pub fetch_payload: FetchPayload,
307    pub parameters: Vec<KeyValuePair>,
308}
309
310#[derive(Debug, Clone, PartialEq, Eq)]
311pub enum FetchPayload {
312    Standalone {
313        track_namespace: TrackNamespace,
314        track_name: Vec<u8>,
315        start_group: VarInt,
316        start_object: VarInt,
317        end_group: VarInt,
318        end_object: VarInt,
319    },
320    Joining {
321        /// The Request ID of the subscription this fetch joins.
322        ///
323        /// Section 8.16 names the field Joining Request ID. Draft-11 spelled
324        /// the same field Joining Subscribe ID, and the two words are not
325        /// interchangeable here: a Request ID is drawn from the one space
326        /// every request shares, so the value identifies a request that
327        /// happens to be a subscription rather than a subscription in a
328        /// space of its own.
329        joining_request_id: VarInt,
330        /// The joining start: an offset for a relative fetch, a group for an
331        /// absolute one.
332        joining_start: VarInt,
333    },
334}
335
336#[derive(Debug, Clone, PartialEq, Eq)]
337pub struct FetchOk {
338    pub request_id: VarInt,
339    pub group_order: GroupOrder,
340    /// Whether the end of the track has been reached.
341    ///
342    /// Held as a raw byte rather than an enum: the draft describes 1 and 0 and
343    /// says nothing about any other value, where it does call an out-of-range
344    /// Group Order, Forward or Content Exists a protocol error. Refusing a 2
345    /// here would be this codec's rule and not the draft's.
346    pub end_of_track: u8,
347    pub end_location: Location,
348    pub parameters: Vec<KeyValuePair>,
349}
350
351#[derive(Debug, Clone, PartialEq, Eq)]
352pub struct FetchError {
353    pub request_id: VarInt,
354    pub error_code: VarInt,
355    pub reason_phrase: Vec<u8>,
356}
357
358#[derive(Debug, Clone, PartialEq, Eq)]
359pub struct FetchCancel {
360    pub request_id: VarInt,
361}
362
363// ============================================================
364// Publish Messages (NEW in draft-12)
365// ============================================================
366
367/// PUBLISH message (type 0x1D).
368#[derive(Debug, Clone, PartialEq, Eq)]
369pub struct Publish {
370    pub request_id: VarInt,
371    pub track_namespace: TrackNamespace,
372    pub track_name: Vec<u8>,
373    pub track_alias: VarInt,
374    pub group_order: GroupOrder,
375    pub content_exists: ContentExists,
376    pub largest_location: Option<Location>,
377    pub forward: Forward,
378    pub parameters: Vec<KeyValuePair>,
379}
380
381/// PUBLISH_OK message (type 0x1E).
382#[derive(Debug, Clone, PartialEq, Eq)]
383pub struct PublishOk {
384    pub request_id: VarInt,
385    pub forward: Forward,
386    pub subscriber_priority: u8,
387    pub group_order: GroupOrder,
388    pub filter_type: VarInt,
389    pub start_group: Option<VarInt>,
390    pub start_object: Option<VarInt>,
391    pub end_group: Option<VarInt>,
392    pub parameters: Vec<KeyValuePair>,
393}
394
395/// PUBLISH_ERROR message (type 0x1F).
396#[derive(Debug, Clone, PartialEq, Eq)]
397pub struct PublishError {
398    pub request_id: VarInt,
399    pub error_code: VarInt,
400    pub reason_phrase: Vec<u8>,
401}
402
403/// Take one byte, or report the end of the buffer instead of panicking.
404fn read_u8(buf: &mut impl Buf) -> Result<u8, CodecError> {
405    if !buf.has_remaining() {
406        return Err(CodecError::UnexpectedEnd);
407    }
408    Ok(buf.get_u8())
409}
410
411/// Read a Group Order from a message that lets the publisher choose.
412///
413/// Every figure draws this field as a single byte, and the values are 0x1
414/// Ascending, 0x2 Descending, and 0x0 for "the original publisher's Group Order
415/// SHOULD be used".
416///
417/// Reading it as a varint instead is invisible for all three legal values —
418/// each is a single byte below 64, where the two encodings coincide — and
419/// diverges on everything a peer may send that is not legal. A two-byte varint
420/// holding 1 passes as Ascending and shifts every field after it, while the
421/// declared message length still adds up.
422fn read_group_order(buf: &mut impl Buf) -> Result<GroupOrder, CodecError> {
423    GroupOrder::from_u8(read_u8(buf)?).ok_or(CodecError::InvalidField)
424}
425
426/// Read a Group Order from a message that must name a real order.
427///
428/// SUBSCRIBE_OK, PUBLISH, PUBLISH_OK and FETCH_OK each say
429/// "Values of 0x0 and those larger than 0x2 are a protocol error": a responder
430/// reports the order it settled on, so deferring to the publisher is not an
431/// answer it can give. SUBSCRIBE and FETCH are the requests, and there
432/// 0x0 is exactly how a subscriber says it has no preference — "the original
433/// publisher's Group Order SHOULD be used". The two readers cannot be merged
434/// without either refusing traffic the requests permit or accepting a reply
435/// that tells the subscriber nothing.
436fn read_group_order_response(buf: &mut impl Buf) -> Result<GroupOrder, CodecError> {
437    match read_group_order(buf)? {
438        GroupOrder::Publisher => Err(CodecError::InvalidField),
439        order => Ok(order),
440    }
441}
442
443/// Read a Forward flag: "Any other value is a protocol error and MUST terminate
444/// the session with a Protocol Violation"
445fn read_forward(buf: &mut impl Buf) -> Result<Forward, CodecError> {
446    match read_u8(buf)? {
447        0 => Ok(Forward::DontForward),
448        1 => Ok(Forward::Forward),
449        other => Err(CodecError::InvalidForward(other)),
450    }
451}
452
453/// Read a Content Exists flag, which carries the same sentence as Forward and
454/// also decides whether a Largest Location follows.
455fn read_content_exists(buf: &mut impl Buf) -> Result<ContentExists, CodecError> {
456    match read_u8(buf)? {
457        0 => Ok(ContentExists::NoLargestLocation),
458        1 => Ok(ContentExists::HasLargestLocation),
459        other => Err(CodecError::InvalidContentExists(other)),
460    }
461}
462
463/// Both halves of a Start Location travel together, and only the two absolute
464/// filters put one on the wire. The Filter Type is still a raw varint on this
465/// draft, so the range check the decoder applies belongs here too.
466fn check_filter(
467    filter_type: VarInt,
468    start_group: &Option<VarInt>,
469    start_object: &Option<VarInt>,
470    end_group: &Option<VarInt>,
471) -> Result<(), CodecError> {
472    let value = filter_type.into_inner();
473    if value == 0 || value > 4 {
474        return Err(CodecError::InvalidFilterType(value));
475    }
476    let wants_start = value == 3 || value == 4;
477    if wants_start != start_group.is_some() || wants_start != start_object.is_some() {
478        return Err(CodecError::InvalidField);
479    }
480    if (value == 4) != end_group.is_some() {
481        return Err(CodecError::InvalidField);
482    }
483    Ok(())
484}
485
486fn check_content(
487    content_exists: ContentExists,
488    largest_location: &Option<Location>,
489) -> Result<(), CodecError> {
490    if (content_exists == ContentExists::HasLargestLocation) != largest_location.is_some() {
491        return Err(CodecError::InvalidField);
492    }
493    Ok(())
494}
495
496/// Hold a TRACK_STATUS Status Code and the fields after it to Section 8.21.
497///
498/// "Status Code: Provides additional information about the status of the track.
499/// It MUST hold one of the following values. Any other value is a malformed
500/// message." Two of those values - 0x01 and 0x02 - add "Subsequent fields MUST
501/// be zero, and any other value is a malformed message".
502///
503/// Applied on both sides. A malformed message is one this codec must not read
504/// and equally must not write: an encoder that emits an unassigned Status Code
505/// hands a conforming peer a message it is required to reject.
506fn check_track_status(status_code: VarInt, largest_location: Location) -> Result<(), CodecError> {
507    let code = crate::draft12::error_codes::TrackStatusCode::from_u64(status_code.into_inner())
508        .ok_or(CodecError::InvalidField)?;
509    if code.requires_zero_location()
510        && (largest_location.group.into_inner() != 0 || largest_location.object.into_inner() != 0)
511    {
512        return Err(CodecError::InvalidField);
513    }
514    Ok(())
515}
516
517/// Refuse a message whose optional fields disagree with the field that decides
518/// whether they are on the wire.
519///
520/// Presence is not a property of the Rust value: the decoder derives it from a
521/// Filter Type, a Content Exists flag or a Fetch Type, and reads exactly the
522/// fields that discriminator names. An encoder that instead writes whatever
523/// happens to be `Some` produces a frame its own reader refuses - short by the
524/// missing fields, so the declared length runs out mid-payload, or long by the
525/// surplus ones, so bytes are left over. Section 8 makes either a session
526/// close, which is why this refuses rather than papering over it.
527fn check_discriminators(message: &ControlMessage) -> Result<(), CodecError> {
528    match message {
529        ControlMessage::Subscribe(m) => {
530            check_filter(m.filter_type, &m.start_group, &m.start_object, &m.end_group)
531        }
532        ControlMessage::PublishOk(m) => {
533            check_filter(m.filter_type, &m.start_group, &m.start_object, &m.end_group)
534        }
535        ControlMessage::SubscribeOk(m) => check_content(m.content_exists, &m.largest_location),
536        ControlMessage::Publish(m) => check_content(m.content_exists, &m.largest_location),
537        ControlMessage::Fetch(m) => {
538            let body_is_standalone = matches!(m.fetch_payload, FetchPayload::Standalone { .. });
539            if body_is_standalone != (m.fetch_type == FetchType::Standalone) {
540                return Err(CodecError::InvalidField);
541            }
542            Ok(())
543        }
544        _ => Ok(()),
545    }
546}
547
548/// Refuse a Group Order of 0x0 on the messages that forbid it.
549///
550/// The decoders refuse it on the way in; without this the codec would still
551/// write a frame its own reader rejects.
552fn check_group_order(message: &ControlMessage) -> Result<(), CodecError> {
553    let order = match message {
554        ControlMessage::SubscribeOk(m) => m.group_order,
555        ControlMessage::FetchOk(m) => m.group_order,
556        ControlMessage::Publish(m) => m.group_order,
557        ControlMessage::PublishOk(m) => m.group_order,
558        _ => return Ok(()),
559    };
560    if order == GroupOrder::Publisher {
561        return Err(CodecError::InvalidField);
562    }
563    Ok(())
564}
565
566// ============================================================
567// Unified Message Enum
568// ============================================================
569
570#[derive(Debug, Clone, PartialEq, Eq)]
571pub enum ControlMessage {
572    ClientSetup(ClientSetup),
573    ServerSetup(ServerSetup),
574    GoAway(GoAway),
575    MaxRequestId(MaxRequestId),
576    RequestsBlocked(RequestsBlocked),
577    Subscribe(Subscribe),
578    SubscribeOk(SubscribeOk),
579    SubscribeError(SubscribeError),
580    SubscribeUpdate(SubscribeUpdate),
581    SubscribeDone(SubscribeDone),
582    Unsubscribe(Unsubscribe),
583    Announce(Announce),
584    AnnounceOk(AnnounceOk),
585    AnnounceError(AnnounceError),
586    AnnounceCancel(AnnounceCancel),
587    Unannounce(Unannounce),
588    SubscribeAnnounces(SubscribeAnnounces),
589    SubscribeAnnouncesOk(SubscribeAnnouncesOk),
590    SubscribeAnnouncesError(SubscribeAnnouncesError),
591    UnsubscribeAnnounces(UnsubscribeAnnounces),
592    TrackStatusRequest(TrackStatusRequest),
593    TrackStatus(TrackStatus),
594    Fetch(Fetch),
595    FetchOk(FetchOk),
596    FetchError(FetchError),
597    FetchCancel(FetchCancel),
598    Publish(Publish),
599    PublishOk(PublishOk),
600    PublishError(PublishError),
601}
602
603fn check_full_track_name(namespace: &TrackNamespace, track_name: &[u8]) -> Result<(), CodecError> {
604    let total = namespace.field_bytes_len().saturating_add(track_name.len());
605    if total > MAX_FULL_TRACK_NAME_LENGTH {
606        return Err(CodecError::TrackNameTooLong);
607    }
608    Ok(())
609}
610
611/// Read a Reason Phrase, holding it to the cap this draft states for a receiver.
612///
613/// "The reason phrase length has a maximum length of 1024 bytes. If an endpoint
614/// receives a length exceeding the maximum, it MUST close the session with a
615/// Protocol Violation". The sentence is about what an endpoint receives, and
616/// receiving was the direction the cap was not applied to: the encoders refused
617/// an over-long phrase and the decoders accepted one.
618fn read_reason_phrase(buf: &mut impl Buf) -> Result<Vec<u8>, CodecError> {
619    let len = VarInt::decode(buf)?.into_inner() as usize;
620    if len > MAX_REASON_PHRASE_LENGTH {
621        return Err(CodecError::ReasonPhraseTooLong);
622    }
623    types::read_bytes(buf, len)
624}
625
626/// Refuse a request whose range ends before it starts.
627///
628/// SUBSCRIBE's AbsoluteRange filter (Section 8.7), SUBSCRIBE_UPDATE
629/// (Section 8.10) and FETCH (Section 8.16) each state it, and the fields
630/// are not spelled the same way in the three places: an End Group is inclusive
631/// on SUBSCRIBE and FETCH and is the last group plus one on SUBSCRIBE_UPDATE,
632/// where zero means open ended, and an End Object is the last object plus one
633/// with zero meaning the whole group. The helpers this calls carry those
634/// conventions, one per shape.
635///
636/// Applied on both sides. A range that ends before it starts selects nothing,
637/// and the peer's only recourse is an error response or a session close, so
638/// writing one is not a way to ask for anything.
639fn check_ranges(message: &ControlMessage) -> Result<(), CodecError> {
640    match message {
641        ControlMessage::Subscribe(m) => match (&m.start_group, &m.end_group) {
642            (Some(start_group), Some(end_group)) => {
643                check_group_range(start_group.into_inner(), end_group.into_inner())
644            }
645            _ => Ok(()),
646        },
647        ControlMessage::SubscribeUpdate(m) => {
648            check_open_ended_group_range(m.start_group.into_inner(), m.end_group.into_inner())
649        }
650        ControlMessage::Fetch(m) => match &m.fetch_payload {
651            FetchPayload::Standalone {
652                start_group, start_object, end_group, end_object, ..
653            } => check_location_range(
654                start_group.into_inner(),
655                start_object.into_inner(),
656                end_group.into_inner(),
657                end_object.into_inner(),
658            ),
659            FetchPayload::Joining { .. } => Ok(()),
660        },
661        _ => Ok(()),
662    }
663}
664
665/// AUTHORIZATION TOKEN, the Version Specific Parameter of Section 8.2.1.1.
666///
667/// The number is this draft's own. Draft-12 assigns AUTHORIZATION TOKEN
668/// "Parameter Type 0x03"; draft-11 put it at 0x01, where this draft has the
669/// setup-side PATH parameter instead. Reading either draft's number into the
670/// other would exempt the wrong type from the repeat rule below.
671const AUTHORIZATION_TOKEN: u64 = 0x03;
672
673/// DELIVERY TIMEOUT, the Version Specific Parameter of Section 8.2.1.2.
674const DELIVERY_TIMEOUT: u64 = 0x02;
675
676/// MAX_CACHE_DURATION, the Version Specific Parameter of Section 8.2.1.3.
677const MAX_CACHE_DURATION: u64 = 0x04;
678
679/// PATH, the Setup Parameter of Section 8.3.2.1.
680const SETUP_PATH: u64 = 0x01;
681
682/// MAX_REQUEST_ID, the Setup Parameter of Section 8.3.2.2.
683const SETUP_MAX_REQUEST_ID: u64 = 0x02;
684
685/// MAX_AUTH_TOKEN_CACHE_SIZE, the Setup Parameter of Section 8.3.2.3.
686const SETUP_MAX_AUTH_TOKEN_CACHE_SIZE: u64 = 0x04;
687
688/// AUTHORIZATION TOKEN, the Setup Parameter of Section 8.3.2.4.
689///
690/// That section defines the setup-side parameter by reference - "See
691/// Section 8.2.1.1" - so it carries the same type number and the same
692/// permission to repeat: "The endpoint can specify one or more tokens in
693/// CLIENT_SETUP or SERVER_SETUP that the peer can use to authorize MOQT session
694/// establishment."
695const SETUP_AUTHORIZATION_TOKEN: u64 = AUTHORIZATION_TOKEN;
696
697/// Every Version Specific Parameter Type Section 8.2.1 names.
698///
699/// This list is what "unknown" means to the receiver's half of the Section 8.2
700/// rule. A type absent from it is one some extension defined, and Section 8.2
701/// requires a receiver to carry a repeat of such a type rather than refuse it.
702const KNOWN_PARAMETERS: &[u64] = &[AUTHORIZATION_TOKEN, DELIVERY_TIMEOUT, MAX_CACHE_DURATION];
703
704/// The Version Specific Parameter Types whose own definition allows a repeat.
705///
706/// Section 8.2.1.1: "The AUTHORIZATION TOKEN parameter MAY be repeated within a
707/// message." It is the only parameter on this draft that says so, and the
708/// exemption it earns holds on both sides.
709const REPEATABLE_PARAMETERS: &[u64] = &[AUTHORIZATION_TOKEN];
710
711/// Every Setup Parameter Type Section 8.3.2 names.
712const KNOWN_SETUP_PARAMETERS: &[u64] =
713    &[SETUP_PATH, SETUP_MAX_REQUEST_ID, SETUP_AUTHORIZATION_TOKEN, SETUP_MAX_AUTH_TOKEN_CACHE_SIZE];
714
715/// The Setup Parameter Types whose own definition allows a repeat.
716///
717/// The two namespaces are kept apart because they do not name the same set: a
718/// PATH lives at 0x01 among Setup Parameters and nothing lives there among
719/// Version Specific Parameters, so one shared list of known types would make a
720/// repeated 0x01 refusable in a message where the draft requires it to be
721/// tolerated.
722const REPEATABLE_SETUP_PARAMETERS: &[u64] = &[SETUP_AUTHORIZATION_TOKEN];
723
724/// Refuse a parameter list a sender is not allowed to put on the wire.
725///
726/// Section 8.2: "Senders MUST NOT repeat the same parameter type in a message
727/// unless the parameter definition explicitly allows multiple instances of that
728/// type to be sent in a single message."
729///
730/// This is the wider half of the rule. It names no exception for types the
731/// sender does not recognise, so every repeat is refused here except the ones
732/// `repeatable` lists. A caller holding a parameter this codec has never heard
733/// of still may not send it twice: code that scans a parameter list for a key
734/// takes whichever copy it meets first, so one frame carrying two values for one
735/// type is read differently by two conforming implementations, and that is true
736/// whoever defined the type.
737fn check_sender_parameters(
738    parameters: &[KeyValuePair],
739    repeatable: &[u64],
740) -> Result<(), CodecError> {
741    for (i, parameter) in parameters.iter().enumerate() {
742        let key = parameter.key.into_inner();
743        if repeatable.contains(&key) {
744            continue;
745        }
746        if parameters[..i].iter().any(|earlier| earlier.key == parameter.key) {
747            return Err(CodecError::DuplicateParameter(key));
748        }
749    }
750    Ok(())
751}
752
753/// Refuse a received parameter list only where Section 8.2 lets a receiver
754/// refuse it.
755///
756/// Section 8.2: "Receivers SHOULD check that there are no unauthorized
757/// duplicate parameters and close the session as a 'Protocol Violation' if
758/// found. Receivers MUST allow duplicates of unknown parameters."
759///
760/// The second sentence is why this is not the mirror of
761/// [`check_sender_parameters`]: a repeat of a type this draft names is refused,
762/// and a repeat of any other type is carried. Making the two sides symmetric
763/// would close sessions over parameters defined by an extension this codec does
764/// not implement - traffic the draft requires an endpoint to tolerate - and the
765/// first sentence is a SHOULD, which does not reach that far.
766fn check_receiver_parameters(
767    parameters: &[KeyValuePair],
768    known: &[u64],
769    repeatable: &[u64],
770) -> Result<(), CodecError> {
771    for (i, parameter) in parameters.iter().enumerate() {
772        let key = parameter.key.into_inner();
773        if repeatable.contains(&key) || !known.contains(&key) {
774            continue;
775        }
776        if parameters[..i].iter().any(|earlier| earlier.key == parameter.key) {
777            return Err(CodecError::DuplicateParameter(key));
778        }
779    }
780    Ok(())
781}
782
783/// Hold every AUTHORIZATION TOKEN parameter to the Token structure it names.
784///
785/// Section 8.2.1.1: "If the Token structure cannot be decoded, the receiver
786/// MUST close the Session with Key-Value Formatting error." That is the answer
787/// Section 1.3.2 gives for any Type whose value does not match the
788/// serialization that Type defines; the Token is the one structure this draft
789/// spells out, and the only parameter value in it that is more than opaque
790/// bytes.
791///
792/// Both namespaces carry the type on this draft, and both reach here.
793///
794/// A type this draft cannot name is left alone. The rule is conditional on the
795/// receiver understanding the Type, and an extension's parameter carries bytes
796/// no rule here describes.
797fn check_authorization_tokens(parameters: &[KeyValuePair]) -> Result<(), CodecError> {
798    for parameter in parameters {
799        let key = parameter.key.into_inner();
800        if key != AUTH_TOKEN_PARAMETER {
801            continue;
802        }
803        match &parameter.value {
804            KvpValue::Bytes(value) => {
805                AuthorizationToken::decode(key, value)?;
806            }
807            // Unreachable from the decoder, which picks the shape from the
808            // type and finds this one length-prefixed. A caller that built the
809            // pair in memory can still get here, and it is the same rule: the
810            // value is not the serialization the type defines.
811            KvpValue::Varint(_) => {
812                return Err(CodecError::KeyValueFormatting {
813                    key,
814                    detail: "its value is a bare varint where the type defines a Token structure",
815                });
816            }
817        }
818    }
819    Ok(())
820}
821
822/// Decode a Version Specific Parameter list.
823fn decode_parameters(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
824    let parameters = KeyValuePair::decode_list(buf)?;
825    check_receiver_parameters(&parameters, KNOWN_PARAMETERS, REPEATABLE_PARAMETERS)?;
826    check_authorization_tokens(&parameters)?;
827    Ok(parameters)
828}
829
830/// Encode a Version Specific Parameter list.
831fn encode_parameters(parameters: &[KeyValuePair], buf: &mut impl BufMut) -> Result<(), CodecError> {
832    check_sender_parameters(parameters, REPEATABLE_PARAMETERS)?;
833    check_authorization_tokens(parameters)?;
834    KeyValuePair::encode_list_checked(parameters, buf)?;
835    Ok(())
836}
837
838/// Decode the Setup Parameters of a CLIENT_SETUP or SERVER_SETUP message.
839fn decode_setup_parameters(buf: &mut impl Buf) -> Result<Vec<KeyValuePair>, CodecError> {
840    let parameters = KeyValuePair::decode_list(buf)?;
841    check_receiver_parameters(&parameters, KNOWN_SETUP_PARAMETERS, REPEATABLE_SETUP_PARAMETERS)?;
842    check_authorization_tokens(&parameters)?;
843    Ok(parameters)
844}
845
846/// Encode the Setup Parameters of a CLIENT_SETUP or SERVER_SETUP message.
847///
848/// The token is in both namespaces from this draft on, so it is held to its
849/// structure in both. See [`encode_parameters`] for why writing one the reader
850/// would refuse is not a way to send it.
851fn encode_setup_parameters(
852    parameters: &[KeyValuePair],
853    buf: &mut impl BufMut,
854) -> Result<(), CodecError> {
855    check_sender_parameters(parameters, REPEATABLE_SETUP_PARAMETERS)?;
856    check_authorization_tokens(parameters)?;
857    KeyValuePair::encode_list_checked(parameters, buf)?;
858    Ok(())
859}
860
861impl ControlMessage {
862    /// Encode this control message to bytes.
863    ///
864    /// Draft-12 framing: type_id(vi) + payload_length(16) + payload.
865    pub fn encode(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
866        check_discriminators(self)?;
867        check_group_order(self)?;
868        check_ranges(self)?;
869        let mut payload = Vec::with_capacity(256);
870        self.encode_payload(&mut payload)?;
871
872        if payload.len() > MAX_MESSAGE_LENGTH {
873            return Err(CodecError::MessageTooLong(payload.len()));
874        }
875
876        VarInt::from_usize(self.message_type().id() as usize).encode(buf);
877        // Draft-12: 16-bit length (big-endian)
878        buf.put_u16(payload.len() as u16);
879        buf.put_slice(&payload);
880        Ok(())
881    }
882
883    /// Decode a control message from bytes.
884    ///
885    /// Draft-12 framing: type_id(vi) + payload_length(16) + payload.
886    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
887        let type_id = VarInt::decode(buf)?.into_inner();
888        let msg_type =
889            MessageType::from_id(type_id).ok_or(CodecError::UnknownMessageType(type_id))?;
890        // Draft-12: 16-bit length (big-endian)
891        if buf.remaining() < 2 {
892            return Err(CodecError::UnexpectedEnd);
893        }
894        let payload_len = buf.get_u16() as usize;
895        if buf.remaining() < payload_len {
896            return Err(CodecError::UnexpectedEnd);
897        }
898        let payload_bytes = buf.copy_to_bytes(payload_len);
899        let mut payload = &payload_bytes[..];
900        let msg = match Self::decode_payload(msg_type, &mut payload) {
901            Ok(msg) => msg,
902            // The fields wanted more bytes than the Length allowed. This buffer
903            // is already bounded by that Length, so running out inside it cannot
904            // mean the message is still arriving - which is what the same error
905            // means everywhere else, and why a reader loops on it rather than
906            // closing. Here there is nothing left to arrive.
907            Err(
908                CodecError::UnexpectedEnd
909                | CodecError::Kvp(crate::kvp::KvpError::UnexpectedEnd)
910                | CodecError::Kvp(crate::kvp::KvpError::VarInt(
911                    crate::varint::VarIntError::UnexpectedEnd,
912                ))
913                | CodecError::VarInt(crate::varint::VarIntError::UnexpectedEnd),
914            ) => {
915                return Err(CodecError::ControlMessageLengthMismatch {
916                    declared: payload_len,
917                    detail: "its fields ran past the end",
918                });
919            }
920            Err(e) => return Err(e),
921        };
922        check_ranges(&msg)?;
923        // The declared length is part of the message, not a hint. Bytes left over
924        // after the fields have been read mean the sender and this reader disagree
925        // about the shape of the message, and guessing which of the two is right
926        // is how a trailing field gets silently dropped.
927        if payload.has_remaining() {
928            return Err(CodecError::ControlMessageLengthMismatch {
929                declared: payload_len,
930                detail: "its fields left bytes unread",
931            });
932        }
933        Ok(msg)
934    }
935
936    fn encode_payload(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
937        match self {
938            ControlMessage::ClientSetup(m) => {
939                VarInt::from_usize(m.supported_versions.len()).encode(buf);
940                for v in &m.supported_versions {
941                    v.encode(buf);
942                }
943                encode_setup_parameters(&m.parameters, buf)?;
944            }
945            ControlMessage::ServerSetup(m) => {
946                m.selected_version.encode(buf);
947                encode_setup_parameters(&m.parameters, buf)?;
948            }
949            ControlMessage::GoAway(m) => {
950                if m.new_session_uri.len() > MAX_GOAWAY_URI_LENGTH {
951                    return Err(CodecError::GoAwayUriTooLong);
952                }
953                VarInt::from_usize(m.new_session_uri.len()).encode(buf);
954                buf.put_slice(&m.new_session_uri);
955            }
956            ControlMessage::MaxRequestId(m) => {
957                m.request_id.encode(buf);
958            }
959            ControlMessage::RequestsBlocked(m) => {
960                m.maximum_request_id.encode(buf);
961            }
962            ControlMessage::Subscribe(m) => {
963                m.request_id.encode(buf);
964                m.track_namespace.validate(TrackNamespaceRules::for_draft(12))?;
965                m.track_namespace.encode(buf);
966                check_full_track_name(&m.track_namespace, &m.track_name)?;
967                VarInt::from_usize(m.track_name.len()).encode(buf);
968                buf.put_slice(&m.track_name);
969                buf.put_u8(m.subscriber_priority);
970                buf.put_u8(m.group_order as u8);
971                buf.put_u8(m.forward as u8);
972                m.filter_type.encode(buf);
973                if let Some(sg) = &m.start_group {
974                    sg.encode(buf);
975                }
976                if let Some(so) = &m.start_object {
977                    so.encode(buf);
978                }
979                if let Some(eg) = &m.end_group {
980                    eg.encode(buf);
981                }
982                encode_parameters(&m.parameters, buf)?;
983            }
984            ControlMessage::SubscribeOk(m) => {
985                m.request_id.encode(buf);
986                m.track_alias.encode(buf);
987                m.expires.encode(buf);
988                buf.put_u8(m.group_order as u8);
989                buf.put_u8(m.content_exists as u8);
990                if let Some(loc) = &m.largest_location {
991                    loc.encode(buf);
992                }
993                encode_parameters(&m.parameters, buf)?;
994            }
995            ControlMessage::SubscribeError(m) => {
996                if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
997                    return Err(CodecError::ReasonPhraseTooLong);
998                }
999                m.request_id.encode(buf);
1000                m.error_code.encode(buf);
1001                VarInt::from_usize(m.reason_phrase.len()).encode(buf);
1002                buf.put_slice(&m.reason_phrase);
1003            }
1004            ControlMessage::SubscribeUpdate(m) => {
1005                m.request_id.encode(buf);
1006                m.start_group.encode(buf);
1007                m.start_object.encode(buf);
1008                m.end_group.encode(buf);
1009                buf.put_u8(m.subscriber_priority);
1010                buf.put_u8(m.forward as u8);
1011                encode_parameters(&m.parameters, buf)?;
1012            }
1013            ControlMessage::SubscribeDone(m) => {
1014                if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1015                    return Err(CodecError::ReasonPhraseTooLong);
1016                }
1017                m.request_id.encode(buf);
1018                m.status_code.encode(buf);
1019                m.stream_count.encode(buf);
1020                VarInt::from_usize(m.reason_phrase.len()).encode(buf);
1021                buf.put_slice(&m.reason_phrase);
1022            }
1023            ControlMessage::Unsubscribe(m) => {
1024                m.request_id.encode(buf);
1025            }
1026            ControlMessage::Announce(m) => {
1027                m.request_id.encode(buf);
1028                m.track_namespace.validate(TrackNamespaceRules::for_draft(12))?;
1029                m.track_namespace.encode(buf);
1030                encode_parameters(&m.parameters, buf)?;
1031            }
1032            ControlMessage::AnnounceOk(m) => {
1033                m.request_id.encode(buf);
1034            }
1035            ControlMessage::AnnounceError(m) => {
1036                if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1037                    return Err(CodecError::ReasonPhraseTooLong);
1038                }
1039                m.request_id.encode(buf);
1040                m.error_code.encode(buf);
1041                VarInt::from_usize(m.reason_phrase.len()).encode(buf);
1042                buf.put_slice(&m.reason_phrase);
1043            }
1044            ControlMessage::AnnounceCancel(m) => {
1045                if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1046                    return Err(CodecError::ReasonPhraseTooLong);
1047                }
1048                m.track_namespace.validate(TrackNamespaceRules::for_draft(12))?;
1049                m.track_namespace.encode(buf);
1050                m.error_code.encode(buf);
1051                VarInt::from_usize(m.reason_phrase.len()).encode(buf);
1052                buf.put_slice(&m.reason_phrase);
1053            }
1054            ControlMessage::Unannounce(m) => {
1055                m.track_namespace.validate(TrackNamespaceRules::for_draft(12))?;
1056                m.track_namespace.encode(buf);
1057            }
1058            ControlMessage::SubscribeAnnounces(m) => {
1059                m.request_id.encode(buf);
1060                m.track_namespace_prefix.validate(TrackNamespaceRules::for_draft(12))?;
1061                m.track_namespace_prefix.encode(buf);
1062                encode_parameters(&m.parameters, buf)?;
1063            }
1064            ControlMessage::SubscribeAnnouncesOk(m) => {
1065                m.request_id.encode(buf);
1066            }
1067            ControlMessage::SubscribeAnnouncesError(m) => {
1068                if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1069                    return Err(CodecError::ReasonPhraseTooLong);
1070                }
1071                m.request_id.encode(buf);
1072                m.error_code.encode(buf);
1073                VarInt::from_usize(m.reason_phrase.len()).encode(buf);
1074                buf.put_slice(&m.reason_phrase);
1075            }
1076            ControlMessage::UnsubscribeAnnounces(m) => {
1077                m.track_namespace_prefix.validate(TrackNamespaceRules::for_draft(12))?;
1078                m.track_namespace_prefix.encode(buf);
1079            }
1080            ControlMessage::TrackStatusRequest(m) => {
1081                m.request_id.encode(buf);
1082                m.track_namespace.validate(TrackNamespaceRules::for_draft(12))?;
1083                m.track_namespace.encode(buf);
1084                check_full_track_name(&m.track_namespace, &m.track_name)?;
1085                VarInt::from_usize(m.track_name.len()).encode(buf);
1086                buf.put_slice(&m.track_name);
1087                encode_parameters(&m.parameters, buf)?;
1088            }
1089            ControlMessage::TrackStatus(m) => {
1090                m.request_id.encode(buf);
1091                check_track_status(m.status_code, m.largest_location)?;
1092                m.status_code.encode(buf);
1093                m.largest_location.encode(buf);
1094                encode_parameters(&m.parameters, buf)?;
1095            }
1096            ControlMessage::Fetch(m) => {
1097                m.request_id.encode(buf);
1098                buf.put_u8(m.subscriber_priority);
1099                buf.put_u8(m.group_order as u8);
1100                VarInt::from_usize(m.fetch_type as usize).encode(buf);
1101                match &m.fetch_payload {
1102                    FetchPayload::Standalone {
1103                        track_namespace,
1104                        track_name,
1105                        start_group,
1106                        start_object,
1107                        end_group,
1108                        end_object,
1109                    } => {
1110                        track_namespace.validate(TrackNamespaceRules::for_draft(12))?;
1111                        track_namespace.encode(buf);
1112                        check_full_track_name(track_namespace, track_name)?;
1113                        VarInt::from_usize(track_name.len()).encode(buf);
1114                        buf.put_slice(track_name);
1115                        start_group.encode(buf);
1116                        start_object.encode(buf);
1117                        end_group.encode(buf);
1118                        end_object.encode(buf);
1119                    }
1120                    FetchPayload::Joining { joining_request_id, joining_start } => {
1121                        joining_request_id.encode(buf);
1122                        joining_start.encode(buf);
1123                    }
1124                }
1125                encode_parameters(&m.parameters, buf)?;
1126            }
1127            ControlMessage::FetchOk(m) => {
1128                m.request_id.encode(buf);
1129                buf.put_u8(m.group_order as u8);
1130                buf.put_u8(m.end_of_track);
1131                m.end_location.encode(buf);
1132                encode_parameters(&m.parameters, buf)?;
1133            }
1134            ControlMessage::FetchError(m) => {
1135                if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1136                    return Err(CodecError::ReasonPhraseTooLong);
1137                }
1138                m.request_id.encode(buf);
1139                m.error_code.encode(buf);
1140                VarInt::from_usize(m.reason_phrase.len()).encode(buf);
1141                buf.put_slice(&m.reason_phrase);
1142            }
1143            ControlMessage::FetchCancel(m) => {
1144                m.request_id.encode(buf);
1145            }
1146            ControlMessage::Publish(m) => {
1147                m.request_id.encode(buf);
1148                m.track_namespace.validate(TrackNamespaceRules::for_draft(12))?;
1149                m.track_namespace.encode(buf);
1150                check_full_track_name(&m.track_namespace, &m.track_name)?;
1151                VarInt::from_usize(m.track_name.len()).encode(buf);
1152                buf.put_slice(&m.track_name);
1153                m.track_alias.encode(buf);
1154                buf.put_u8(m.group_order as u8);
1155                buf.put_u8(m.content_exists as u8);
1156                if let Some(loc) = &m.largest_location {
1157                    loc.encode(buf);
1158                }
1159                buf.put_u8(m.forward as u8);
1160                encode_parameters(&m.parameters, buf)?;
1161            }
1162            ControlMessage::PublishOk(m) => {
1163                m.request_id.encode(buf);
1164                buf.put_u8(m.forward as u8);
1165                buf.put_u8(m.subscriber_priority);
1166                buf.put_u8(m.group_order as u8);
1167                m.filter_type.encode(buf);
1168                if let Some(sg) = &m.start_group {
1169                    sg.encode(buf);
1170                }
1171                if let Some(so) = &m.start_object {
1172                    so.encode(buf);
1173                }
1174                if let Some(eg) = &m.end_group {
1175                    eg.encode(buf);
1176                }
1177                encode_parameters(&m.parameters, buf)?;
1178            }
1179            ControlMessage::PublishError(m) => {
1180                if m.reason_phrase.len() > MAX_REASON_PHRASE_LENGTH {
1181                    return Err(CodecError::ReasonPhraseTooLong);
1182                }
1183                m.request_id.encode(buf);
1184                m.error_code.encode(buf);
1185                VarInt::from_usize(m.reason_phrase.len()).encode(buf);
1186                buf.put_slice(&m.reason_phrase);
1187            }
1188        }
1189        Ok(())
1190    }
1191
1192    fn decode_payload(msg_type: MessageType, buf: &mut impl Buf) -> Result<Self, CodecError> {
1193        match msg_type {
1194            MessageType::ClientSetup => {
1195                let num_versions = VarInt::decode(buf)?.into_inner() as usize;
1196                // Not a rule this draft states. It says only that the server
1197                // "MUST reply with one of the versions offered by the client"
1198                // and that a peer with no version in common "MUST close the
1199                // session" - outcomes of negotiation rather than parse errors,
1200                // and a CLIENT_SETUP offering nothing decodes cleanly under the
1201                // figure. It is refused here because there is no version a
1202                // reply could name, so the session is already over and the
1203                // early close is the more useful answer than a well-formed
1204                // message no caller can act on.
1205                if num_versions == 0 {
1206                    return Err(CodecError::InvalidField);
1207                }
1208                let mut supported_versions = crate::types::reserve_bounded(num_versions, buf);
1209                for _ in 0..num_versions {
1210                    supported_versions.push(VarInt::decode(buf)?);
1211                }
1212                let parameters = decode_setup_parameters(buf)?;
1213                Ok(ControlMessage::ClientSetup(ClientSetup { supported_versions, parameters }))
1214            }
1215            MessageType::ServerSetup => {
1216                let selected_version = VarInt::decode(buf)?;
1217                let parameters = decode_setup_parameters(buf)?;
1218                Ok(ControlMessage::ServerSetup(ServerSetup { selected_version, parameters }))
1219            }
1220            MessageType::GoAway => {
1221                let uri_len = VarInt::decode(buf)?.into_inner() as usize;
1222                if uri_len > MAX_GOAWAY_URI_LENGTH {
1223                    return Err(CodecError::GoAwayUriTooLong);
1224                }
1225                let uri = types::read_bytes(buf, uri_len)?;
1226                Ok(ControlMessage::GoAway(GoAway { new_session_uri: uri }))
1227            }
1228            MessageType::MaxRequestId => {
1229                let request_id = VarInt::decode(buf)?;
1230                Ok(ControlMessage::MaxRequestId(MaxRequestId { request_id }))
1231            }
1232            MessageType::RequestsBlocked => {
1233                let maximum_request_id = VarInt::decode(buf)?;
1234                Ok(ControlMessage::RequestsBlocked(RequestsBlocked { maximum_request_id }))
1235            }
1236            MessageType::Subscribe => {
1237                let request_id = VarInt::decode(buf)?;
1238                let track_namespace = TrackNamespace::decode(buf)?;
1239                let track_name_len = VarInt::decode(buf)?.into_inner() as usize;
1240                let track_name = types::read_bytes(buf, track_name_len)?;
1241                check_full_track_name(&track_namespace, &track_name)?;
1242                if buf.remaining() < 1 {
1243                    return Err(CodecError::UnexpectedEnd);
1244                }
1245                let subscriber_priority = buf.get_u8();
1246                let group_order = read_group_order(buf)?;
1247                let forward = read_forward(buf)?;
1248                let filter_type = VarInt::decode(buf)?;
1249                let ft_val = filter_type.into_inner();
1250                if ft_val == 0 || ft_val > 4 {
1251                    return Err(CodecError::InvalidFilterType(ft_val));
1252                }
1253                let (start_group, start_object) = if ft_val == 3 || ft_val == 4 {
1254                    (Some(VarInt::decode(buf)?), Some(VarInt::decode(buf)?))
1255                } else {
1256                    (None, None)
1257                };
1258                let end_group = if ft_val == 4 { Some(VarInt::decode(buf)?) } else { None };
1259                let parameters = decode_parameters(buf)?;
1260                Ok(ControlMessage::Subscribe(Subscribe {
1261                    request_id,
1262                    track_namespace,
1263                    track_name,
1264                    subscriber_priority,
1265                    group_order,
1266                    forward,
1267                    filter_type,
1268                    start_group,
1269                    start_object,
1270                    end_group,
1271                    parameters,
1272                }))
1273            }
1274            MessageType::SubscribeOk => {
1275                let request_id = VarInt::decode(buf)?;
1276                let track_alias = VarInt::decode(buf)?;
1277                let expires = VarInt::decode(buf)?;
1278                let group_order = read_group_order_response(buf)?;
1279                let content_exists = read_content_exists(buf)?;
1280                let largest_location = if content_exists == ContentExists::HasLargestLocation {
1281                    Some(Location::decode(buf)?)
1282                } else {
1283                    None
1284                };
1285                let parameters = decode_parameters(buf)?;
1286                Ok(ControlMessage::SubscribeOk(SubscribeOk {
1287                    request_id,
1288                    track_alias,
1289                    expires,
1290                    group_order,
1291                    content_exists,
1292                    largest_location,
1293                    parameters,
1294                }))
1295            }
1296            MessageType::SubscribeError => {
1297                let request_id = VarInt::decode(buf)?;
1298                let error_code = VarInt::decode(buf)?;
1299                let reason_phrase = read_reason_phrase(buf)?;
1300                Ok(ControlMessage::SubscribeError(SubscribeError {
1301                    request_id,
1302                    error_code,
1303                    reason_phrase,
1304                }))
1305            }
1306            MessageType::SubscribeUpdate => {
1307                let request_id = VarInt::decode(buf)?;
1308                let start_group = VarInt::decode(buf)?;
1309                let start_object = VarInt::decode(buf)?;
1310                let end_group = VarInt::decode(buf)?;
1311                if buf.remaining() < 1 {
1312                    return Err(CodecError::UnexpectedEnd);
1313                }
1314                let subscriber_priority = buf.get_u8();
1315                let forward = read_forward(buf)?;
1316                let parameters = decode_parameters(buf)?;
1317                Ok(ControlMessage::SubscribeUpdate(SubscribeUpdate {
1318                    request_id,
1319                    start_group,
1320                    start_object,
1321                    end_group,
1322                    subscriber_priority,
1323                    forward,
1324                    parameters,
1325                }))
1326            }
1327            MessageType::SubscribeDone => {
1328                let request_id = VarInt::decode(buf)?;
1329                let status_code = VarInt::decode(buf)?;
1330                let stream_count = VarInt::decode(buf)?;
1331                let reason_phrase = read_reason_phrase(buf)?;
1332                Ok(ControlMessage::SubscribeDone(SubscribeDone {
1333                    request_id,
1334                    status_code,
1335                    stream_count,
1336                    reason_phrase,
1337                }))
1338            }
1339            MessageType::Unsubscribe => {
1340                let request_id = VarInt::decode(buf)?;
1341                Ok(ControlMessage::Unsubscribe(Unsubscribe { request_id }))
1342            }
1343            MessageType::Announce => {
1344                let request_id = VarInt::decode(buf)?;
1345                let track_namespace = TrackNamespace::decode(buf)?;
1346                let parameters = decode_parameters(buf)?;
1347                Ok(ControlMessage::Announce(Announce { request_id, track_namespace, parameters }))
1348            }
1349            MessageType::AnnounceOk => {
1350                let request_id = VarInt::decode(buf)?;
1351                Ok(ControlMessage::AnnounceOk(AnnounceOk { request_id }))
1352            }
1353            MessageType::AnnounceError => {
1354                let request_id = VarInt::decode(buf)?;
1355                let error_code = VarInt::decode(buf)?;
1356                let reason_phrase = read_reason_phrase(buf)?;
1357                Ok(ControlMessage::AnnounceError(AnnounceError {
1358                    request_id,
1359                    error_code,
1360                    reason_phrase,
1361                }))
1362            }
1363            MessageType::AnnounceCancel => {
1364                let track_namespace = TrackNamespace::decode(buf)?;
1365                let error_code = VarInt::decode(buf)?;
1366                let reason_phrase = read_reason_phrase(buf)?;
1367                Ok(ControlMessage::AnnounceCancel(AnnounceCancel {
1368                    track_namespace,
1369                    error_code,
1370                    reason_phrase,
1371                }))
1372            }
1373            MessageType::Unannounce => {
1374                let track_namespace = TrackNamespace::decode(buf)?;
1375                Ok(ControlMessage::Unannounce(Unannounce { track_namespace }))
1376            }
1377            MessageType::SubscribeAnnounces => {
1378                let request_id = VarInt::decode(buf)?;
1379                let track_namespace_prefix = TrackNamespace::decode(buf)?;
1380                let parameters = decode_parameters(buf)?;
1381                Ok(ControlMessage::SubscribeAnnounces(SubscribeAnnounces {
1382                    request_id,
1383                    track_namespace_prefix,
1384                    parameters,
1385                }))
1386            }
1387            MessageType::SubscribeAnnouncesOk => {
1388                let request_id = VarInt::decode(buf)?;
1389                Ok(ControlMessage::SubscribeAnnouncesOk(SubscribeAnnouncesOk { request_id }))
1390            }
1391            MessageType::SubscribeAnnouncesError => {
1392                let request_id = VarInt::decode(buf)?;
1393                let error_code = VarInt::decode(buf)?;
1394                let reason_phrase = read_reason_phrase(buf)?;
1395                Ok(ControlMessage::SubscribeAnnouncesError(SubscribeAnnouncesError {
1396                    request_id,
1397                    error_code,
1398                    reason_phrase,
1399                }))
1400            }
1401            MessageType::UnsubscribeAnnounces => {
1402                let track_namespace_prefix = TrackNamespace::decode(buf)?;
1403                Ok(ControlMessage::UnsubscribeAnnounces(UnsubscribeAnnounces {
1404                    track_namespace_prefix,
1405                }))
1406            }
1407            MessageType::TrackStatusRequest => {
1408                let request_id = VarInt::decode(buf)?;
1409                let track_namespace = TrackNamespace::decode(buf)?;
1410                let track_name_len = VarInt::decode(buf)?.into_inner() as usize;
1411                let track_name = types::read_bytes(buf, track_name_len)?;
1412                check_full_track_name(&track_namespace, &track_name)?;
1413                let parameters = decode_parameters(buf)?;
1414                Ok(ControlMessage::TrackStatusRequest(TrackStatusRequest {
1415                    request_id,
1416                    track_namespace,
1417                    track_name,
1418                    parameters,
1419                }))
1420            }
1421            MessageType::TrackStatus => {
1422                let request_id = VarInt::decode(buf)?;
1423                let status_code = VarInt::decode(buf)?;
1424                let largest_location = Location::decode(buf)?;
1425                check_track_status(status_code, largest_location)?;
1426                let parameters = decode_parameters(buf)?;
1427                Ok(ControlMessage::TrackStatus(TrackStatus {
1428                    request_id,
1429                    status_code,
1430                    largest_location,
1431                    parameters,
1432                }))
1433            }
1434            MessageType::Fetch => {
1435                let request_id = VarInt::decode(buf)?;
1436                if buf.remaining() < 1 {
1437                    return Err(CodecError::UnexpectedEnd);
1438                }
1439                let subscriber_priority = buf.get_u8();
1440                let group_order = read_group_order(buf)?;
1441                let fetch_type_val = VarInt::decode(buf)?.into_inner();
1442                let fetch_type = FetchType::from_u64(fetch_type_val)
1443                    .ok_or(CodecError::InvalidFetchType(fetch_type_val))?;
1444                let fetch_payload = match fetch_type {
1445                    FetchType::Standalone => {
1446                        let track_namespace = TrackNamespace::decode(buf)?;
1447                        let track_name_len = VarInt::decode(buf)?.into_inner() as usize;
1448                        let track_name = types::read_bytes(buf, track_name_len)?;
1449                        check_full_track_name(&track_namespace, &track_name)?;
1450                        let start_group = VarInt::decode(buf)?;
1451                        let start_object = VarInt::decode(buf)?;
1452                        let end_group = VarInt::decode(buf)?;
1453                        let end_object = VarInt::decode(buf)?;
1454                        FetchPayload::Standalone {
1455                            track_namespace,
1456                            track_name,
1457                            start_group,
1458                            start_object,
1459                            end_group,
1460                            end_object,
1461                        }
1462                    }
1463                    FetchType::RelativeJoining | FetchType::AbsoluteJoining => {
1464                        let joining_request_id = VarInt::decode(buf)?;
1465                        let joining_start = VarInt::decode(buf)?;
1466                        FetchPayload::Joining { joining_request_id, joining_start }
1467                    }
1468                };
1469                let parameters = decode_parameters(buf)?;
1470                Ok(ControlMessage::Fetch(Fetch {
1471                    request_id,
1472                    subscriber_priority,
1473                    group_order,
1474                    fetch_type,
1475                    fetch_payload,
1476                    parameters,
1477                }))
1478            }
1479            MessageType::FetchOk => {
1480                let request_id = VarInt::decode(buf)?;
1481                let group_order = read_group_order_response(buf)?;
1482                let end_of_track = read_u8(buf)?;
1483                let end_location = Location::decode(buf)?;
1484                let parameters = decode_parameters(buf)?;
1485                Ok(ControlMessage::FetchOk(FetchOk {
1486                    request_id,
1487                    group_order,
1488                    end_of_track,
1489                    end_location,
1490                    parameters,
1491                }))
1492            }
1493            MessageType::FetchError => {
1494                let request_id = VarInt::decode(buf)?;
1495                let error_code = VarInt::decode(buf)?;
1496                let reason_phrase = read_reason_phrase(buf)?;
1497                Ok(ControlMessage::FetchError(FetchError { request_id, error_code, reason_phrase }))
1498            }
1499            MessageType::FetchCancel => {
1500                let request_id = VarInt::decode(buf)?;
1501                Ok(ControlMessage::FetchCancel(FetchCancel { request_id }))
1502            }
1503            MessageType::Publish => {
1504                let request_id = VarInt::decode(buf)?;
1505                let track_namespace = TrackNamespace::decode(buf)?;
1506                let track_name_len = VarInt::decode(buf)?.into_inner() as usize;
1507                let track_name = types::read_bytes(buf, track_name_len)?;
1508                check_full_track_name(&track_namespace, &track_name)?;
1509                let track_alias = VarInt::decode(buf)?;
1510                let group_order = read_group_order_response(buf)?;
1511                let content_exists = read_content_exists(buf)?;
1512                let largest_location = if content_exists == ContentExists::HasLargestLocation {
1513                    Some(Location::decode(buf)?)
1514                } else {
1515                    None
1516                };
1517                let forward = read_forward(buf)?;
1518                let parameters = decode_parameters(buf)?;
1519                Ok(ControlMessage::Publish(Publish {
1520                    request_id,
1521                    track_namespace,
1522                    track_name,
1523                    track_alias,
1524                    group_order,
1525                    content_exists,
1526                    largest_location,
1527                    forward,
1528                    parameters,
1529                }))
1530            }
1531            MessageType::PublishOk => {
1532                let request_id = VarInt::decode(buf)?;
1533                let forward = read_forward(buf)?;
1534                if buf.remaining() < 1 {
1535                    return Err(CodecError::UnexpectedEnd);
1536                }
1537                let subscriber_priority = buf.get_u8();
1538                let group_order = read_group_order_response(buf)?;
1539                let filter_type = VarInt::decode(buf)?;
1540                let ft_val = filter_type.into_inner();
1541                if ft_val == 0 || ft_val > 4 {
1542                    return Err(CodecError::InvalidFilterType(ft_val));
1543                }
1544                let (start_group, start_object) = if ft_val == 3 || ft_val == 4 {
1545                    (Some(VarInt::decode(buf)?), Some(VarInt::decode(buf)?))
1546                } else {
1547                    (None, None)
1548                };
1549                let end_group = if ft_val == 4 { Some(VarInt::decode(buf)?) } else { None };
1550                let parameters = decode_parameters(buf)?;
1551                Ok(ControlMessage::PublishOk(PublishOk {
1552                    request_id,
1553                    forward,
1554                    subscriber_priority,
1555                    group_order,
1556                    filter_type,
1557                    start_group,
1558                    start_object,
1559                    end_group,
1560                    parameters,
1561                }))
1562            }
1563            MessageType::PublishError => {
1564                let request_id = VarInt::decode(buf)?;
1565                let error_code = VarInt::decode(buf)?;
1566                let reason_phrase = read_reason_phrase(buf)?;
1567                Ok(ControlMessage::PublishError(PublishError {
1568                    request_id,
1569                    error_code,
1570                    reason_phrase,
1571                }))
1572            }
1573        }
1574    }
1575
1576    pub fn message_type(&self) -> MessageType {
1577        match self {
1578            ControlMessage::ClientSetup(_) => MessageType::ClientSetup,
1579            ControlMessage::ServerSetup(_) => MessageType::ServerSetup,
1580            ControlMessage::GoAway(_) => MessageType::GoAway,
1581            ControlMessage::MaxRequestId(_) => MessageType::MaxRequestId,
1582            ControlMessage::RequestsBlocked(_) => MessageType::RequestsBlocked,
1583            ControlMessage::Subscribe(_) => MessageType::Subscribe,
1584            ControlMessage::SubscribeOk(_) => MessageType::SubscribeOk,
1585            ControlMessage::SubscribeError(_) => MessageType::SubscribeError,
1586            ControlMessage::SubscribeUpdate(_) => MessageType::SubscribeUpdate,
1587            ControlMessage::SubscribeDone(_) => MessageType::SubscribeDone,
1588            ControlMessage::Unsubscribe(_) => MessageType::Unsubscribe,
1589            ControlMessage::Announce(_) => MessageType::Announce,
1590            ControlMessage::AnnounceOk(_) => MessageType::AnnounceOk,
1591            ControlMessage::AnnounceError(_) => MessageType::AnnounceError,
1592            ControlMessage::AnnounceCancel(_) => MessageType::AnnounceCancel,
1593            ControlMessage::Unannounce(_) => MessageType::Unannounce,
1594            ControlMessage::SubscribeAnnounces(_) => MessageType::SubscribeAnnounces,
1595            ControlMessage::SubscribeAnnouncesOk(_) => MessageType::SubscribeAnnouncesOk,
1596            ControlMessage::SubscribeAnnouncesError(_) => MessageType::SubscribeAnnouncesError,
1597            ControlMessage::UnsubscribeAnnounces(_) => MessageType::UnsubscribeAnnounces,
1598            ControlMessage::TrackStatusRequest(_) => MessageType::TrackStatusRequest,
1599            ControlMessage::TrackStatus(_) => MessageType::TrackStatus,
1600            ControlMessage::Fetch(_) => MessageType::Fetch,
1601            ControlMessage::FetchOk(_) => MessageType::FetchOk,
1602            ControlMessage::FetchError(_) => MessageType::FetchError,
1603            ControlMessage::FetchCancel(_) => MessageType::FetchCancel,
1604            ControlMessage::Publish(_) => MessageType::Publish,
1605            ControlMessage::PublishOk(_) => MessageType::PublishOk,
1606            ControlMessage::PublishError(_) => MessageType::PublishError,
1607        }
1608    }
1609}
moqtap_codec/draft12/message.rs

moqtap_codec/draft12/
message.rs
