Skip to main content

moqtap_codec/draft14/
data_stream.rs

1//! Draft-14 data streams: subgroup streams, fetch streams, datagrams.
2//!
3//! Draft-14 carries object data in three wire shapes:
4//!
5//! * **Subgroup stream**: starts with a Type byte `0x10..=0x1D`
6//!   whose bit-flags determine whether a Subgroup ID field is present,
7//!   whether the subgroup ID is zero or the first Object ID, whether
8//!   extension headers are present, and whether the stream ends at a
9//!   group boundary. Object IDs are delta-encoded relative to the
10//!   previous Object ID in the same stream.
11//!
12//! * **Fetch stream**: Type `0x05`, Request ID, then a sequence
13//!   of self-describing objects until FIN.
14//!
15//! * **Datagram**: Type byte `0x00..=0x07` or `0x20..=0x21`
16//!   with bit-flags for End of Group, Extensions Present, Object ID
17//!   Present, and Status vs Payload.
18
19use bytes::{Buf, BufMut};
20
21use super::types::ObjectStatus;
22use crate::error::CodecError;
23use crate::object_properties::{ObjectProperty, PropertyValue};
24use crate::varint::VarInt;
25
26/// Advance `buf` past `len` bytes without copying them.
27fn skip(buf: &mut impl Buf, len: u64) -> Result<(), CodecError> {
28    let len = usize::try_from(len).map_err(|_| CodecError::UnexpectedEnd)?;
29    if buf.remaining() < len {
30        return Err(CodecError::UnexpectedEnd);
31    }
32    buf.advance(len);
33    Ok(())
34}
35
36/// Hold an object to the rule binding extension headers to Object Status.
37///
38/// Draft-14 Section 10.2.1.2: "Any Object may have extension headers except
39/// those with Object Status 'Object Does Not Exist'. If an endpoint receives a
40/// non-existent Object containing extension headers it MUST close the session
41/// with a PROTOCOL_VIOLATION."
42///
43/// The rule names one status and no others, so extensions beside End of Group
44/// or End of Track stay legal and are left alone here. The reasoning behind the
45/// exception is that an object nobody has cannot carry metadata about itself:
46/// a relay that forwards the extensions of a non-existent object is inventing
47/// provenance for something that was never published.
48///
49/// All three carriers reach this, in both directions — subgroup streams, fetch
50/// streams and status datagrams each pair a status with an extension block.
51///
52/// Reported under [`CodecError::ExtensionsOnNonExistentObject`], which is this
53/// rule and nothing else. [`CodecError::InvalidField`] is too coarse for it:
54/// shared with a dozen unrelated malformations the draft does not answer with a
55/// close, it leaves a caller unable to act on the sentence above.
56fn check_extensions_against_status(
57    status: Option<u64>,
58    extension_headers_len: u64,
59) -> Result<(), CodecError> {
60    if status == Some(ObjectStatus::ObjectDoesNotExist.as_u64()) && extension_headers_len != 0 {
61        // The length is reported as it appeared on the wire; saturating rather
62        // than truncating means a declared length above `usize::MAX` — which
63        // cannot have been read, but can have been declared — is never reported
64        // as some smaller, plausible number.
65        return Err(CodecError::ExtensionsOnNonExistentObject(
66            usize::try_from(extension_headers_len).unwrap_or(usize::MAX),
67        ));
68    }
69    Ok(())
70}
71
72// ============================================================
73// Subgroup stream (Type 0x10..=0x1D)
74// ============================================================
75
76/// Subgroup stream type byte.
77///
78/// The 12 defined types encode four independent boolean fields in the
79/// low nibble:
80///
81/// * bit 0 (`0x01`) — Extensions Present
82/// * bit 1 (`0x02`) — Subgroup ID derives from first Object ID
83///   (only meaningful when bit 2 is clear)
84/// * bit 2 (`0x04`) — Subgroup ID Field Present (explicit Subgroup ID varint)
85/// * bit 3 (`0x08`) — Contains End of Group
86#[derive(Debug, Clone, Copy, PartialEq, Eq)]
87pub struct SubgroupStreamType(u8);
88
89impl SubgroupStreamType {
90    /// The raw wire byte.
91    pub fn as_u8(self) -> u8 {
92        self.0
93    }
94
95    /// Create a [`SubgroupStreamType`] from its raw byte, validating
96    /// that it is one of the 12 defined values. `0x16` and `0x17` fall
97    /// inside the `0x10..=0x1D` range but are not defined, so a range
98    /// check alone is not enough.
99    pub fn from_u8(v: u8) -> Option<Self> {
100        if (0x10..=0x15).contains(&v) || (0x18..=0x1D).contains(&v) {
101            Some(SubgroupStreamType(v))
102        } else {
103            None
104        }
105    }
106
107    /// Build a subgroup stream type from its component flags.
108    ///
109    /// `subgroup_id_is_first_object` and `subgroup_id_field_present` are
110    /// mutually exclusive — if both are set, the resulting type has the
111    /// "Subgroup ID Field Present" bit set (bit 2 wins).
112    pub fn from_flags(
113        subgroup_id_field_present: bool,
114        subgroup_id_is_first_object: bool,
115        extensions_present: bool,
116        end_of_group: bool,
117    ) -> Self {
118        let mut v: u8 = 0x10;
119        if extensions_present {
120            v |= 0x01;
121        }
122        if subgroup_id_field_present {
123            v |= 0x04;
124        } else if subgroup_id_is_first_object {
125            v |= 0x02;
126        }
127        if end_of_group {
128            v |= 0x08;
129        }
130        SubgroupStreamType(v)
131    }
132
133    /// True if the header carries an explicit Subgroup ID varint.
134    pub fn has_subgroup_id_field(self) -> bool {
135        self.0 & 0x04 != 0
136    }
137
138    /// True if the subgroup ID is defined to equal the first Object ID
139    /// in the stream (applies only when [`Self::has_subgroup_id_field`]
140    /// is false).
141    pub fn subgroup_id_is_first_object(self) -> bool {
142        !self.has_subgroup_id_field() && (self.0 & 0x02 != 0)
143    }
144
145    /// True if every object in the stream carries extension headers.
146    pub fn extensions_present(self) -> bool {
147        self.0 & 0x01 != 0
148    }
149
150    /// True if the last object on the stream (prior to FIN) is the end
151    /// of its group.
152    pub fn contains_end_of_group(self) -> bool {
153        self.0 & 0x08 != 0
154    }
155}
156
157/// Subgroup stream header.
158///
159/// Wire order: type byte, Track Alias, Group ID, Subgroup ID (only for
160/// stream types that carry one), Publisher Priority.
161#[derive(Debug, Clone, PartialEq, Eq)]
162pub struct SubgroupHeader {
163    /// Type byte identifying the flag set for this stream.
164    pub stream_type: SubgroupStreamType,
165    /// Track alias.
166    pub track_alias: VarInt,
167    /// Group ID.
168    pub group_id: VarInt,
169    /// Explicit Subgroup ID — present only when the stream type sets
170    /// `Subgroup ID Field Present = Yes`. For types where the subgroup
171    /// ID is implicit (0 or the first Object ID), the effective
172    /// subgroup ID is resolved on the receive side by the reader.
173    pub subgroup_id: Option<VarInt>,
174    /// Publisher priority.
175    pub publisher_priority: u8,
176}
177
178impl SubgroupHeader {
179    /// Encode the header including the leading stream type byte.
180    ///
181    /// Driven by `stream_type`, and silent about a `subgroup_id` that
182    /// disagrees with it in either direction: a `None` under a type whose
183    /// Subgroup ID Field Present column reads Yes is written as zero, which
184    /// names a different subgroup rather than no subgroup, and a `Some` under
185    /// a type whose column reads No is dropped. Both write a well-formed
186    /// stream describing something other than what the caller built.
187    /// [`Self::encode_checked`] refuses that shape instead.
188    pub fn encode(&self, buf: &mut impl BufMut) {
189        VarInt::from_u64(self.stream_type.as_u8() as u64).unwrap().encode(buf);
190        self.track_alias.encode(buf);
191        self.group_id.encode(buf);
192        if self.stream_type.has_subgroup_id_field() {
193            let sg = self.subgroup_id.unwrap_or_else(|| VarInt::from_u64(0).unwrap());
194            sg.encode(buf);
195        }
196        buf.put_u8(self.publisher_priority);
197    }
198
199    /// Encode, refusing a header whose Subgroup ID disagrees with its own type
200    /// byte.
201    ///
202    /// Section 10.4.2 gives the SUBGROUP_HEADER type table a Subgroup ID Field
203    /// Present column, and that column - not the value in hand - decides
204    /// whether the field is on the wire. [`Self::decode`] therefore only ever
205    /// produces a header where the two agree, so this refuses exactly the
206    /// shapes a caller assembled by hand: an absent ID under a type that
207    /// carries one, and a present ID under a type that does not.
208    ///
209    /// Nothing about the *value* is refused. Zero is a legal Subgroup ID, so a
210    /// `Some(0)` under a type that carries the field is written as the caller
211    /// asked.
212    ///
213    /// # Errors
214    ///
215    /// [`CodecError::InvalidField`] if the presence of `subgroup_id` does not
216    /// match what `stream_type` puts on the wire.
217    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
218        if self.stream_type.has_subgroup_id_field() != self.subgroup_id.is_some() {
219            return Err(CodecError::InvalidField);
220        }
221        self.encode(buf);
222        Ok(())
223    }
224
225    /// Decode a subgroup header (leading type byte + remaining fields).
226    ///
227    /// Errors with [`CodecError::UnknownStreamType`] when the stream table does
228    /// not assign the leading type, which this draft answers with a close, and
229    /// with [`CodecError::InvalidField`] for `0x05`, which it assigns to a
230    /// fetch stream. `stream_type_error` draws that line.
231    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
232        let type_val = VarInt::decode(buf)?.into_inner();
233        if type_val > 0xFF {
234            return Err(stream_type_error(type_val));
235        }
236        let stream_type = SubgroupStreamType::from_u8(type_val as u8)
237            .ok_or_else(|| stream_type_error(type_val))?;
238        let track_alias = VarInt::decode(buf)?;
239        let group_id = VarInt::decode(buf)?;
240        let subgroup_id =
241            if stream_type.has_subgroup_id_field() { Some(VarInt::decode(buf)?) } else { None };
242        if buf.remaining() < 1 {
243            return Err(CodecError::UnexpectedEnd);
244        }
245        let publisher_priority = buf.get_u8();
246        Ok(SubgroupHeader { stream_type, track_alias, group_id, subgroup_id, publisher_priority })
247    }
248}
249
250/// One object within a subgroup stream, with the Object ID already
251/// resolved from its delta encoding.
252#[derive(Debug, Clone, PartialEq, Eq)]
253pub struct SubgroupObject {
254    /// Resolved Object ID (delta decoded to an absolute value).
255    pub object_id: VarInt,
256    /// Raw extension-header bytes. Empty when the stream type has
257    /// `Extensions Present = No`, or when present but the length was 0.
258    /// The content is a sequence of Key-Value-Pairs but is
259    /// left opaque here — relays and subscribers that do not understand
260    /// specific extensions must forward or ignore the bytes unchanged.
261    pub extension_headers: Vec<u8>,
262    /// Object Status when `payload.is_empty()` and the object was sent
263    /// with an explicit status code; `None` when a non-empty payload
264    /// follows (status is implicitly [`ObjectStatus::Normal`]).
265    pub status: Option<ObjectStatus>,
266    /// Object payload. Empty when `status` is `Some(..)`.
267    pub payload: Vec<u8>,
268}
269
270/// The framing of one subgroup object, without its payload.
271///
272/// Produced by [`SubgroupObjectReader::read_object_meta`] for callers that
273/// forward an object's bytes verbatim and never inspect the payload.
274#[derive(Debug, Clone, Copy, PartialEq, Eq)]
275pub struct SubgroupObjectMeta {
276    /// Resolved Object ID (delta decoded to an absolute value).
277    pub object_id: u64,
278    /// Byte length of the extension-header block's contents, excluding its
279    /// length prefix.
280    pub extension_headers_len: u64,
281    /// Declared payload length. Zero when `status` is `Some`.
282    pub payload_length: u64,
283    /// Object Status wire code, present only when the payload is empty.
284    pub status: Option<u64>,
285    /// Total bytes this object occupies on the wire, prefix fields included.
286    pub wire_len: u64,
287}
288
289/// Whether an object carrying a given status may hold a non-empty payload.
290///
291/// Section 10.2.1.1 states the rule in one sentence — "Any object with a
292/// status code other than zero MUST have an empty payload" — so on this draft
293/// the answer falls out of the code being zero or not, and every status but
294/// [`ObjectStatus::Normal`] forbids a payload.
295///
296/// It is worth a type all the same, because that arithmetic is not something a
297/// consumer can safely perform on a raw wire code. A code this draft does not
298/// assign is not *non-zero, and therefore forbidden*: it is a code with no
299/// meaning at all, and no payload rule attaches to it. Handing back a
300/// `PayloadPermission` keeps the two apart, and lets a caller ask the question
301/// without restating the rule — or, worse, restating it slightly differently.
302#[derive(Debug, Clone, Copy, PartialEq, Eq)]
303pub enum PayloadPermission {
304    /// The status permits a payload but does not require one: a zero-length
305    /// object with such a status is well formed, and this draft's encodings
306    /// have a way to spell it.
307    Permitted,
308    /// An object with such a status has an empty payload, and one carrying
309    /// bytes is malformed.
310    Forbidden,
311}
312
313impl PayloadPermission {
314    /// The permission this draft gives objects carrying `status`.
315    ///
316    /// Written as a match over every assigned status rather than as a test for
317    /// zero, so that a status added to [`ObjectStatus`] later cannot quietly
318    /// inherit *not Normal, therefore forbidden* — it stops the crate
319    /// compiling until its own answer is written down.
320    pub fn for_status(status: ObjectStatus) -> Self {
321        match status {
322            ObjectStatus::Normal => PayloadPermission::Permitted,
323            ObjectStatus::ObjectDoesNotExist => PayloadPermission::Forbidden,
324            ObjectStatus::EndOfGroup => PayloadPermission::Forbidden,
325            ObjectStatus::EndOfTrack => PayloadPermission::Forbidden,
326        }
327    }
328
329    /// `true` for [`PayloadPermission::Permitted`].
330    ///
331    /// The permission answers on its own, with no payload length in hand,
332    /// which is the point of asking the status rather than the framing.
333    pub fn permits(self) -> bool {
334        matches!(self, PayloadPermission::Permitted)
335    }
336}
337
338impl SubgroupObject {
339    /// The status this object resolves to.
340    ///
341    /// The wire carries a status field only on an empty object, so an object
342    /// holding bytes is [`ObjectStatus::Normal`] whatever `status` says.
343    pub fn status(&self) -> ObjectStatus {
344        if self.payload.is_empty() {
345            self.status.unwrap_or(ObjectStatus::Normal)
346        } else {
347            ObjectStatus::Normal
348        }
349    }
350
351    /// Whether this object's status permits it a non-empty payload.
352    ///
353    /// Answered from the status alone. `payload` is not consulted: on a status
354    /// that permits a payload it says only whether this particular object took
355    /// the offer, and on one that forbids a payload a non-empty payload is the
356    /// malformation this reports, not evidence about the rule.
357    pub fn permits_payload(&self) -> bool {
358        PayloadPermission::for_status(self.status()).permits()
359    }
360}
361
362impl SubgroupObjectMeta {
363    /// Whether this object's status permits it a non-empty payload.
364    ///
365    /// The same question [`SubgroupObject::permits_payload`] answers, from the
366    /// declared length and the wire code rather than from the bytes. A relay
367    /// that forwards an object verbatim reads it through
368    /// [`SubgroupObjectReader::read_object_meta`] and never copies the payload,
369    /// so asking this must not require having it.
370    ///
371    /// An absent status answers [`PayloadPermission::Permitted`] rather than
372    /// `None`. A meta has no status only when its payload length is non-zero,
373    /// so the object has a status — the encoding just does not spell it.
374    ///
375    /// `None` means the code is one this draft leaves unassigned, and so one it
376    /// gives no payload rule for. That is not reachable through
377    /// [`SubgroupObjectReader::read_object_meta`], which refuses such a code
378    /// before it can reach the field, but the fields here are public and a meta
379    /// assembled by hand — by a relay carrying a status across from a draft
380    /// that numbers them differently, say — can hold anything a varint can. The
381    /// answer there is that this draft has none, not that the payload is
382    /// forbidden.
383    pub fn payload_permission(&self) -> Option<PayloadPermission> {
384        match self.status {
385            None => Some(PayloadPermission::Permitted),
386            Some(code) => ObjectStatus::from_u64(code).map(PayloadPermission::for_status),
387        }
388    }
389}
390
391/// Prior Group ID Gap, Extension Header Type 0x3C (Section 11.1).
392pub const PRIOR_GROUP_ID_GAP: u64 = 0x3C;
393/// Prior Object ID Gap, Extension Header Type 0x3E (Section 11.3).
394pub const PRIOR_OBJECT_ID_GAP: u64 = 0x3E;
395
396/// The Prior Group ID Gap and Prior Object ID Gap one Object carries.
397#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
398pub struct PriorGaps {
399    /// Prior Group ID Gap, if present.
400    pub group: Option<u64>,
401    /// Prior Object ID Gap, if present.
402    pub object: Option<u64>,
403}
404
405impl PriorGaps {
406    /// Read the two gaps from an Object's parsed extension headers, refusing
407    /// the conditions one Object shows on its own.
408    ///
409    /// Section 11.1 makes the Track malformed when "An Object contains more
410    /// than one instance of Prior Group ID Gap" and when "An Object has a Prior
411    /// Group ID Gap larger than the Group ID". Section 11.3 says the same two
412    /// for Prior Object ID Gap against the Object ID. Each is
413    /// [`CodecError::MalformedTrack`].
414    pub fn from_properties(
415        properties: &[ObjectProperty<'_>],
416        group_id: u64,
417        object_id: u64,
418    ) -> Result<Self, CodecError> {
419        let mut gaps = Self::default();
420        collect_gaps(properties, &mut gaps)?;
421        if let Some(gap) = gaps.group.filter(|&gap| gap > group_id) {
422            return Err(CodecError::MalformedTrack {
423                detail: format!("Prior Group ID Gap {gap} is larger than Group ID {group_id}"),
424            });
425        }
426        if let Some(gap) = gaps.object.filter(|&gap| gap > object_id) {
427            return Err(CodecError::MalformedTrack {
428                detail: format!("Prior Object ID Gap {gap} is larger than Object ID {object_id}"),
429            });
430        }
431        Ok(gaps)
432    }
433}
434
435fn collect_gaps(properties: &[ObjectProperty<'_>], gaps: &mut PriorGaps) -> Result<(), CodecError> {
436    for property in properties {
437        let slot = match (&property.value, property.property_type) {
438            (PropertyValue::Immutable { properties, .. }, _) => {
439                collect_gaps(properties, gaps)?;
440                continue;
441            }
442            (PropertyValue::Varint(value), PRIOR_GROUP_ID_GAP) => {
443                (&mut gaps.group, *value, "Group")
444            }
445            (PropertyValue::Varint(value), PRIOR_OBJECT_ID_GAP) => {
446                (&mut gaps.object, *value, "Object")
447            }
448            _ => continue,
449        };
450        let (field, value, which) = slot;
451        if field.replace(value).is_some() {
452            return Err(CodecError::MalformedTrack {
453                detail: format!("an Object contains more than one Prior {which} ID Gap"),
454            });
455        }
456    }
457    Ok(())
458}
459
460/// The Prior Group ID Gap and Prior Object ID Gap conditions of
461/// draft-14 Section 11.1 and draft-14 Section 11.3 that one fetch stream
462/// shows.
463///
464/// [`PriorGaps::from_properties`] refuses what one Object shows. Across the
465/// stream this refuses "A Group contains more than one Object with different
466/// values for Prior Group ID Gap" for the Objects of one Group, "An endpoint
467/// receives an Object with a Prior Group ID Gap covering an Object it
468/// previously received", "An endpoint receives an Object with a Group ID
469/// within a previously communicated gap", and the same two for Prior Object ID
470/// Gap, for the Group and the Object before on the stream.
471/// draft-14 Section 9.16.3: "Within each group, objects are sent in Object ID
472/// order; subgroup ID is not used for ordering."
473///
474/// [`FetchObject::decode`] reads one Object and applies none of this: a reader
475/// of a fetch stream checks each Object with one `FetchGaps`.
476#[derive(Debug, Clone, PartialEq, Eq)]
477// Boxed so that the fetch readers holding one stay small.
478pub struct FetchGaps(Box<crate::prior_gaps::FetchGapTracker>);
479
480impl Default for FetchGaps {
481    fn default() -> Self {
482        Self::new()
483    }
484}
485
486impl FetchGaps {
487    /// Track the Objects of one fetch stream.
488    pub fn new() -> Self {
489        Self(Box::new(crate::prior_gaps::FetchGapTracker::new(
490            crate::prior_gaps::GapRules::Draft14To15,
491        )))
492    }
493
494    /// Check the next Object on the stream, from its resolved Group ID and
495    /// Object ID and its extension block, refusing a gap condition with
496    /// [`CodecError::MalformedTrack`] and a block that does not parse as
497    /// [`crate::object_properties::parse_object_properties`] does. Records the
498    /// Object only when it passes.
499    pub fn check(
500        &mut self,
501        group_id: u64,
502        object_id: u64,
503        extension_block: &[u8],
504    ) -> Result<PriorGaps, CodecError> {
505        let properties = crate::object_properties::parse_object_properties(
506            crate::version::DraftVersion::Draft14,
507            extension_block,
508        )?;
509        let gaps = PriorGaps::from_properties(&properties, group_id, object_id)?;
510        self.0.check(group_id, object_id, gaps.group.as_slice(), gaps.object)?;
511        Ok(gaps)
512    }
513}
514
515/// Stateful reader for the object fields on a subgroup stream.
516///
517/// Object IDs on a subgroup stream are delta-encoded against the
518/// previous Object ID, and whether extension headers are present is
519/// fixed by the enclosing [`SubgroupHeader`]'s stream type. This reader
520/// carries that context across successive `read_object` calls.
521#[derive(Debug, Clone)]
522pub struct SubgroupObjectReader {
523    extensions_present: bool,
524    prev_object_id: Option<u64>,
525    group_id: u64,
526    group_gap: Option<u64>,
527}
528
529impl SubgroupObjectReader {
530    /// Create a reader from a parsed subgroup header.
531    pub fn new(header: &SubgroupHeader) -> Self {
532        Self {
533            extensions_present: header.stream_type.extensions_present(),
534            prev_object_id: None,
535            group_id: header.group_id.into_inner(),
536            group_gap: None,
537        }
538    }
539
540    /// Decode the next object from `buf`. Caller is responsible for
541    /// ensuring the buffer contains a complete object (draft-14 objects
542    /// are length-delimited by the payload-length field, so the buffer
543    /// boundary is known once the header portion has been consumed).
544    ///
545    /// Refuses, with [`CodecError::MalformedTrack`], the Prior Group ID Gap
546    /// and Prior Object ID Gap conditions of Section 11.1 and Section 11.3
547    /// this stream shows: those [`PriorGaps::from_properties`] refuses, "A
548    /// Group contains more than one Object with different values for Prior
549    /// Group ID Gap", and "An endpoint receives an Object with a Prior Object
550    /// ID Gap covering an Object it previously received" for an Object earlier
551    /// on this stream. [`Self::read_object_meta`] does not read extension
552    /// headers and applies none of them.
553    pub fn read_object(&mut self, buf: &mut impl Buf) -> Result<SubgroupObject, CodecError> {
554        let delta = VarInt::decode(buf)?.into_inner();
555        let prior_object_id = self.prev_object_id;
556        let object_id_val = match self.prev_object_id {
557            None => delta,
558            Some(prev) => prev
559                .checked_add(1)
560                .and_then(|v| v.checked_add(delta))
561                .ok_or(CodecError::InvalidField)?,
562        };
563        self.prev_object_id = Some(object_id_val);
564        let object_id = VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?;
565
566        let extension_headers = if self.extensions_present {
567            let ext_len = VarInt::decode(buf)?.into_inner() as usize;
568            crate::types::read_bytes(buf, ext_len)?
569        } else {
570            Vec::new()
571        };
572
573        let payload_length = VarInt::decode(buf)?.into_inner() as usize;
574        let (status, payload) = if payload_length == 0 {
575            let status_val = VarInt::decode(buf)?.into_inner();
576            let status = ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?;
577            (Some(status), Vec::new())
578        } else {
579            let payload = crate::types::read_bytes(buf, payload_length)?;
580            (None, payload)
581        };
582        check_extensions_against_status(
583            status.map(|s| s.as_u64()),
584            extension_headers.len() as u64,
585        )?;
586
587        let properties = crate::object_properties::parse_object_properties(
588            crate::version::DraftVersion::Draft14,
589            &extension_headers,
590        )?;
591        let gaps = PriorGaps::from_properties(&properties, self.group_id, object_id_val)?;
592        self.check_stream_gaps(gaps, prior_object_id, object_id_val)?;
593        Ok(SubgroupObject { object_id, extension_headers, status, payload })
594    }
595
596    fn check_stream_gaps(
597        &mut self,
598        gaps: PriorGaps,
599        prior_object_id: Option<u64>,
600        object_id: u64,
601    ) -> Result<(), CodecError> {
602        if let Some(gap) = gaps.group {
603            match self.group_gap {
604                Some(seen) if seen != gap => {
605                    return Err(CodecError::MalformedTrack {
606                        detail: format!(
607                            "two Objects in Group {} carry different Prior Group ID Gap values",
608                            self.group_id
609                        ),
610                    });
611                }
612                _ => self.group_gap = Some(gap),
613            }
614        }
615        // Object IDs rise along a subgroup stream, so the latest earlier
616        // Object is the only one a gap below this Object can reach.
617        if let (Some(gap), Some(prior)) = (gaps.object, prior_object_id) {
618            if gap > 0 && prior >= object_id - gap {
619                return Err(CodecError::MalformedTrack {
620                    detail: format!(
621                        "Prior Object ID Gap {gap} of Object {object_id} covers Object {prior}"
622                    ),
623                });
624            }
625        }
626        Ok(())
627    }
628
629    /// Decode the next object's framing without copying its payload.
630    ///
631    /// Consumes exactly the bytes [`Self::read_object`] consumes and leaves
632    /// the same delta state behind, so the two are interchangeable on a
633    /// given stream.
634    pub fn read_object_meta(
635        &mut self,
636        buf: &mut impl Buf,
637    ) -> Result<SubgroupObjectMeta, CodecError> {
638        let start = buf.remaining();
639        let delta = VarInt::decode(buf)?.into_inner();
640        let object_id_val = match self.prev_object_id {
641            None => delta,
642            Some(prev) => prev
643                .checked_add(1)
644                .and_then(|v| v.checked_add(delta))
645                .ok_or(CodecError::InvalidField)?,
646        };
647        self.prev_object_id = Some(object_id_val);
648        let object_id =
649            VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?.into_inner();
650
651        let extension_headers_len = if self.extensions_present {
652            let ext_len = VarInt::decode(buf)?.into_inner();
653            skip(buf, ext_len)?;
654            ext_len
655        } else {
656            0
657        };
658
659        let payload_length = VarInt::decode(buf)?.into_inner();
660        let status = if payload_length == 0 {
661            let status_val = VarInt::decode(buf)?.into_inner();
662            Some(ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?.as_u64())
663        } else {
664            skip(buf, payload_length)?;
665            None
666        };
667        check_extensions_against_status(status, extension_headers_len)?;
668
669        Ok(SubgroupObjectMeta {
670            object_id,
671            extension_headers_len,
672            payload_length,
673            status,
674            wire_len: (start - buf.remaining()) as u64,
675        })
676    }
677
678    /// Serialize a subgroup object using the reader's delta state. Intended
679    /// for senders that want to build a stream incrementally — tracks
680    /// `prev_object_id` so successive calls produce correct deltas.
681    ///
682    /// Returns an error if `object.object_id <= prev_object_id`, which
683    /// would produce an invalid delta.
684    ///
685    /// It also refuses the three shapes the stream cannot carry, rather than
686    /// writing whichever half fits and dropping the rest:
687    ///
688    /// * A non-empty payload beside a status other than Normal. Section
689    ///   10.2.1.1: "Any object with a status code other than zero MUST have an
690    ///   empty payload." A truncated payload is worse than a refusal — the
691    ///   receiver has no way to tell that anything was there.
692    /// * Extension headers on an object whose stream type says the subgroup has
693    ///   none. The type byte is fixed for the whole stream by the header, so
694    ///   this object cannot opt in, and its extensions would simply vanish.
695    /// * Extension headers on an Object Does Not Exist status, per Section
696    ///   10.2.1.2.
697    ///
698    /// Normal beside a non-empty payload is not one of those and is written as
699    /// an ordinary payload-bearing object: it is the status such an object
700    /// already has, so naming it asks for the bytes leaving it out asks for.
701    /// Normal beside an empty payload keeps the explicit status form, which is
702    /// the only way to send a zero-length object at all.
703    pub fn write_object(
704        &mut self,
705        object: &SubgroupObject,
706        buf: &mut impl BufMut,
707    ) -> Result<(), CodecError> {
708        let explicit_status = match object.status {
709            Some(status) if status != ObjectStatus::Normal => {
710                if !object.payload.is_empty() {
711                    return Err(CodecError::InvalidField);
712                }
713                Some(status)
714            }
715            Some(ObjectStatus::Normal) if object.payload.is_empty() => Some(ObjectStatus::Normal),
716            _ => None,
717        };
718        if !self.extensions_present && !object.extension_headers.is_empty() {
719            return Err(CodecError::InvalidField);
720        }
721        check_extensions_against_status(
722            explicit_status.map(|s| s.as_u64()),
723            object.extension_headers.len() as u64,
724        )?;
725
726        let oid = object.object_id.into_inner();
727        let delta = match self.prev_object_id {
728            None => oid,
729            Some(prev) => oid
730                .checked_sub(prev)
731                .and_then(|v| v.checked_sub(1))
732                .ok_or(CodecError::InvalidField)?,
733        };
734        VarInt::from_u64(delta).map_err(|_| CodecError::InvalidField)?.encode(buf);
735        if self.extensions_present {
736            VarInt::from_u64(object.extension_headers.len() as u64)
737                .map_err(|_| CodecError::InvalidField)?
738                .encode(buf);
739            buf.put_slice(&object.extension_headers);
740        }
741        if let Some(status) = explicit_status {
742            VarInt::from_u64(0).unwrap().encode(buf);
743            VarInt::from_u64(status.as_u64()).unwrap().encode(buf);
744        } else {
745            VarInt::from_u64(object.payload.len() as u64)
746                .map_err(|_| CodecError::InvalidField)?
747                .encode(buf);
748            buf.put_slice(&object.payload);
749        }
750        self.prev_object_id = Some(oid);
751        Ok(())
752    }
753}
754
755// ============================================================
756// Fetch stream (Type 0x05)
757// ============================================================
758
759/// Which failure a leading unidirectional stream type that is not the one a
760/// reader wants is.
761///
762/// Section 10: "An endpoint that receives an unknown stream or datagram type
763/// MUST close the session." One sentence, two tables. The stream table assigns
764/// 0x05 for FETCH_HEADER and the range 0x10 to 0x1D for SUBGROUP_HEADER;
765/// everything outside them is unknown at the head of a stream, and the session
766/// ends.
767///
768/// The two assigned kinds are what the [`CodecError::InvalidField`] arm is for:
769/// a fetch stream reaching the subgroup reader, or a subgroup stream reaching
770/// the fetch reader, is a value this draft defines, and the disagreement is
771/// with the reader that was called rather than with the draft. Reporting it as
772/// unknown would end sessions over streams draft-14 permits.
773///
774/// Values above 0xFF land here too. None of them is assigned — this draft
775/// carries its control messages on a bidirectional stream and so has no
776/// multi-byte unidirectional type the way drafts 17 and later do.
777fn stream_type_error(raw: u64) -> CodecError {
778    let assigned = raw == FETCH_STREAM_TYPE as u64
779        || (raw <= 0xFF && SubgroupStreamType::from_u8(raw as u8).is_some());
780    if assigned {
781        CodecError::InvalidField
782    } else {
783        CodecError::UnknownStreamType(raw)
784    }
785}
786
787/// Draft-14 fetch stream type byte.
788pub const FETCH_STREAM_TYPE: u8 = 0x05;
789
790/// Fetch stream header: the type byte `0x05` followed by the Request ID.
791#[derive(Debug, Clone, PartialEq, Eq)]
792pub struct FetchHeader {
793    /// Request ID from the originating FETCH control message.
794    pub request_id: VarInt,
795}
796
797impl FetchHeader {
798    /// Encode the header including the leading type byte.
799    pub fn encode(&self, buf: &mut impl BufMut) {
800        VarInt::from_u64(FETCH_STREAM_TYPE as u64).unwrap().encode(buf);
801        self.request_id.encode(buf);
802    }
803
804    /// Decode the header.
805    ///
806    /// Errors with [`CodecError::UnknownStreamType`] when the stream table does
807    /// not assign the leading type, which this draft answers with a close, and
808    /// with [`CodecError::InvalidField`] for the subgroup types, which it does
809    /// assign. `stream_type_error` draws that line.
810    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
811        let type_val = VarInt::decode(buf)?.into_inner();
812        if type_val != FETCH_STREAM_TYPE as u64 {
813            return Err(stream_type_error(type_val));
814        }
815        let request_id = VarInt::decode(buf)?;
816        Ok(FetchHeader { request_id })
817    }
818}
819
820/// One object carried on a fetch stream.
821///
822/// Every object on a fetch stream is self-describing — unlike subgroup
823/// streams, there is no delta encoding and extension headers are always
824/// length-prefixed (the length is zero when absent).
825#[derive(Debug, Clone, PartialEq, Eq)]
826pub struct FetchObject {
827    /// Group ID.
828    pub group_id: VarInt,
829    /// Subgroup ID. For objects whose Forwarding Preference is Datagram,
830    /// this is set to the Object ID.
831    pub subgroup_id: VarInt,
832    /// Object ID.
833    pub object_id: VarInt,
834    /// Publisher priority.
835    pub publisher_priority: u8,
836    /// Raw extension-header bytes (opaque sequence of Key-Value-Pairs).
837    pub extension_headers: Vec<u8>,
838    /// Object status when `payload.is_empty()`, otherwise `None`.
839    pub status: Option<ObjectStatus>,
840    /// Object payload.
841    pub payload: Vec<u8>,
842}
843
844/// The framing of one fetch object, without its payload.
845///
846/// Produced by [`FetchObject::decode_meta`] for callers that forward an
847/// object's bytes verbatim and never inspect the payload.
848#[derive(Debug, Clone, Copy, PartialEq, Eq)]
849pub struct FetchObjectMeta {
850    /// Group ID.
851    pub group_id: u64,
852    /// Subgroup ID.
853    pub subgroup_id: u64,
854    /// Object ID.
855    pub object_id: u64,
856    /// Publisher priority.
857    pub publisher_priority: u8,
858    /// Byte length of the extension-header block's contents, excluding its
859    /// length prefix.
860    pub extension_headers_len: u64,
861    /// Declared payload length. Zero when `status` is `Some`.
862    pub payload_length: u64,
863    /// Object Status wire code, present only when the payload is empty.
864    pub status: Option<u64>,
865    /// Total bytes this object occupies on the wire, prefix fields included.
866    pub wire_len: u64,
867}
868
869impl FetchObjectMeta {
870    /// Whether this object's status permits it a non-empty payload.
871    ///
872    /// The fetch-stream twin of [`SubgroupObjectMeta::payload_permission`], and
873    /// it answers on the same terms: `None` for a code this draft does not
874    /// assign, and [`PayloadPermission::Permitted`] for an absent status, which
875    /// a meta has only when its payload length is non-zero.
876    pub fn payload_permission(&self) -> Option<PayloadPermission> {
877        match self.status {
878            None => Some(PayloadPermission::Permitted),
879            Some(code) => ObjectStatus::from_u64(code).map(PayloadPermission::for_status),
880        }
881    }
882}
883
884impl FetchObject {
885    /// The status this object resolves to.
886    ///
887    /// The wire carries a status field only on an empty object, so an object
888    /// holding bytes is [`ObjectStatus::Normal`] whatever `status` says.
889    pub fn status(&self) -> ObjectStatus {
890        if self.payload.is_empty() {
891            self.status.unwrap_or(ObjectStatus::Normal)
892        } else {
893            ObjectStatus::Normal
894        }
895    }
896
897    /// Whether this object's status permits it a non-empty payload.
898    ///
899    /// Answered from the status alone, for the reason
900    /// [`SubgroupObject::permits_payload`] gives. Note that
901    /// [`Self::encode_checked`] refuses the pairing this reports on, so a
902    /// `false` here is a value that will not be written rather than one
903    /// already on the wire.
904    pub fn permits_payload(&self) -> bool {
905        PayloadPermission::for_status(self.status()).permits()
906    }
907
908    /// Encode one fetch object, refusing a value the wire shape cannot carry.
909    ///
910    /// A fetch object holds a status and a payload in the same value, and the
911    /// wire form has room for only one: the Object Status field is written only
912    /// when Object Payload Length is zero. [`Self::encode`] settles that by
913    /// following the status and dropping the payload, which loses the payload
914    /// without saying so. This refuses instead, before any byte is written, so
915    /// a rejected object leaves `buf` untouched.
916    ///
917    /// Section 10.2.1.1: "Any object with a status code other than zero MUST
918    /// have an empty payload." Normal beside a payload is therefore not a
919    /// disagreement — it is the status a payload-bearing object already has —
920    /// and it is written as an ordinary payload-bearing object.
921    ///
922    /// Extension headers on an Object Does Not Exist status are refused for the
923    /// separate reason given in Section 10.2.1.2.
924    ///
925    /// Unlike a subgroup stream, a fetch object always carries its Extension
926    /// Headers Length field, so a length of zero is an ordinary object with no
927    /// extensions and is written as such.
928    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
929        if matches!(self.status, Some(status) if status != ObjectStatus::Normal)
930            && !self.payload.is_empty()
931        {
932            return Err(CodecError::InvalidField);
933        }
934        check_extensions_against_status(
935            self.status.map(|s| s.as_u64()),
936            self.extension_headers.len() as u64,
937        )?;
938        self.encode(buf);
939        Ok(())
940    }
941
942    /// Encode one fetch object.
943    ///
944    /// Infallible because the status is taken as the authority on framing, and
945    /// lossy for the same reason: a payload set beside a status is discarded
946    /// here without a word. Prefer [`Self::encode_checked`], which refuses that
947    /// combination rather than resolving it.
948    pub fn encode(&self, buf: &mut impl BufMut) {
949        self.group_id.encode(buf);
950        self.subgroup_id.encode(buf);
951        self.object_id.encode(buf);
952        buf.put_u8(self.publisher_priority);
953        VarInt::from_u64(self.extension_headers.len() as u64).unwrap().encode(buf);
954        buf.put_slice(&self.extension_headers);
955        if let Some(status) = self.status {
956            VarInt::from_u64(0).unwrap().encode(buf);
957            VarInt::from_u64(status.as_u64()).unwrap().encode(buf);
958        } else {
959            VarInt::from_u64(self.payload.len() as u64).unwrap().encode(buf);
960            buf.put_slice(&self.payload);
961        }
962    }
963
964    /// Decode one fetch object.
965    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
966        let group_id = VarInt::decode(buf)?;
967        let subgroup_id = VarInt::decode(buf)?;
968        let object_id = VarInt::decode(buf)?;
969        if buf.remaining() < 1 {
970            return Err(CodecError::UnexpectedEnd);
971        }
972        let publisher_priority = buf.get_u8();
973        let ext_len = VarInt::decode(buf)?.into_inner() as usize;
974        let extension_headers = crate::types::read_bytes(buf, ext_len)?;
975        let payload_length = VarInt::decode(buf)?.into_inner() as usize;
976        let (status, payload) = if payload_length == 0 {
977            let status_val = VarInt::decode(buf)?.into_inner();
978            let status = ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?;
979            (Some(status), Vec::new())
980        } else {
981            (None, crate::types::read_bytes(buf, payload_length)?)
982        };
983        check_extensions_against_status(
984            status.map(|s| s.as_u64()),
985            extension_headers.len() as u64,
986        )?;
987        Ok(FetchObject {
988            group_id,
989            subgroup_id,
990            object_id,
991            publisher_priority,
992            extension_headers,
993            status,
994            payload,
995        })
996    }
997
998    /// Decode one fetch object's framing without copying its payload.
999    ///
1000    /// Consumes exactly the bytes [`Self::decode`] consumes.
1001    pub fn decode_meta(buf: &mut impl Buf) -> Result<FetchObjectMeta, CodecError> {
1002        let start = buf.remaining();
1003        let group_id = VarInt::decode(buf)?.into_inner();
1004        let subgroup_id = VarInt::decode(buf)?.into_inner();
1005        let object_id = VarInt::decode(buf)?.into_inner();
1006        if buf.remaining() < 1 {
1007            return Err(CodecError::UnexpectedEnd);
1008        }
1009        let publisher_priority = buf.get_u8();
1010        let extension_headers_len = VarInt::decode(buf)?.into_inner();
1011        skip(buf, extension_headers_len)?;
1012        let payload_length = VarInt::decode(buf)?.into_inner();
1013        let status = if payload_length == 0 {
1014            let status_val = VarInt::decode(buf)?.into_inner();
1015            Some(ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?.as_u64())
1016        } else {
1017            skip(buf, payload_length)?;
1018            None
1019        };
1020        check_extensions_against_status(status, extension_headers_len)?;
1021        Ok(FetchObjectMeta {
1022            group_id,
1023            subgroup_id,
1024            object_id,
1025            publisher_priority,
1026            extension_headers_len,
1027            payload_length,
1028            status,
1029            wire_len: (start - buf.remaining()) as u64,
1030        })
1031    }
1032}
1033
1034// ============================================================
1035// Datagram (Type 0x00..=0x07, 0x20..=0x21)
1036// ============================================================
1037
1038/// Datagram type byte.
1039///
1040/// Bit layout (low nibble):
1041///
1042/// * bit 0 (`0x01`) — Extensions Present
1043/// * bit 1 (`0x02`) — End of Group
1044/// * bit 2 (`0x04`) — Object ID **absent** (when set, Object ID = 0)
1045///
1046/// Status variants use the high nibble (`0x20..=0x21`). Only types
1047/// `0x00..=0x07`, `0x20`, `0x21` are defined.
1048#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1049pub struct DatagramType(u8);
1050
1051impl DatagramType {
1052    /// Raw wire byte.
1053    pub fn as_u8(self) -> u8 {
1054        self.0
1055    }
1056
1057    /// Validate and wrap a raw wire byte.
1058    pub fn from_u8(v: u8) -> Option<Self> {
1059        if (0x00..=0x07).contains(&v) || v == 0x20 || v == 0x21 {
1060            Some(DatagramType(v))
1061        } else {
1062            None
1063        }
1064    }
1065
1066    /// Build a payload-bearing datagram type (`0x00..=0x07`).
1067    pub fn payload(object_id_present: bool, extensions_present: bool, end_of_group: bool) -> Self {
1068        let mut v: u8 = 0x00;
1069        if extensions_present {
1070            v |= 0x01;
1071        }
1072        if end_of_group {
1073            v |= 0x02;
1074        }
1075        if !object_id_present {
1076            v |= 0x04;
1077        }
1078        DatagramType(v)
1079    }
1080
1081    /// Build a status-only datagram type (`0x20` or `0x21`).
1082    pub fn status(extensions_present: bool) -> Self {
1083        if extensions_present {
1084            DatagramType(0x21)
1085        } else {
1086            DatagramType(0x20)
1087        }
1088    }
1089
1090    /// True when the datagram carries an Object Status instead of a
1091    /// payload (types `0x20` / `0x21`).
1092    pub fn is_status(self) -> bool {
1093        self.0 >= 0x20
1094    }
1095
1096    /// True when the datagram carries an explicit Object ID field.
1097    pub fn object_id_present(self) -> bool {
1098        // Bit 2 is only meaningful in the 0x00..=0x07 range; status
1099        // variants (0x20/0x21) always carry an Object ID.
1100        if self.is_status() {
1101            true
1102        } else {
1103            self.0 & 0x04 == 0
1104        }
1105    }
1106
1107    /// True if the last object of the group is conveyed.
1108    pub fn end_of_group(self) -> bool {
1109        !self.is_status() && (self.0 & 0x02 != 0)
1110    }
1111
1112    /// True if extension headers are present in this datagram.
1113    pub fn extensions_present(self) -> bool {
1114        self.0 & 0x01 != 0
1115    }
1116}
1117
1118/// Datagram carrying a single object.
1119#[derive(Debug, Clone, PartialEq, Eq)]
1120pub struct DatagramObject {
1121    /// Datagram type byte.
1122    pub datagram_type: DatagramType,
1123    /// Track alias.
1124    pub track_alias: VarInt,
1125    /// Group ID.
1126    pub group_id: VarInt,
1127    /// Object ID. Defaults to 0 when
1128    /// [`DatagramType::object_id_present`] is false.
1129    pub object_id: VarInt,
1130    /// Publisher priority.
1131    pub publisher_priority: u8,
1132    /// Raw extension-header bytes (empty unless
1133    /// [`DatagramType::extensions_present`] is true).
1134    pub extension_headers: Vec<u8>,
1135    /// Object status (only present for status-type datagrams).
1136    pub status: Option<ObjectStatus>,
1137    /// Object payload (empty for status-type datagrams).
1138    pub payload: Vec<u8>,
1139}
1140
1141impl DatagramObject {
1142    /// Encode the datagram in full, refusing a field the type byte cannot
1143    /// carry.
1144    ///
1145    /// Draft-14 Section 10.3.1 states the framing rule outright: "The Object
1146    /// Status field and Object Payload are mutually exclusive." Types 0x00
1147    /// through 0x07 carry a payload and omit the status field; types 0x20 and
1148    /// 0x21 carry a status field and have no payload. This value can hold both
1149    /// at once, and [`Self::encode`] resolves the disagreement by writing
1150    /// whichever one the type byte announces and discarding the other without a
1151    /// word. An End of Group marker written under a payload type does not
1152    /// arrive late or malformed — it does not arrive at all, and the receiver
1153    /// sees an ordinary object in its place; a payload written under a status
1154    /// type vanishes the same way.
1155    ///
1156    /// [`ObjectStatus::Normal`] under a payload type is not that case and is
1157    /// accepted. Section 10.2.1.1 says "Any object with a status code other
1158    /// than zero MUST have an empty payload", so Normal is the status a
1159    /// payload-bearing object already has, and stating it asks for exactly the
1160    /// bytes leaving it out asks for.
1161    ///
1162    /// Errors with [`CodecError::InvalidField`] on either lossy combination,
1163    /// before any byte is written, so a refused datagram leaves `buf`
1164    /// untouched.
1165    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1166        if self.datagram_type.is_status() {
1167            if !self.payload.is_empty() {
1168                return Err(CodecError::InvalidField);
1169            }
1170        } else if matches!(self.status, Some(status) if status != ObjectStatus::Normal) {
1171            return Err(CodecError::InvalidField);
1172        }
1173        // The extension block is the other field the type byte governs, and it
1174        // is governed in both directions. A type that announces extensions must
1175        // carry some, because Section 10.3.1 makes a declared length of 0 a
1176        // session-closing offence on receipt; a type that announces none cannot
1177        // carry any, and `encode` would drop them in silence.
1178        if self.datagram_type.extensions_present() {
1179            if self.extension_headers.is_empty() {
1180                return Err(CodecError::InvalidField);
1181            }
1182        } else if !self.extension_headers.is_empty() {
1183            return Err(CodecError::InvalidField);
1184        }
1185        check_extensions_against_status(
1186            self.status.map(|s| s.as_u64()),
1187            self.extension_headers.len() as u64,
1188        )?;
1189        self.encode(buf);
1190        Ok(())
1191    }
1192
1193    /// Encode the datagram in full.
1194    ///
1195    /// The type byte is taken as the authority on framing, which is what makes
1196    /// this infallible — and what makes it lossy when the value disagrees with
1197    /// itself. A `status` set under a payload type, or a `payload` set under a
1198    /// status type, is discarded here without a word. Prefer
1199    /// [`Self::encode_checked`], which refuses those combinations instead of
1200    /// resolving them.
1201    pub fn encode(&self, buf: &mut impl BufMut) {
1202        VarInt::from_u64(self.datagram_type.as_u8() as u64).unwrap().encode(buf);
1203        self.track_alias.encode(buf);
1204        self.group_id.encode(buf);
1205        if self.datagram_type.object_id_present() {
1206            self.object_id.encode(buf);
1207        }
1208        buf.put_u8(self.publisher_priority);
1209        if self.datagram_type.extensions_present() {
1210            VarInt::from_u64(self.extension_headers.len() as u64).unwrap().encode(buf);
1211            buf.put_slice(&self.extension_headers);
1212        }
1213        if self.datagram_type.is_status() {
1214            let status = self.status.unwrap_or(ObjectStatus::Normal);
1215            VarInt::from_u64(status.as_u64()).unwrap().encode(buf);
1216        } else {
1217            buf.put_slice(&self.payload);
1218        }
1219    }
1220
1221    /// Decode a datagram. The buffer must contain the full datagram —
1222    /// payload-bearing types extend to the end of the QUIC datagram,
1223    /// which the caller is responsible for delimiting.
1224    ///
1225    /// [`Self::decode_object`] also applies the Object rules this read does
1226    /// not.
1227    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1228        let type_val = VarInt::decode(buf)?.into_inner();
1229        if type_val > 0xFF {
1230            return Err(CodecError::UnknownDatagramType(type_val));
1231        }
1232        let datagram_type = DatagramType::from_u8(type_val as u8)
1233            .ok_or(CodecError::UnknownDatagramType(type_val))?;
1234        let track_alias = VarInt::decode(buf)?;
1235        let group_id = VarInt::decode(buf)?;
1236        let object_id = if datagram_type.object_id_present() {
1237            VarInt::decode(buf)?
1238        } else {
1239            VarInt::from_u64(0).unwrap()
1240        };
1241        if buf.remaining() < 1 {
1242            return Err(CodecError::UnexpectedEnd);
1243        }
1244        let publisher_priority = buf.get_u8();
1245        let extension_headers = if datagram_type.extensions_present() {
1246            let ext_len = VarInt::decode(buf)?.into_inner() as usize;
1247            // Section 10.3.1: "If an endpoint receives a datagram with
1248            // Extensions Present as 'Yes' and a Extension Headers Length of 0,
1249            // it MUST close the session with PROTOCOL_VIOLATION." A datagram
1250            // with no extensions has a type byte that says so, and the two
1251            // spellings of "no extensions" are not interchangeable here.
1252            //
1253            // Subgroup streams say the opposite in Section 10.4.2 — there the
1254            // type byte is fixed for the whole stream, so an object with no
1255            // extensions has nowhere to say it but a length of 0. Only the
1256            // datagram carries this rule.
1257            if ext_len == 0 {
1258                return Err(CodecError::InvalidField);
1259            }
1260            crate::types::read_bytes(buf, ext_len)?
1261        } else {
1262            Vec::new()
1263        };
1264        let (status, payload) = if datagram_type.is_status() {
1265            let status_val = VarInt::decode(buf)?.into_inner();
1266            let status = ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?;
1267            (Some(status), Vec::new())
1268        } else {
1269            let remaining = buf.remaining();
1270            (None, crate::types::read_bytes(buf, remaining)?)
1271        };
1272        check_extensions_against_status(
1273            status.map(|s| s.as_u64()),
1274            extension_headers.len() as u64,
1275        )?;
1276        Ok(DatagramObject {
1277            datagram_type,
1278            track_alias,
1279            group_id,
1280            object_id,
1281            publisher_priority,
1282            extension_headers,
1283            status,
1284            payload,
1285        })
1286    }
1287
1288    /// [`Self::decode`], then the Object rules a whole datagram is checked
1289    /// against.
1290    ///
1291    /// Refuses an extension block that does not parse, as
1292    /// [`crate::object_properties::parse_object_properties`] does, and, with
1293    /// [`CodecError::MalformedTrack`], the Prior Group ID Gap and Prior Object
1294    /// ID Gap conditions one Object shows (draft-14 Section 11.1,
1295    /// draft-14 Section 11.3), through [`PriorGaps::from_properties`].
1296    ///
1297    /// It also refuses bytes after a status datagram's Object Status, with
1298    /// [`CodecError::PayloadNotPermitted`], which [`Self::decode`] leaves in
1299    /// `buf`. draft-14 Section 10.2.1.1: "Any object with a status code other
1300    /// than zero MUST have an empty payload." draft-14 Section 10.3.1: "For
1301    /// Type values 0x20 and 0x21, the Object Status field is present and there
1302    /// is no Object Payload."
1303    pub fn decode_object(buf: &mut impl Buf) -> Result<Self, CodecError> {
1304        let object = Self::decode(buf)?;
1305        if let Some(status) = object.status {
1306            if buf.has_remaining() {
1307                return Err(CodecError::PayloadNotPermitted {
1308                    status: status.as_u64(),
1309                    len: buf.remaining(),
1310                    detail: "its type states a status in place of a payload",
1311                });
1312            }
1313        }
1314        let properties = crate::object_properties::parse_object_properties(
1315            crate::version::DraftVersion::Draft14,
1316            &object.extension_headers,
1317        )?;
1318        PriorGaps::from_properties(
1319            &properties,
1320            object.group_id.into_inner(),
1321            object.object_id.into_inner(),
1322        )?;
1323        Ok(object)
1324    }
1325}
1326
1327#[cfg(test)]
1328mod tests {
1329    use super::*;
1330
1331    fn vi(v: u64) -> VarInt {
1332        VarInt::from_u64(v).unwrap()
1333    }
1334
1335    // ── SubgroupStreamType flag helpers ─────────────────────
1336
1337    #[test]
1338    fn subgroup_type_0x10_all_off() {
1339        let t = SubgroupStreamType::from_u8(0x10).unwrap();
1340        assert!(!t.has_subgroup_id_field());
1341        assert!(!t.subgroup_id_is_first_object());
1342        assert!(!t.extensions_present());
1343        assert!(!t.contains_end_of_group());
1344    }
1345
1346    #[test]
1347    fn subgroup_type_0x15_explicit_with_ext() {
1348        let t = SubgroupStreamType::from_u8(0x15).unwrap();
1349        assert!(t.has_subgroup_id_field());
1350        assert!(!t.subgroup_id_is_first_object());
1351        assert!(t.extensions_present());
1352        assert!(!t.contains_end_of_group());
1353    }
1354
1355    #[test]
1356    fn subgroup_type_0x1d_all_on() {
1357        let t = SubgroupStreamType::from_u8(0x1D).unwrap();
1358        assert!(t.has_subgroup_id_field());
1359        assert!(t.extensions_present());
1360        assert!(t.contains_end_of_group());
1361    }
1362
1363    #[test]
1364    fn subgroup_type_0x12_first_object() {
1365        let t = SubgroupStreamType::from_u8(0x12).unwrap();
1366        assert!(!t.has_subgroup_id_field());
1367        assert!(t.subgroup_id_is_first_object());
1368        assert!(!t.extensions_present());
1369    }
1370
1371    #[test]
1372    fn subgroup_type_rejects_undefined() {
1373        for bad in [0x00u8, 0x0F, 0x16, 0x17, 0x1E, 0x1F, 0x20] {
1374            assert!(SubgroupStreamType::from_u8(bad).is_none(), "0x{bad:02x} should be rejected");
1375        }
1376    }
1377
1378    #[test]
1379    fn subgroup_type_from_flags_roundtrip() {
1380        for &f_sg in &[false, true] {
1381            for &f_first in &[false, true] {
1382                for &f_ext in &[false, true] {
1383                    for &f_eog in &[false, true] {
1384                        let t = SubgroupStreamType::from_flags(f_sg, f_first, f_ext, f_eog);
1385                        assert_eq!(t.has_subgroup_id_field(), f_sg);
1386                        // subgroup_id_is_first_object only meaningful when
1387                        // explicit field is absent
1388                        if !f_sg {
1389                            assert_eq!(t.subgroup_id_is_first_object(), f_first);
1390                        }
1391                        assert_eq!(t.extensions_present(), f_ext);
1392                        assert_eq!(t.contains_end_of_group(), f_eog);
1393                    }
1394                }
1395            }
1396        }
1397    }
1398
1399    // ── SubgroupHeader round-trip ───────────────────────────
1400
1401    #[test]
1402    fn subgroup_header_roundtrip_0x10() {
1403        let h = SubgroupHeader {
1404            stream_type: SubgroupStreamType::from_u8(0x10).unwrap(),
1405            track_alias: vi(1),
1406            group_id: vi(0),
1407            subgroup_id: None,
1408            publisher_priority: 128,
1409        };
1410        let mut buf = Vec::new();
1411        h.encode(&mut buf);
1412        assert_eq!(buf[0], 0x10);
1413        let decoded = SubgroupHeader::decode(&mut &buf[..]).unwrap();
1414        assert_eq!(decoded, h);
1415    }
1416
1417    #[test]
1418    fn subgroup_header_roundtrip_explicit_subgroup() {
1419        let h = SubgroupHeader {
1420            stream_type: SubgroupStreamType::from_u8(0x14).unwrap(),
1421            track_alias: vi(5),
1422            group_id: vi(10),
1423            subgroup_id: Some(vi(2)),
1424            publisher_priority: 64,
1425        };
1426        let mut buf = Vec::new();
1427        h.encode(&mut buf);
1428        let decoded = SubgroupHeader::decode(&mut &buf[..]).unwrap();
1429        assert_eq!(decoded, h);
1430    }
1431
1432    #[test]
1433    fn subgroup_header_decode_rejects_bad_type() {
1434        // 0x16 falls in the gap between the two ranges Table 4 assigns,
1435        // 0x10-0x15 and 0x18-0x1D, so no table names it. Draft-14 states no
1436        // list of invalid Types the way drafts 16 and later do — the value is
1437        // simply one it does not have, which Section 10 answers by ending the
1438        // session.
1439        let buf = [0x16u8, 0x01, 0x00, 0x80];
1440        let err = SubgroupHeader::decode(&mut &buf[..]).unwrap_err();
1441        assert!(
1442            matches!(err, CodecError::UnknownStreamType(0x16)),
1443            "an unassigned Type must be named as one, got {err:?}"
1444        );
1445    }
1446
1447    #[test]
1448    fn subgroup_header_decode_does_not_call_the_fetch_type_unknown() {
1449        // 0x05 is FETCH_HEADER, which Table 4 assigns. A subgroup reader
1450        // refuses it, but the disagreement is with the caller rather than with
1451        // the draft, so it must not reach for the rule that ends the session.
1452        let buf = [0x05u8, 0x01, 0x00, 0x80];
1453        let err = SubgroupHeader::decode(&mut &buf[..]).unwrap_err();
1454        assert!(
1455            matches!(err, CodecError::InvalidField),
1456            "a fetch stream at the subgroup reader must be refused without naming the \
1457             unknown-stream-type rule, got {err:?}"
1458        );
1459    }
1460
1461    // ── Subgroup object reader (delta + extensions) ─────────
1462
1463    #[test]
1464    fn subgroup_reader_delta_sequential_ids() {
1465        // Type 0x10: no subgroup field, no extensions, no eog
1466        let header = SubgroupHeader {
1467            stream_type: SubgroupStreamType::from_u8(0x10).unwrap(),
1468            track_alias: vi(1),
1469            group_id: vi(0),
1470            subgroup_id: None,
1471            publisher_priority: 0,
1472        };
1473
1474        let mut write = SubgroupObjectReader::new(&header);
1475        let mut buf = Vec::new();
1476        for i in 0..3u64 {
1477            let obj = SubgroupObject {
1478                object_id: vi(i),
1479                extension_headers: vec![],
1480                status: None,
1481                payload: vec![0xAA + i as u8; 4],
1482            };
1483            write.write_object(&obj, &mut buf).unwrap();
1484        }
1485
1486        let mut read = SubgroupObjectReader::new(&header);
1487        let mut cursor = &buf[..];
1488        let o0 = read.read_object(&mut cursor).unwrap();
1489        assert_eq!(o0.object_id.into_inner(), 0);
1490        assert_eq!(o0.payload, vec![0xAA; 4]);
1491        let o1 = read.read_object(&mut cursor).unwrap();
1492        assert_eq!(o1.object_id.into_inner(), 1);
1493        let o2 = read.read_object(&mut cursor).unwrap();
1494        assert_eq!(o2.object_id.into_inner(), 2);
1495    }
1496
1497    #[test]
1498    fn subgroup_reader_delta_sparse_ids() {
1499        // Object IDs 5, 10, 11 — deltas are 5, 4, 0
1500        let header = SubgroupHeader {
1501            stream_type: SubgroupStreamType::from_u8(0x10).unwrap(),
1502            track_alias: vi(1),
1503            group_id: vi(0),
1504            subgroup_id: None,
1505            publisher_priority: 0,
1506        };
1507        let mut write = SubgroupObjectReader::new(&header);
1508        let mut buf = Vec::new();
1509        for &id in &[5u64, 10, 11] {
1510            write
1511                .write_object(
1512                    &SubgroupObject {
1513                        object_id: vi(id),
1514                        extension_headers: vec![],
1515                        status: None,
1516                        payload: vec![1, 2, 3],
1517                    },
1518                    &mut buf,
1519                )
1520                .unwrap();
1521        }
1522        let mut read = SubgroupObjectReader::new(&header);
1523        let mut cursor = &buf[..];
1524        assert_eq!(read.read_object(&mut cursor).unwrap().object_id.into_inner(), 5);
1525        assert_eq!(read.read_object(&mut cursor).unwrap().object_id.into_inner(), 10);
1526        assert_eq!(read.read_object(&mut cursor).unwrap().object_id.into_inner(), 11);
1527    }
1528
1529    #[test]
1530    fn subgroup_reader_with_extensions() {
1531        // Type 0x11: extensions present
1532        let header = SubgroupHeader {
1533            stream_type: SubgroupStreamType::from_u8(0x11).unwrap(),
1534            track_alias: vi(1),
1535            group_id: vi(5),
1536            subgroup_id: None,
1537            publisher_priority: 0,
1538        };
1539        let mut write = SubgroupObjectReader::new(&header);
1540        let mut buf = Vec::new();
1541        write
1542            .write_object(
1543                &SubgroupObject {
1544                    object_id: vi(0),
1545                    extension_headers: vec![0x3c, 0x02],
1546                    status: None,
1547                    payload: vec![0xFF],
1548                },
1549                &mut buf,
1550            )
1551            .unwrap();
1552        let mut read = SubgroupObjectReader::new(&header);
1553        let o = read.read_object(&mut &buf[..]).unwrap();
1554        assert_eq!(o.extension_headers, vec![0x3c, 0x02]);
1555        assert_eq!(o.payload, vec![0xFF]);
1556    }
1557
1558    #[test]
1559    fn subgroup_reader_status_object() {
1560        let header = SubgroupHeader {
1561            stream_type: SubgroupStreamType::from_u8(0x10).unwrap(),
1562            track_alias: vi(1),
1563            group_id: vi(0),
1564            subgroup_id: None,
1565            publisher_priority: 0,
1566        };
1567        let mut write = SubgroupObjectReader::new(&header);
1568        let mut buf = Vec::new();
1569        write
1570            .write_object(
1571                &SubgroupObject {
1572                    object_id: vi(7),
1573                    extension_headers: vec![],
1574                    status: Some(ObjectStatus::EndOfGroup),
1575                    payload: vec![],
1576                },
1577                &mut buf,
1578            )
1579            .unwrap();
1580        let mut read = SubgroupObjectReader::new(&header);
1581        let o = read.read_object(&mut &buf[..]).unwrap();
1582        assert_eq!(o.object_id.into_inner(), 7);
1583        assert_eq!(o.status, Some(ObjectStatus::EndOfGroup));
1584        assert!(o.payload.is_empty());
1585    }
1586
1587    #[test]
1588    fn subgroup_reader_meta_matches_read_object() {
1589        // Type 0x11: extensions present, so every field is exercised.
1590        let header = SubgroupHeader {
1591            stream_type: SubgroupStreamType::from_u8(0x11).unwrap(),
1592            track_alias: vi(1),
1593            group_id: vi(0),
1594            subgroup_id: None,
1595            publisher_priority: 0,
1596        };
1597        let mut write = SubgroupObjectReader::new(&header);
1598        let mut buf = Vec::new();
1599        for (id, status) in
1600            [(0u64, None), (4, Some(ObjectStatus::EndOfGroup)), (9, None)].into_iter()
1601        {
1602            write
1603                .write_object(
1604                    &SubgroupObject {
1605                        object_id: vi(id),
1606                        extension_headers: vec![0x0A, 0x0B],
1607                        status,
1608                        payload: if status.is_some() { vec![] } else { vec![0xEE; 3] },
1609                    },
1610                    &mut buf,
1611                )
1612                .unwrap();
1613        }
1614
1615        let mut full = SubgroupObjectReader::new(&header);
1616        let mut meta = SubgroupObjectReader::new(&header);
1617        let mut full_cursor = &buf[..];
1618        let mut meta_cursor = &buf[..];
1619        for _ in 0..3 {
1620            let before = meta_cursor.remaining();
1621            let o = full.read_object(&mut full_cursor).unwrap();
1622            let m = meta.read_object_meta(&mut meta_cursor).unwrap();
1623            assert_eq!(m.object_id, o.object_id.into_inner());
1624            assert_eq!(m.extension_headers_len, o.extension_headers.len() as u64);
1625            assert_eq!(m.payload_length, o.payload.len() as u64);
1626            assert_eq!(m.status, o.status.map(|s| s.as_u64()));
1627            assert_eq!(m.wire_len, (before - meta_cursor.remaining()) as u64);
1628            assert_eq!(full_cursor.remaining(), meta_cursor.remaining());
1629        }
1630        assert!(meta_cursor.is_empty());
1631    }
1632
1633    #[test]
1634    fn subgroup_reader_meta_short_buffer_is_unexpected_end() {
1635        let header = SubgroupHeader {
1636            stream_type: SubgroupStreamType::from_u8(0x10).unwrap(),
1637            track_alias: vi(1),
1638            group_id: vi(0),
1639            subgroup_id: None,
1640            publisher_priority: 0,
1641        };
1642        let mut write = SubgroupObjectReader::new(&header);
1643        let mut buf = Vec::new();
1644        write
1645            .write_object(
1646                &SubgroupObject {
1647                    object_id: vi(0),
1648                    extension_headers: vec![],
1649                    status: None,
1650                    payload: vec![1, 2, 3, 4],
1651                },
1652                &mut buf,
1653            )
1654            .unwrap();
1655
1656        for cut in 1..buf.len() {
1657            let mut read = SubgroupObjectReader::new(&header);
1658            let err = read.read_object_meta(&mut &buf[..cut]).unwrap_err();
1659            assert!(
1660                matches!(err, CodecError::UnexpectedEnd | CodecError::VarInt(_)),
1661                "cut {cut} gave {err:?}"
1662            );
1663        }
1664    }
1665
1666    // ── FetchHeader + FetchObject ───────────────────────────
1667
1668    #[test]
1669    fn fetch_header_roundtrip() {
1670        let h = FetchHeader { request_id: vi(99) };
1671        let mut buf = Vec::new();
1672        h.encode(&mut buf);
1673        assert_eq!(buf[0], 0x05);
1674        assert_eq!(FetchHeader::decode(&mut &buf[..]).unwrap(), h);
1675    }
1676
1677    #[test]
1678    fn fetch_header_rejects_wrong_type() {
1679        let buf = [0x10u8, 0x05];
1680        assert!(FetchHeader::decode(&mut &buf[..]).is_err());
1681    }
1682
1683    #[test]
1684    fn fetch_object_roundtrip_with_payload() {
1685        let obj = FetchObject {
1686            group_id: vi(3),
1687            subgroup_id: vi(1),
1688            object_id: vi(7),
1689            publisher_priority: 200,
1690            extension_headers: vec![0xAA, 0xBB],
1691            status: None,
1692            payload: vec![1, 2, 3, 4],
1693        };
1694        let mut buf = Vec::new();
1695        obj.encode(&mut buf);
1696        assert_eq!(FetchObject::decode(&mut &buf[..]).unwrap(), obj);
1697    }
1698
1699    #[test]
1700    fn fetch_object_roundtrip_status() {
1701        let obj = FetchObject {
1702            group_id: vi(3),
1703            subgroup_id: vi(1),
1704            object_id: vi(8),
1705            publisher_priority: 200,
1706            extension_headers: vec![],
1707            status: Some(ObjectStatus::ObjectDoesNotExist),
1708            payload: vec![],
1709        };
1710        let mut buf = Vec::new();
1711        obj.encode(&mut buf);
1712        assert_eq!(FetchObject::decode(&mut &buf[..]).unwrap(), obj);
1713    }
1714
1715    #[test]
1716    fn fetch_object_meta_matches_decode() {
1717        for obj in [
1718            FetchObject {
1719                group_id: vi(3),
1720                subgroup_id: vi(1),
1721                object_id: vi(7),
1722                publisher_priority: 200,
1723                extension_headers: vec![0xAA, 0xBB],
1724                status: None,
1725                payload: vec![1, 2, 3, 4],
1726            },
1727            FetchObject {
1728                group_id: vi(4),
1729                subgroup_id: vi(0),
1730                object_id: vi(8),
1731                publisher_priority: 1,
1732                extension_headers: vec![],
1733                status: Some(ObjectStatus::EndOfTrack),
1734                payload: vec![],
1735            },
1736        ] {
1737            let mut buf = Vec::new();
1738            obj.encode(&mut buf);
1739            let mut decode_cursor = &buf[..];
1740            let mut meta_cursor = &buf[..];
1741            let decoded = FetchObject::decode(&mut decode_cursor).unwrap();
1742            let meta = FetchObject::decode_meta(&mut meta_cursor).unwrap();
1743            assert_eq!(meta.group_id, decoded.group_id.into_inner());
1744            assert_eq!(meta.subgroup_id, decoded.subgroup_id.into_inner());
1745            assert_eq!(meta.object_id, decoded.object_id.into_inner());
1746            assert_eq!(meta.publisher_priority, decoded.publisher_priority);
1747            assert_eq!(meta.extension_headers_len, decoded.extension_headers.len() as u64);
1748            assert_eq!(meta.payload_length, decoded.payload.len() as u64);
1749            assert_eq!(meta.status, decoded.status.map(|s| s.as_u64()));
1750            assert_eq!(meta.wire_len, buf.len() as u64);
1751            assert!(meta_cursor.is_empty());
1752            assert_eq!(decode_cursor.remaining(), meta_cursor.remaining());
1753        }
1754    }
1755
1756    // ── DatagramType ────────────────────────────────────────
1757
1758    #[test]
1759    fn datagram_type_variants() {
1760        let t0 = DatagramType::from_u8(0x00).unwrap();
1761        assert!(t0.object_id_present());
1762        assert!(!t0.extensions_present());
1763        assert!(!t0.end_of_group());
1764        assert!(!t0.is_status());
1765
1766        let t7 = DatagramType::from_u8(0x07).unwrap();
1767        assert!(!t7.object_id_present()); // bit 2 set
1768        assert!(t7.extensions_present());
1769        assert!(t7.end_of_group());
1770        assert!(!t7.is_status());
1771
1772        let t20 = DatagramType::from_u8(0x20).unwrap();
1773        assert!(t20.is_status());
1774        assert!(!t20.extensions_present());
1775        // Status datagrams always carry Object ID
1776        assert!(t20.object_id_present());
1777
1778        let t21 = DatagramType::from_u8(0x21).unwrap();
1779        assert!(t21.is_status());
1780        assert!(t21.extensions_present());
1781    }
1782
1783    #[test]
1784    fn datagram_type_rejects_undefined() {
1785        for bad in [0x08u8, 0x10, 0x1F, 0x22, 0x80] {
1786            assert!(DatagramType::from_u8(bad).is_none(), "0x{bad:02x}");
1787        }
1788    }
1789
1790    // ── DatagramObject round-trip ───────────────────────────
1791
1792    #[test]
1793    fn datagram_object_0x00_roundtrip() {
1794        let d = DatagramObject {
1795            datagram_type: DatagramType::from_u8(0x00).unwrap(),
1796            track_alias: vi(1),
1797            group_id: vi(2),
1798            object_id: vi(3),
1799            publisher_priority: 100,
1800            extension_headers: vec![],
1801            status: None,
1802            payload: vec![0xDE, 0xAD, 0xBE, 0xEF],
1803        };
1804        let mut buf = Vec::new();
1805        d.encode(&mut buf);
1806        assert_eq!(DatagramObject::decode(&mut &buf[..]).unwrap(), d);
1807    }
1808
1809    #[test]
1810    fn datagram_object_0x04_no_object_id() {
1811        // 0x04: no object id field, implicit 0
1812        let d = DatagramObject {
1813            datagram_type: DatagramType::from_u8(0x04).unwrap(),
1814            track_alias: vi(1),
1815            group_id: vi(2),
1816            object_id: vi(0),
1817            publisher_priority: 100,
1818            extension_headers: vec![],
1819            status: None,
1820            payload: vec![0xAA],
1821        };
1822        let mut buf = Vec::new();
1823        d.encode(&mut buf);
1824        let decoded = DatagramObject::decode(&mut &buf[..]).unwrap();
1825        assert_eq!(decoded, d);
1826    }
1827
1828    #[test]
1829    fn datagram_object_0x21_status_with_extensions() {
1830        let d = DatagramObject {
1831            datagram_type: DatagramType::from_u8(0x21).unwrap(),
1832            track_alias: vi(9),
1833            group_id: vi(4),
1834            object_id: vi(11),
1835            publisher_priority: 50,
1836            extension_headers: vec![0xCA, 0xFE],
1837            status: Some(ObjectStatus::EndOfTrack),
1838            payload: vec![],
1839        };
1840        let mut buf = Vec::new();
1841        d.encode(&mut buf);
1842        assert_eq!(DatagramObject::decode(&mut &buf[..]).unwrap(), d);
1843    }
1844
1845    /// A datagram under `type_byte` holding both `status` and `payload`.
1846    fn datagram(type_byte: u8, status: Option<ObjectStatus>, payload: Vec<u8>) -> DatagramObject {
1847        DatagramObject {
1848            datagram_type: DatagramType::from_u8(type_byte).unwrap(),
1849            track_alias: vi(1),
1850            group_id: vi(0),
1851            object_id: vi(0),
1852            publisher_priority: 128,
1853            extension_headers: vec![],
1854            status,
1855            payload,
1856        }
1857    }
1858
1859    /// Neither of the two fields the type byte cannot carry is dropped in
1860    /// silence; both are refused.
1861    ///
1862    /// Draft-14 Section 10.3.1: "The Object Status field and Object Payload are
1863    /// mutually exclusive." Types 0x00 through 0x07 write a payload and no
1864    /// status; types 0x20 and 0x21 write a status and no payload. A
1865    /// [`DatagramObject`] can hold both at once, and [`DatagramObject::encode`]
1866    /// resolves that by writing whichever the type byte announces and
1867    /// discarding the other — the loss this gate exists for. The middle of each
1868    /// half observes the discard directly, so the gate states `encode`'s silent
1869    /// discard as well as `encode_checked`'s refusal.
1870    ///
1871    /// Normal beside a payload is exempt and checked at the end. Section
1872    /// 10.2.1.1 says "Any object with a status code other than zero MUST have
1873    /// an empty payload", so Normal is the status a payload-bearing object
1874    /// already has, and naming it asks for the same bytes as leaving it out.
1875    ///
1876    /// Fails if `encode_checked` accepts a non-Normal status under a payload
1877    /// type or a payload under a status type, or refuses Normal beside a
1878    /// payload.
1879    #[test]
1880    fn encode_checked_refuses_the_field_the_type_byte_cannot_carry() {
1881        for &status in ObjectStatus::ALL {
1882            if status == ObjectStatus::Normal {
1883                continue;
1884            }
1885
1886            // A status under a payload type: the status is what would go.
1887            let object = datagram(0x00, Some(status), vec![0xDE, 0xAD]);
1888            let mut refused = Vec::new();
1889            let result = object.encode_checked(&mut refused);
1890            assert!(
1891                matches!(result, Err(CodecError::InvalidField)),
1892                "encode_checked must refuse {status:?} under a payload type; got {result:?}"
1893            );
1894            assert!(refused.is_empty(), "a refused {status:?} datagram still wrote {refused:?}");
1895
1896            let mut dropped = Vec::new();
1897            object.encode(&mut dropped);
1898            let decoded = DatagramObject::decode(&mut &dropped[..])
1899                .unwrap_or_else(|e| panic!("the lossy encoding of {status:?} must parse: {e:?}"));
1900            assert_eq!(decoded.status, None, "{status:?} is exactly what `encode` loses here");
1901            assert_eq!(decoded.payload, vec![0xDE, 0xAD]);
1902
1903            // The same status under a status type is representable, so it is
1904            // written and read back unchanged.
1905            let mut carried = Vec::new();
1906            datagram(0x20, Some(status), vec![]).encode_checked(&mut carried).unwrap_or_else(|e| {
1907                panic!("encode_checked refused a status-type {status:?}: {e:?}")
1908            });
1909            let decoded = DatagramObject::decode(&mut &carried[..]).unwrap();
1910            assert_eq!(decoded.status, Some(status), "{status:?} lost its status");
1911        }
1912
1913        // A payload under a status type: now the payload is what would go.
1914        let object = datagram(0x20, Some(ObjectStatus::EndOfGroup), vec![0xDE, 0xAD]);
1915        let mut refused = Vec::new();
1916        let result = object.encode_checked(&mut refused);
1917        assert!(
1918            matches!(result, Err(CodecError::InvalidField)),
1919            "encode_checked must refuse a payload under a status type; got {result:?}"
1920        );
1921        assert!(refused.is_empty(), "a refused datagram still wrote {refused:?}");
1922
1923        let mut dropped = Vec::new();
1924        object.encode(&mut dropped);
1925        let decoded = DatagramObject::decode(&mut &dropped[..]).unwrap();
1926        assert!(decoded.payload.is_empty(), "the payload is exactly what `encode` loses here");
1927        assert_eq!(decoded.status, Some(ObjectStatus::EndOfGroup));
1928
1929        // Normal beside a payload asks for the bytes a payload datagram
1930        // already writes, so it is accepted and writes exactly those.
1931        let mut named = Vec::new();
1932        datagram(0x00, Some(ObjectStatus::Normal), vec![0xDE, 0xAD])
1933            .encode_checked(&mut named)
1934            .unwrap_or_else(|e| {
1935                panic!("encode_checked refused a Normal status under a payload type: {e:?}")
1936            });
1937        let mut unnamed = Vec::new();
1938        datagram(0x00, None, vec![0xDE, 0xAD]).encode_checked(&mut unnamed).unwrap();
1939        assert_eq!(named, unnamed, "naming Normal must ask for the bytes leaving it out asks for");
1940    }
1941}
1942
1943#[cfg(test)]
1944mod datagram_gap_tests {
1945    use super::*;
1946
1947    fn datagram(group: u64, object: u64, block: &[u8]) -> Vec<u8> {
1948        let mut bytes = Vec::new();
1949        DatagramObject {
1950            datagram_type: DatagramType::payload(true, true, false),
1951            track_alias: VarInt::from_u64(1).unwrap(),
1952            group_id: VarInt::from_u64(group).unwrap(),
1953            object_id: VarInt::from_u64(object).unwrap(),
1954            publisher_priority: 0x80,
1955            extension_headers: block.to_vec(),
1956            status: None,
1957            payload: vec![0xAA],
1958        }
1959        .encode(&mut bytes);
1960        bytes
1961    }
1962
1963    fn is_malformed_track(bytes: &[u8]) -> bool {
1964        matches!(
1965            DatagramObject::decode_object(&mut &bytes[..]),
1966            Err(CodecError::MalformedTrack { .. })
1967        )
1968    }
1969
1970    #[test]
1971    fn decode_object_refuses_the_gap_conditions_one_datagram_shows() {
1972        for bytes in [
1973            datagram(1, 0, &[0x3C, 2]),
1974            datagram(5, 0, &[0x3E, 1]),
1975            datagram(5, 5, &[0x3C, 1, 0x3C, 1]),
1976        ] {
1977            assert!(is_malformed_track(&bytes));
1978            assert!(DatagramObject::decode(&mut &bytes[..]).is_ok());
1979        }
1980        let wire = datagram(2, 1, &[0x3C, 2, 0x3E, 1]);
1981        assert!(DatagramObject::decode_object(&mut &wire[..]).is_ok());
1982    }
1983
1984    #[test]
1985    fn decode_object_refuses_a_payload_after_a_status_datagram() {
1986        let mut wire = Vec::new();
1987        DatagramObject {
1988            datagram_type: DatagramType::from_u8(0x20).unwrap(),
1989            track_alias: VarInt::from_u64(1).unwrap(),
1990            group_id: VarInt::from_u64(0).unwrap(),
1991            object_id: VarInt::from_u64(0).unwrap(),
1992            publisher_priority: 0x80,
1993            extension_headers: Vec::new(),
1994            status: Some(ObjectStatus::Normal),
1995            payload: Vec::new(),
1996        }
1997        .encode(&mut wire);
1998        assert!(DatagramObject::decode_object(&mut &wire[..]).is_ok());
1999        wire.push(0xAA);
2000        assert!(DatagramObject::decode(&mut &wire[..]).is_ok());
2001        assert!(matches!(
2002            DatagramObject::decode_object(&mut &wire[..]),
2003            Err(CodecError::PayloadNotPermitted { len: 1, .. })
2004        ));
2005    }
2006}