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::types::read_bytes;
11use crate::varint::VarInt;
12use bytes::{Buf, BufMut};
13
14/// Stream type IDs for draft-12 data streams.
15///
16/// # Draft-12 contradicts itself about where the subgroup types sit
17///
18/// Three places in this draft name the SUBGROUP_HEADER Type range, and they do
19/// not agree. Two of them carry draft-11's answer:
20///
21/// - Section 9, Table 10, the table of unidirectional stream types, whose
22/// SUBGROUP_HEADER row reads 0x08-0x0D.
23/// - Section 9.4.2, Figure 35, the header's own layout:
24/// `Type (i) = 0x8..0xD`.
25///
26/// The third does not. Section 9.4.2 says "There are 12 defined Type values for
27/// SUBGROUP_HEADER" and Table 13, immediately under that figure, lists them:
28/// 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x18, 0x19, 0x1A, 0x1B, 0x1C and 0x1D.
29///
30/// **This enum implements Table 13**, so a stream opening with 0x08 through
31/// 0x0D is [`CodecError::UnknownStreamType`] and closes the session.
32///
33/// Table 13 is the surviving half, for the same reason and by the same
34/// mechanism as the datagram-status contradiction documented on
35/// [`DatagramType`] — a code-point update that missed a spot:
36///
37/// - The range 0x08-0x0D holds six values, and this draft defines twelve
38/// types. Draft-11 Section 9.4.2 Table 11 lists exactly six, at 0x08
39/// through 0x0D, and the twelve here are those six crossed with the End Of
40/// Group bit draft-12 added — its own change log says so: "Use bits in
41/// SUBGROUP_HEADER and DATAGRAM* types to compress subgroup ID and
42/// extensions". So 0x08-0x0D and `0x8..0xD` are draft-11's range left
43/// behind, and they cannot hold what this draft defines.
44/// - Table 13 is the only one of the three that says what each value *means*.
45/// The other two give a range and nothing else, so following either would
46/// leave every framing decision — whether a Subgroup ID field is on the
47/// wire, whether objects carry extensions, whether the stream ends the
48/// group — with nothing to read it from.
49/// - Draft-14 keeps Table 13 unchanged and corrects the other two to match:
50/// its Table 10 reads "0x10-0x1D" and its Figure reads
51/// `Type (i) = 0x10..0x1D`. That is the disagreement being resolved in
52/// favour of Table 13 by the working group, one draft later.
53///
54/// Draft-13 repeats all three halves unchanged, and carries the same note.
55///
56/// The cost of being wrong is asymmetric and points the same way. Accepting
57/// 0x08-0x0D as well would mean parsing a stream under framing no table
58/// assigns it, which is the failure this codebase refuses elsewhere — an
59/// out-of-range Type aliased onto a valid one produces objects with plausible,
60/// wrong contents. Refusing them closes a session with a peer that followed the
61/// stale half of its own draft, which is visible, reportable, and what
62/// draft-14 says the peer should not have done.
63#[derive(Debug, Clone, Copy, PartialEq, Eq)]
64#[repr(u64)]
65pub enum StreamType {
66 /// Fetch response stream (0x05).
67 Fetch = 0x05,
68 /// Subgroup: subgroup_id=0, no extensions (0x10).
69 SubgroupZero = 0x10,
70 /// Subgroup: subgroup_id=0, with extensions (0x11).
71 SubgroupZeroExt = 0x11,
72 /// Subgroup: subgroup_id=first object ID, no extensions (0x12).
73 SubgroupFirstObj = 0x12,
74 /// Subgroup: subgroup_id=first object ID, with extensions (0x13).
75 SubgroupFirstObjExt = 0x13,
76 /// Subgroup: explicit subgroup_id, no extensions (0x14).
77 SubgroupExplicit = 0x14,
78 /// Subgroup: explicit subgroup_id, with extensions (0x15).
79 SubgroupExplicitExt = 0x15,
80 /// Subgroup: subgroup_id=0, contains end of group, no extensions (0x18).
81 SubgroupZeroEog = 0x18,
82 /// Subgroup: subgroup_id=0, contains end of group, with extensions (0x19).
83 SubgroupZeroEogExt = 0x19,
84 /// Subgroup: subgroup_id=first object ID, contains end of group, no extensions (0x1A).
85 SubgroupFirstObjEog = 0x1A,
86 /// Subgroup: subgroup_id=first object ID, contains end of group, with extensions (0x1B).
87 SubgroupFirstObjEogExt = 0x1B,
88 /// Subgroup: explicit subgroup_id, contains end of group, no extensions (0x1C).
89 SubgroupExplicitEog = 0x1C,
90 /// Subgroup: explicit subgroup_id, contains end of group, with extensions (0x1D).
91 SubgroupExplicitEogExt = 0x1D,
92}
93
94/// Hold an object to the rule that a non-existent object carries no extensions.
95///
96/// Section 9.2.1.2: "Any Object may have extension headers except those with
97/// Object Status 'Object Does Not Exist'. If an endpoint receives a non-existent
98/// Object containing extension headers it MUST close the session with a Protocol
99/// Violation."
100///
101/// The sentence is about a receiver, and it reaches all three carriers that can
102/// announce a status: an object on a subgroup stream, an object on a fetch
103/// stream, and a status datagram. A plain datagram has no status field, so it
104/// is the only carrier that cannot break the rule.
105///
106/// Reported under [`CodecError::ExtensionsOnNonExistentObject`], which is this
107/// rule and nothing else. [`CodecError::InvalidField`] is too coarse for it:
108/// shared with a dozen unrelated malformations the draft does not answer with a
109/// close, it leaves a caller unable to act on the sentence above.
110fn check_extensions_against_status(
111 status: ObjectStatus,
112 extensions: &[u8],
113) -> Result<(), CodecError> {
114 if status == ObjectStatus::ObjectDoesNotExist && !extensions.is_empty() {
115 return Err(CodecError::ExtensionsOnNonExistentObject(extensions.len()));
116 }
117 Ok(())
118}
119
120impl StreamType {
121 pub fn from_id(id: u64) -> Option<Self> {
122 match id {
123 0x05 => Some(StreamType::Fetch),
124 0x10 => Some(StreamType::SubgroupZero),
125 0x11 => Some(StreamType::SubgroupZeroExt),
126 0x12 => Some(StreamType::SubgroupFirstObj),
127 0x13 => Some(StreamType::SubgroupFirstObjExt),
128 0x14 => Some(StreamType::SubgroupExplicit),
129 0x15 => Some(StreamType::SubgroupExplicitExt),
130 0x18 => Some(StreamType::SubgroupZeroEog),
131 0x19 => Some(StreamType::SubgroupZeroEogExt),
132 0x1A => Some(StreamType::SubgroupFirstObjEog),
133 0x1B => Some(StreamType::SubgroupFirstObjEogExt),
134 0x1C => Some(StreamType::SubgroupExplicitEog),
135 0x1D => Some(StreamType::SubgroupExplicitEogExt),
136 _ => None,
137 }
138 }
139
140 pub fn is_subgroup(&self) -> bool {
141 matches!(
142 self,
143 StreamType::SubgroupZero
144 | StreamType::SubgroupZeroExt
145 | StreamType::SubgroupFirstObj
146 | StreamType::SubgroupFirstObjExt
147 | StreamType::SubgroupExplicit
148 | StreamType::SubgroupExplicitExt
149 | StreamType::SubgroupZeroEog
150 | StreamType::SubgroupZeroEogExt
151 | StreamType::SubgroupFirstObjEog
152 | StreamType::SubgroupFirstObjEogExt
153 | StreamType::SubgroupExplicitEog
154 | StreamType::SubgroupExplicitEogExt
155 )
156 }
157
158 pub fn has_extensions(&self) -> bool {
159 matches!(
160 self,
161 StreamType::SubgroupZeroExt
162 | StreamType::SubgroupFirstObjExt
163 | StreamType::SubgroupExplicitExt
164 | StreamType::SubgroupZeroEogExt
165 | StreamType::SubgroupFirstObjEogExt
166 | StreamType::SubgroupExplicitEogExt
167 )
168 }
169
170 /// True if this subgroup stream type indicates the stream contains the end of its group.
171 pub fn contains_end_of_group(&self) -> bool {
172 matches!(
173 self,
174 StreamType::SubgroupZeroEog
175 | StreamType::SubgroupZeroEogExt
176 | StreamType::SubgroupFirstObjEog
177 | StreamType::SubgroupFirstObjEogExt
178 | StreamType::SubgroupExplicitEog
179 | StreamType::SubgroupExplicitEogExt
180 )
181 }
182
183 /// True if this subgroup stream type puts an explicit Subgroup ID on the
184 /// wire.
185 ///
186 /// The Subgroup ID Field Present column of the SUBGROUP_HEADER type table
187 /// in Section 9.4.2. The other two columns of that row say what the
188 /// Subgroup ID *is* where the field is absent — zero, or the first
189 /// Object's ID — so this is only about the field, never about the value.
190 pub fn writes_subgroup_id(&self) -> bool {
191 matches!(
192 self,
193 StreamType::SubgroupExplicit
194 | StreamType::SubgroupExplicitExt
195 | StreamType::SubgroupExplicitEog
196 | StreamType::SubgroupExplicitEogExt
197 )
198 }
199}
200
201/// Datagram wire types (separate namespace from QUIC stream types).
202///
203/// The two namespaces overlap in draft-12 and cannot share one enum: 0x05 is
204/// FETCH_HEADER among stream types and OBJECT_DATAGRAM_STATUS with extensions
205/// among datagram types.
206///
207/// Draft-12 contradicts itself about where the status types sit. Its table of
208/// datagram types gives OBJECT_DATAGRAM 0x00 through 0x03 and
209/// OBJECT_DATAGRAM_STATUS 0x04 through 0x05, and the four OBJECT_DATAGRAM
210/// values are spelled out one by one as the End Of Group bit crossed with the
211/// Extensions bit — the End Of Group bit being what this draft added. But the
212/// sentence under the OBJECT_DATAGRAM_STATUS figure still reads "the set of
213/// values from 0x02 to 0x03", which is draft-11's range from before the bit
214/// existed and cannot be squared with the table above it. Draft-13 repeats both
215/// halves unchanged; draft-14 keeps the table's answer and records the sentence
216/// as a missed code-point update. The table is therefore the surviving half and
217/// the values below follow it.
218#[derive(Debug, Clone, Copy, PartialEq, Eq)]
219#[repr(u64)]
220pub enum DatagramType {
221 /// Object datagram, no extensions (0x00).
222 Datagram = 0x00,
223 /// Object datagram, with extensions (0x01).
224 DatagramExt = 0x01,
225 /// Object datagram carrying the end of its group, no extensions (0x02).
226 DatagramEog = 0x02,
227 /// Object datagram carrying the end of its group, with extensions (0x03).
228 DatagramEogExt = 0x03,
229 /// Object datagram status, no extensions (0x04).
230 DatagramStatus = 0x04,
231 /// Object datagram status, with extensions (0x05).
232 DatagramStatusExt = 0x05,
233}
234
235impl DatagramType {
236 pub fn from_id(id: u64) -> Option<Self> {
237 match id {
238 0x00 => Some(DatagramType::Datagram),
239 0x01 => Some(DatagramType::DatagramExt),
240 0x02 => Some(DatagramType::DatagramEog),
241 0x03 => Some(DatagramType::DatagramEogExt),
242 0x04 => Some(DatagramType::DatagramStatus),
243 0x05 => Some(DatagramType::DatagramStatusExt),
244 _ => None,
245 }
246 }
247
248 pub fn has_extensions(&self) -> bool {
249 matches!(
250 self,
251 DatagramType::DatagramExt
252 | DatagramType::DatagramEogExt
253 | DatagramType::DatagramStatusExt
254 )
255 }
256
257 pub fn is_status(&self) -> bool {
258 matches!(self, DatagramType::DatagramStatus | DatagramType::DatagramStatusExt)
259 }
260
261 pub fn is_end_of_group(&self) -> bool {
262 matches!(self, DatagramType::DatagramEog | DatagramType::DatagramEogExt)
263 }
264}
265
266/// Which failure a leading unidirectional stream type that is not the one a
267/// reader wants is.
268///
269/// Section 9: "An endpoint that receives an unknown stream or datagram type
270/// MUST close the session." One sentence, two tables, and on this draft the two
271/// tables collide: 0x05 is FETCH_HEADER in the stream table and
272/// OBJECT_DATAGRAM_STATUS with extensions in the datagram table. Which table
273/// was consulted is therefore part of the answer, not a detail, and it is why
274/// [`CodecError::UnknownStreamType`] and [`CodecError::UnknownDatagramType`]
275/// are separate variants rather than one.
276///
277/// The stream table assigns 0x05 and the range 0x10 to 0x1D. Everything outside
278/// them is unknown at the head of a stream, and the session ends.
279fn stream_type_error(raw: u64) -> CodecError {
280 if StreamType::from_id(raw).is_some() {
281 CodecError::InvalidField
282 } else {
283 CodecError::UnknownStreamType(raw)
284 }
285}
286
287/// Which failure a leading datagram type that is not one a reader wants is.
288///
289/// The datagram half of the sentence quoted on `stream_type_error`, read
290/// against the other table: 0x00 to 0x05 are what it assigns, and everything
291/// else arriving as a datagram is unknown.
292///
293/// Answered from [`DatagramType`] alone, never from [`StreamType`]. A value in
294/// both tables means one thing as a datagram and another as a stream, and
295/// consulting the wrong one turns an assigned datagram type into an unknown one
296/// or the reverse.
297fn datagram_type_error(raw: u64) -> CodecError {
298 if DatagramType::from_id(raw).is_some() {
299 CodecError::InvalidField
300 } else {
301 CodecError::UnknownDatagramType(raw)
302 }
303}
304
305// ── Extension helpers ─────────────────────────────────────────
306
307fn read_extension_bytes(buf: &mut impl Buf, byte_len: u64) -> Result<Vec<u8>, CodecError> {
308 read_bytes(buf, byte_len as usize)
309}
310
311// ============================================================
312// Subgroup stream header
313// ============================================================
314
315#[derive(Debug, Clone, PartialEq, Eq)]
316pub struct SubgroupHeader {
317 pub stream_type: StreamType,
318 pub track_alias: VarInt,
319 pub group_id: VarInt,
320 pub subgroup_id: VarInt,
321 pub publisher_priority: u8,
322}
323
324impl SubgroupHeader {
325 /// Encode a subgroup stream header including its leading stream-type
326 /// field, so the bytes form the start of a data stream a peer can read.
327 ///
328 /// [`Self::encode`] writes the body alone, which is what a caller wants
329 /// once the stream is already open and what a caller must not use for its
330 /// first write. It is also the half that cannot stand on its own here,
331 /// because the stream type is what says whether a Subgroup ID follows it
332 /// and whether the objects on the stream carry extension headers.
333 pub fn encode_stream(&self, buf: &mut impl BufMut) {
334 VarInt::from_usize(self.stream_type as usize).encode(buf);
335 self.encode(buf);
336 }
337
338 /// Encode the header body, without its leading stream-type field.
339 ///
340 /// Driven by the stream type, and silent about a `subgroup_id` it decides
341 /// not to write: on a type whose Subgroup ID Field Present column reads No
342 /// the field is dropped, and the peer reads the subgroup the *type* names -
343 /// zero, or the first Object's ID - rather than the one in hand. Nothing is
344 /// malformed about the result, which is what makes it worth refusing rather
345 /// than tolerating. [`Self::encode_checked`] refuses it.
346 pub fn encode(&self, buf: &mut impl BufMut) {
347 self.track_alias.encode(buf);
348 self.group_id.encode(buf);
349 if self.stream_type.writes_subgroup_id() {
350 self.subgroup_id.encode(buf);
351 }
352 buf.put_u8(self.publisher_priority);
353 }
354
355 /// Encode the header body, refusing a Subgroup ID this stream type has
356 /// nowhere to put.
357 ///
358 /// [`Self::decode_with_type`] leaves the field at zero for every type that
359 /// does not carry it, so a decoded header always passes: the refusal is for
360 /// a header assembled by hand, where a caller set an ID the type will
361 /// discard.
362 ///
363 /// A zero is accepted under any type. It is what the decoder produces, and
364 /// on a Subgroup ID Value column reading `0` it is also the truth, so
365 /// refusing it would refuse the ordinary case to catch nothing.
366 ///
367 /// # Errors
368 ///
369 /// [`CodecError::InvalidField`] if a non-zero Subgroup ID sits under a
370 /// stream type that writes no Subgroup ID field.
371 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
372 if !self.stream_type.writes_subgroup_id() && self.subgroup_id.into_inner() != 0 {
373 return Err(CodecError::InvalidField);
374 }
375 self.encode(buf);
376 Ok(())
377 }
378
379 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
380 Self::decode_with_type(StreamType::SubgroupExplicit, buf)
381 }
382
383 pub fn decode_with_type(
384 stream_type: StreamType,
385 buf: &mut impl Buf,
386 ) -> Result<Self, CodecError> {
387 let track_alias = VarInt::decode(buf)?;
388 let group_id = VarInt::decode(buf)?;
389 let subgroup_id = match stream_type {
390 StreamType::SubgroupZero
391 | StreamType::SubgroupZeroExt
392 | StreamType::SubgroupZeroEog
393 | StreamType::SubgroupZeroEogExt => VarInt::from_usize(0),
394 StreamType::SubgroupExplicit
395 | StreamType::SubgroupExplicitExt
396 | StreamType::SubgroupExplicitEog
397 | StreamType::SubgroupExplicitEogExt => VarInt::decode(buf)?,
398 StreamType::SubgroupFirstObj
399 | StreamType::SubgroupFirstObjExt
400 | StreamType::SubgroupFirstObjEog
401 | StreamType::SubgroupFirstObjEogExt => VarInt::from_usize(0),
402 _ => return Err(CodecError::InvalidField),
403 };
404 if buf.remaining() < 1 {
405 return Err(CodecError::UnexpectedEnd);
406 }
407 let publisher_priority = buf.get_u8();
408 Ok(Self { stream_type, track_alias, group_id, subgroup_id, publisher_priority })
409 }
410
411 /// Decode a subgroup header from the start of a data stream, consuming
412 /// the leading stream type varint and using it to select the variant.
413 ///
414 /// Errors with [`CodecError::UnknownStreamType`] when the stream table does
415 /// not assign the leading type, which this draft answers with a close, and
416 /// with [`CodecError::InvalidField`] when it does assign it but not to a
417 /// subgroup. `stream_type_error` draws that line.
418 pub fn decode_stream(buf: &mut impl Buf) -> Result<Self, CodecError> {
419 let raw = VarInt::decode(buf)?.into_inner();
420 let stream_type = StreamType::from_id(raw).ok_or_else(|| stream_type_error(raw))?;
421 if !stream_type.is_subgroup() {
422 return Err(stream_type_error(raw));
423 }
424 Self::decode_with_type(stream_type, buf)
425 }
426}
427
428// ============================================================
429// Object header within subgroup
430// ============================================================
431
432#[derive(Debug, Clone, PartialEq, Eq)]
433pub struct ObjectHeader {
434 pub object_id: VarInt,
435 pub extension_headers_length: VarInt,
436 pub extensions: Vec<u8>,
437 pub payload_length: VarInt,
438 pub object_status: ObjectStatus,
439}
440
441impl ObjectHeader {
442 /// Encode the object header in the framing that carries no extension
443 /// block.
444 ///
445 /// Lossy, and lossy in a way the caller cannot see: an object holding
446 /// extension headers is written without them and without a word. Which
447 /// framing is correct is not a property of the object at all - Section
448 /// 9.4.2 gives the stream's type an Extensions Present column, and every
449 /// object on the stream follows it - so this entry point can only guess,
450 /// and it guesses "absent". Prefer [`Self::encode_with_extensions`], which
451 /// is told, or [`Self::encode_checked`], which refuses what it would
452 /// otherwise drop.
453 pub fn encode(&self, buf: &mut impl BufMut) {
454 self.encode_with_extensions(false, buf);
455 }
456
457 /// Encode the header, refusing a status the framing cannot carry.
458 ///
459 /// Section 9.4.2 puts the Object Status field on the wire only when the
460 /// Object Payload Length is zero, and Section 9.2.1.1 says "Any object
461 /// with a status code other than zero MUST have an empty payload". A
462 /// non-zero status paired with a non-zero payload length therefore has no
463 /// encoding at all: [`Self::encode`] drops the status and the peer reads an
464 /// ordinary object, which is a different object from the one the caller
465 /// described. This refuses instead.
466 ///
467 /// The datagram types on this draft already refuse the same pairing. These
468 /// two did not, and they are the ones a publisher writes on every stream.
469 ///
470 /// Extension headers are refused here rather than dropped, for a reason
471 /// the status rule does not share: this entry point writes the framing
472 /// that has no Extension Headers Length field, so the bytes have nowhere
473 /// to go. Writing them anyway is not an option and losing them silently
474 /// puts a stream on the wire that no reader can follow - a reader on an
475 /// extensions-bearing stream takes the Object Payload Length as the
476 /// extension length and every object after it is misread. A caller that
477 /// knows the stream's framing wants
478 /// [`Self::encode_checked_with_extensions`].
479 ///
480 /// # Errors
481 ///
482 /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
483 /// a non-zero Object Payload Length, or if the object carries extension
484 /// headers this framing cannot write.
485 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
486 self.encode_checked_with_extensions(false, buf)
487 }
488
489 /// Encode the object header into a stream whose type has already settled
490 /// whether objects carry an extension block, refusing what that framing
491 /// cannot express.
492 ///
493 /// `has_extensions` is the stream's answer, not the object's: Section
494 /// 9.4.2 fixes it for the whole stream from the SUBGROUP_HEADER type, so
495 /// an object with no extensions on a stream that carries them still writes
496 /// a length of zero, and that is the one direction not refused here. The
497 /// other direction has no encoding, so it is refused.
498 ///
499 /// # Errors
500 ///
501 /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
502 /// a non-zero Object Payload Length, or if `has_extensions` is `false`
503 /// while the object carries extension headers.
504 pub fn encode_checked_with_extensions(
505 &self,
506 has_extensions: bool,
507 buf: &mut impl BufMut,
508 ) -> Result<(), CodecError> {
509 if self.payload_length.into_inner() != 0 && self.object_status as usize != 0 {
510 return Err(CodecError::InvalidField);
511 }
512 if !has_extensions && !self.extensions.is_empty() {
513 return Err(CodecError::InvalidField);
514 }
515 self.encode_with_extensions(has_extensions, buf);
516 Ok(())
517 }
518
519 /// Encode the object header, writing the extension block only when the
520 /// stream's type says objects carry one.
521 ///
522 /// Infallible, and so unable to say that a `false` here discards the
523 /// extension headers the object holds. [`Self::encode_checked_with_extensions`]
524 /// is the same write with that refusal in front of it.
525 pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
526 self.object_id.encode(buf);
527 if has_extensions {
528 VarInt::from_usize(self.extensions.len()).encode(buf);
529 buf.put_slice(&self.extensions);
530 }
531 self.payload_length.encode(buf);
532 if self.payload_length.into_inner() == 0 {
533 VarInt::from_usize(self.object_status as usize).encode(buf);
534 }
535 }
536
537 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
538 Self::decode_with_extensions(false, buf)
539 }
540
541 pub fn decode_with_extensions(
542 has_extensions: bool,
543 buf: &mut impl Buf,
544 ) -> Result<Self, CodecError> {
545 let object_id = VarInt::decode(buf)?;
546 let (extension_headers_length, extensions) = if has_extensions {
547 let ehl = VarInt::decode(buf)?;
548 let ext = read_extension_bytes(buf, ehl.into_inner())?;
549 (ehl, ext)
550 } else {
551 (VarInt::from_usize(0), Vec::new())
552 };
553 let payload_length = VarInt::decode(buf)?;
554 let object_status = if payload_length.into_inner() == 0 {
555 let sv = VarInt::decode(buf)?.into_inner();
556 ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?
557 } else {
558 ObjectStatus::Normal
559 };
560 check_extensions_against_status(object_status, &extensions)?;
561 Ok(Self { object_id, extension_headers_length, extensions, payload_length, object_status })
562 }
563}
564
565// ============================================================
566// Datagram (types 0x00 through 0x03)
567// ============================================================
568
569#[derive(Debug, Clone, PartialEq, Eq)]
570pub struct DatagramHeader {
571 pub track_alias: VarInt,
572 pub group_id: VarInt,
573 pub object_id: VarInt,
574 pub publisher_priority: u8,
575 pub extension_headers_length: VarInt,
576 pub extensions: Vec<u8>,
577 /// Whether this object is the last one in its group.
578 ///
579 /// Draft-12 added this flag and put it in the datagram type rather than in
580 /// the header body, so it is not written or read by the methods below;
581 /// [`DatagramType::is_end_of_group`] is where it lives on the wire.
582 pub end_of_group: bool,
583}
584
585impl DatagramHeader {
586 /// Encode the datagram header in the framing that carries no extension
587 /// block.
588 ///
589 /// Lossy in the same way the subgroup object header is: an extension block
590 /// this value holds is dropped, because the framing being written has no
591 /// field for it. The type byte decides which framing is right, and this
592 /// entry point does not write the type byte, so it cannot consult it.
593 /// [`Datagram::encode`] does both together and never disagrees with itself;
594 /// this is the piece for a caller that has already written the type.
595 pub fn encode(&self, buf: &mut impl BufMut) {
596 self.encode_with_extensions(false, buf);
597 }
598
599 pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
600 self.track_alias.encode(buf);
601 self.group_id.encode(buf);
602 self.object_id.encode(buf);
603 buf.put_u8(self.publisher_priority);
604 if has_extensions {
605 VarInt::from_usize(self.extensions.len()).encode(buf);
606 buf.put_slice(&self.extensions);
607 }
608 }
609
610 /// Encode the datagram header, refusing what this framing cannot carry.
611 ///
612 /// No status is ever refused, and that is a fact about this draft rather
613 /// than a check left out. This is the OBJECT_DATAGRAM of Section 9.3.1,
614 /// whose layout carries no Object Status field at all; a datagram that
615 /// states a status is the separate OBJECT_DATAGRAM_STATUS message, modelled
616 /// here as [`DatagramStatusHeader`]. So there is no status for
617 /// [`Self::encode`] to drop, and nothing for Section 9.2.1.1's "Any object
618 /// with a status code other than zero MUST have an empty payload" to rule
619 /// on: an object framed this way has status zero by construction.
620 ///
621 /// The extension block is a different matter. [`Self::encode`] writes the
622 /// framing without one, so a block this value holds has nowhere to go, and
623 /// dropping it silently is what puts a datagram on the wire describing
624 /// something other than what the caller built. That is refused here.
625 ///
626 /// The fallible signature is also what lets one entry point span every
627 /// draft. `dispatch::AnyDatagramHeader::encode` calls this on all thirteen,
628 /// and the drafts whose payload-bearing datagram *does* carry a status field
629 /// need somewhere to say no.
630 ///
631 /// # Errors
632 ///
633 /// [`CodecError::InvalidField`] if the value carries extension headers,
634 /// which this framing has no field for.
635 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
636 if !self.extensions.is_empty() {
637 return Err(CodecError::InvalidField);
638 }
639 self.encode(buf);
640 Ok(())
641 }
642
643 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
644 Self::decode_with_extensions(false, buf)
645 }
646
647 pub fn decode_with_extensions(
648 has_extensions: bool,
649 buf: &mut impl Buf,
650 ) -> Result<Self, CodecError> {
651 let track_alias = VarInt::decode(buf)?;
652 let group_id = VarInt::decode(buf)?;
653 let object_id = VarInt::decode(buf)?;
654 if buf.remaining() < 1 {
655 return Err(CodecError::UnexpectedEnd);
656 }
657 let publisher_priority = buf.get_u8();
658 let (extension_headers_length, extensions) = if has_extensions {
659 let ehl = VarInt::decode(buf)?;
660 // A datagram whose type says extensions are present must actually carry
661 // some: receiving one with an Extension Headers Length of 0 closes the
662 // session. The opposite holds on a subgroup stream, where the type byte is
663 // fixed for the whole stream and an object with no extensions has no other
664 // way to say so, which is why this check belongs to the datagram readers
665 // alone.
666 if ehl.into_inner() == 0 {
667 return Err(CodecError::InvalidField);
668 }
669 let ext = read_extension_bytes(buf, ehl.into_inner())?;
670 (ehl, ext)
671 } else {
672 (VarInt::from_usize(0), Vec::new())
673 };
674 Ok(Self {
675 track_alias,
676 group_id,
677 object_id,
678 publisher_priority,
679 extension_headers_length,
680 extensions,
681 end_of_group: false,
682 })
683 }
684}
685
686// ============================================================
687// Datagram Status (types 0x04, 0x05)
688// ============================================================
689
690#[derive(Debug, Clone, PartialEq, Eq)]
691pub struct DatagramStatusHeader {
692 pub track_alias: VarInt,
693 pub group_id: VarInt,
694 pub object_id: VarInt,
695 pub publisher_priority: u8,
696 pub extension_headers_length: VarInt,
697 pub extensions: Vec<u8>,
698 pub object_status: ObjectStatus,
699}
700
701impl DatagramStatusHeader {
702 pub fn encode(&self, buf: &mut impl BufMut) {
703 self.encode_with_extensions(false, buf);
704 }
705
706 /// Encode the status datagram header, refusing an extension block this
707 /// framing cannot carry.
708 ///
709 /// The same one-sided rule the payload-bearing header obeys, and the same
710 /// reason for it: [`Self::encode`] writes the framing without a block, so
711 /// bytes held here would be dropped rather than written.
712 ///
713 /// # Errors
714 ///
715 /// [`CodecError::InvalidField`] if the value carries extension headers.
716 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
717 if !self.extensions.is_empty() {
718 return Err(CodecError::InvalidField);
719 }
720 self.encode(buf);
721 Ok(())
722 }
723
724 pub fn encode_with_extensions(&self, has_extensions: bool, buf: &mut impl BufMut) {
725 self.track_alias.encode(buf);
726 self.group_id.encode(buf);
727 self.object_id.encode(buf);
728 buf.put_u8(self.publisher_priority);
729 if has_extensions {
730 VarInt::from_usize(self.extensions.len()).encode(buf);
731 buf.put_slice(&self.extensions);
732 }
733 VarInt::from_usize(self.object_status as usize).encode(buf);
734 }
735
736 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
737 Self::decode_with_extensions(false, buf)
738 }
739
740 pub fn decode_with_extensions(
741 has_extensions: bool,
742 buf: &mut impl Buf,
743 ) -> Result<Self, CodecError> {
744 let track_alias = VarInt::decode(buf)?;
745 let group_id = VarInt::decode(buf)?;
746 let object_id = VarInt::decode(buf)?;
747 if buf.remaining() < 1 {
748 return Err(CodecError::UnexpectedEnd);
749 }
750 let publisher_priority = buf.get_u8();
751 let (extension_headers_length, extensions) = if has_extensions {
752 let ehl = VarInt::decode(buf)?;
753 // A datagram whose type says extensions are present must actually carry
754 // some: receiving one with an Extension Headers Length of 0 closes the
755 // session. The opposite holds on a subgroup stream, where the type byte is
756 // fixed for the whole stream and an object with no extensions has no other
757 // way to say so, which is why this check belongs to the datagram readers
758 // alone.
759 if ehl.into_inner() == 0 {
760 return Err(CodecError::InvalidField);
761 }
762 let ext = read_extension_bytes(buf, ehl.into_inner())?;
763 (ehl, ext)
764 } else {
765 (VarInt::from_usize(0), Vec::new())
766 };
767 let sv = VarInt::decode(buf)?.into_inner();
768 let object_status = ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?;
769 check_extensions_against_status(object_status, &extensions)?;
770 Ok(Self {
771 track_alias,
772 group_id,
773 object_id,
774 publisher_priority,
775 extension_headers_length,
776 extensions,
777 object_status,
778 })
779 }
780}
781
782// ============================================================
783// Datagram framing
784// ============================================================
785
786/// One datagram, of whichever shape its type field names.
787///
788/// A MoQT datagram opens with a variable-length integer naming its type, and
789/// that integer is what says which of the layouts above follows it, and whether an extension block sits inside it.
790/// Neither [`DatagramHeader`] nor [`DatagramStatusHeader`] reads or writes it, so neither can be handed
791/// the first byte of a datagram a peer sent, and neither produces bytes a peer
792/// can read. This is the entry point that does both.
793///
794/// The payload of a payload-bearing datagram runs to the end of the QUIC
795/// datagram, so it is not part of this value: [`Self::decode`] stops at the end
796/// of the header and leaves the payload in the buffer, and a caller appends the
797/// payload after [`Self::encode`].
798#[derive(Debug, Clone, PartialEq, Eq)]
799pub enum Datagram {
800 /// An object carrying a payload.
801 Payload(DatagramHeader),
802 /// An object stating a status, with no payload.
803 Status(DatagramStatusHeader),
804}
805
806impl Datagram {
807 /// Whether this datagram states an Object Status instead of carrying a
808 /// payload.
809 pub fn is_status(&self) -> bool {
810 matches!(self, Self::Status(_))
811 }
812
813 /// The type field this value writes.
814 ///
815 /// The extensions bit is taken from the extension bytes themselves rather
816 /// than from the declared length beside them, which is what keeps the type
817 /// and the body from contradicting each other: a datagram whose type
818 /// announces extensions and then declares a length of 0 closes the session
819 /// on receipt, and one that announces none has nowhere to put them. The end
820 /// of group bit has no home in the body at all, so it comes from the header
821 /// flag and goes nowhere else.
822 pub fn datagram_type(&self) -> DatagramType {
823 match self {
824 Self::Payload(header) => match (header.end_of_group, header.extensions.is_empty()) {
825 (false, true) => DatagramType::Datagram,
826 (false, false) => DatagramType::DatagramExt,
827 (true, true) => DatagramType::DatagramEog,
828 (true, false) => DatagramType::DatagramEogExt,
829 },
830 Self::Status(header) => {
831 if header.extensions.is_empty() {
832 DatagramType::DatagramStatus
833 } else {
834 DatagramType::DatagramStatusExt
835 }
836 }
837 }
838 }
839
840 /// Decode a datagram from its first byte, type field included.
841 ///
842 /// Errors with [`CodecError::UnknownDatagramType`] when the datagram table
843 /// does not assign the leading type, which this draft answers with a close.
844 /// `datagram_type_error` settles it against that table alone — 0x05 is
845 /// assigned in both tables here and means different things in each.
846 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
847 let raw = VarInt::decode(buf)?.into_inner();
848 let datagram_type = DatagramType::from_id(raw).ok_or_else(|| datagram_type_error(raw))?;
849 let has_extensions = datagram_type.has_extensions();
850 if datagram_type.is_status() {
851 return Ok(Self::Status(DatagramStatusHeader::decode_with_extensions(
852 has_extensions,
853 buf,
854 )?));
855 }
856 let mut header = DatagramHeader::decode_with_extensions(has_extensions, buf)?;
857 header.end_of_group = datagram_type.is_end_of_group();
858 Ok(Self::Payload(header))
859 }
860
861 /// Encode the datagram, type field included.
862 pub fn encode(&self, buf: &mut impl BufMut) {
863 let datagram_type = self.datagram_type();
864 let has_extensions = datagram_type.has_extensions();
865 VarInt::from_usize(datagram_type as usize).encode(buf);
866 match self {
867 Self::Payload(header) => header.encode_with_extensions(has_extensions, buf),
868 Self::Status(header) => header.encode_with_extensions(has_extensions, buf),
869 }
870 }
871
872 /// Encode the datagram, refusing a header the framing it names cannot
873 /// carry.
874 ///
875 /// The body is built before anything reaches `buf`, so a refused datagram
876 /// leaves `buf` untouched rather than a type field with no body under it.
877 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
878 let mut body = Vec::with_capacity(64);
879 let datagram_type = self.datagram_type();
880 let has_extensions = datagram_type.has_extensions();
881 match self {
882 Self::Payload(header) => header.encode_with_extensions(has_extensions, &mut body),
883 Self::Status(header) => {
884 check_extensions_against_status(header.object_status, &header.extensions)?;
885 header.encode_with_extensions(has_extensions, &mut body);
886 }
887 }
888 VarInt::from_usize(datagram_type as usize).encode(buf);
889 buf.put_slice(&body);
890 Ok(())
891 }
892}
893
894// ============================================================
895// Fetch stream (type 0x05)
896// ============================================================
897
898#[derive(Debug, Clone, PartialEq, Eq)]
899pub struct FetchHeader {
900 pub request_id: VarInt,
901}
902
903#[derive(Debug, Clone, PartialEq, Eq)]
904pub struct FetchObjectHeader {
905 pub group_id: VarInt,
906 pub subgroup_id: VarInt,
907 pub object_id: VarInt,
908 pub publisher_priority: u8,
909 pub extension_headers_length: VarInt,
910 pub extensions: Vec<u8>,
911 pub payload_length: VarInt,
912 pub object_status: ObjectStatus,
913}
914
915impl FetchHeader {
916 /// Encode a fetch stream header including its leading stream-type field,
917 /// so the bytes form the start of a data stream a peer can read.
918 ///
919 /// [`Self::encode`] writes the body alone, which is what a caller wants
920 /// once the stream is already open and what a caller must not use for its
921 /// first write. The read side has had [`Self::decode_stream`] all along,
922 /// so without this the codec could not round-trip its own fetch stream
923 /// through its own reader.
924 pub fn encode_stream(&self, buf: &mut impl BufMut) {
925 VarInt::from_usize(StreamType::Fetch as usize).encode(buf);
926 self.encode(buf);
927 }
928
929 pub fn encode(&self, buf: &mut impl BufMut) {
930 self.request_id.encode(buf);
931 }
932
933 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
934 let request_id = VarInt::decode(buf)?;
935 Ok(Self { request_id })
936 }
937
938 /// Decode a fetch header from the start of a data stream, consuming the
939 /// leading stream type varint.
940 ///
941 /// Errors with [`CodecError::UnknownStreamType`] when the stream table does
942 /// not assign the leading type, which this draft answers with a close, and
943 /// with [`CodecError::InvalidField`] when it is assigned but is not
944 /// [`StreamType::Fetch`]. `stream_type_error` draws that line.
945 pub fn decode_stream(buf: &mut impl Buf) -> Result<Self, CodecError> {
946 let stream_type = VarInt::decode(buf)?.into_inner();
947 if stream_type != StreamType::Fetch as u64 {
948 return Err(stream_type_error(stream_type));
949 }
950 Self::decode(buf)
951 }
952}
953
954impl FetchObjectHeader {
955 pub fn encode(&self, buf: &mut impl BufMut) {
956 self.group_id.encode(buf);
957 self.subgroup_id.encode(buf);
958 self.object_id.encode(buf);
959 buf.put_u8(self.publisher_priority);
960 VarInt::from_usize(self.extensions.len()).encode(buf);
961 buf.put_slice(&self.extensions);
962 self.payload_length.encode(buf);
963 if self.payload_length.into_inner() == 0 {
964 VarInt::from_usize(self.object_status as usize).encode(buf);
965 }
966 }
967
968 /// Encode the header, refusing a status the framing cannot carry.
969 ///
970 /// Section 9.4.4 puts the Object Status field on the wire only when the
971 /// Object Payload Length is zero, and Section 9.2.1.1 says "Any object
972 /// with a status code other than zero MUST have an empty payload". A
973 /// non-zero status paired with a non-zero payload length therefore has no
974 /// encoding at all: [`Self::encode`] drops the status and the peer reads an
975 /// ordinary object, which is a different object from the one the caller
976 /// described. This refuses instead.
977 ///
978 /// The datagram types on this draft already refuse the same pairing. These
979 /// two did not, and they are the ones a publisher writes on every stream.
980 ///
981 /// # Errors
982 ///
983 /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
984 /// a non-zero Object Payload Length.
985 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
986 if self.payload_length.into_inner() != 0 && self.object_status as usize != 0 {
987 return Err(CodecError::InvalidField);
988 }
989 self.encode(buf);
990 Ok(())
991 }
992
993 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
994 let group_id = VarInt::decode(buf)?;
995 let subgroup_id = VarInt::decode(buf)?;
996 let object_id = VarInt::decode(buf)?;
997 if buf.remaining() < 1 {
998 return Err(CodecError::UnexpectedEnd);
999 }
1000 let publisher_priority = buf.get_u8();
1001 let extension_headers_length = VarInt::decode(buf)?;
1002 let extensions = read_extension_bytes(buf, extension_headers_length.into_inner())?;
1003 let payload_length = VarInt::decode(buf)?;
1004 let object_status = if payload_length.into_inner() == 0 {
1005 let sv = VarInt::decode(buf)?.into_inner();
1006 ObjectStatus::from_u64(sv).ok_or(CodecError::InvalidField)?
1007 } else {
1008 ObjectStatus::Normal
1009 };
1010 check_extensions_against_status(object_status, &extensions)?;
1011 Ok(Self {
1012 group_id,
1013 subgroup_id,
1014 object_id,
1015 publisher_priority,
1016 extension_headers_length,
1017 extensions,
1018 payload_length,
1019 object_status,
1020 })
1021 }
1022}
1023
1024#[cfg(test)]
1025mod tests {
1026 use super::*;
1027
1028 /// A payload datagram for track 1, group 0, object 7, priority 128, with
1029 /// no extension block — the shape [`DatagramHeader::encode`] writes.
1030 fn payload_datagram() -> DatagramHeader {
1031 DatagramHeader {
1032 track_alias: VarInt::from_usize(1),
1033 group_id: VarInt::from_usize(0),
1034 object_id: VarInt::from_usize(7),
1035 publisher_priority: 128,
1036 extension_headers_length: VarInt::from_usize(0),
1037 extensions: Vec::new(),
1038 end_of_group: false,
1039 }
1040 }
1041
1042 /// The same datagram on the message that does carry a status.
1043 fn status_datagram(status: ObjectStatus) -> DatagramStatusHeader {
1044 DatagramStatusHeader {
1045 track_alias: VarInt::from_usize(1),
1046 group_id: VarInt::from_usize(0),
1047 object_id: VarInt::from_usize(7),
1048 publisher_priority: 128,
1049 extension_headers_length: VarInt::from_usize(0),
1050 extensions: Vec::new(),
1051 object_status: status,
1052 }
1053 }
1054
1055 /// Draft-12's payload datagram has no status for the encoder to drop, and
1056 /// so nothing for the fallible encode to refuse.
1057 ///
1058 /// [`DatagramHeader`] is the OBJECT_DATAGRAM of Section 9.3.1, whose layout
1059 /// carries no Object Status field at all. A datagram that states a status
1060 /// is the separate OBJECT_DATAGRAM_STATUS message of Section 9.3.2, modelled
1061 /// here as [`DatagramStatusHeader`]. The refusal drafts 07, 08 and 14-19
1062 /// need on this path therefore has nothing to bite on, and this gate holds
1063 /// [`DatagramHeader::encode_checked`] to writing exactly what
1064 /// [`DatagramHeader::encode`] writes and never refusing — the alternative
1065 /// being a codec that answers `Err` for a datagram every draft-12 publisher
1066 /// is entitled to send.
1067 ///
1068 /// The second half is what makes the first half safe rather than merely
1069 /// permissive. Every status draft-12 assigns travels intact on the message
1070 /// that can express one, so leaving the payload datagram unchecked loses
1071 /// nothing; if a status could ride this header, the check being skipped
1072 /// here would be the check drafts 07 and 18 need.
1073 ///
1074 /// # What this catches, observed by making each change and running it
1075 ///
1076 /// Making `encode_checked` refuse unconditionally, as an over-eager copy of
1077 /// the drafts that do need a check would:
1078 ///
1079 /// ```text
1080 /// draft-12's payload datagram has no status to refuse: InvalidField
1081 /// ```
1082 ///
1083 /// Making `encode_checked` return `Ok(())` without writing anything:
1084 ///
1085 /// ```text
1086 /// assertion `left == right` failed: the fallible encode must write exactly what `encode` writes
1087 /// left: []
1088 /// right: [1, 0, 7, 128]
1089 /// ```
1090 ///
1091 /// Making `DatagramStatusHeader::encode_with_extensions` write a constant
1092 /// `ObjectStatus::Normal` instead of the header's own status, so the
1093 /// message that is supposed to carry a status stops doing so:
1094 ///
1095 /// ```text
1096 /// assertion `left == right` failed: ObjectDoesNotExist must survive on the message that carries a status
1097 /// left: Normal
1098 /// right: ObjectDoesNotExist
1099 /// ```
1100 #[test]
1101 fn a_payload_datagram_has_no_status_to_refuse() {
1102 let header = payload_datagram();
1103
1104 let mut checked = Vec::new();
1105 header.encode_checked(&mut checked).unwrap_or_else(|e| {
1106 panic!("draft-12's payload datagram has no status to refuse: {e:?}")
1107 });
1108
1109 let mut plain = Vec::new();
1110 header.encode(&mut plain);
1111 assert_eq!(checked, plain, "the fallible encode must write exactly what `encode` writes");
1112
1113 let decoded = DatagramHeader::decode(&mut &checked[..])
1114 .expect("the bytes encode_checked wrote must parse back");
1115 assert_eq!(decoded, header, "the payload datagram did not survive its own encoding");
1116
1117 for &status in ObjectStatus::ALL {
1118 let mut buf = Vec::new();
1119 status_datagram(status).encode(&mut buf);
1120 let decoded = DatagramStatusHeader::decode(&mut &buf[..])
1121 .unwrap_or_else(|e| panic!("the status datagram for {status:?} must parse: {e:?}"));
1122 assert_eq!(
1123 decoded.object_status, status,
1124 "{status:?} must survive on the message that carries a status"
1125 );
1126 }
1127 }
1128}