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

moqtap_codec/draft12/
message.rs
