Skip to main content

moqtap_codec/draft09/
data_stream.rs

1//! Draft-09 data stream header encoding and decoding.
2//!
3//! Changes from draft-08:
4//! - `extension_count` → `extension_headers_length` (byte length, not count)
5//! - Datagram (0x01): no `payload_length` or `object_status`; payload is remaining bytes
6//! - DatagramStatus (0x02): gains `extension_headers_length` field
7
8use super::types::ObjectStatus;
9use crate::error::CodecError;
10use crate::types::read_bytes;
11use crate::varint::VarInt;
12use bytes::{Buf, BufMut};
13
14/// Every type ID this draft's data plane assigns, from two separate tables.
15///
16/// The draft keeps unidirectional stream types and datagram types in different
17/// tables and different number spaces: SUBGROUP_HEADER and FETCH_HEADER name
18/// unidirectional streams, OBJECT_DATAGRAM and OBJECT_DATAGRAM_STATUS name
19/// datagrams. The numbers happen not to collide on this draft, which is what
20/// lets one enum hold all four.
21///
22/// [`StreamType::from_id`] therefore answers across both tables and cannot say
23/// which one a value came from, which makes it the wrong question to ask about
24/// a stream or a datagram in hand. `stream_type_error` and
25/// `datagram_type_error` ask it per table, and the difference is not
26/// cosmetic: a value assigned in the other table is unknown in this one, and
27/// this draft answers an unknown type by ending the session.
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29#[repr(u64)]
30pub enum StreamType {
31    /// Datagram with payload (0x01).
32    Datagram = 0x01,
33    /// Datagram with status only, no payload (0x02).
34    DatagramStatus = 0x02,
35    /// Subgroup stream type (0x04).
36    Subgroup = 0x04,
37    /// Fetch stream type (0x05).
38    Fetch = 0x05,
39}
40
41impl StreamType {
42    /// Convert a raw stream type ID to a `StreamType`, if valid.
43    pub fn from_id(id: u64) -> Option<Self> {
44        match id {
45            0x01 => Some(StreamType::Datagram),
46            0x02 => Some(StreamType::DatagramStatus),
47            0x04 => Some(StreamType::Subgroup),
48            0x05 => Some(StreamType::Fetch),
49            _ => None,
50        }
51    }
52}
53
54/// Which failure a leading unidirectional stream type that is not the one a
55/// reader wants is.
56///
57/// Section 8: "An endpoint that receives an unknown stream or datagram type
58/// MUST close the session." One sentence, two tables. The stream table assigns
59/// SUBGROUP_HEADER and FETCH_HEADER; everything outside those two is unknown at
60/// the head of a stream, and the session ends.
61///
62/// The datagram types are among the values outside them. This draft is the
63/// first to give datagrams a table of their own, with numbering independent of
64/// the stream table, so a datagram type says nothing about a stream and is not
65/// a value the stream table assigns. Draft-07, which numbered both in one
66/// table, is the only draft where that is not so.
67///
68/// A stream announcing the *other* assigned stream type is a different matter
69/// and not that rule. The value is one this draft defines, and the disagreement
70/// is with the reader that was called rather than with the draft, so the
71/// session survives it.
72fn stream_type_error(raw: u64) -> CodecError {
73    if raw == StreamType::Subgroup as u64 || raw == StreamType::Fetch as u64 {
74        CodecError::InvalidField
75    } else {
76        CodecError::UnknownStreamType(raw)
77    }
78}
79
80/// Which failure a leading datagram type that is not one a reader wants is.
81///
82/// The datagram half of the sentence quoted on `stream_type_error`, read
83/// against the other table: OBJECT_DATAGRAM and OBJECT_DATAGRAM_STATUS are what
84/// it assigns, and everything else arriving as a datagram is unknown.
85///
86/// Both assigned values are decodable here, so the [`CodecError::InvalidField`]
87/// arm is reached only by the stream types — 0x04 and 0x05 — delivered as
88/// datagrams. Those are values this draft defines, and refusing them without
89/// closing is the same judgement the stream side makes about a datagram type.
90fn datagram_type_error(raw: u64) -> CodecError {
91    if raw == StreamType::Datagram as u64 || raw == StreamType::DatagramStatus as u64 {
92        CodecError::InvalidField
93    } else {
94        CodecError::UnknownDatagramType(raw)
95    }
96}
97
98// ── Extension helpers (length-based, not count-based) ─────────
99
100/// Read `byte_len` bytes of raw extension data from the buffer.
101fn read_extension_bytes(buf: &mut impl Buf, byte_len: u64) -> Result<Vec<u8>, CodecError> {
102    read_bytes(buf, byte_len as usize)
103}
104
105/// Encode extension bytes to the buffer (just writes the raw bytes).
106fn encode_extensions(extensions: &[u8], buf: &mut impl BufMut) {
107    buf.put_slice(extensions);
108}
109
110// ============================================================
111// Subgroup stream (type 0x04)
112// ============================================================
113
114/// Subgroup stream header (follows the stream type varint).
115#[derive(Debug, Clone, PartialEq, Eq)]
116pub struct SubgroupHeader {
117    /// Track alias identifying the subscription.
118    pub track_alias: VarInt,
119    /// Group identifier.
120    pub group_id: VarInt,
121    /// Subgroup identifier within the group.
122    pub subgroup_id: VarInt,
123    /// Publisher priority for delivery ordering.
124    pub publisher_priority: u8,
125}
126
127/// Object within a subgroup stream (draft-09).
128///
129/// Encoding: object_id(vi), extension_headers_length(vi), [extensions...],
130///   payload_length(vi),
131///   if payload_length == 0: object_status(vi)
132///   else: payload bytes
133#[derive(Debug, Clone, PartialEq, Eq)]
134pub struct ObjectHeader {
135    /// Object identifier within the subgroup.
136    pub object_id: VarInt,
137    /// Total byte length of extension headers, as it arrived.
138    ///
139    /// Advisory on encode. Every encoder here writes
140    /// `VarInt::from_usize(self.extensions.len())` and never reads this field,
141    /// so a hand-built header whose stated length disagrees with the bytes
142    /// beside it goes on the wire with the derived length and is read back
143    /// consistent. Decoding always sets the two together, so a value that came
144    /// off the wire never disagrees.
145    ///
146    /// It is kept because it is what the peer stated, which is not always
147    /// recoverable from the bytes: a non-minimal varint length encodes the same
148    /// number in more bytes, and a relay that must forward the block unchanged
149    /// has to know which it saw.
150    pub extension_headers_length: VarInt,
151    /// Raw extension bytes (opaque).
152    pub extensions: Vec<u8>,
153    /// Length of the object payload in bytes.
154    pub payload_length: VarInt,
155    /// Status of this object.
156    pub object_status: ObjectStatus,
157}
158
159impl SubgroupHeader {
160    /// Encode a subgroup stream header including its leading stream-type
161    /// field, so the bytes form the start of a data stream a peer can read.
162    ///
163    /// [`Self::encode`] writes the body alone, which is what a caller wants
164    /// once the stream is already open and what a caller must not use for its
165    /// first write.
166    pub fn encode_stream(&self, buf: &mut impl BufMut) {
167        VarInt::from_usize(StreamType::Subgroup as usize).encode(buf);
168        self.encode(buf);
169    }
170
171    /// Encode the subgroup header into the buffer.
172    pub fn encode(&self, buf: &mut impl BufMut) {
173        self.track_alias.encode(buf);
174        self.group_id.encode(buf);
175        self.subgroup_id.encode(buf);
176        buf.put_u8(self.publisher_priority);
177    }
178
179    /// Decode a subgroup header from the buffer.
180    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
181        let track_alias = VarInt::decode(buf)?;
182        let group_id = VarInt::decode(buf)?;
183        let subgroup_id = VarInt::decode(buf)?;
184        if buf.remaining() < 1 {
185            return Err(CodecError::UnexpectedEnd);
186        }
187        let publisher_priority = buf.get_u8();
188        Ok(Self { track_alias, group_id, subgroup_id, publisher_priority })
189    }
190
191    /// Decode a subgroup header from the start of a data stream, consuming
192    /// the leading stream type varint.
193    ///
194    /// Errors with [`CodecError::UnknownStreamType`] when the stream type is
195    /// one the stream table does not assign, which this draft answers with a
196    /// close, and with [`CodecError::InvalidField`] when it is assigned but is
197    /// not [`StreamType::Subgroup`]. `stream_type_error` draws that line.
198    pub fn decode_stream(buf: &mut impl Buf) -> Result<Self, CodecError> {
199        let stream_type = VarInt::decode(buf)?.into_inner();
200        if stream_type != StreamType::Subgroup as u64 {
201            return Err(stream_type_error(stream_type));
202        }
203        Self::decode(buf)
204    }
205}
206
207/// Refuse an end-of-track object that does not end at object zero.
208///
209/// Section 8.1.1.1 describes Object Status 0x5 as "end of Track. GroupID is one
210/// greater than the largest group produced in this track and the ObjectId is
211/// zero", and states the consequence: "An object with this status that has a
212/// Group ID less than or equal to any other Group ID, or an Object ID other
213/// than zero, is a protocol error, and the receiver MUST terminate the
214/// session."
215///
216/// Only the Object ID half is answerable here. The Group ID half compares
217/// against the largest group produced on the track, which no single header
218/// carries and no reader of one header can know.
219///
220/// Applied on both sides. A receiver is required to close the session over
221/// this, so writing one is not a way to send it.
222fn check_end_of_track(object_id: VarInt, status: ObjectStatus) -> Result<(), CodecError> {
223    if status == ObjectStatus::EndOfTrack && object_id.into_inner() != 0 {
224        return Err(CodecError::EndOfTrackObjectId(object_id.into_inner()));
225    }
226    Ok(())
227}
228
229impl ObjectHeader {
230    /// Encode the object header into the buffer.
231    pub fn encode(&self, buf: &mut impl BufMut) {
232        self.object_id.encode(buf);
233        VarInt::from_usize(self.extensions.len()).encode(buf);
234        encode_extensions(&self.extensions, buf);
235        self.payload_length.encode(buf);
236        if self.payload_length.into_inner() == 0 {
237            VarInt::from_usize(self.object_status as usize).encode(buf);
238        }
239    }
240
241    /// Encode the header, refusing a status the framing cannot carry.
242    ///
243    /// Section 8.4.1 puts the Object Status field on the wire only when the
244    /// Object Payload Length is zero, and Section 8.1.1.1 says "Any object
245    /// with a status code other than zero MUST have an empty payload". A
246    /// non-zero status paired with a non-zero payload length therefore has no
247    /// encoding at all: [`Self::encode`] drops the status and the peer reads an
248    /// ordinary object, which is a different object from the one the caller
249    /// described. This refuses instead.
250    ///
251    /// The datagram types on this draft already refuse the same pairing. These
252    /// two did not, and they are the ones a publisher writes on every stream.
253    ///
254    /// # Errors
255    ///
256    /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
257    /// a non-zero Object Payload Length.
258    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
259        check_end_of_track(self.object_id, self.object_status)?;
260        if self.payload_length.into_inner() != 0 && self.object_status as usize != 0 {
261            return Err(CodecError::InvalidField);
262        }
263        self.encode(buf);
264        Ok(())
265    }
266
267    /// Decode an object header from the buffer.
268    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
269        let object_id = VarInt::decode(buf)?;
270        let extension_headers_length = VarInt::decode(buf)?;
271        let extensions = read_extension_bytes(buf, extension_headers_length.into_inner())?;
272        let payload_length = VarInt::decode(buf)?;
273        let object_status = if payload_length.into_inner() == 0 {
274            let status_val = VarInt::decode(buf)?.into_inner();
275            ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?
276        } else {
277            ObjectStatus::Normal
278        };
279        check_end_of_track(object_id, object_status)?;
280        Ok(Self { object_id, extension_headers_length, extensions, payload_length, object_status })
281    }
282}
283
284// ============================================================
285// Datagram (type 0x01)
286// ============================================================
287
288/// Datagram header with payload (draft-09, type 0x01).
289///
290/// Draft-09 change: no `payload_length` or `object_status` fields.
291/// Payload is the remaining bytes in the datagram.
292///
293/// Encoding (after type varint):
294///   track_alias(vi), group_id(vi), object_id(vi),
295///   publisher_priority(u8), extension_headers_length(vi), [extensions...],
296///   [remaining bytes = payload]
297#[derive(Debug, Clone, PartialEq, Eq)]
298pub struct DatagramHeader {
299    /// Track alias identifying the subscription.
300    pub track_alias: VarInt,
301    /// Group identifier.
302    pub group_id: VarInt,
303    /// Object identifier within the group.
304    pub object_id: VarInt,
305    /// Publisher priority for delivery ordering.
306    pub publisher_priority: u8,
307    /// Total byte length of extension headers, as it arrived.
308    ///
309    /// Advisory on encode. Every encoder here writes
310    /// `VarInt::from_usize(self.extensions.len())` and never reads this field,
311    /// so a hand-built header whose stated length disagrees with the bytes
312    /// beside it goes on the wire with the derived length and is read back
313    /// consistent. Decoding always sets the two together, so a value that came
314    /// off the wire never disagrees.
315    ///
316    /// It is kept because it is what the peer stated, which is not always
317    /// recoverable from the bytes: a non-minimal varint length encodes the same
318    /// number in more bytes, and a relay that must forward the block unchanged
319    /// has to know which it saw.
320    pub extension_headers_length: VarInt,
321    /// Raw extension bytes (opaque).
322    pub extensions: Vec<u8>,
323}
324
325impl DatagramHeader {
326    /// Encode the datagram header into the buffer.
327    pub fn encode(&self, buf: &mut impl BufMut) {
328        self.track_alias.encode(buf);
329        self.group_id.encode(buf);
330        self.object_id.encode(buf);
331        buf.put_u8(self.publisher_priority);
332        VarInt::from_usize(self.extensions.len()).encode(buf);
333        encode_extensions(&self.extensions, buf);
334    }
335
336    /// Encode the datagram header, refusing a status the framing cannot carry.
337    ///
338    /// Nothing here is ever refused, and that is a fact about draft-09 rather
339    /// than a check left out. This is the OBJECT_DATAGRAM of Section 8.2, whose
340    /// layout carries no Object Status field at all; a datagram that states a
341    /// status is the separate OBJECT_DATAGRAM_STATUS message of Section 8.3,
342    /// modelled here as [`DatagramStatusHeader`]. So there is no status for
343    /// [`Self::encode`] to drop, and nothing for Section 8.1.1.1's "Any object
344    /// with a status code other than zero MUST have an empty payload" to rule
345    /// on: an object framed this way has status zero by construction.
346    ///
347    /// The fallible signature is what lets one entry point span every draft.
348    /// `dispatch::AnyDatagramHeader::encode` calls this on all thirteen, and
349    /// the drafts whose payload-bearing datagram *does* carry a status field
350    /// need somewhere to say no.
351    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
352        self.encode(buf);
353        Ok(())
354    }
355
356    /// Decode a datagram header from the buffer.
357    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
358        let track_alias = VarInt::decode(buf)?;
359        let group_id = VarInt::decode(buf)?;
360        let object_id = VarInt::decode(buf)?;
361        if buf.remaining() < 1 {
362            return Err(CodecError::UnexpectedEnd);
363        }
364        let publisher_priority = buf.get_u8();
365        let extension_headers_length = VarInt::decode(buf)?;
366        let extensions = read_extension_bytes(buf, extension_headers_length.into_inner())?;
367        Ok(Self {
368            track_alias,
369            group_id,
370            object_id,
371            publisher_priority,
372            extension_headers_length,
373            extensions,
374        })
375    }
376}
377
378// ============================================================
379// Datagram Status (type 0x02)
380// ============================================================
381
382/// Datagram status header (draft-09, type 0x02).
383///
384/// Draft-09 change: gains `extension_headers_length` field.
385#[derive(Debug, Clone, PartialEq, Eq)]
386pub struct DatagramStatusHeader {
387    /// Track alias identifying the subscription.
388    pub track_alias: VarInt,
389    /// Group identifier.
390    pub group_id: VarInt,
391    /// Object identifier within the group.
392    pub object_id: VarInt,
393    /// Publisher priority for delivery ordering.
394    pub publisher_priority: u8,
395    /// Total byte length of extension headers, as it arrived.
396    ///
397    /// Advisory on encode. Every encoder here writes
398    /// `VarInt::from_usize(self.extensions.len())` and never reads this field,
399    /// so a hand-built header whose stated length disagrees with the bytes
400    /// beside it goes on the wire with the derived length and is read back
401    /// consistent. Decoding always sets the two together, so a value that came
402    /// off the wire never disagrees.
403    ///
404    /// It is kept because it is what the peer stated, which is not always
405    /// recoverable from the bytes: a non-minimal varint length encodes the same
406    /// number in more bytes, and a relay that must forward the block unchanged
407    /// has to know which it saw.
408    pub extension_headers_length: VarInt,
409    /// Raw extension bytes (opaque).
410    pub extensions: Vec<u8>,
411    /// Object status code.
412    pub object_status: ObjectStatus,
413}
414
415impl DatagramStatusHeader {
416    /// Encode the datagram status header into the buffer.
417    pub fn encode(&self, buf: &mut impl BufMut) {
418        self.track_alias.encode(buf);
419        self.group_id.encode(buf);
420        self.object_id.encode(buf);
421        buf.put_u8(self.publisher_priority);
422        VarInt::from_usize(self.extensions.len()).encode(buf);
423        encode_extensions(&self.extensions, buf);
424        VarInt::from_usize(self.object_status as usize).encode(buf);
425    }
426
427    /// Encode the datagram status header, refusing an end-of-track object that
428    /// does not end at object zero.
429    ///
430    /// # Errors
431    ///
432    /// [`CodecError::EndOfTrackObjectId`] if an end-of-track status is paired
433    /// with a non-zero Object ID.
434    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
435        check_end_of_track(self.object_id, self.object_status)?;
436        self.encode(buf);
437        Ok(())
438    }
439
440    /// Decode a datagram status header from the buffer.
441    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
442        let track_alias = VarInt::decode(buf)?;
443        let group_id = VarInt::decode(buf)?;
444        let object_id = VarInt::decode(buf)?;
445        if buf.remaining() < 1 {
446            return Err(CodecError::UnexpectedEnd);
447        }
448        let publisher_priority = buf.get_u8();
449        let extension_headers_length = VarInt::decode(buf)?;
450        let extensions = read_extension_bytes(buf, extension_headers_length.into_inner())?;
451        let status_val = VarInt::decode(buf)?.into_inner();
452        let object_status = ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?;
453        check_end_of_track(object_id, object_status)?;
454        Ok(Self {
455            track_alias,
456            group_id,
457            object_id,
458            publisher_priority,
459            extension_headers_length,
460            extensions,
461            object_status,
462        })
463    }
464}
465
466// ============================================================
467// Datagram framing
468// ============================================================
469
470/// One datagram, of whichever shape its type field names.
471///
472/// A MoQT datagram opens with a variable-length integer naming its type, and
473/// that integer is what says which of the layouts above follows it.
474/// Neither [`DatagramHeader`] nor [`DatagramStatusHeader`] reads or writes it, so neither can be handed
475/// the first byte of a datagram a peer sent, and neither produces bytes a peer
476/// can read. This is the entry point that does both.
477///
478/// The payload of a payload-bearing datagram runs to the end of the QUIC
479/// datagram, so it is not part of this value: [`Self::decode`] stops at the end
480/// of the header and leaves the payload in the buffer, and a caller appends the
481/// payload after [`Self::encode`].
482#[derive(Debug, Clone, PartialEq, Eq)]
483pub enum Datagram {
484    /// An object carrying a payload.
485    Payload(DatagramHeader),
486    /// An object stating a status, with no payload.
487    Status(DatagramStatusHeader),
488}
489
490impl Datagram {
491    /// Whether this datagram states an Object Status instead of carrying a
492    /// payload.
493    pub fn is_status(&self) -> bool {
494        matches!(self, Self::Status(_))
495    }
496
497    /// The type field this value writes.
498    pub fn datagram_type(&self) -> StreamType {
499        match self {
500            Self::Payload(_) => StreamType::Datagram,
501            Self::Status(_) => StreamType::DatagramStatus,
502        }
503    }
504
505    /// Decode a datagram from its first byte, type field included.
506    ///
507    /// Errors with [`CodecError::UnknownDatagramType`] when the leading type is
508    /// one the datagram table does not assign, which this draft answers with a
509    /// close, and with [`CodecError::InvalidField`] for the stream types, which
510    /// it does assign but not to a datagram. `datagram_type_error` draws that
511    /// line.
512    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
513        let raw = VarInt::decode(buf)?.into_inner();
514        match StreamType::from_id(raw) {
515            Some(StreamType::Datagram) => Ok(Self::Payload(DatagramHeader::decode(buf)?)),
516            Some(StreamType::DatagramStatus) => {
517                Ok(Self::Status(DatagramStatusHeader::decode(buf)?))
518            }
519            _ => Err(datagram_type_error(raw)),
520        }
521    }
522
523    /// Encode the datagram, type field included.
524    pub fn encode(&self, buf: &mut impl BufMut) {
525        VarInt::from_usize(self.datagram_type() as usize).encode(buf);
526        match self {
527            Self::Payload(header) => header.encode(buf),
528            Self::Status(header) => header.encode(buf),
529        }
530    }
531
532    /// Encode the datagram, refusing a header the framing it names cannot
533    /// carry.
534    ///
535    /// The body is built before anything reaches `buf`, so a refused datagram
536    /// leaves `buf` untouched rather than a type field with no body under it.
537    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
538        let mut body = Vec::with_capacity(64);
539        match self {
540            Self::Payload(header) => header.encode_checked(&mut body)?,
541            Self::Status(header) => header.encode_checked(&mut body)?,
542        }
543        VarInt::from_usize(self.datagram_type() as usize).encode(buf);
544        buf.put_slice(&body);
545        Ok(())
546    }
547}
548
549// ============================================================
550// Fetch stream (type 0x05)
551// ============================================================
552
553/// Fetch stream header (follows the stream type varint).
554#[derive(Debug, Clone, PartialEq, Eq)]
555pub struct FetchHeader {
556    /// Subscribe ID this fetch responds to.
557    pub subscribe_id: VarInt,
558}
559
560/// Object within a fetch stream (draft-09).
561///
562/// Uses `extension_headers_length` instead of `extension_count`.
563#[derive(Debug, Clone, PartialEq, Eq)]
564pub struct FetchObjectHeader {
565    /// Group identifier.
566    pub group_id: VarInt,
567    /// Subgroup identifier within the group.
568    pub subgroup_id: VarInt,
569    /// Object identifier within the subgroup.
570    pub object_id: VarInt,
571    /// Publisher priority for delivery ordering.
572    pub publisher_priority: u8,
573    /// Total byte length of extension headers, as it arrived.
574    ///
575    /// Advisory on encode. Every encoder here writes
576    /// `VarInt::from_usize(self.extensions.len())` and never reads this field,
577    /// so a hand-built header whose stated length disagrees with the bytes
578    /// beside it goes on the wire with the derived length and is read back
579    /// consistent. Decoding always sets the two together, so a value that came
580    /// off the wire never disagrees.
581    ///
582    /// It is kept because it is what the peer stated, which is not always
583    /// recoverable from the bytes: a non-minimal varint length encodes the same
584    /// number in more bytes, and a relay that must forward the block unchanged
585    /// has to know which it saw.
586    pub extension_headers_length: VarInt,
587    /// Raw extension bytes (opaque).
588    pub extensions: Vec<u8>,
589    /// Status of this object.
590    pub object_status: ObjectStatus,
591    /// Length of the object payload in bytes.
592    pub payload_length: VarInt,
593}
594
595impl FetchHeader {
596    /// Encode a fetch stream header including its leading stream-type field,
597    /// so the bytes form the start of a data stream a peer can read.
598    ///
599    /// [`Self::encode`] writes the body alone, which is what a caller wants
600    /// once the stream is already open and what a caller must not use for its
601    /// first write. The read side has had [`Self::decode_stream`] all along,
602    /// so without this the codec could not round-trip its own fetch stream
603    /// through its own reader.
604    pub fn encode_stream(&self, buf: &mut impl BufMut) {
605        VarInt::from_usize(StreamType::Fetch as usize).encode(buf);
606        self.encode(buf);
607    }
608
609    /// Encode the fetch header into the buffer.
610    pub fn encode(&self, buf: &mut impl BufMut) {
611        self.subscribe_id.encode(buf);
612    }
613
614    /// Decode a fetch header from the buffer.
615    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
616        let subscribe_id = VarInt::decode(buf)?;
617        Ok(Self { subscribe_id })
618    }
619
620    /// Decode a fetch header from the start of a data stream, consuming the
621    /// leading stream type varint.
622    ///
623    /// Errors with [`CodecError::UnknownStreamType`] when the stream type is
624    /// one the stream table does not assign, which this draft answers with a
625    /// close, and with [`CodecError::InvalidField`] when it is assigned but is
626    /// not [`StreamType::Fetch`]. `stream_type_error` draws that line.
627    pub fn decode_stream(buf: &mut impl Buf) -> Result<Self, CodecError> {
628        let stream_type = VarInt::decode(buf)?.into_inner();
629        if stream_type != StreamType::Fetch as u64 {
630            return Err(stream_type_error(stream_type));
631        }
632        Self::decode(buf)
633    }
634}
635
636impl FetchObjectHeader {
637    /// Encode the fetch object header into the buffer.
638    pub fn encode(&self, buf: &mut impl BufMut) {
639        self.group_id.encode(buf);
640        self.subgroup_id.encode(buf);
641        self.object_id.encode(buf);
642        buf.put_u8(self.publisher_priority);
643        VarInt::from_usize(self.extensions.len()).encode(buf);
644        encode_extensions(&self.extensions, buf);
645        self.payload_length.encode(buf);
646        if self.payload_length.into_inner() == 0 {
647            VarInt::from_usize(self.object_status as usize).encode(buf);
648        }
649    }
650
651    /// Encode the header, refusing a status the framing cannot carry.
652    ///
653    /// Section 8.4.3 puts the Object Status field on the wire only when the
654    /// Object Payload Length is zero, and Section 8.1.1.1 says "Any object
655    /// with a status code other than zero MUST have an empty payload". A
656    /// non-zero status paired with a non-zero payload length therefore has no
657    /// encoding at all: [`Self::encode`] drops the status and the peer reads an
658    /// ordinary object, which is a different object from the one the caller
659    /// described. This refuses instead.
660    ///
661    /// The datagram types on this draft already refuse the same pairing. These
662    /// two did not, and they are the ones a publisher writes on every stream.
663    ///
664    /// # Errors
665    ///
666    /// [`CodecError::InvalidField`] if a non-zero Object Status is paired with
667    /// a non-zero Object Payload Length.
668    pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
669        check_end_of_track(self.object_id, self.object_status)?;
670        if self.payload_length.into_inner() != 0 && self.object_status as usize != 0 {
671            return Err(CodecError::InvalidField);
672        }
673        self.encode(buf);
674        Ok(())
675    }
676
677    /// Decode a fetch object header from the buffer.
678    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
679        let group_id = VarInt::decode(buf)?;
680        let subgroup_id = VarInt::decode(buf)?;
681        let object_id = VarInt::decode(buf)?;
682        if buf.remaining() < 1 {
683            return Err(CodecError::UnexpectedEnd);
684        }
685        let publisher_priority = buf.get_u8();
686        let extension_headers_length = VarInt::decode(buf)?;
687        let extensions = read_extension_bytes(buf, extension_headers_length.into_inner())?;
688        let payload_length = VarInt::decode(buf)?;
689        let object_status = if payload_length.into_inner() == 0 {
690            let status_val = VarInt::decode(buf)?.into_inner();
691            ObjectStatus::from_u64(status_val).ok_or(CodecError::InvalidField)?
692        } else {
693            ObjectStatus::Normal
694        };
695        check_end_of_track(object_id, object_status)?;
696        Ok(Self {
697            group_id,
698            subgroup_id,
699            object_id,
700            publisher_priority,
701            extension_headers_length,
702            extensions,
703            object_status,
704            payload_length,
705        })
706    }
707}