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}