moqtap_codec/draft13/data_stream.rs
1//! Draft-13 data stream header encoding and decoding.
2//!
3//! - Subgroup stream type IDs: 0x10-0x15, 0x18-0x1D
4//! - Fetch stream type: 0x05
5//! - Datagram types (separate namespace): 0x00-0x05
6
7use super::types::ObjectStatus;
8use crate::error::CodecError;
9use crate::object_properties::{parse_object_properties, ObjectProperty, PropertyValue};
10use crate::types::read_bytes;
11use crate::varint::VarInt;
12use crate::version::DraftVersion;
13use bytes::{Buf, BufMut};
14
15/// Stream type IDs for draft-13 data streams.
16///
17/// # Draft-13 contradicts itself about where the subgroup types sit
18///
19/// Draft-13 repeats draft-12's disagreement word for word, in the same three
20/// places. Two carry draft-11's answer:
21///
22/// - Section 9, Table 10, the table of unidirectional stream types, whose
23/// SUBGROUP_HEADER row reads 0x08-0x0D.
24/// - Section 9.4.2, Figure 33, the header's own layout:
25/// `Type (i) = 0x8..0xD`.
26///
27/// The third does not. Section 9.4.2 says "There are 12 defined Type values for
28/// SUBGROUP_HEADER" and Table 13, immediately under that figure, lists them:
29/// 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x18, 0x19, 0x1A, 0x1B, 0x1C and 0x1D.
30///
31/// **This enum implements Table 13**, so a stream opening with 0x08 through
32/// 0x0D is [`CodecError::UnknownStreamType`] and closes the session.
33///
34/// Table 13 is the surviving half: a code-point update that missed a spot.
35/// The datagram-status disagreement recorded on [`DatagramType`] has the same
36/// shape and is settled the same way there.
37///
38/// - The range 0x08-0x0D holds six values, and this draft defines twelve
39/// types. Draft-11 Section 9.4.2 Table 11 lists exactly six, at 0x08
40/// through 0x0D, and the twelve here are those six crossed with the End Of
41/// Group bit draft-12 added. So 0x08-0x0D and `0x8..0xD` are draft-11's
42/// range left behind, and they cannot hold what this draft defines.
43/// - Table 13 is the only one of the three that says what each value *means*.
44/// The other two give a range and nothing else, so following either would
45/// leave every framing decision — whether a Subgroup ID field is on the
46/// wire, whether objects carry extensions, whether the stream ends the
47/// group — with nothing to read it from.
48/// - Draft-14 keeps Table 13 unchanged and corrects the other two to match:
49/// its Table 10 reads "0x10-0x1D" and its Figure reads
50/// `Type (i) = 0x10..0x1D`. That is the disagreement being resolved in
51/// favour of Table 13 by the working group, one draft later.
52///
53/// The cost of being wrong is asymmetric and points the same way. Accepting
54/// 0x08-0x0D as well would mean parsing a stream under framing no table
55/// assigns it, which is the failure this codebase refuses elsewhere — an
56/// out-of-range Type aliased onto a valid one produces objects with plausible,
57/// wrong contents. Refusing them closes a session with a peer that followed the
58/// stale half of its own draft, which is visible, reportable, and what
59/// draft-14 says the peer should not have done.
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61#[repr(u64)]
62pub enum StreamType {
63 Fetch = 0x05,
64 SubgroupZero = 0x10,
65 SubgroupZeroExt = 0x11,
66 SubgroupFirstObj = 0x12,
67 SubgroupFirstObjExt = 0x13,
68 SubgroupExplicit = 0x14,
69 SubgroupExplicitExt = 0x15,
70 SubgroupZeroEog = 0x18,
71 SubgroupZeroEogExt = 0x19,
72 SubgroupFirstObjEog = 0x1A,
73 SubgroupFirstObjEogExt = 0x1B,
74 SubgroupExplicitEog = 0x1C,
75 SubgroupExplicitEogExt = 0x1D,
76}
77
78/// Hold an object to the rule that a non-existent object carries no extensions.
79///
80/// Section 9.2.1.2: "Any Object may have extension headers except those with
81/// Object Status 'Object Does Not Exist'. If an endpoint receives a non-existent
82/// Object containing extension headers it MUST close the session with a Protocol
83/// Violation."
84///
85/// The sentence is about a receiver, and it reaches all three carriers that can
86/// announce a status: an object on a subgroup stream, an object on a fetch
87/// stream, and a status datagram. A plain datagram has no status field, so it
88/// is the only carrier that cannot break the rule.
89///
90/// Reported under [`CodecError::ExtensionsOnNonExistentObject`], which is this
91/// rule and nothing else. [`CodecError::InvalidField`] is too coarse for it:
92/// shared with a dozen unrelated malformations the draft does not answer with a
93/// close, it leaves a caller unable to act on the sentence above.
94fn check_extensions_against_status(
95 status: ObjectStatus,
96 extensions: &[u8],
97) -> Result<(), CodecError> {
98 if status == ObjectStatus::ObjectDoesNotExist && !extensions.is_empty() {
99 return Err(CodecError::ExtensionsOnNonExistentObject(extensions.len()));
100 }
101 Ok(())
102}
103
104impl StreamType {
105 pub fn from_id(id: u64) -> Option<Self> {
106 match id {
107 0x05 => Some(StreamType::Fetch),
108 0x10 => Some(StreamType::SubgroupZero),
109 0x11 => Some(StreamType::SubgroupZeroExt),
110 0x12 => Some(StreamType::SubgroupFirstObj),
111 0x13 => Some(StreamType::SubgroupFirstObjExt),
112 0x14 => Some(StreamType::SubgroupExplicit),
113 0x15 => Some(StreamType::SubgroupExplicitExt),
114 0x18 => Some(StreamType::SubgroupZeroEog),
115 0x19 => Some(StreamType::SubgroupZeroEogExt),
116 0x1A => Some(StreamType::SubgroupFirstObjEog),
117 0x1B => Some(StreamType::SubgroupFirstObjEogExt),
118 0x1C => Some(StreamType::SubgroupExplicitEog),
119 0x1D => Some(StreamType::SubgroupExplicitEogExt),
120 _ => None,
121 }
122 }
123
124 pub fn is_subgroup(&self) -> bool {
125 matches!(
126 self,
127 StreamType::SubgroupZero
128 | StreamType::SubgroupZeroExt
129 | StreamType::SubgroupFirstObj
130 | StreamType::SubgroupFirstObjExt
131 | StreamType::SubgroupExplicit
132 | StreamType::SubgroupExplicitExt
133 | StreamType::SubgroupZeroEog
134 | StreamType::SubgroupZeroEogExt
135 | StreamType::SubgroupFirstObjEog
136 | StreamType::SubgroupFirstObjEogExt
137 | StreamType::SubgroupExplicitEog
138 | StreamType::SubgroupExplicitEogExt
139 )
140 }
141
142 pub fn has_extensions(&self) -> bool {
143 matches!(
144 self,
145 StreamType::SubgroupZeroExt
146 | StreamType::SubgroupFirstObjExt
147 | StreamType::SubgroupExplicitExt
148 | StreamType::SubgroupZeroEogExt
149 | StreamType::SubgroupFirstObjEogExt
150 | StreamType::SubgroupExplicitEogExt
151 )
152 }
153
154 pub fn contains_end_of_group(&self) -> bool {
155 matches!(
156 self,
157 StreamType::SubgroupZeroEog
158 | StreamType::SubgroupZeroEogExt
159 | StreamType::SubgroupFirstObjEog
160 | StreamType::SubgroupFirstObjEogExt
161 | StreamType::SubgroupExplicitEog
162 | StreamType::SubgroupExplicitEogExt
163 )
164 }
165
166 /// True if this subgroup stream type puts an explicit Subgroup ID on the
167 /// wire.
168 ///
169 /// The Subgroup ID Field Present column of the SUBGROUP_HEADER type table
170 /// in Section 9.4.2. The other two columns of that row say what the
171 /// Subgroup ID *is* where the field is absent — zero, or the first
172 /// Object's ID — so this is only about the field, never about the value.
173 pub fn writes_subgroup_id(&self) -> bool {
174 matches!(
175 self,
176 StreamType::SubgroupExplicit
177 | StreamType::SubgroupExplicitExt
178 | StreamType::SubgroupExplicitEog
179 | StreamType::SubgroupExplicitEogExt
180 )
181 }
182}
183
184/// Datagram wire types (separate namespace from QUIC stream types).
185///
186/// The two namespaces overlap in draft-13 and cannot share one enum: 0x05 is
187/// FETCH_HEADER among stream types and OBJECT_DATAGRAM_STATUS with extensions
188/// among datagram types.
189///
190/// The values follow the table in draft-13 Section 9: 0x00-0x03 are
191/// OBJECT_DATAGRAM and 0x04-0x05 are OBJECT_DATAGRAM_STATUS. The two close rules
192/// for an empty extension block still carry draft-11's code points, and are
193/// read through the table:
194///
195/// - draft-13 Section 9.3.2: "The Type field takes the form 0b0000001X (or
196/// the set of values from 0x02 to 0x03)" and "If an endpoint receives a
197/// datagram with Type 0x03 and Extension Headers Length is 0, it MUST close
198/// the session with Protocol Violation." draft-11 Section 9 assigns
199/// 0x02-0x03 to OBJECT_DATAGRAM_STATUS. draft-14 Appendix A.1 lists
200/// "Corrected missed code point updates to Object Datagram Status (#1082)",
201/// and that change rewrites the sentence to "Type 0x05". It names 0x05.
202/// - draft-13 Section 9.3.1: "If an endpoint receives a datagram with Type
203/// 0x01 and Extension Headers Length is 0, it MUST close the session with
204/// Protocol Violation." 0x01 is the only OBJECT_DATAGRAM Type with
205/// extensions in draft-11 Section 9.2. draft-13 Section 9.3.1 Table 12
206/// adds 0x03, End Of Group with Extensions Present "Yes", and the sentence
207/// was not updated. draft-14 Section 10.3.1 states the rule for every such
208/// Type: "If an endpoint receives a datagram with Extensions Present as
209/// "Yes" and a Extension Headers Length of 0, it MUST close the session
210/// with PROTOCOL_VIOLATION." It names 0x01 and 0x03.
211///
212/// So 0x01, 0x03 and 0x05 carry extensions, and each is refused with an
213/// Extension Headers Length of 0. 0x00, 0x02 and 0x04 carry no Extension
214/// Headers Length. The client's `EmptyDatagramExtensions` closes the session
215/// over the same three Types.
216#[derive(Debug, Clone, Copy, PartialEq, Eq)]
217#[repr(u64)]
218pub enum DatagramType {
219 Datagram = 0x00,
220 DatagramExt = 0x01,
221 DatagramEog = 0x02,
222 DatagramEogExt = 0x03,
223 DatagramStatus = 0x04,
224 DatagramStatusExt = 0x05,
225}
226
227impl DatagramType {
228 pub fn from_id(id: u64) -> Option<Self> {
229 match id {
230 0x00 => Some(DatagramType::Datagram),
231 0x01 => Some(DatagramType::DatagramExt),
232 0x02 => Some(DatagramType::DatagramEog),
233 0x03 => Some(DatagramType::DatagramEogExt),
234 0x04 => Some(DatagramType::DatagramStatus),
235 0x05 => Some(DatagramType::DatagramStatusExt),
236 _ => None,
237 }
238 }
239
240 pub fn has_extensions(&self) -> bool {
241 matches!(
242 self,
243 DatagramType::DatagramExt
244 | DatagramType::DatagramEogExt
245 | DatagramType::DatagramStatusExt
246 )
247 }
248
249 pub fn is_status(&self) -> bool {
250 matches!(self, DatagramType::DatagramStatus | DatagramType::DatagramStatusExt)
251 }
252
253 pub fn is_end_of_group(&self) -> bool {
254 matches!(self, DatagramType::DatagramEog | DatagramType::DatagramEogExt)
255 }
256}
257
258/// Which failure a leading unidirectional stream type that is not the one a
259/// reader wants is.
260///
261/// Section 9: "An endpoint that receives an unknown stream or datagram type
262/// MUST close the session." One sentence, two tables, and on this draft the two
263/// tables collide: 0x05 is FETCH_HEADER in the stream table and
264/// OBJECT_DATAGRAM_STATUS with extensions in the datagram table. Which table
265/// was consulted is therefore part of the answer, not a detail, and it is why
266/// [`CodecError::UnknownStreamType`] and [`CodecError::UnknownDatagramType`]
267/// are separate variants rather than one.
268///
269/// The stream table assigns 0x05 and the range 0x10 to 0x1D. Everything outside
270/// them is unknown at the head of a stream, and the session ends.
271fn stream_type_error(raw: u64) -> CodecError {
272 if StreamType::from_id(raw).is_some() {
273 CodecError::InvalidField
274 } else {
275 CodecError::UnknownStreamType(raw)
276 }
277}
278
279/// Which failure a leading datagram type that is not one a reader wants is.
280///
281/// The datagram half of the sentence quoted on `stream_type_error`, read
282/// against the other table: 0x00 to 0x05 are what it assigns, and everything
283/// else arriving as a datagram is unknown.
284///
285/// Answered from [`DatagramType`] alone, never from [`StreamType`]. A value in
286/// both tables means one thing as a datagram and another as a stream, and
287/// consulting the wrong one turns an assigned datagram type into an unknown one
288/// or the reverse.
289fn datagram_type_error(raw: u64) -> CodecError {
290 if DatagramType::from_id(raw).is_some() {
291 CodecError::InvalidField
292 } else {
293 CodecError::UnknownDatagramType(raw)
294 }
295}
296
297fn read_extension_bytes(buf: &mut impl Buf, byte_len: u64) -> Result<Vec<u8>, CodecError> {
298 read_bytes(buf, byte_len as usize)
299}
300
301// ============================================================
302// Subgroup stream header
303// ============================================================
304
305#[derive(Debug, Clone, PartialEq, Eq)]
306pub struct SubgroupHeader {
307 pub stream_type: StreamType,
308 pub track_alias: VarInt,
309 pub group_id: VarInt,
310 pub subgroup_id: VarInt,
311 pub publisher_priority: u8,
312}
313
314impl SubgroupHeader {
315 /// Encode a subgroup stream header including its leading stream-type
316 /// field, so the bytes form the start of a data stream a peer can read.
317 ///
318 /// [`Self::encode`] writes the body alone, which is what a caller wants
319 /// once the stream is already open and what a caller must not use for its
320 /// first write. It is also the half that cannot stand on its own here,
321 /// because the stream type is what says whether a Subgroup ID follows it
322 /// and whether the objects on the stream carry extension headers.
323 pub fn encode_stream(&self, buf: &mut impl BufMut) {
324 VarInt::from_usize(self.stream_type as usize).encode(buf);
325 self.encode(buf);
326 }
327
328 /// Encode the header body, without its leading stream-type field.
329 ///
330 /// Driven by the stream type, and silent about a `subgroup_id` it decides
331 /// not to write: on a type whose Subgroup ID Field Present column reads No
332 /// the field is dropped, and the peer reads the subgroup the *type* names -
333 /// zero, or the first Object's ID - rather than the one in hand. Nothing is
334 /// malformed about the result, which is what makes it worth refusing rather
335 /// than tolerating. [`Self::encode_checked`] refuses it.
336 pub fn encode(&self, buf: &mut impl BufMut) {
337 self.track_alias.encode(buf);
338 self.group_id.encode(buf);
339 if self.stream_type.writes_subgroup_id() {
340 self.subgroup_id.encode(buf);
341 }
342 buf.put_u8(self.publisher_priority);
343 }
344
345 /// Encode the header body, refusing a Subgroup ID this stream type has
346 /// nowhere to put.
347 ///
348 /// [`Self::decode_with_type`] leaves the field at zero for every type that
349 /// does not carry it, so a decoded header always passes: the refusal is for
350 /// a header assembled by hand, where a caller set an ID the type will
351 /// discard.
352 ///
353 /// A zero is accepted under any type. It is what the decoder produces, and
354 /// on a Subgroup ID Value column reading `0` it is also the truth, so
355 /// refusing it would refuse the ordinary case to catch nothing.
356 ///
357 /// # Errors
358 ///
359 /// [`CodecError::InvalidField`] if a non-zero Subgroup ID sits under a
360 /// stream type that writes no Subgroup ID field.
361 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
362 if !self.stream_type.writes_subgroup_id() && self.subgroup_id.into_inner() != 0 {
363 return Err(CodecError::InvalidField);
364 }
365 self.encode(buf);
366 Ok(())
367 }
368
369 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
370 Self::decode_with_type(StreamType::SubgroupExplicit, buf)
371 }
372
373 pub fn decode_with_type(
374 stream_type: StreamType,
375 buf: &mut impl Buf,
376 ) -> Result<Self, CodecError> {
377 let track_alias = VarInt::decode(buf)?;
378 let group_id = VarInt::decode(buf)?;
379 let subgroup_id = match stream_type {
380 StreamType::SubgroupZero
381 | StreamType::SubgroupZeroExt
382 | StreamType::SubgroupZeroEog
383 | StreamType::SubgroupZeroEogExt => VarInt::from_usize(0),
384 StreamType::SubgroupExplicit
385 | StreamType::SubgroupExplicitExt
386 | StreamType::SubgroupExplicitEog
387 | StreamType::SubgroupExplicitEogExt => VarInt::decode(buf)?,
388 StreamType::SubgroupFirstObj
389 | StreamType::SubgroupFirstObjExt
390 | StreamType::SubgroupFirstObjEog
391 | StreamType::SubgroupFirstObjEogExt => VarInt::from_usize(0),
392 _ => return Err(CodecError::InvalidField),
393 };
394 if buf.remaining() < 1 {
395 return Err(CodecError::UnexpectedEnd);
396 }
397 let publisher_priority = buf.get_u8();
398 Ok(Self { stream_type, track_alias, group_id, subgroup_id, publisher_priority })
399 }
400
401 /// Decode a subgroup header from the start of a data stream, consuming
402 /// the leading stream type varint and using it to select the variant.
403 ///
404 /// Errors with [`CodecError::UnknownStreamType`] when the stream table does
405 /// not assign the leading type, which this draft answers with a close, and
406 /// with [`CodecError::InvalidField`] when it does assign it but not to a
407 /// subgroup. `stream_type_error` draws that line.
408 pub fn decode_stream(buf: &mut impl Buf) -> Result<Self, CodecError> {
409 let raw = VarInt::decode(buf)?.into_inner();
410 let stream_type = StreamType::from_id(raw).ok_or_else(|| stream_type_error(raw))?;
411 if !stream_type.is_subgroup() {
412 return Err(stream_type_error(raw));
413 }
414 Self::decode_with_type(stream_type, buf)
415 }
416}
417
418// ============================================================
419// Object header within subgroup
420// ============================================================
421
422#[derive(Debug, Clone, PartialEq, Eq)]
423pub struct ObjectHeader {
424 pub object_id: VarInt,
425 pub extension_headers_length: VarInt,
426 pub extensions: Vec<u8>,
427 pub payload_length: VarInt,
428 pub object_status: ObjectStatus,
429}
430
431impl ObjectHeader {
432 /// Encode the object header in the framing that carries no extension
433 /// block.
434 ///
435 /// Lossy, and lossy in a way the caller cannot see: an object holding
436 /// extension headers is written without them and without a word. Which
437 /// framing is correct is not a property of the object at all - Section
438 /// 9.4.2 gives the stream's type an Extensions Present column, and every
439 /// object on the stream follows it - so this entry point can only guess,
440 /// and it guesses "absent". Prefer [`Self::encode_with_extensions`], which
441 /// is told, or [`Self::encode_checked`], which refuses what it would
442 /// otherwise drop.
443 pub fn encode(&self, buf: &mut impl BufMut) {
444 self.encode_with_extensions(false, buf);
445 }
446
447 /// Encode the header, refusing a status the framing cannot carry.
448 ///
449 /// Section 9.4.2 puts the Object Status field on the wire only when the
450 /// Object Payload Length is zero, and Section 9.2.1.1 says "Any object
451 /// with a status code other than zero MUST have an empty payload". A
452 /// non-zero status paired with a non-zero payload length therefore has no
453 /// encoding at all: [`Self::encode`] drops the status and the peer reads an
454 /// ordinary object, which is a different object from the one the caller
455 /// described. This refuses instead.
456 ///
457 /// The datagram types on this draft already refuse the same pairing. These
458 /// two did not, and they are the ones a publisher writes on every stream.
459 ///
460 /// Extension headers are refused here rather than dropped, for a reason
461 /// the status rule does not share: this entry point writes the framing
462 /// that has no Extension Headers Length field, so the bytes have nowhere
463 /// to go. Writing them anyway is not an option and losing them silently
464 /// puts a stream on the wire that no reader can follow - a reader on an
465 /// extensions-bearing stream takes the Object Payload Length as the
466 /// extension length and every object after it is misread. A caller that
467 /// knows the stream's framing wants
468 /// [`Self::encode_checked_with_extensions`].
469 ///
470 /// # Errors
471 ///
472 /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
473 /// a non-zero Object Payload Length, or if the object carries extension
474 /// headers this framing cannot write.
475 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
476 self.encode_checked_with_extensions(false, buf)
477 }
478
479 /// Encode the object header into a stream whose type has already settled
480 /// whether objects carry an extension block, refusing what that framing
481 /// cannot express.
482 ///
483 /// `has_extensions` is the stream's answer, not the object's: Section
484 /// 9.4.2 fixes it for the whole stream from the SUBGROUP_HEADER type, so
485 /// an object with no extensions on a stream that carries them still writes
486 /// a length of zero, and that is the one direction not refused here. The
487 /// other direction has no encoding, so it is refused.
488 ///
489 /// # Errors
490 ///
491 /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
492 /// a non-zero Object Payload Length, or if `has_extensions` is `false`
493 /// while the object carries extension headers.
494 pub fn encode_checked_with_extensions(
495 &self,
496 has_extensions: bool,
497 buf: &mut impl BufMut,
498 ) -> Result<(), CodecError> {
499 if self.payload_length.into_inner() != 0 && self.object_status as usize != 0 {
500 return Err(CodecError::InvalidField);
501 }
502 if !has_extensions && !self.extensions.is_empty() {
503 return Err(CodecError::InvalidField);
504 }
505 self.encode_with_extensions(has_extensions, buf);
506 Ok(())
507 }
508
509 /// Encode the object header, writing the extension block only when the
510 /// stream's type says objects carry one.
511 ///
512 /// Infallible, and so unable to say that a `false` here discards the
513 /// extension headers the object holds. [`Self::encode_checked_with_extensions`]
514 /// is the same write with that refusal in front of it.
515 pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
516 self.object_id.encode(buf);
517 if has_extensions {
518 VarInt::from_usize(self.extensions.len()).encode(buf);
519 buf.put_slice(&self.extensions);
520 }
521 self.payload_length.encode(buf);
522 if self.payload_length.into_inner() == 0 {
523 VarInt::from_usize(self.object_status as usize).encode(buf);
524 }
525 }
526
527 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
528 Self::decode_with_extensions(false, buf)
529 }
530
531 pub fn decode_with_extensions(
532 has_extensions: bool,
533 buf: &mut impl Buf,
534 ) -> Result<Self, CodecError> {
535 let object_id = VarInt::decode(buf)?;
536 let (extension_headers_length, extensions) = if has_extensions {
537 let ehl = VarInt::decode(buf)?;
538 let ext = read_extension_bytes(buf, ehl.into_inner())?;
539 (ehl, ext)
540 } else {
541 (VarInt::from_usize(0), Vec::new())
542 };
543 let payload_length = VarInt::decode(buf)?;
544 let object_status = if payload_length.into_inner() == 0 {
545 let sv = VarInt::decode(buf)?.into_inner();
546 ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?
547 } else {
548 ObjectStatus::Normal
549 };
550 check_extensions_against_status(object_status, &extensions)?;
551 Ok(Self { object_id, extension_headers_length, extensions, payload_length, object_status })
552 }
553}
554
555// ============================================================
556// Prior Group ID Gap
557// ============================================================
558
559/// Prior Group ID Gap, Extension Header Type 0x40 (draft-13 Section 10.1).
560pub const PRIOR_GROUP_ID_GAP: u64 = 0x40;
561
562/// The Prior Group ID Gap values one Object carries.
563///
564/// draft-13 Section 10.1: "A track with a Group that contains more than one
565/// Object with different values for Prior Group ID Gap or has a Prior Group ID
566/// Gap larger than the Group ID is considered malformed." Malformed Tracks are
567/// draft-13 Section 2.5. This draft does not limit an Object to one instance,
568/// so every value is kept. The least is held to the Group ID, for the reason
569/// [`SubgroupGaps`] gives.
570#[derive(Debug, Clone, Default, PartialEq, Eq)]
571pub struct PriorGaps {
572 /// Each Prior Group ID Gap, in wire order.
573 pub group: Vec<u64>,
574}
575
576impl PriorGaps {
577 /// Read the gaps from an Object's parsed extension headers, refusing them
578 /// with [`CodecError::MalformedTrack`] when every one is larger than
579 /// `group_id`.
580 pub fn from_properties(
581 properties: &[ObjectProperty<'_>],
582 group_id: u64,
583 ) -> Result<Self, CodecError> {
584 let group: Vec<u64> = properties
585 .iter()
586 .filter_map(|property| match (property.property_type, &property.value) {
587 (PRIOR_GROUP_ID_GAP, PropertyValue::Varint(value)) => Some(*value),
588 _ => None,
589 })
590 .collect();
591 if let Some(gap) = group.iter().copied().min().filter(|&gap| gap > group_id) {
592 return Err(CodecError::MalformedTrack {
593 detail: format!("Prior Group ID Gap {gap} is larger than Group ID {group_id}"),
594 });
595 }
596 Ok(Self { group })
597 }
598
599 /// [`Self::from_properties`] on an extension block, refusing a block that
600 /// does not parse as [`crate::object_properties::parse_object_properties`]
601 /// does.
602 pub fn from_extensions(extensions: &[u8], group_id: u64) -> Result<Self, CodecError> {
603 let properties = parse_object_properties(DraftVersion::Draft13, extensions)?;
604 Self::from_properties(&properties, group_id)
605 }
606}
607
608/// The Prior Group ID Gap values the Objects of one subgroup stream carried.
609///
610/// A subgroup stream carries one Group, so its Objects are where "a Group that
611/// contains more than one Object with different values for Prior Group ID Gap"
612/// (draft-13 Section 10.1) shows on one stream.
613///
614/// This draft does not say whether an Object may carry 0x40 more than once, or
615/// which value is that Object's when it does. draft-13 Section 9.2.1.2 says
616/// only "Object Extension Headers are serialized as Key-Value-Pairs", and the
617/// no-repeat rule of draft-13 Section 8.2, "Senders MUST NOT repeat the same
618/// parameter type in a message", covers message Parameters. draft-13 Section 12
619/// leaves it open: "List which headers can be repeated in the table." So the
620/// Objects carry different values only when no value is carried by every one of
621/// them: then, whichever of its values is taken as each Object's, two of them
622/// differ. While a value is common to all, taking it as each Object's makes
623/// them equal, and they are not refused. The values one Object carries are not
624/// compared with each other, since the condition needs "more than one Object".
625///
626/// "has a Prior Group ID Gap larger than the Group ID" (draft-13 Section 10.1)
627/// also depends on which value is the Object's. An Object carrying a value
628/// larger than the Group ID and one that is not has a gap larger than the Group
629/// ID under one reading and not under another, so it is refused only when every
630/// value it carries is larger.
631#[derive(Debug, Clone, PartialEq, Eq)]
632pub struct SubgroupGaps {
633 group_id: u64,
634 /// The values every earlier Object that carried a gap carried, or `None`
635 /// before the first such Object.
636 common: Option<Vec<u64>>,
637}
638
639impl SubgroupGaps {
640 /// Track the Objects of the stream `header` opens.
641 pub fn new(header: &SubgroupHeader) -> Self {
642 Self { group_id: header.group_id.into_inner(), common: None }
643 }
644
645 /// Check the next Object on this stream, refusing with
646 /// [`CodecError::MalformedTrack`] a gap larger than the Group ID, or
647 /// values that leave no value common to every Object on the stream that
648 /// carried a gap. Records the Object only when it passes.
649 pub fn check(&mut self, object: &ObjectHeader) -> Result<PriorGaps, CodecError> {
650 let gaps = PriorGaps::from_extensions(&object.extensions, self.group_id)?;
651 if gaps.group.is_empty() {
652 return Ok(gaps);
653 }
654 let common: Vec<u64> = match &self.common {
655 Some(common) => common.iter().copied().filter(|v| gaps.group.contains(v)).collect(),
656 None => {
657 let mut values = gaps.group.clone();
658 values.sort_unstable();
659 values.dedup();
660 values
661 }
662 };
663 if common.is_empty() {
664 return Err(CodecError::MalformedTrack {
665 detail: format!(
666 "the Objects in Group {} carry no Prior Group ID Gap value in common",
667 self.group_id
668 ),
669 });
670 }
671 self.common = Some(common);
672 Ok(gaps)
673 }
674}
675
676/// The Prior Group ID Gap conditions of draft-13 Section 10.1 that one
677/// fetch stream shows.
678///
679/// [`PriorGaps::from_extensions`] refuses a gap larger than the Group ID. Across
680/// the stream this refuses Objects of one Group with no value in common, read
681/// as [`SubgroupGaps`] reads them, and "an Object with a Group ID within a
682/// previously communicated gap" for the Group before on the stream. An Object
683/// whose values differ has communicated at least the least of them, whichever
684/// is that Object's. draft-13 Section 8.16: "Within each group,
685/// objects are sent in Object ID order; subgroup ID is not used for ordering."
686///
687/// [`FetchObjectHeader::decode`] reads one Object and applies none of this: a
688/// reader of a fetch stream checks each Object with one `FetchGaps`.
689#[derive(Debug, Clone, PartialEq, Eq)]
690// Boxed so that the fetch readers holding one stay small.
691pub struct FetchGaps(Box<crate::prior_gaps::FetchGapTracker>);
692
693impl Default for FetchGaps {
694 fn default() -> Self {
695 Self::new()
696 }
697}
698
699impl FetchGaps {
700 /// Track the Objects of one fetch stream.
701 pub fn new() -> Self {
702 Self(Box::new(crate::prior_gaps::FetchGapTracker::new(
703 crate::prior_gaps::GapRules::Draft11To13,
704 )))
705 }
706
707 /// Check the next Object on the stream, from its Group ID, Object ID and
708 /// extension block, refusing a gap condition with
709 /// [`CodecError::MalformedTrack`] and a block that does not parse as
710 /// [`parse_object_properties`] does. Records the Object only when it
711 /// passes.
712 pub fn check(
713 &mut self,
714 group_id: u64,
715 object_id: u64,
716 extensions: &[u8],
717 ) -> Result<PriorGaps, CodecError> {
718 let gaps = PriorGaps::from_extensions(extensions, group_id)?;
719 self.0.check(group_id, object_id, &gaps.group, None)?;
720 Ok(gaps)
721 }
722}
723
724// ============================================================
725// Datagram (types 0x00, 0x01)
726// ============================================================
727
728#[derive(Debug, Clone, PartialEq, Eq)]
729pub struct DatagramHeader {
730 pub track_alias: VarInt,
731 pub group_id: VarInt,
732 pub object_id: VarInt,
733 pub publisher_priority: u8,
734 pub extension_headers_length: VarInt,
735 pub extensions: Vec<u8>,
736 pub end_of_group: bool,
737}
738
739impl DatagramHeader {
740 /// Encode the datagram header in the framing that carries no extension
741 /// block.
742 ///
743 /// Lossy in the same way the subgroup object header is: an extension block
744 /// this value holds is dropped, because the framing being written has no
745 /// field for it. The type byte decides which framing is right, and this
746 /// entry point does not write the type byte, so it cannot consult it.
747 /// [`Datagram::encode`] does both together and never disagrees with itself;
748 /// this is the piece for a caller that has already written the type.
749 pub fn encode(&self, buf: &mut impl BufMut) {
750 self.encode_with_extensions(false, buf);
751 }
752
753 pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
754 self.track_alias.encode(buf);
755 self.group_id.encode(buf);
756 self.object_id.encode(buf);
757 buf.put_u8(self.publisher_priority);
758 if has_extensions {
759 VarInt::from_usize(self.extensions.len()).encode(buf);
760 buf.put_slice(&self.extensions);
761 }
762 }
763
764 /// Encode the datagram header, refusing what this framing cannot carry.
765 ///
766 /// No status is ever refused, and that is a fact about this draft rather
767 /// than a check left out. This is the OBJECT_DATAGRAM of Section 9.3.1,
768 /// whose layout carries no Object Status field at all; a datagram that
769 /// states a status is the separate OBJECT_DATAGRAM_STATUS message, modelled
770 /// here as [`DatagramStatusHeader`]. So there is no status for
771 /// [`Self::encode`] to drop, and nothing for Section 9.2.1.1's "Any object
772 /// with a status code other than zero MUST have an empty payload" to rule
773 /// on: an object framed this way has status zero by construction.
774 ///
775 /// The extension block is a different matter. [`Self::encode`] writes the
776 /// framing without one, so a block this value holds has nowhere to go, and
777 /// dropping it silently is what puts a datagram on the wire describing
778 /// something other than what the caller built. That is refused here.
779 ///
780 /// The fallible signature is also what lets one entry point span every
781 /// draft. `dispatch::AnyDatagramHeader::encode` calls this on all thirteen,
782 /// and the drafts whose payload-bearing datagram *does* carry a status field
783 /// need somewhere to say no.
784 ///
785 /// # Errors
786 ///
787 /// [`CodecError::InvalidField`] if the value carries extension headers,
788 /// which this framing has no field for.
789 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
790 if !self.extensions.is_empty() {
791 return Err(CodecError::InvalidField);
792 }
793 self.encode(buf);
794 Ok(())
795 }
796
797 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
798 Self::decode_with_extensions(false, buf)
799 }
800
801 pub fn decode_with_extensions(
802 has_extensions: bool,
803 buf: &mut impl Buf,
804 ) -> Result<Self, CodecError> {
805 let track_alias = VarInt::decode(buf)?;
806 let group_id = VarInt::decode(buf)?;
807 let object_id = VarInt::decode(buf)?;
808 if buf.remaining() < 1 {
809 return Err(CodecError::UnexpectedEnd);
810 }
811 let publisher_priority = buf.get_u8();
812 let (extension_headers_length, extensions) = if has_extensions {
813 let ehl = VarInt::decode(buf)?;
814 // A datagram whose type says extensions are present must actually carry
815 // some: receiving one with an Extension Headers Length of 0 closes the
816 // session. The opposite holds on a subgroup stream, where the type byte is
817 // fixed for the whole stream and an object with no extensions has no other
818 // way to say so, which is why this check belongs to the datagram readers
819 // alone.
820 if ehl.into_inner() == 0 {
821 return Err(CodecError::InvalidField);
822 }
823 let ext = read_extension_bytes(buf, ehl.into_inner())?;
824 (ehl, ext)
825 } else {
826 (VarInt::from_usize(0), Vec::new())
827 };
828 Ok(Self {
829 track_alias,
830 group_id,
831 object_id,
832 publisher_priority,
833 extension_headers_length,
834 extensions,
835 end_of_group: false,
836 })
837 }
838}
839
840// ============================================================
841// Datagram Status (types 0x04, 0x05)
842// ============================================================
843
844#[derive(Debug, Clone, PartialEq, Eq)]
845pub struct DatagramStatusHeader {
846 pub track_alias: VarInt,
847 pub group_id: VarInt,
848 pub object_id: VarInt,
849 pub publisher_priority: u8,
850 pub extension_headers_length: VarInt,
851 pub extensions: Vec<u8>,
852 pub object_status: ObjectStatus,
853}
854
855impl DatagramStatusHeader {
856 pub fn encode(&self, buf: &mut impl BufMut) {
857 self.encode_with_extensions(false, buf);
858 }
859
860 /// Encode the status datagram header, refusing an extension block this
861 /// framing cannot carry.
862 ///
863 /// The same one-sided rule the payload-bearing header obeys, and the same
864 /// reason for it: [`Self::encode`] writes the framing without a block, so
865 /// bytes held here would be dropped rather than written.
866 ///
867 /// # Errors
868 ///
869 /// [`CodecError::InvalidField`] if the value carries extension headers.
870 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
871 if !self.extensions.is_empty() {
872 return Err(CodecError::InvalidField);
873 }
874 self.encode(buf);
875 Ok(())
876 }
877
878 pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
879 self.track_alias.encode(buf);
880 self.group_id.encode(buf);
881 self.object_id.encode(buf);
882 buf.put_u8(self.publisher_priority);
883 if has_extensions {
884 VarInt::from_usize(self.extensions.len()).encode(buf);
885 buf.put_slice(&self.extensions);
886 }
887 VarInt::from_usize(self.object_status as usize).encode(buf);
888 }
889
890 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
891 Self::decode_with_extensions(false, buf)
892 }
893
894 pub fn decode_with_extensions(
895 has_extensions: bool,
896 buf: &mut impl Buf,
897 ) -> Result<Self, CodecError> {
898 let track_alias = VarInt::decode(buf)?;
899 let group_id = VarInt::decode(buf)?;
900 let object_id = VarInt::decode(buf)?;
901 if buf.remaining() < 1 {
902 return Err(CodecError::UnexpectedEnd);
903 }
904 let publisher_priority = buf.get_u8();
905 let (extension_headers_length, extensions) = if has_extensions {
906 let ehl = VarInt::decode(buf)?;
907 // A datagram whose type says extensions are present must actually carry
908 // some: receiving one with an Extension Headers Length of 0 closes the
909 // session. The opposite holds on a subgroup stream, where the type byte is
910 // fixed for the whole stream and an object with no extensions has no other
911 // way to say so, which is why this check belongs to the datagram readers
912 // alone.
913 if ehl.into_inner() == 0 {
914 return Err(CodecError::InvalidField);
915 }
916 let ext = read_extension_bytes(buf, ehl.into_inner())?;
917 (ehl, ext)
918 } else {
919 (VarInt::from_usize(0), Vec::new())
920 };
921 let sv = VarInt::decode(buf)?.into_inner();
922 let object_status = ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?;
923 check_extensions_against_status(object_status, &extensions)?;
924 Ok(Self {
925 track_alias,
926 group_id,
927 object_id,
928 publisher_priority,
929 extension_headers_length,
930 extensions,
931 object_status,
932 })
933 }
934}
935
936// ============================================================
937// Datagram framing
938// ============================================================
939
940/// One datagram, of whichever shape its type field names.
941///
942/// A MoQT datagram opens with a variable-length integer naming its type, and
943/// that integer is what says which of the layouts above follows it, and whether an extension block sits inside it.
944/// Neither [`DatagramHeader`] nor [`DatagramStatusHeader`] reads or writes it, so neither can be handed
945/// the first byte of a datagram a peer sent, and neither produces bytes a peer
946/// can read. This is the entry point that does both.
947///
948/// The payload of a payload-bearing datagram runs to the end of the QUIC
949/// datagram, so it is not part of this value: [`Self::decode`] stops at the end
950/// of the header and leaves the payload in the buffer, and a caller appends the
951/// payload after [`Self::encode`].
952#[derive(Debug, Clone, PartialEq, Eq)]
953pub enum Datagram {
954 /// An object carrying a payload.
955 Payload(DatagramHeader),
956 /// An object stating a status, with no payload.
957 Status(DatagramStatusHeader),
958}
959
960impl Datagram {
961 /// Whether this datagram states an Object Status instead of carrying a
962 /// payload.
963 pub fn is_status(&self) -> bool {
964 matches!(self, Self::Status(_))
965 }
966
967 /// The type field this value writes.
968 ///
969 /// The extensions bit is taken from the extension bytes themselves rather
970 /// than from the declared length beside them, which is what keeps the type
971 /// and the body from contradicting each other: a datagram whose type
972 /// announces extensions and then declares a length of 0 closes the session
973 /// on receipt, and one that announces none has nowhere to put them. The end
974 /// of group bit has no home in the body at all, so it comes from the header
975 /// flag and goes nowhere else.
976 pub fn datagram_type(&self) -> DatagramType {
977 match self {
978 Self::Payload(header) => match (header.end_of_group, header.extensions.is_empty()) {
979 (false, true) => DatagramType::Datagram,
980 (false, false) => DatagramType::DatagramExt,
981 (true, true) => DatagramType::DatagramEog,
982 (true, false) => DatagramType::DatagramEogExt,
983 },
984 Self::Status(header) => {
985 if header.extensions.is_empty() {
986 DatagramType::DatagramStatus
987 } else {
988 DatagramType::DatagramStatusExt
989 }
990 }
991 }
992 }
993
994 /// Decode a datagram from its first byte, type field included.
995 ///
996 /// Errors with [`CodecError::UnknownDatagramType`] when the datagram table
997 /// does not assign the leading type, which this draft answers with a close.
998 /// `datagram_type_error` settles it against that table alone — 0x05 is
999 /// assigned in both tables here and means different things in each.
1000 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1001 let raw = VarInt::decode(buf)?.into_inner();
1002 let datagram_type = DatagramType::from_id(raw).ok_or_else(|| datagram_type_error(raw))?;
1003 let has_extensions = datagram_type.has_extensions();
1004 if datagram_type.is_status() {
1005 return Ok(Self::Status(DatagramStatusHeader::decode_with_extensions(
1006 has_extensions,
1007 buf,
1008 )?));
1009 }
1010 let mut header = DatagramHeader::decode_with_extensions(has_extensions, buf)?;
1011 header.end_of_group = datagram_type.is_end_of_group();
1012 Ok(Self::Payload(header))
1013 }
1014
1015 /// Decode one whole datagram: [`Self::decode`], then the payload that runs
1016 /// to the end of `buf`, which must hold exactly one transport datagram.
1017 ///
1018 /// Unlike [`Self::decode`], this refuses an extension block that does not
1019 /// parse, as [`crate::object_properties::parse_object_properties`] does,
1020 /// and, with [`CodecError::MalformedTrack`], a Prior Group ID Gap larger
1021 /// than the datagram's Group ID, through [`PriorGaps::from_extensions`].
1022 /// draft-13 Section 9.3 carries the Group ID and the extension headers in
1023 /// one datagram, so this half of the draft-13 Section 10.1 rule shows in it.
1024 ///
1025 /// It also refuses bytes after a status datagram's Object Status, with
1026 /// [`CodecError::PayloadNotPermitted`]. draft-13 Section 9.3.2: "An
1027 /// OBJECT_DATAGRAM_STATUS is similar to OBJECT_DATAGRAM except it conveys
1028 /// an Object Status and has no payload."
1029 pub fn decode_object(buf: &mut impl Buf) -> Result<(Self, Vec<u8>), CodecError> {
1030 let datagram = Self::decode(buf)?;
1031 let (group_id, extensions) = match &datagram {
1032 Self::Payload(header) => (header.group_id.into_inner(), &header.extensions),
1033 Self::Status(header) => (header.group_id.into_inner(), &header.extensions),
1034 };
1035 PriorGaps::from_extensions(extensions, group_id)?;
1036 let payload = read_bytes(buf, buf.remaining())?;
1037 if let Self::Status(header) = &datagram {
1038 if !payload.is_empty() {
1039 return Err(CodecError::PayloadNotPermitted {
1040 status: header.object_status.as_u64(),
1041 len: payload.len(),
1042 detail: "its type states a status in place of a payload",
1043 });
1044 }
1045 }
1046 Ok((datagram, payload))
1047 }
1048
1049 /// Encode the datagram, type field included.
1050 pub fn encode(&self, buf: &mut impl BufMut) {
1051 let datagram_type = self.datagram_type();
1052 let has_extensions = datagram_type.has_extensions();
1053 VarInt::from_usize(datagram_type as usize).encode(buf);
1054 match self {
1055 Self::Payload(header) => header.encode_with_extensions(has_extensions, buf),
1056 Self::Status(header) => header.encode_with_extensions(has_extensions, buf),
1057 }
1058 }
1059
1060 /// Encode the datagram, refusing a header the framing it names cannot
1061 /// carry.
1062 ///
1063 /// The body is built before anything reaches `buf`, so a refused datagram
1064 /// leaves `buf` untouched rather than a type field with no body under it.
1065 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1066 let mut body = Vec::with_capacity(64);
1067 let datagram_type = self.datagram_type();
1068 let has_extensions = datagram_type.has_extensions();
1069 match self {
1070 Self::Payload(header) => header.encode_with_extensions(has_extensions, &mut body),
1071 Self::Status(header) => {
1072 check_extensions_against_status(header.object_status, &header.extensions)?;
1073 header.encode_with_extensions(has_extensions, &mut body);
1074 }
1075 }
1076 VarInt::from_usize(datagram_type as usize).encode(buf);
1077 buf.put_slice(&body);
1078 Ok(())
1079 }
1080}
1081
1082// ============================================================
1083// Fetch stream (type 0x05)
1084// ============================================================
1085
1086#[derive(Debug, Clone, PartialEq, Eq)]
1087pub struct FetchHeader {
1088 pub request_id: VarInt,
1089}
1090
1091#[derive(Debug, Clone, PartialEq, Eq)]
1092pub struct FetchObjectHeader {
1093 pub group_id: VarInt,
1094 pub subgroup_id: VarInt,
1095 pub object_id: VarInt,
1096 pub publisher_priority: u8,
1097 pub extension_headers_length: VarInt,
1098 pub extensions: Vec<u8>,
1099 pub payload_length: VarInt,
1100 pub object_status: ObjectStatus,
1101}
1102
1103impl FetchHeader {
1104 /// Encode a fetch stream header including its leading stream-type field,
1105 /// so the bytes form the start of a data stream a peer can read.
1106 ///
1107 /// [`Self::encode`] writes the body alone, which is what a caller wants
1108 /// once the stream is already open and what a caller must not use for its
1109 /// first write. The read side has had [`Self::decode_stream`] all along,
1110 /// so without this the codec could not round-trip its own fetch stream
1111 /// through its own reader.
1112 pub fn encode_stream(&self, buf: &mut impl BufMut) {
1113 VarInt::from_usize(StreamType::Fetch as usize).encode(buf);
1114 self.encode(buf);
1115 }
1116
1117 pub fn encode(&self, buf: &mut impl BufMut) {
1118 self.request_id.encode(buf);
1119 }
1120
1121 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1122 let request_id = VarInt::decode(buf)?;
1123 Ok(Self { request_id })
1124 }
1125
1126 /// Decode a fetch header from the start of a data stream, consuming the
1127 /// leading stream type varint.
1128 ///
1129 /// Errors with [`CodecError::UnknownStreamType`] when the stream table does
1130 /// not assign the leading type, which this draft answers with a close, and
1131 /// with [`CodecError::InvalidField`] when it is assigned but is not
1132 /// [`StreamType::Fetch`]. `stream_type_error` draws that line.
1133 pub fn decode_stream(buf: &mut impl Buf) -> Result<Self, CodecError> {
1134 let stream_type = VarInt::decode(buf)?.into_inner();
1135 if stream_type != StreamType::Fetch as u64 {
1136 return Err(stream_type_error(stream_type));
1137 }
1138 Self::decode(buf)
1139 }
1140}
1141
1142impl FetchObjectHeader {
1143 pub fn encode(&self, buf: &mut impl BufMut) {
1144 self.group_id.encode(buf);
1145 self.subgroup_id.encode(buf);
1146 self.object_id.encode(buf);
1147 buf.put_u8(self.publisher_priority);
1148 VarInt::from_usize(self.extensions.len()).encode(buf);
1149 buf.put_slice(&self.extensions);
1150 self.payload_length.encode(buf);
1151 if self.payload_length.into_inner() == 0 {
1152 VarInt::from_usize(self.object_status as usize).encode(buf);
1153 }
1154 }
1155
1156 /// Encode the header, refusing a status the framing cannot carry.
1157 ///
1158 /// Section 9.4.4 puts the Object Status field on the wire only when the
1159 /// Object Payload Length is zero, and Section 9.2.1.1 says "Any object
1160 /// with a status code other than zero MUST have an empty payload". A
1161 /// non-zero status paired with a non-zero payload length therefore has no
1162 /// encoding at all: [`Self::encode`] drops the status and the peer reads an
1163 /// ordinary object, which is a different object from the one the caller
1164 /// described. This refuses instead.
1165 ///
1166 /// The datagram types on this draft already refuse the same pairing. These
1167 /// two did not, and they are the ones a publisher writes on every stream.
1168 ///
1169 /// # Errors
1170 ///
1171 /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
1172 /// a non-zero Object Payload Length.
1173 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1174 if self.payload_length.into_inner() != 0 && self.object_status as usize != 0 {
1175 return Err(CodecError::InvalidField);
1176 }
1177 self.encode(buf);
1178 Ok(())
1179 }
1180
1181 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1182 let group_id = VarInt::decode(buf)?;
1183 let subgroup_id = VarInt::decode(buf)?;
1184 let object_id = VarInt::decode(buf)?;
1185 if buf.remaining() < 1 {
1186 return Err(CodecError::UnexpectedEnd);
1187 }
1188 let publisher_priority = buf.get_u8();
1189 let extension_headers_length = VarInt::decode(buf)?;
1190 let extensions = read_extension_bytes(buf, extension_headers_length.into_inner())?;
1191 let payload_length = VarInt::decode(buf)?;
1192 let object_status = if payload_length.into_inner() == 0 {
1193 let sv = VarInt::decode(buf)?.into_inner();
1194 ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?
1195 } else {
1196 ObjectStatus::Normal
1197 };
1198 check_extensions_against_status(object_status, &extensions)?;
1199 Ok(Self {
1200 group_id,
1201 subgroup_id,
1202 object_id,
1203 publisher_priority,
1204 extension_headers_length,
1205 extensions,
1206 payload_length,
1207 object_status,
1208 })
1209 }
1210}
1211
1212#[cfg(test)]
1213mod prior_gap_tests {
1214 use super::*;
1215
1216 fn header(group: usize) -> SubgroupHeader {
1217 SubgroupHeader {
1218 stream_type: StreamType::SubgroupZeroExt,
1219 track_alias: VarInt::from_usize(1),
1220 group_id: VarInt::from_usize(group),
1221 subgroup_id: VarInt::from_usize(0),
1222 publisher_priority: 0x80,
1223 }
1224 }
1225
1226 fn gaps_block(gaps: &[usize]) -> Vec<u8> {
1227 let mut block = Vec::new();
1228 for &gap in gaps {
1229 VarInt::from_usize(PRIOR_GROUP_ID_GAP as usize).encode(&mut block);
1230 VarInt::from_usize(gap).encode(&mut block);
1231 }
1232 block
1233 }
1234
1235 fn object(id: usize, gaps: &[usize]) -> ObjectHeader {
1236 let extensions = gaps_block(gaps);
1237 ObjectHeader {
1238 object_id: VarInt::from_usize(id),
1239 extension_headers_length: VarInt::from_usize(extensions.len()),
1240 extensions,
1241 payload_length: VarInt::from_usize(1),
1242 object_status: ObjectStatus::Normal,
1243 }
1244 }
1245
1246 fn is_malformed_track<T>(result: Result<T, CodecError>) -> bool {
1247 matches!(result, Err(CodecError::MalformedTrack { .. }))
1248 }
1249
1250 #[test]
1251 fn gap_larger_than_the_group_id_is_refused() {
1252 assert!(is_malformed_track(SubgroupGaps::new(&header(1)).check(&object(0, &[2]))));
1253 let gaps = SubgroupGaps::new(&header(2)).check(&object(0, &[2])).unwrap();
1254 assert_eq!(gaps.group, vec![2]);
1255 }
1256
1257 #[test]
1258 fn objects_with_different_gaps_are_refused() {
1259 let mut gaps = SubgroupGaps::new(&header(5));
1260 gaps.check(&object(0, &[2])).unwrap();
1261 gaps.check(&object(1, &[])).unwrap();
1262 gaps.check(&object(2, &[2])).unwrap();
1263 assert!(is_malformed_track(gaps.check(&object(3, &[3]))));
1264 }
1265
1266 #[test]
1267 fn values_within_one_object_are_not_compared() {
1268 let mut gaps = SubgroupGaps::new(&header(5));
1269 assert_eq!(gaps.check(&object(0, &[2, 3])).unwrap().group, vec![2, 3]);
1270 assert_eq!(gaps.check(&object(1, &[3, 2, 3])).unwrap().group, vec![3, 2, 3]);
1271 }
1272
1273 #[test]
1274 fn objects_with_a_common_value_are_accepted() {
1275 let mut gaps = SubgroupGaps::new(&header(5));
1276 gaps.check(&object(0, &[2, 3])).unwrap();
1277 gaps.check(&object(1, &[3, 4])).unwrap();
1278 gaps.check(&object(2, &[])).unwrap();
1279 gaps.check(&object(3, &[3])).unwrap();
1280 }
1281
1282 #[test]
1283 fn objects_that_share_no_value_are_refused() {
1284 let mut gaps = SubgroupGaps::new(&header(5));
1285 gaps.check(&object(0, &[2, 3])).unwrap();
1286 assert!(is_malformed_track(gaps.check(&object(1, &[4, 5]))));
1287 gaps.check(&object(2, &[3, 4])).unwrap();
1288 assert!(is_malformed_track(gaps.check(&object(3, &[4]))));
1289 }
1290
1291 #[test]
1292 fn objects_with_no_value_common_to_all_are_refused() {
1293 let mut gaps = SubgroupGaps::new(&header(5));
1294 gaps.check(&object(0, &[1, 2])).unwrap();
1295 gaps.check(&object(1, &[2, 3])).unwrap();
1296 assert!(is_malformed_track(gaps.check(&object(2, &[1, 3]))));
1297 }
1298
1299 #[test]
1300 fn a_gap_larger_than_the_group_id_under_every_reading_is_refused() {
1301 assert!(is_malformed_track(SubgroupGaps::new(&header(5)).check(&object(0, &[7, 6]))));
1302 SubgroupGaps::new(&header(5)).check(&object(0, &[2, 6])).unwrap();
1303 }
1304
1305 #[test]
1306 fn a_refused_object_is_not_recorded() {
1307 let mut gaps = SubgroupGaps::new(&header(5));
1308 assert!(is_malformed_track(gaps.check(&object(0, &[6]))));
1309 gaps.check(&object(1, &[3])).unwrap();
1310 }
1311
1312 fn datagram(group: usize, gaps: &[usize]) -> Vec<u8> {
1313 let extensions = gaps_block(gaps);
1314 let mut bytes = Vec::new();
1315 Datagram::Payload(DatagramHeader {
1316 track_alias: VarInt::from_usize(1),
1317 group_id: VarInt::from_usize(group),
1318 object_id: VarInt::from_usize(0),
1319 publisher_priority: 0x80,
1320 extension_headers_length: VarInt::from_usize(extensions.len()),
1321 extensions,
1322 end_of_group: false,
1323 })
1324 .encode(&mut bytes);
1325 bytes.push(0xAA);
1326 bytes
1327 }
1328
1329 #[test]
1330 fn decode_object_refuses_a_gap_larger_than_the_datagram_group_id() {
1331 let wire = datagram(1, &[2]);
1332 assert!(is_malformed_track(Datagram::decode_object(&mut &wire[..])));
1333 assert!(Datagram::decode(&mut &wire[..]).is_ok());
1334 let wire = datagram(2, &[2]);
1335 let (decoded, payload) = Datagram::decode_object(&mut &wire[..]).unwrap();
1336 assert_eq!(decoded.datagram_type(), DatagramType::DatagramExt);
1337 assert_eq!(payload, vec![0xAA]);
1338 }
1339
1340 #[test]
1341 fn decode_object_refuses_a_payload_after_a_status_datagram() {
1342 let mut wire = Vec::new();
1343 Datagram::Status(DatagramStatusHeader {
1344 track_alias: VarInt::from_usize(1),
1345 group_id: VarInt::from_usize(0),
1346 object_id: VarInt::from_usize(0),
1347 publisher_priority: 0x80,
1348 extension_headers_length: VarInt::from_usize(0),
1349 extensions: Vec::new(),
1350 object_status: ObjectStatus::Normal,
1351 })
1352 .encode(&mut wire);
1353 let (_, payload) = Datagram::decode_object(&mut &wire[..]).unwrap();
1354 assert!(payload.is_empty());
1355 wire.push(0xAA);
1356 assert!(matches!(
1357 Datagram::decode_object(&mut &wire[..]),
1358 Err(CodecError::PayloadNotPermitted { len: 1, .. })
1359 ));
1360 }
1361}