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