moqtap_codec/draft16/data_stream.rs
1//! Draft-16 data stream header encoding and decoding.
2//!
3//! Draft-16 subgroup type-byte flag layout:
4//! - `& 0x01`: extensions present on objects
5//! - `& 0x06`: SUBGROUP_ID_MODE. Section 10.4.2: "The SUBGROUP_ID_MODE field
6//! (bits 1-2, mask 0x06) is a two-bit field that determines the encoding of
7//! the Subgroup ID. To extract this value, perform a bitwise AND with mask
8//! 0x06 and right-shift by 1 bit". Mode 0 puts no Subgroup ID on the wire
9//! and it is zero, mode 1 takes it from the first object, mode 2 reads it as
10//! a field after the Group ID, and mode 3 is reserved
11//! - `& 0x08`: end-of-group marker
12//! - `& 0x20`: no publisher_priority (0x30+ types)
13//!
14//! Because the mode is one field and not two independent bits, the three
15//! carriers are mutually exclusive: a Type setting both `0x02` and `0x04` names
16//! the reserved mode, not two carriers at once. This module described those bits
17//! separately and read them separately, which let a single Type answer to two
18//! carriers.
19//!
20//! Draft-15 spells the same three carriers out of the same two bits, as a pair
21//! of table columns rather than a named field, and the two drafts end up
22//! admitting the same twenty-four Types — draft-16 by excluding the reserved
23//! mode from the ranges `0x10`-`0x1F` and `0x30`-`0x3F`, draft-15 by listing
24//! them.
25//!
26//! Draft-16 datagram type-byte flag layout:
27//! - `0x01`: extensions present (byte-length-prefixed blob)
28//! - `0x02`: end-of-group
29//! - `0x04`: no object_id (object_id = 0 implied)
30//! - `0x08`: default priority (priority omitted, inherited)
31//! - `0x20`: status datagram (carries object_status instead of payload)
32//!
33//! Draft-16 fetch objects are framed differently again: a per-object
34//! Serialization Flags varint decides which of Group ID, Subgroup ID, Object ID,
35//! Priority and Extensions are on the wire, and a field left off is taken from
36//! the object before it. See [`FetchObjectHeader`] for the flag layout and
37//! [`FetchObjectReader`] for the inheritance. A fetch object carries no Object
38//! Status at all — draft-16 Section 10.2.1.1 puts that field on subscription
39//! deliveries only.
40//!
41//! Extension headers in draft-16 are byte-length-prefixed opaque blobs
42//! (not count-prefixed as in draft-14).
43
44use super::types::ObjectStatus;
45use crate::error::CodecError;
46use crate::object_properties::{ObjectProperty, PropertyValue};
47use crate::varint::VarInt;
48use bytes::{Buf, BufMut};
49
50/// Advance `buf` past `len` bytes without copying them.
51fn skip(buf: &mut impl Buf, len: u64) -> Result<(), CodecError> {
52 let len = usize::try_from(len).map_err(|_| CodecError::UnexpectedEnd)?;
53 if buf.remaining() < len {
54 return Err(CodecError::UnexpectedEnd);
55 }
56 buf.advance(len);
57 Ok(())
58}
59
60/// Turn a wire Object Status code into the status draft-16 gives it, refusing
61/// any code the draft does not assign.
62///
63/// Draft-16 Section 10.2.1.1 lists the codes an object may carry and says any
64/// other value SHOULD be treated as a protocol error and the session closed
65/// with a PROTOCOL_VIOLATION. Every place this module reads a status runs the
66/// wire code through here. Draft-16 is where the set narrowed to three: 0x1,
67/// which drafts 07-15 assign to Object Does Not Exist, is refused here.
68///
69/// [`SubgroupObject::object_status`] and [`DatagramHeader::object_status`]
70/// then store the [`ObjectStatus`] this returns rather than the raw code, so
71/// the refusal is not something a future decode site can forget: those fields
72/// cannot hold an unassigned value at all, in either direction, and the encode
73/// paths need no check of their own.
74/// [`SubgroupObjectMeta::status`] deliberately keeps the raw code — it is a
75/// decode-only view that never feeds an encoder — but it is filtered through
76/// here too, so the two readers agree byte for byte on what parses.
77fn decoded_status(code: u64) -> Result<ObjectStatus, CodecError> {
78 ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)
79}
80
81/// Whether an Object at `status` is allowed to carry `extensions_len` bytes of
82/// extension headers.
83///
84/// Draft-16 Section 10.2.1.2: "Any Object with status Normal can have extension
85/// headers", with a reference to Section 2.5 inside the sentence, and "If an
86/// endpoint receives extension headers on Objects with status that is not
87/// Normal, it MUST close the session with a PROTOCOL_VIOLATION."
88///
89/// Draft-16 is the draft where the status set narrowed to Normal, End of Group
90/// and End of Track, so the rule reaches two codes rather than the single
91/// "Object Does Not Exist" earlier drafts name, draft-15 Section 10.2.1.1 among
92/// them. It reaches every carrier that can announce a status: an Object on a
93/// subgroup stream and a status datagram. An Object on a fetch stream carries
94/// no status field on this draft — Section 10.2.1.1 puts the field on
95/// subscription deliveries only — so it is the one carrier that cannot break
96/// the rule.
97///
98/// So this is `false` for exactly one shape: a non-empty extension block on an
99/// Object whose status is not Normal. An Object with no extensions is fine at
100/// any status, and an Object at Normal may carry any extensions.
101///
102/// `status` is taken as a raw code so the two subgroup readers can share this:
103/// one resolves the status into [`ObjectStatus`] and the other keeps the wire
104/// code. `None` means the Object carried a payload, which is Normal by
105/// definition and always permitted.
106///
107/// [`SubgroupObjectReader::read_object`] and
108/// [`SubgroupObjectReader::write_object`] refuse an Object this answers `false`
109/// for; [`SubgroupObjectReader::read_object_meta`] frames it.
110fn extensions_permitted_at(status: Option<u64>, extensions_len: u64) -> bool {
111 match status {
112 None => true,
113 Some(code) => extensions_len == 0 || code == ObjectStatus::Normal.as_u64(),
114 }
115}
116
117// ── Subgroup and datagram Type validation ─────────────────────
118//
119// Both Types are read as a single byte rather than as a varint. Every Type
120// draft-16 assigns to either is below 0x40, which is exactly the one-byte range
121// of the varint encoding this draft uses, and both forms independently forbid
122// bits 6 and 7 — so a first byte that begins a longer varint is a Type the
123// draft rules out anyway.
124//
125// Reading a full varint and narrowing it to `u8` was the defect: a Type of
126// 0x110, spelled as the two-byte varint 0x41 0x10, truncated to 0x10 and was
127// accepted as a valid subgroup header. An out-of-range Type aliased onto a
128// valid one instead of being refused, and the stream behind it was parsed under
129// framing its sender never asked for.
130
131/// Bit 4, which every subgroup header Type sets.
132const SUBGROUP_BASE_BIT: u8 = 0x10;
133/// Bits 6 and 7, which the subgroup header's 0b00X1XXXX form leaves clear.
134///
135/// Narrower than the equivalent on draft-19, whose form is 0b0XX1XXXX and which
136/// therefore admits 0x50..0x5F and 0x70..0x7F as well. Draft-16 Section 10.4.2
137/// names only "the ranges 0x10..0x1F and 0x30..0x3F".
138const SUBGROUP_FORM_FORBIDDEN_BITS: u8 = 0xC0;
139/// Bits 1-2, the SUBGROUP_ID_MODE field.
140const SUBGROUP_ID_MODE_MASK: u8 = 0x06;
141/// The SUBGROUP_ID_MODE value draft-16 reserves, once the mask is applied and
142/// the field shifted down.
143const SUBGROUP_ID_MODE_RESERVED: u8 = 0b11;
144
145/// Refuse a subgroup header Type value draft-16 Section 10.4.2 lists as invalid.
146///
147/// The section gives two lists and says of both that an endpoint receiving a
148/// stream header with such a Type MUST close the session with a
149/// PROTOCOL_VIOLATION:
150///
151/// - "Type values with SUBGROUP_ID_MODE set to 0b11: 0x16, 0x17, 0x1E, 0x1F,
152/// 0x36, 0x37, 0x3E, 0x3F. This mode is reserved for future use."
153/// - "Type values that do not match the form 0b00X1XXXX (i.e., Type values
154/// outside the ranges 0x10..0x1F and 0x30..0x3F, or values where bit 4 is
155/// not set)."
156///
157/// The reserved mode is worth separating from a merely unassigned code point,
158/// because it is not decodable rather than merely unknown. The other three modes
159/// each say whether a Subgroup ID field follows the Group ID; 0b11 says nothing,
160/// so a decoder has to guess, and a wrong guess shifts every later field by the
161/// width of that varint. This module read those eight Types as though an
162/// explicit Subgroup ID were present, which turned a header the draft says to
163/// reject into objects with plausible, wrong contents.
164///
165/// Which of the two lists a Type failed is what `stream_type_error` reports,
166/// and the answer is not the same error: one is a Type no table assigns and the
167/// other is a Type this draft assigns a form to and then forbids.
168fn validate_subgroup_type(raw: u64) -> Result<(), CodecError> {
169 if subgroup_type_is_valid(raw) {
170 Ok(())
171 } else {
172 Err(stream_type_error(raw))
173 }
174}
175
176/// The unidirectional stream Type draft-16 Section 10.4.4 gives a fetch stream.
177const FETCH_STREAM_TYPE: u64 = 0x05;
178
179/// Refuse a Type field spelled in more than one byte, before anything narrows
180/// it to a byte.
181///
182/// Returns `Ok(None)` when the next Type is a single byte and the caller should
183/// read it itself, `Ok(Some(err))` when it is wider and `refusal` has named the
184/// failure, and `Err` only when the buffer does not hold the whole field yet.
185///
186/// Every Type draft-16 assigns is below 0x40 and so occupies one byte under
187/// this draft's variable-length integer encoding. A wider spelling is therefore
188/// one of two things, and neither may be read as a header: a Type this draft
189/// does not assign, or a non-minimal spelling of one it does. The second is the
190/// dangerous one — narrowing a two-byte 0x4001 to its low octet turns it into
191/// the assigned Type 0x01, so a peer could name any Type it liked and have it
192/// parsed as another.
193///
194/// The full varint is decoded before `refusal` sees it, so the refusal reports
195/// the number the field actually carried rather than its first byte.
196fn wide_type_refusal(
197 buf: &mut impl Buf,
198 refusal: fn(u64) -> CodecError,
199) -> Result<Option<CodecError>, CodecError> {
200 if !buf.has_remaining() {
201 return Err(CodecError::UnexpectedEnd);
202 }
203 // Under the draft-16 encoding the top two bits of the first byte give the
204 // field's length, so a first byte below 0x40 is the whole of it.
205 if buf.chunk()[0] < 0x40 {
206 return Ok(None);
207 }
208 let raw = VarInt::decode(buf)?.into_inner();
209 Ok(Some(refusal(raw)))
210}
211
212/// The refusal a datagram Type deserves, as a plain function so
213/// `wide_type_refusal` can take it.
214fn datagram_type_refusal(raw: u64) -> CodecError {
215 match validate_datagram_type(raw) {
216 Ok(()) => CodecError::InvalidField,
217 Err(e) => e,
218 }
219}
220
221/// Whether `raw` is a subgroup Type draft-16 admits: inside the form, and not
222/// the reserved SUBGROUP_ID_MODE.
223fn subgroup_type_is_valid(raw: u64) -> bool {
224 raw <= 0xFF && {
225 let t = raw as u8;
226 t & SUBGROUP_FORM_FORBIDDEN_BITS == 0
227 && t & SUBGROUP_BASE_BIT != 0
228 && (t & SUBGROUP_ID_MODE_MASK) >> 1 != SUBGROUP_ID_MODE_RESERVED
229 }
230}
231
232/// Whether `raw` sits inside the subgroup form but names the reserved
233/// SUBGROUP_ID_MODE — the first of Section 10.4.2's two lists.
234fn subgroup_type_is_reserved_mode(raw: u64) -> bool {
235 raw <= 0xFF && {
236 let t = raw as u8;
237 t & SUBGROUP_FORM_FORBIDDEN_BITS == 0
238 && t & SUBGROUP_BASE_BIT != 0
239 && (t & SUBGROUP_ID_MODE_MASK) >> 1 == SUBGROUP_ID_MODE_RESERVED
240 }
241}
242
243/// Which failure a leading unidirectional stream Type that is not the one a
244/// reader wants is.
245///
246/// Draft-16 states two rules about such a Type and answers both with a close,
247/// and telling them apart is the whole job of this function.
248///
249/// Section 10 is about the table: "An endpoint that receives an unknown stream
250/// or datagram type MUST close the session." A Type the table does not assign
251/// is [`CodecError::UnknownStreamType`]. The table has two entries — 0x05 for
252/// FETCH_HEADER and the subgroup form — because draft-16 carries its control
253/// messages on a bidirectional stream and defines no padding stream, so
254/// FETCH_HEADER is the only assigned value a subgroup reader can be handed.
255///
256/// Section 10.4.2 is about the subgroup form specifically, and the eight Types
257/// inside it that name the reserved SUBGROUP_ID_MODE. Those are not unknown —
258/// the form is assigned and the draft lists the values outright — but they are
259/// unreadable, and they are [`CodecError::InvalidStreamTypeValue`].
260///
261/// A fetch stream at the subgroup reader is neither. The value is one this
262/// draft defines, the disagreement is with the reader that was called, and the
263/// session survives it.
264fn stream_type_error(raw: u64) -> CodecError {
265 if raw == FETCH_STREAM_TYPE || subgroup_type_is_valid(raw) {
266 CodecError::InvalidField
267 } else if subgroup_type_is_reserved_mode(raw) {
268 CodecError::InvalidStreamTypeValue {
269 raw,
270 detail: "its SUBGROUP_ID_MODE is 0b11, which this draft reserves",
271 }
272 } else {
273 CodecError::UnknownStreamType(raw)
274 }
275}
276
277/// The END_OF_GROUP bit of a datagram Type.
278const DATAGRAM_END_OF_GROUP_BIT: u8 = 0x02;
279/// The STATUS bit of a datagram Type.
280const DATAGRAM_STATUS_BIT: u8 = 0x20;
281/// Bits 4, 6 and 7, which the datagram's 0b00X0XXXX form leaves clear.
282const DATAGRAM_FORM_FORBIDDEN_BITS: u8 = 0xD0;
283
284/// Refuse a datagram Type value draft-16 Section 10.3.1 lists as invalid.
285///
286/// The section gives two lists and says of both that an endpoint receiving a
287/// datagram with such a Type MUST close the session with a PROTOCOL_VIOLATION:
288///
289/// - "Type values with both the STATUS bit (0x20) and END_OF_GROUP bit (0x02)
290/// set: 0x22, 0x23, 0x26, 0x27, 0x2A, 0x2B, 0x2E, 0x2F. An object status
291/// message cannot signal end of group."
292/// - "Type values that do not match the form 0b00X0XXXX (i.e., Type values
293/// outside the ranges 0x00..0x0F and 0x20..0x2F)."
294///
295/// That leaves 24 of the 256 byte values valid. Without this check every other
296/// one is accepted, and bit 4 is the bit that separates a datagram Type from a
297/// subgroup stream header Type — so an unchecked datagram Type could name a
298/// stream header and be parsed as a datagram anyway.
299///
300/// The two lists are not the same failure. A Type outside the form is one no
301/// table assigns, and so is [`CodecError::UnknownDatagramType`]; a Type inside
302/// the form setting STATUS and END_OF_GROUP together is one this draft names
303/// and forbids, and so is [`CodecError::InvalidDatagramTypeValue`].
304fn validate_datagram_type(raw: u64) -> Result<(), CodecError> {
305 if datagram_type_is_valid(raw) {
306 Ok(())
307 } else if datagram_type_is_status_end_of_group(raw) {
308 Err(CodecError::InvalidDatagramTypeValue {
309 raw,
310 detail: "it sets both the STATUS bit and the END_OF_GROUP bit",
311 })
312 } else {
313 Err(CodecError::UnknownDatagramType(raw))
314 }
315}
316
317/// Whether `raw` is a datagram Type draft-16 admits.
318fn datagram_type_is_valid(raw: u64) -> bool {
319 raw <= 0xFF && {
320 let t = raw as u8;
321 t & DATAGRAM_FORM_FORBIDDEN_BITS == 0
322 && !(t & DATAGRAM_STATUS_BIT != 0 && t & DATAGRAM_END_OF_GROUP_BIT != 0)
323 }
324}
325
326/// Whether `raw` sits inside the datagram form but sets STATUS and END_OF_GROUP
327/// together — the first of Section 10.3.1's two lists.
328fn datagram_type_is_status_end_of_group(raw: u64) -> bool {
329 raw <= 0xFF && {
330 let t = raw as u8;
331 t & DATAGRAM_FORM_FORBIDDEN_BITS == 0
332 && t & DATAGRAM_STATUS_BIT != 0
333 && t & DATAGRAM_END_OF_GROUP_BIT != 0
334 }
335}
336
337#[derive(Debug, Clone, PartialEq, Eq)]
338pub struct SubgroupHeader {
339 pub header_type: u8,
340 pub track_alias: VarInt,
341 pub group_id: VarInt,
342 pub subgroup_id: VarInt,
343 pub publisher_priority: Option<u8>,
344}
345
346impl SubgroupHeader {
347 pub fn has_extensions(&self) -> bool {
348 self.header_type & 0x01 != 0
349 }
350
351 /// Whether SUBGROUP_ID_MODE is 1: the Subgroup ID is the first object's
352 /// Object ID and is not transmitted on the wire.
353 ///
354 /// Reads the two bits as the one field Section 10.4.2 defines, rather than
355 /// as the single bit `0x02`. The single-bit reading answered `true` for a
356 /// Type naming the reserved mode 3, where `0x02` and `0x04` are both set —
357 /// so this predicate and [`Self::has_explicit_subgroup_id`] answered `true`
358 /// together, for a state one two-bit field cannot be in.
359 pub fn subgroup_id_from_first_object(&self) -> bool {
360 (self.header_type & SUBGROUP_ID_MODE_MASK) >> 1 == 1
361 }
362
363 /// Whether SUBGROUP_ID_MODE is 2: an explicit Subgroup ID field follows the
364 /// Group ID.
365 ///
366 /// [`Self::decode`] refuses the reserved mode before reading any field, so
367 /// on a decoded header this agrees with the single-bit `0x04` reading it
368 /// replaces. The two differ only on a header built by hand, which is the
369 /// only way to hold a reserved-mode Type at all.
370 pub fn has_explicit_subgroup_id(&self) -> bool {
371 (self.header_type & SUBGROUP_ID_MODE_MASK) >> 1 == 2
372 }
373
374 pub fn has_end_of_group(&self) -> bool {
375 self.header_type & 0x08 != 0
376 }
377
378 pub fn has_priority(&self) -> bool {
379 self.header_type & 0x20 == 0
380 }
381
382 /// Serialize the header exactly as its Type byte describes it.
383 ///
384 /// Infallible, and so willing to write a Type draft-16 Section 10.4.2 tells
385 /// an endpoint to reject — including a reserved-mode Type this module's own
386 /// [`Self::decode`] refuses to read back. Prefer [`Self::encode_checked`],
387 /// which refuses those Types instead.
388 ///
389 /// A reserved-mode Type gets no Subgroup ID field, because mode 3 defines
390 /// none: the draft says what the other three modes put on the wire and
391 /// nothing about this one. Writing the field there was an artifact of
392 /// reading `0x04` on its own, and it disagreed with every neighbouring
393 /// draft — 15, 17, 18 and 19 all leave it off.
394 ///
395 /// Every field is driven by the Type byte, because that is the only thing
396 /// [`Self::decode`] and the peer have to go on — the Priority byte
397 /// included. Driving that one off `publisher_priority` being `Some`
398 /// instead desyncs the stream: a Type with DEFAULT_PRIORITY (`0x20`) set
399 /// beside a `Some` writes a stray byte the reader takes for the first
400 /// Object's Object ID Delta, and a Type with the bit clear beside a `None`
401 /// leaves the reader taking that Delta for the priority. Either way every
402 /// later field shifts — which is precisely the disagreement
403 /// [`Self::encode_checked`]'s doc comment says this pair must not have.
404 ///
405 /// A `None` under a Type whose bit is clear therefore writes 128 rather
406 /// than dropping the byte. Draft-16 Section 11.1.1.1: "A subscription has
407 /// Publisher Priorty 128 if this extension is omitted", so 128 is this
408 /// draft's own name for an unstated priority and not an invented filler.
409 /// Drafts 17-23 write the same value in the same place. Use
410 /// [`Self::encode_checked`] to be told about the disagreement rather than
411 /// having it resolved silently.
412 pub fn encode(&self, buf: &mut impl BufMut) {
413 VarInt::from_usize(self.header_type as usize).encode(buf);
414 self.track_alias.encode(buf);
415 self.group_id.encode(buf);
416 if self.has_explicit_subgroup_id() {
417 self.subgroup_id.encode(buf);
418 }
419 if self.has_priority() {
420 buf.put_u8(self.publisher_priority.unwrap_or(128));
421 }
422 }
423
424 /// Serialize the header, refusing a Type value draft-16 forbids.
425 ///
426 /// Errors with [`CodecError::InvalidField`] for exactly the Types Section
427 /// 10.4.2 lists as invalid — the same set [`Self::decode`] refuses — before
428 /// any byte is written, so a refused header leaves `buf` untouched.
429 ///
430 /// The check belongs on this side as well because the two halves would
431 /// otherwise disagree about which streams exist: a reserved-mode header
432 /// written by [`Self::encode`] cannot be read back by [`Self::decode`], and
433 /// a codec used to rewrite captured traffic would emit a stream it could not
434 /// then parse.
435 ///
436 /// That argument covers the Priority byte too. `publisher_priority`
437 /// disagreeing with the Type byte's DEFAULT_PRIORITY bit (`0x20`) is the
438 /// same failure one field along: a header that does not mean what it says,
439 /// whose bytes the peer reads under framing its writer never asked for.
440 /// [`Self::encode`] does not desync the stream over it — the byte follows
441 /// the Type byte — but resolving the disagreement is not the same as being
442 /// told about it, and the caller who set the wrong half is the only one who
443 /// can fix it. Same check, same reason, as draft-15 Section 10.4.2's
444 /// encoder.
445 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
446 validate_subgroup_type(self.header_type as u64)?;
447 if self.has_priority() != self.publisher_priority.is_some() {
448 return Err(CodecError::InvalidField);
449 }
450 self.encode(buf);
451 Ok(())
452 }
453
454 /// Decode a subgroup header, Type field included.
455 ///
456 /// A Type spelled in more than one byte is refused before it is narrowed,
457 /// by `wide_type_refusal`. Every Type draft-16 assigns fits a single
458 /// byte, so a wider spelling is either a Type this draft does not have or a
459 /// non-minimal spelling of one it does, and both have to be refused rather
460 /// than truncated into a Type that happens to be valid.
461 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
462 if let Some(err) = wide_type_refusal(buf, stream_type_error)? {
463 return Err(err);
464 }
465 let header_type = buf.get_u8();
466 validate_subgroup_type(header_type as u64)?;
467 let track_alias = VarInt::decode(buf)?;
468 let group_id = VarInt::decode(buf)?;
469 // `validate_subgroup_type` has already refused the reserved mode, so
470 // the field is read for mode 2 and for nothing else. Reading the mask
471 // rather than the `0x04` bit makes that independent of the check above
472 // instead of contingent on it.
473 let subgroup_id = if (header_type & SUBGROUP_ID_MODE_MASK) >> 1 == 2 {
474 VarInt::decode(buf)?
475 } else {
476 VarInt::from_usize(0)
477 };
478 let publisher_priority = if header_type & 0x20 == 0 {
479 if buf.remaining() < 1 {
480 return Err(CodecError::UnexpectedEnd);
481 }
482 Some(buf.get_u8())
483 } else {
484 None
485 };
486 Ok(Self { header_type, track_alias, group_id, subgroup_id, publisher_priority })
487 }
488}
489
490/// One object within a draft-16 subgroup stream with its Object ID
491/// already resolved from the delta encoding. See
492/// [`SubgroupObjectReader`] for stateful encode/decode.
493#[derive(Debug, Clone, PartialEq, Eq)]
494pub struct SubgroupObject {
495 pub object_id: VarInt,
496 /// Raw extension-header bytes, excluding the byte-length prefix that
497 /// precedes them on the wire. Empty when the stream header does not
498 /// set the extensions-present bit, or when the block is present but
499 /// zero-length. Opaque: [`SubgroupObjectReader::write_object`] re-emits
500 /// the prefix and these bytes verbatim.
501 pub extension_headers: Vec<u8>,
502 pub payload_length: VarInt,
503 /// Object status; `Some` when `payload_length == 0`.
504 ///
505 /// The wire field is a varint, so it can carry any value up to 2^62-1;
506 /// draft-16 Section 10.2.1.1 assigns three of them and says a peer SHOULD
507 /// treat the rest as a protocol error. This field is typed to the assigned
508 /// set, so it refuses to hold the codes the draft leaves unassigned —
509 /// including 0x1, Object Does Not Exist, which drafts 07-15 assign and
510 /// draft-16 dropped. That makes the refusal a property of the struct
511 /// rather than of any one code path: an encoder cannot be handed a status
512 /// the draft does not define, and does not have to check.
513 ///
514 /// `None` on a zero-length object means the same thing as
515 /// [`ObjectStatus::Normal`] and encodes as it; the wire field is not
516 /// optional once `payload_length` is zero.
517 pub object_status: Option<ObjectStatus>,
518 pub payload: Vec<u8>,
519}
520
521impl SubgroupObject {
522 /// The status this Object resolves to.
523 ///
524 /// The wire carries a status field only on a zero-length Object, so an
525 /// Object holding bytes is [`ObjectStatus::Normal`] whatever
526 /// [`Self::object_status`] says — draft-16 Section 10.2.1.1: "This status is
527 /// implicit for any non-zero length object."
528 pub fn status(&self) -> ObjectStatus {
529 if self.payload_length.into_inner() == 0 {
530 self.object_status.unwrap_or(ObjectStatus::Normal)
531 } else {
532 ObjectStatus::Normal
533 }
534 }
535
536 /// Whether this Object's status is allowed to carry the extension headers
537 /// it has.
538 ///
539 /// `false` for exactly one shape: a non-empty extension block on an Object
540 /// whose status is not [`ObjectStatus::Normal`]. See
541 /// `extensions_permitted_at` for the rule.
542 ///
543 /// [`SubgroupObjectReader::read_object`] and
544 /// [`SubgroupObjectReader::write_object`] refuse an object this answers
545 /// `false` for; [`SubgroupObjectReader::read_object_meta`] frames it.
546 pub fn extensions_permitted(&self) -> bool {
547 extensions_permitted_at(
548 self.object_status.map(|s| s.as_u64()),
549 self.extension_headers.len() as u64,
550 )
551 }
552}
553
554/// Whether an Object carrying a given status is allowed a non-empty payload.
555///
556/// Draft-16 Section 10.2.1.1 states the rule in one sentence — "Any object with
557/// a status code other than zero MUST have an empty payload" — and the same
558/// section makes Normal (0x0) "implicit for any non-zero length object". So the
559/// permission follows from the status code alone, with no payload length in
560/// hand, which is what makes it worth naming: a caller that has only an
561/// object's framing can ask whether the bytes it is about to forward are
562/// allowed there at all.
563#[derive(Debug, Clone, Copy, PartialEq, Eq)]
564pub enum PayloadPermission {
565 /// The status permits a payload but does not require one: a zero-length
566 /// object with such a status is well formed, and draft-16's encodings can
567 /// spell it.
568 Permitted,
569 /// An object with such a status has an empty payload, and one carrying
570 /// bytes is malformed.
571 Forbidden,
572}
573
574impl PayloadPermission {
575 /// `true` for [`PayloadPermission::Permitted`].
576 pub fn permits(self) -> bool {
577 matches!(self, PayloadPermission::Permitted)
578 }
579}
580
581/// The framing of one draft-16 subgroup object, without its payload.
582///
583/// Produced by [`SubgroupObjectReader::read_object_meta`] for callers that
584/// forward an object's bytes verbatim and never inspect the payload.
585#[derive(Debug, Clone, Copy, PartialEq, Eq)]
586pub struct SubgroupObjectMeta {
587 /// Resolved absolute Object ID.
588 pub object_id: u64,
589 /// Byte length of the extension-header block's contents, excluding its
590 /// length prefix.
591 pub extension_headers_len: u64,
592 /// Declared payload length. Zero when `status` is `Some`.
593 pub payload_length: u64,
594 /// Object status wire code, present only when the payload is empty.
595 ///
596 /// Kept as the raw code, unlike [`SubgroupObject::object_status`]: a meta
597 /// is produced by [`SubgroupObjectReader::read_object_meta`] and is never
598 /// an encode input, and a relay that reads a status on one draft may hand
599 /// it to a draft that numbers the same value differently. The code is
600 /// still one draft-16 assigns — `read_object_meta` refuses the others.
601 pub status: Option<u64>,
602 /// Total bytes this object occupies on the wire, prefix fields included.
603 pub wire_len: u64,
604}
605
606impl SubgroupObjectMeta {
607 /// Whether draft-16 allows this object a non-empty payload, or `None` when
608 /// the answer does not exist.
609 ///
610 /// Draft-16 Section 10.2.1.1 gives the rule for the three codes it assigns:
611 /// Normal (0x0) permits a payload, and "any object with a status code other
612 /// than zero MUST have an empty payload", which covers End of Group (0x3)
613 /// and End of Track (0x4). No status at all means the object carried bytes
614 /// — the status field is on the wire only when `payload_length` is zero —
615 /// and the same section makes Normal "implicit for any non-zero length
616 /// object", so that case answers [`PayloadPermission::Permitted`] as well.
617 ///
618 /// `None` is reserved for a `status` the draft does not assign. A meta read
619 /// off the wire cannot hold one — [`SubgroupObjectReader::read_object_meta`]
620 /// refuses those bytes — but the field is public and holds the raw code
621 /// precisely so a value carried over from another draft's numbering can sit
622 /// in it. Draft-16 states no payload rule for a code it never assigned, and
623 /// there is no safe direction to guess in: answering `Permitted` would wave
624 /// through a payload the peer may have to reject, and `Forbidden` would
625 /// discard one the peer may accept.
626 pub fn payload_permission(&self) -> Option<PayloadPermission> {
627 match self.status {
628 None => Some(PayloadPermission::Permitted),
629 Some(code) => match ObjectStatus::from_u64(code)? {
630 ObjectStatus::Normal => Some(PayloadPermission::Permitted),
631 ObjectStatus::EndOfGroup | ObjectStatus::EndOfTrack => {
632 Some(PayloadPermission::Forbidden)
633 }
634 },
635 }
636 }
637
638 /// Whether this Object's status is allowed to carry the extension block it
639 /// declares.
640 ///
641 /// The meta form of [`SubgroupObject::extensions_permitted`], answering
642 /// from the declared block length rather than its contents, so a relay that
643 /// forwards bytes verbatim can report the violation without copying the
644 /// block.
645 pub fn extensions_permitted(&self) -> bool {
646 extensions_permitted_at(self.status, self.extension_headers_len)
647 }
648}
649
650/// Prior Group ID Gap, Extension Header Type 0x3C (Section 11.3).
651pub const PRIOR_GROUP_ID_GAP: u64 = 0x3C;
652/// Prior Object ID Gap, Extension Header Type 0x3E (Section 11.4).
653pub const PRIOR_OBJECT_ID_GAP: u64 = 0x3E;
654
655/// The Prior Group ID Gap and Prior Object ID Gap one Object carries.
656#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
657pub struct PriorGaps {
658 /// Prior Group ID Gap, if present.
659 pub group: Option<u64>,
660 /// Prior Object ID Gap, if present.
661 pub object: Option<u64>,
662}
663
664impl PriorGaps {
665 /// Read the two gaps from an Object's parsed extension headers, refusing
666 /// the conditions one Object shows on its own.
667 ///
668 /// Section 11.3 makes the Track malformed when "An Object contains more
669 /// than one instance of Prior Group ID Gap" and when "An Object has a Prior
670 /// Group ID Gap larger than the Group ID". Section 11.4 says the same two
671 /// for Prior Object ID Gap against the Object ID. Each is
672 /// [`CodecError::MalformedTrack`].
673 pub fn from_properties(
674 properties: &[ObjectProperty<'_>],
675 group_id: u64,
676 object_id: u64,
677 ) -> Result<Self, CodecError> {
678 let mut gaps = Self::default();
679 collect_gaps(properties, &mut gaps)?;
680 if let Some(gap) = gaps.group.filter(|&gap| gap > group_id) {
681 return Err(CodecError::MalformedTrack {
682 detail: format!("Prior Group ID Gap {gap} is larger than Group ID {group_id}"),
683 });
684 }
685 if let Some(gap) = gaps.object.filter(|&gap| gap > object_id) {
686 return Err(CodecError::MalformedTrack {
687 detail: format!("Prior Object ID Gap {gap} is larger than Object ID {object_id}"),
688 });
689 }
690 Ok(gaps)
691 }
692}
693
694fn collect_gaps(properties: &[ObjectProperty<'_>], gaps: &mut PriorGaps) -> Result<(), CodecError> {
695 for property in properties {
696 let slot = match (&property.value, property.property_type) {
697 (PropertyValue::Immutable { properties, .. }, _) => {
698 collect_gaps(properties, gaps)?;
699 continue;
700 }
701 (PropertyValue::Varint(value), PRIOR_GROUP_ID_GAP) => {
702 (&mut gaps.group, *value, "Group")
703 }
704 (PropertyValue::Varint(value), PRIOR_OBJECT_ID_GAP) => {
705 (&mut gaps.object, *value, "Object")
706 }
707 _ => continue,
708 };
709 let (field, value, which) = slot;
710 if field.replace(value).is_some() {
711 return Err(CodecError::MalformedTrack {
712 detail: format!("an Object contains more than one Prior {which} ID Gap"),
713 });
714 }
715 }
716 Ok(())
717}
718
719/// The Prior Group ID Gap and Prior Object ID Gap conditions of
720/// draft-16 Section 11.3 and draft-16 Section 11.4 that one fetch stream
721/// shows.
722///
723/// [`PriorGaps::from_properties`] refuses what one Object shows. Across the
724/// stream this refuses "A Group contains more than one Object with different
725/// values for Prior Group ID Gap" for the Objects of one Group, and "An
726/// endpoint receives an Object with a Group ID within a previously communicated
727/// gap" and its Prior Object ID Gap twin, for the Group and the Object before
728/// on the stream. It does not refuse "An endpoint receives an Object with a
729/// Prior Group ID Gap covering an Object it previously received" or its Prior
730/// Object ID Gap twin: draft-16 Section 2.1 says of an Object known to exist,
731/// "From this state, it can transition to not existing, but not vice versa", so
732/// whether such a gap makes the Track malformed is ambiguous.
733/// draft-16 Section 9.16.3: "Within each group, objects are sent in Object ID
734/// order; subgroup ID is not used for ordering."
735///
736/// [`FetchObjectReader::resolve`] applies it to each Object it reads.
737///
738/// End of Range indicators are not Objects; the caller passes Objects only.
739#[derive(Debug, Clone, PartialEq, Eq)]
740// Boxed so that the fetch readers holding one stay small.
741pub struct FetchGaps(Box<crate::prior_gaps::FetchGapTracker>);
742
743impl Default for FetchGaps {
744 fn default() -> Self {
745 Self::new()
746 }
747}
748
749impl FetchGaps {
750 /// Track the Objects of one fetch stream.
751 pub fn new() -> Self {
752 Self(Box::new(crate::prior_gaps::FetchGapTracker::new(
753 crate::prior_gaps::GapRules::Draft16To17,
754 )))
755 }
756
757 /// Check the next Object on the stream, from its resolved Group ID and
758 /// Object ID and its extension block, refusing a gap condition with
759 /// [`CodecError::MalformedTrack`] and a block that does not parse as
760 /// [`crate::object_properties::parse_object_properties`] does. Records the
761 /// Object only when it passes.
762 pub fn check(
763 &mut self,
764 group_id: u64,
765 object_id: u64,
766 extension_block: &[u8],
767 ) -> Result<PriorGaps, CodecError> {
768 let properties = crate::object_properties::parse_object_properties(
769 crate::version::DraftVersion::Draft16,
770 extension_block,
771 )?;
772 let gaps = PriorGaps::from_properties(&properties, group_id, object_id)?;
773 self.0.check(group_id, object_id, gaps.group.as_slice(), gaps.object)?;
774 Ok(gaps)
775 }
776}
777
778/// Stateful reader/writer for draft-16 subgroup objects. Mirrors the
779/// draft-15 semantics (delta-encoded object IDs and header-typed
780/// extension presence).
781#[derive(Debug, Clone)]
782pub struct SubgroupObjectReader {
783 extensions_present: bool,
784 prev_object_id: Option<u64>,
785 group_id: u64,
786 group_gap: Option<u64>,
787}
788
789impl SubgroupObjectReader {
790 pub fn new(header: &SubgroupHeader) -> Self {
791 Self {
792 extensions_present: header.has_extensions(),
793 prev_object_id: None,
794 group_id: header.group_id.into_inner(),
795 group_gap: None,
796 }
797 }
798
799 /// Decode the next object whole, payload included.
800 ///
801 /// Refuses, with [`CodecError::InvalidField`], extension headers on an
802 /// object whose status is not Normal: Section 10.2.1.2, "If an endpoint
803 /// receives extension headers on Objects with status that is not Normal, it
804 /// MUST close the session with a PROTOCOL_VIOLATION."
805 /// [`Self::read_object_meta`] frames such an object without judging it.
806 ///
807 /// Refuses, with [`CodecError::MalformedTrack`], the Prior Group ID Gap and
808 /// Prior Object ID Gap conditions of Section 11.3 and Section 11.4 this stream
809 /// shows: those [`PriorGaps::from_properties`] refuses, and "A Group contains
810 /// more than one Object with different values for Prior Group ID Gap". It does
811 /// not refuse "An endpoint receives an Object with a Prior Object ID Gap
812 /// covering an Object it previously received": draft-16 Section 2.1 says of an
813 /// Object known to exist, "From this state, it can transition to not existing,
814 /// but not vice versa", so whether such a gap makes the Track malformed is
815 /// ambiguous. [`Self::read_object_meta`] does not read extension headers and
816 /// applies none of them.
817 pub fn read_object(&mut self, buf: &mut impl Buf) -> Result<SubgroupObject, CodecError> {
818 let delta = VarInt::decode(buf)?.into_inner();
819 // The first object's field is its absolute Object ID; every later
820 // object encodes the gap to its predecessor, biased by one because
821 // two objects on a subgroup stream cannot share an ID.
822 let object_id_val = match self.prev_object_id {
823 None => delta,
824 Some(prev) => prev
825 .checked_add(1)
826 .and_then(|v| v.checked_add(delta))
827 .ok_or(CodecError::InvalidField)?,
828 };
829 self.prev_object_id = Some(object_id_val);
830 let object_id = VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?;
831
832 // Draft-16: extensions are a byte-length-prefixed opaque blob.
833 let extension_headers = if self.extensions_present {
834 let ext_len = VarInt::decode(buf)?.into_inner() as usize;
835 crate::types::read_bytes(buf, ext_len)?
836 } else {
837 Vec::new()
838 };
839
840 let payload_length_vi = VarInt::decode(buf)?;
841 let payload_length_val = payload_length_vi.into_inner() as usize;
842 let (object_status, payload) = if payload_length_val == 0 {
843 let status = decoded_status(VarInt::decode(buf)?.into_inner())?;
844 (Some(status), Vec::new())
845 } else {
846 let payload = crate::types::read_bytes(buf, payload_length_val)?;
847 (None, payload)
848 };
849 let properties = crate::object_properties::parse_object_properties(
850 crate::version::DraftVersion::Draft16,
851 &extension_headers,
852 )?;
853 let gaps = PriorGaps::from_properties(&properties, self.group_id, object_id_val)?;
854 self.check_stream_gaps(gaps)?;
855 let object = SubgroupObject {
856 object_id,
857 extension_headers,
858 payload_length: payload_length_vi,
859 object_status,
860 payload,
861 };
862 if !object.extensions_permitted() {
863 return Err(CodecError::InvalidField);
864 }
865 Ok(object)
866 }
867
868 fn check_stream_gaps(&mut self, gaps: PriorGaps) -> Result<(), CodecError> {
869 if let Some(gap) = gaps.group {
870 match self.group_gap {
871 Some(seen) if seen != gap => {
872 return Err(CodecError::MalformedTrack {
873 detail: format!(
874 "two Objects in Group {} carry different Prior Group ID Gap values",
875 self.group_id
876 ),
877 });
878 }
879 _ => self.group_gap = Some(gap),
880 }
881 }
882 Ok(())
883 }
884
885 /// Decode the next object's framing without copying its payload.
886 ///
887 /// Consumes exactly the bytes [`Self::read_object`] consumes and leaves
888 /// the same delta state behind, so the two are interchangeable on a
889 /// given stream.
890 pub fn read_object_meta(
891 &mut self,
892 buf: &mut impl Buf,
893 ) -> Result<SubgroupObjectMeta, CodecError> {
894 let start = buf.remaining();
895 let delta = VarInt::decode(buf)?.into_inner();
896 let object_id_val = match self.prev_object_id {
897 None => delta,
898 Some(prev) => prev
899 .checked_add(1)
900 .and_then(|v| v.checked_add(delta))
901 .ok_or(CodecError::InvalidField)?,
902 };
903 self.prev_object_id = Some(object_id_val);
904 let object_id =
905 VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?.into_inner();
906
907 let extension_headers_len = if self.extensions_present {
908 let ext_len = VarInt::decode(buf)?.into_inner();
909 skip(buf, ext_len)?;
910 ext_len
911 } else {
912 0
913 };
914
915 let payload_length = VarInt::decode(buf)?.into_inner();
916 let status = if payload_length == 0 {
917 let code = VarInt::decode(buf)?.into_inner();
918 decoded_status(code)?;
919 Some(code)
920 } else {
921 skip(buf, payload_length)?;
922 None
923 };
924 Ok(SubgroupObjectMeta {
925 object_id,
926 extension_headers_len,
927 payload_length,
928 status,
929 wire_len: (start - buf.remaining()) as u64,
930 })
931 }
932
933 /// Serialize an object, producing the correct delta encoding.
934 ///
935 /// Errors with [`CodecError::InvalidField`] when `object.object_id` is
936 /// not strictly greater than the previously written object's ID, since
937 /// no valid delta exists for that case.
938 ///
939 /// A zero `payload_length` makes this a status object, and the status
940 /// field is then mandatory on the wire: an absent
941 /// [`SubgroupObject::object_status`] is written as
942 /// [`ObjectStatus::Normal`]. The code written is always one draft-16
943 /// assigns, because the field cannot hold any other, so the bytes this
944 /// produces are always bytes [`SubgroupObjectReader::read_object`] accepts.
945 ///
946 /// Errors with [`CodecError::InvalidField`] when `payload_length` is not
947 /// exactly `payload.len()`. The declared length is written ahead of the
948 /// payload, so a mismatch is a frame [`Self::read_object`] cannot parse
949 /// and one no caller could fix by appending bytes.
950 pub fn write_object(
951 &mut self,
952 object: &SubgroupObject,
953 buf: &mut impl BufMut,
954 ) -> Result<(), CodecError> {
955 // A declared length that disagrees with the payload framed under it
956 // produces bytes no reader can parse and no caller can repair: the
957 // length is already on the wire ahead of the payload. Checked before
958 // anything is written, so a refused object leaves `buf` untouched
959 // rather than half an object the next read would run into.
960 //
961 // Zero is not "an empty payload" here; it is the marker that puts a
962 // status code where the payload would go, so an object carrying bytes
963 // under it is asking for two framings at once.
964 let declared = object.payload_length.into_inner();
965 if declared != object.payload.len() as u64 {
966 return Err(CodecError::InvalidField);
967 }
968
969 // Extension headers on a non-Normal status are a frame `read_object`
970 // refuses (Section 10.2.1.2), so they are not written either.
971 if !object.extensions_permitted() {
972 return Err(CodecError::InvalidField);
973 }
974
975 let oid = object.object_id.into_inner();
976 let delta = match self.prev_object_id {
977 None => oid,
978 Some(prev) => oid
979 .checked_sub(prev)
980 .and_then(|v| v.checked_sub(1))
981 .ok_or(CodecError::InvalidField)?,
982 };
983 VarInt::from_u64(delta).map_err(|_| CodecError::InvalidField)?.encode(buf);
984 if self.extensions_present {
985 let ext_len = object.extension_headers.len();
986 VarInt::from_usize(ext_len).encode(buf);
987 buf.put_slice(&object.extension_headers);
988 }
989 object.payload_length.encode(buf);
990 if object.payload_length.into_inner() == 0 {
991 // Zero-length means a status object, and the status is not
992 // optional on the wire; an unset one is Normal.
993 let status = object.object_status.unwrap_or(ObjectStatus::Normal);
994 VarInt::from_usize(status.as_u64() as usize).encode(buf);
995 } else {
996 buf.put_slice(&object.payload);
997 }
998 self.prev_object_id = Some(oid);
999 Ok(())
1000 }
1001}
1002
1003#[derive(Debug, Clone, PartialEq, Eq)]
1004pub struct DatagramHeader {
1005 pub datagram_type: u8,
1006 pub track_alias: VarInt,
1007 pub group_id: VarInt,
1008 pub object_id: VarInt,
1009 /// Publisher priority — `None` when the DEFAULT_PRIORITY flag is set and
1010 /// the priority is inherited from the subscription's control message.
1011 pub publisher_priority: Option<u8>,
1012 /// Opaque extension-headers blob (only when flag 0x01 is set).
1013 pub extension_headers: Vec<u8>,
1014 /// Object status (only when the `0x20` status flag is set).
1015 ///
1016 /// The wire field is a varint and can carry any value up to 2^62-1;
1017 /// draft-16 Section 10.2.1.1 assigns three of them and says a peer SHOULD
1018 /// treat the rest as a protocol error. This field is typed to the assigned
1019 /// set, so it cannot hold 0x1, 0x2 or anything from 0x5 up —
1020 /// [`Self::encode`] therefore needs no check and cannot emit a datagram
1021 /// that [`Self::decode`] would reject.
1022 ///
1023 /// `None` while the status flag is set encodes as
1024 /// [`ObjectStatus::Normal`]: once the flag is set the field is present on
1025 /// the wire, so there is nothing for `None` to mean but the default.
1026 pub object_status: Option<ObjectStatus>,
1027}
1028
1029impl DatagramHeader {
1030 pub fn has_extensions(&self) -> bool {
1031 self.datagram_type & 0x01 != 0
1032 }
1033
1034 pub fn is_end_of_group(&self) -> bool {
1035 self.datagram_type & 0x02 != 0
1036 }
1037
1038 pub fn has_object_id(&self) -> bool {
1039 self.datagram_type & 0x04 == 0
1040 }
1041
1042 /// When set, publisher_priority is omitted on the wire and inherited
1043 /// from the subscription / control-message context.
1044 pub fn has_default_priority(&self) -> bool {
1045 self.datagram_type & 0x08 != 0
1046 }
1047
1048 pub fn is_status(&self) -> bool {
1049 self.datagram_type & 0x20 != 0
1050 }
1051
1052 /// Whether this datagram's status is allowed to carry the extension headers
1053 /// it has.
1054 ///
1055 /// The same rule the subgroup form obeys — draft-16 Section 10.3.1 builds
1056 /// the datagram's Extensions field out of the structure Section 10.2.1.2
1057 /// defines, and that section is where the rule sits. See
1058 /// [`SubgroupObject::extensions_permitted`] for why [`Self::decode`]
1059 /// reports this instead of refusing it.
1060 ///
1061 /// The block is on the wire only when the type byte sets the EXTENSIONS
1062 /// bit, so contents held here with the bit clear are not written and do not
1063 /// count against the rule.
1064 pub fn extensions_permitted(&self) -> bool {
1065 extensions_permitted_at(
1066 self.object_status.map(|s| s.as_u64()),
1067 if self.has_extensions() { self.extension_headers.len() as u64 } else { 0 },
1068 )
1069 }
1070
1071 /// Encode the datagram header, refusing a status the framing cannot carry.
1072 ///
1073 /// A datagram states a status only when its type byte sets the STATUS bit
1074 /// (0x20). With the bit clear there is no status field on the wire, so an
1075 /// `object_status` of anything but [`ObjectStatus::Normal`] has nowhere to
1076 /// go: [`Self::encode`] drops it, and the datagram parses back as an
1077 /// ordinary payload object. An End of Group marker written that way does
1078 /// not arrive late or malformed — it does not arrive at all, and the
1079 /// receiver sees a normal object in its place.
1080 ///
1081 /// Draft-16 Section 10.3.1 puts the framing side plainly — "The STATUS bit
1082 /// (0x20) indicates whether the datagram contains an Object Status or
1083 /// Object Payload" — and Section 10.2.1.1 the conformance side: "Any object
1084 /// with a status code other than zero MUST have an empty payload." Between
1085 /// them there is no datagram that carries a non-zero status and a payload,
1086 /// so the pair being refused here is not one this encoder merely declines
1087 /// to spell.
1088 ///
1089 /// [`ObjectStatus::Normal`] with the bit clear is not that case and is
1090 /// accepted. It is the status the encoding elides for every datagram that
1091 /// carries a payload, so stating it asks for exactly the bytes leaving it
1092 /// out asks for, and nothing is lost.
1093 ///
1094 /// Errors with [`CodecError::InvalidField`] on the lossy combination,
1095 /// before any byte is written, so a refused header leaves `buf` untouched.
1096 ///
1097 /// Three further refusals, each of them a datagram this codec would
1098 /// otherwise emit and then decline to read back:
1099 ///
1100 /// * A Type value Section 10.3.1 lists as invalid. See
1101 /// `validate_datagram_type`.
1102 /// * An EXTENSIONS bit set over an empty extension block. Section 10.3.1:
1103 /// "If an endpoint receives a datagram with the EXTENSIONS bit set and an
1104 /// Extension Headers Length of 0, it MUST close the session with a
1105 /// PROTOCOL_VIOLATION." [`Self::encode`] writes the length prefix from the
1106 /// block's own length, so it spells exactly the datagram the peer must
1107 /// close the session over. This is a datagram rule only: on a subgroup
1108 /// stream the same section says "Objects with no extensions set Extension
1109 /// Headers Length to 0", because the Type byte is fixed for the whole
1110 /// stream and an Object with no extensions has no other way to say so.
1111 /// * Extensions on an Object whose status is not Normal (Section 10.2.1.2).
1112 pub fn encode_checked(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1113 validate_datagram_type(self.datagram_type as u64)?;
1114 if !self.is_status() && matches!(self.object_status, Some(s) if s != ObjectStatus::Normal) {
1115 return Err(CodecError::InvalidField);
1116 }
1117 if self.has_extensions() && self.extension_headers.is_empty() {
1118 return Err(CodecError::InvalidField);
1119 }
1120 // Section 10.2.1.2. [`Self::decode`] reports this rather than refusing
1121 // it, because the datagram it describes is well framed and a codec that
1122 // could not read one could not reproduce a capture containing it.
1123 // Writing one is the other direction and has no such excuse: a
1124 // conforming peer answers with a PROTOCOL_VIOLATION, so emitting one
1125 // costs the session and not merely the datagram.
1126 if !self.extensions_permitted() {
1127 return Err(CodecError::InvalidField);
1128 }
1129 self.encode(buf);
1130 Ok(())
1131 }
1132
1133 /// Encode the datagram header to `buf`.
1134 ///
1135 /// When the type byte sets the status flag the status field is written
1136 /// unconditionally, defaulting to [`ObjectStatus::Normal`]. Omitting it
1137 /// would truncate the datagram: [`Self::decode`] reads a status whenever
1138 /// the flag is set, and answers [`CodecError::UnexpectedEnd`] when the
1139 /// bytes stop first.
1140 ///
1141 /// The type byte is taken as the authority on framing, which is what makes
1142 /// this infallible — and what makes it lossy when the struct disagrees with
1143 /// itself. An `object_status` set while the type byte leaves the STATUS bit
1144 /// clear is discarded here without a word. Prefer [`Self::encode_checked`],
1145 /// which refuses that combination instead of resolving it.
1146 pub fn encode(&self, buf: &mut impl BufMut) {
1147 VarInt::from_usize(self.datagram_type as usize).encode(buf);
1148 self.track_alias.encode(buf);
1149 self.group_id.encode(buf);
1150 if self.has_object_id() {
1151 self.object_id.encode(buf);
1152 }
1153 if !self.has_default_priority() {
1154 buf.put_u8(self.publisher_priority.unwrap_or(128));
1155 }
1156 if self.has_extensions() {
1157 VarInt::from_usize(self.extension_headers.len()).encode(buf);
1158 buf.put_slice(&self.extension_headers);
1159 }
1160 if self.is_status() {
1161 let status = self.object_status.unwrap_or(ObjectStatus::Normal);
1162 VarInt::from_usize(status.as_u64() as usize).encode(buf);
1163 }
1164 }
1165
1166 /// Decode a datagram header, Type field included.
1167 ///
1168 /// [`Self::decode_object`] is the read of a whole datagram, and applies
1169 /// the Object rules this header-only read does not.
1170 ///
1171 /// A Type spelled in more than one byte is refused first, for the reason
1172 /// given on [`SubgroupHeader::decode`].
1173 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1174 if let Some(err) = wide_type_refusal(buf, datagram_type_refusal)? {
1175 return Err(err);
1176 }
1177 let datagram_type = buf.get_u8();
1178 validate_datagram_type(datagram_type as u64)?;
1179 let track_alias = VarInt::decode(buf)?;
1180 let group_id = VarInt::decode(buf)?;
1181 let object_id =
1182 if datagram_type & 0x04 == 0 { VarInt::decode(buf)? } else { VarInt::from_usize(0) };
1183 let publisher_priority = if datagram_type & 0x08 != 0 {
1184 None
1185 } else {
1186 if buf.remaining() < 1 {
1187 return Err(CodecError::UnexpectedEnd);
1188 }
1189 Some(buf.get_u8())
1190 };
1191 let extension_headers = if datagram_type & 0x01 != 0 {
1192 let ext_len = VarInt::decode(buf)?.into_inner() as usize;
1193 // A datagram whose type says extensions are present must actually carry
1194 // some: receiving one with an Extension Headers Length of 0 closes the
1195 // session. The opposite holds on a subgroup stream, where the type byte is
1196 // fixed for the whole stream and an object with no extensions has no other
1197 // way to say so, which is why this check belongs to the datagram readers
1198 // alone.
1199 if ext_len == 0 {
1200 return Err(CodecError::InvalidField);
1201 }
1202 crate::types::read_bytes(buf, ext_len)?
1203 } else {
1204 Vec::new()
1205 };
1206 let object_status = if datagram_type & 0x20 != 0 {
1207 Some(decoded_status(VarInt::decode(buf)?.into_inner())?)
1208 } else {
1209 None
1210 };
1211 Ok(Self {
1212 datagram_type,
1213 track_alias,
1214 group_id,
1215 object_id,
1216 publisher_priority,
1217 extension_headers,
1218 object_status,
1219 })
1220 }
1221
1222 /// Decode one whole datagram: the header, then the payload that runs to the
1223 /// end of `buf`, which must hold exactly one transport datagram.
1224 ///
1225 /// Unlike [`Self::decode`], this refuses an extension block that does not
1226 /// parse, as [`crate::object_properties::parse_object_properties`] does,
1227 /// and, with [`CodecError::MalformedTrack`], the Prior Group ID Gap and
1228 /// Prior Object ID Gap conditions one Object shows (draft-16 Section 11.3,
1229 /// draft-16 Section 11.4), through [`PriorGaps::from_properties`].
1230 ///
1231 /// It also refuses bytes after a status datagram's Object Status, with
1232 /// [`CodecError::PayloadNotPermitted`]. draft-16 Section 10.3.1, of the
1233 /// STATUS bit: "When set to 1, the Object Status field is present and there
1234 /// is no Object Payload."
1235 pub fn decode_object(buf: &mut impl Buf) -> Result<(Self, Vec<u8>), CodecError> {
1236 let header = Self::decode(buf)?;
1237 let properties = crate::object_properties::parse_object_properties(
1238 crate::version::DraftVersion::Draft16,
1239 &header.extension_headers,
1240 )?;
1241 PriorGaps::from_properties(
1242 &properties,
1243 header.group_id.into_inner(),
1244 header.object_id.into_inner(),
1245 )?;
1246 let payload = crate::types::read_bytes(buf, buf.remaining())?;
1247 if let Some(status) = header.object_status {
1248 if !payload.is_empty() {
1249 return Err(CodecError::PayloadNotPermitted {
1250 status: status.as_u64(),
1251 len: payload.len(),
1252 detail: "its type states a status in place of a payload",
1253 });
1254 }
1255 }
1256 Ok((header, payload))
1257 }
1258}
1259
1260#[derive(Debug, Clone, PartialEq, Eq)]
1261pub struct FetchHeader {
1262 pub request_id: VarInt,
1263}
1264
1265impl FetchHeader {
1266 pub fn encode(&self, buf: &mut impl BufMut) {
1267 VarInt::from_usize(FETCH_STREAM_TYPE as usize).encode(buf);
1268 self.request_id.encode(buf);
1269 }
1270
1271 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1272 let stream_type = VarInt::decode(buf)?.into_inner();
1273 if stream_type != FETCH_STREAM_TYPE {
1274 return Err(stream_type_error(stream_type));
1275 }
1276 let request_id = VarInt::decode(buf)?;
1277 Ok(Self { request_id })
1278 }
1279}
1280
1281/// The two Serialization Flags values draft-16 gives a meaning of their own
1282/// instead of reading as a bit field, from Section 10.4.4 Table 4.
1283///
1284/// The bits of Serialization Flags are flags only "when less than 128"; Table 4
1285/// assigns two larger values, and of everything else at or above 128 the
1286/// section says "Any other value is a PROTOCOL_VIOLATION". That is why
1287/// [`FetchObjectHeader::decode`] refuses an unassigned large value outright
1288/// rather than carrying it through as an unknown flag word — there is no
1289/// framing to parse behind it.
1290///
1291/// Both markers stand in for a span of Objects rather than one: Section 10.4.4.2
1292/// reads them as covering every Location between the previous serialized
1293/// Object, if any, and this one, inclusive.
1294#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1295#[repr(u64)]
1296pub enum FetchEndOfRange {
1297 /// End of Non-Existent Range (0x8C): every Object in the covered span is
1298 /// known not to exist. Section 10.4.4.2 adds that a publisher SHOULD NOT
1299 /// use this except to split a span it will not serialize into the part
1300 /// known absent and the part whose status is unknown.
1301 NonExistent = 0x8c,
1302 /// End of Unknown Range (0x10C): the status of every Object in the covered
1303 /// span is unknown.
1304 Unknown = 0x10c,
1305}
1306
1307impl FetchEndOfRange {
1308 /// Both values Table 4 assigns, in ascending wire order.
1309 ///
1310 /// This is exactly the set [`FetchEndOfRange::from_u64`] accepts above the
1311 /// flag range.
1312 pub const ALL: &[FetchEndOfRange] = &[FetchEndOfRange::NonExistent, FetchEndOfRange::Unknown];
1313
1314 /// The Table 4 meaning of `v`, or `None` when `v` is a flag word (anything
1315 /// below 128) or a large value the table leaves unassigned.
1316 ///
1317 /// The two answers `None` covers are not the same thing, and callers must
1318 /// not merge them: a flag word is ordinary framing, an unassigned large
1319 /// value is a protocol violation. [`FetchObjectHeader::decode`] separates
1320 /// them by testing the 128 boundary itself.
1321 pub fn from_u64(v: u64) -> Option<Self> {
1322 match v {
1323 0x8c => Some(FetchEndOfRange::NonExistent),
1324 0x10c => Some(FetchEndOfRange::Unknown),
1325 _ => None,
1326 }
1327 }
1328
1329 /// Return the wire value.
1330 pub fn as_u64(self) -> u64 {
1331 self as u64
1332 }
1333}
1334
1335/// How an Object on a draft-16 fetch stream states its Subgroup ID.
1336///
1337/// The two least significant bits of Serialization Flags form this field;
1338/// draft-16 Section 10.4.4.1 Table 5 lists the four readings. Three of them put
1339/// no Subgroup ID on the wire.
1340#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1341pub enum FetchSubgroupMode {
1342 /// 0x00 — the Subgroup ID is zero.
1343 Zero,
1344 /// 0x01 — the Subgroup ID is the prior Object's Subgroup ID.
1345 SameAsPrior,
1346 /// 0x02 — the Subgroup ID is the prior Object's Subgroup ID plus one.
1347 PriorPlusOne,
1348 /// 0x03 — the Subgroup ID field is present on the wire.
1349 Present,
1350}
1351
1352/// One Object as it is framed on a draft-16 fetch stream, before any field is
1353/// resolved against the Object before it.
1354///
1355/// Draft-16 Section 10.4.4 Figure 31 gives the layout: a Serialization Flags
1356/// varint, then Group ID, Subgroup ID, Object ID, Publisher Priority and
1357/// Extensions, each present only when the flags say so, then an Object Payload
1358/// Length and the payload itself. Every optional field here is `Some` exactly
1359/// when its bytes were on the wire, so [`Self::encode`] can put back the same
1360/// bytes [`Self::decode`] took off — including the difference between an absent
1361/// extensions block and a present, zero-length one.
1362///
1363/// The payload is not part of this struct: `payload_length` bytes follow it on
1364/// the stream. That mirrors [`SubgroupObjectMeta`], and it is what lets a relay
1365/// that forwards bytes verbatim read the framing without copying the payload.
1366///
1367/// Two things separate this from the fetch object of drafts 07-13. There is no
1368/// Object Status field — Section 10.2.1.1 says the status "is only present in
1369/// objects that are delivered via a SUBSCRIPTION, and is absent in Objects
1370/// delivered via a FETCH" — so a zero `payload_length` is simply an empty
1371/// object, with no status varint behind it. And the fields that are present are
1372/// absolute: the flags decide presence, and Table 5 and Table 6 decide what an
1373/// absent field inherits, but a field that is on the wire carries its own value
1374/// rather than a delta. Draft-16 spells "Object ID Delta" out by name where it
1375/// means one, on subgroup streams; Figure 31 says "Object ID".
1376///
1377/// Resolving the absent fields needs the Object before this one, which a single
1378/// header does not have. [`FetchObjectReader`] carries that state.
1379#[derive(Debug, Clone, PartialEq, Eq)]
1380pub struct FetchObjectHeader {
1381 /// Serialization Flags exactly as they appeared on the wire: a flag word
1382 /// below 128, or one of the [`FetchEndOfRange`] markers.
1383 pub serialization_flags: VarInt,
1384 /// Group ID, when the flags put it on the wire.
1385 pub group_id: Option<VarInt>,
1386 /// Subgroup ID, present only under [`FetchSubgroupMode::Present`].
1387 pub subgroup_id: Option<VarInt>,
1388 /// Object ID, when the flags put it on the wire.
1389 pub object_id: Option<VarInt>,
1390 /// Publisher Priority, when the flags put it on the wire.
1391 pub publisher_priority: Option<u8>,
1392 /// Raw extension-header bytes, excluding the byte-length prefix that
1393 /// precedes them on the wire, and `None` when the flags carry no extensions
1394 /// block at all. Opaque: [`Self::encode`] re-emits the prefix and these
1395 /// bytes verbatim. The block's own shape is the one Section 10.2.1.2
1396 /// defines for every draft-16 object.
1397 pub extensions: Option<Vec<u8>>,
1398 /// Declared Object Payload Length. The payload follows on the stream and is
1399 /// not held here.
1400 pub payload_length: VarInt,
1401}
1402
1403impl FetchObjectHeader {
1404 /// Serialization Flags as a plain integer.
1405 fn flags(&self) -> u64 {
1406 self.serialization_flags.into_inner()
1407 }
1408
1409 /// The Table 4 marker this object is, or `None` when its Serialization
1410 /// Flags are an ordinary flag word.
1411 pub fn end_of_range(&self) -> Option<FetchEndOfRange> {
1412 FetchEndOfRange::from_u64(self.flags())
1413 }
1414
1415 /// Bit 0x40: the Object's Forwarding Preference is Datagram, so it has no
1416 /// Subgroup ID.
1417 ///
1418 /// Draft-16 Section 10.4.4.1 requires the publisher to set this bit for such
1419 /// an Object, says it SHOULD then zero the two least significant bits, and
1420 /// requires the subscriber to ignore them — so a Subgroup ID is not read
1421 /// even when those bits spell [`FetchSubgroupMode::Present`]. Ignoring them
1422 /// is a framing decision, not a cosmetic one: reading a Subgroup ID there
1423 /// would consume a varint that belongs to the next field.
1424 ///
1425 /// Never true for a [`FetchEndOfRange`] marker, whose value is not a flag
1426 /// word.
1427 pub fn is_datagram(&self) -> bool {
1428 self.end_of_range().is_none() && self.flags() & 0x40 != 0
1429 }
1430
1431 /// The Table 5 reading of the two least significant bits.
1432 ///
1433 /// Answers [`FetchSubgroupMode::Zero`] for a [`FetchEndOfRange`] marker and
1434 /// for a Datagram-forwarded Object, neither of which has a Subgroup ID on
1435 /// the wire: Section 10.4.4.2 lists Subgroup ID among the fields an End of
1436 /// Range does not carry, and Section 10.4.4.1 says a Datagram Object has
1437 /// none at all.
1438 pub fn subgroup_mode(&self) -> FetchSubgroupMode {
1439 if self.end_of_range().is_some() || self.is_datagram() {
1440 return FetchSubgroupMode::Zero;
1441 }
1442 match self.flags() & 0x03 {
1443 0x00 => FetchSubgroupMode::Zero,
1444 0x01 => FetchSubgroupMode::SameAsPrior,
1445 0x02 => FetchSubgroupMode::PriorPlusOne,
1446 _ => FetchSubgroupMode::Present,
1447 }
1448 }
1449
1450 /// Whether a Group ID field is on the wire (bit 0x08).
1451 ///
1452 /// Always true for a [`FetchEndOfRange`] marker: Section 10.4.4.2 says both
1453 /// the Group ID and the Object ID fields are present.
1454 pub fn has_group_id(&self) -> bool {
1455 self.end_of_range().is_some() || self.flags() & 0x08 != 0
1456 }
1457
1458 /// Whether a Subgroup ID field is on the wire, which is
1459 /// [`FetchSubgroupMode::Present`] and nothing else.
1460 pub fn has_subgroup_id(&self) -> bool {
1461 matches!(self.subgroup_mode(), FetchSubgroupMode::Present)
1462 }
1463
1464 /// Whether an Object ID field is on the wire (bit 0x04).
1465 ///
1466 /// Always true for a [`FetchEndOfRange`] marker, per Section 10.4.4.2.
1467 pub fn has_object_id(&self) -> bool {
1468 self.end_of_range().is_some() || self.flags() & 0x04 != 0
1469 }
1470
1471 /// Whether a Publisher Priority byte is on the wire (bit 0x10).
1472 ///
1473 /// Never true for a [`FetchEndOfRange`] marker: Section 10.4.4.2 lists
1474 /// Priority among the fields it does not carry.
1475 pub fn has_priority(&self) -> bool {
1476 self.end_of_range().is_none() && self.flags() & 0x10 != 0
1477 }
1478
1479 /// Whether an Extensions block is on the wire (bit 0x20).
1480 ///
1481 /// Never true for a [`FetchEndOfRange`] marker, per Section 10.4.4.2.
1482 pub fn has_extensions(&self) -> bool {
1483 self.end_of_range().is_none() && self.flags() & 0x20 != 0
1484 }
1485
1486 /// Whether these flags read any field off the Object before this one.
1487 ///
1488 /// Draft-16 Section 10.4.4.1 closes the section on flags with: "If the first
1489 /// Object in the FETCH response uses a flag that references fields in the
1490 /// prior Object, the Subscriber MUST close the session with a
1491 /// PROTOCOL_VIOLATION." Four of the readings do that — an absent Group ID,
1492 /// Object ID or Priority each names the prior Object in Table 5 or Table 6,
1493 /// as do the two middle Subgroup ID modes.
1494 ///
1495 /// [`FetchSubgroupMode::Zero`] does not: it states a value outright. Nor
1496 /// does an absent Extensions block, which means the Object has none rather
1497 /// than the ones before it. A [`FetchEndOfRange`] marker references nothing
1498 /// either — its Group ID and Object ID are always present, and Section
1499 /// 10.4.4.2 gives it no Priority or Extensions to inherit.
1500 ///
1501 /// [`FetchObjectReader::resolve`] refuses the first Object of a stream when
1502 /// this is true of a field it would have to produce a value for. It is
1503 /// exposed separately because the draft's rule is wider than that: it also
1504 /// covers an absent Priority, for which Section 11.1.1.1 supplies a default
1505 /// that makes resolution possible anyway.
1506 pub fn references_prior_object(&self) -> bool {
1507 if self.end_of_range().is_some() {
1508 return false;
1509 }
1510 !self.has_group_id()
1511 || !self.has_object_id()
1512 || !self.has_priority()
1513 || matches!(
1514 self.subgroup_mode(),
1515 FetchSubgroupMode::SameAsPrior | FetchSubgroupMode::PriorPlusOne
1516 )
1517 }
1518
1519 /// Encode the object's framing, refusing a struct that disagrees with its
1520 /// own Serialization Flags.
1521 ///
1522 /// The flags are the authority on which fields are on the wire, so a field
1523 /// that is `Some` while its flag is clear has nowhere to go, and one that is
1524 /// `None` while its flag is set leaves a hole the reader would fill from the
1525 /// bytes of the next field. Either way the result is a stream
1526 /// [`Self::decode`] cannot read back as what was handed in, so both are
1527 /// refused here.
1528 ///
1529 /// Errors with [`CodecError::InvalidField`] on any such disagreement, and on
1530 /// a Serialization Flags value at or above 128 that Table 4 does not assign,
1531 /// before a single byte is written — a refused object leaves `buf` untouched
1532 /// rather than half an object the next read would run into.
1533 ///
1534 /// The payload is not written: `payload_length` bytes of it belong on the
1535 /// stream immediately after these.
1536 pub fn encode(&self, buf: &mut impl BufMut) -> Result<(), CodecError> {
1537 let flags = self.flags();
1538 if flags >= 128 && FetchEndOfRange::from_u64(flags).is_none() {
1539 return Err(CodecError::InvalidField);
1540 }
1541 if self.group_id.is_some() != self.has_group_id()
1542 || self.subgroup_id.is_some() != self.has_subgroup_id()
1543 || self.object_id.is_some() != self.has_object_id()
1544 || self.publisher_priority.is_some() != self.has_priority()
1545 || self.extensions.is_some() != self.has_extensions()
1546 {
1547 return Err(CodecError::InvalidField);
1548 }
1549
1550 self.serialization_flags.encode(buf);
1551 if let Some(group_id) = self.group_id {
1552 group_id.encode(buf);
1553 }
1554 if let Some(subgroup_id) = self.subgroup_id {
1555 subgroup_id.encode(buf);
1556 }
1557 if let Some(object_id) = self.object_id {
1558 object_id.encode(buf);
1559 }
1560 if let Some(priority) = self.publisher_priority {
1561 buf.put_u8(priority);
1562 }
1563 if let Some(extensions) = &self.extensions {
1564 VarInt::from_usize(extensions.len()).encode(buf);
1565 buf.put_slice(extensions);
1566 }
1567 self.payload_length.encode(buf);
1568 Ok(())
1569 }
1570
1571 /// Decode one object's framing, leaving `buf` positioned at its payload.
1572 ///
1573 /// Errors with [`CodecError::InvalidField`] when Serialization Flags is at
1574 /// or above 128 and is not one of the two values Table 4 assigns — draft-16
1575 /// Section 10.4.4 makes any other such value a PROTOCOL_VIOLATION, and there
1576 /// is no way to guess which fields follow it.
1577 pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
1578 let serialization_flags = VarInt::decode(buf)?;
1579 let flags = serialization_flags.into_inner();
1580 if flags >= 128 && FetchEndOfRange::from_u64(flags).is_none() {
1581 return Err(CodecError::InvalidField);
1582 }
1583 // Presence is settled before any field is read, so the field order of
1584 // Figure 31 is followed exactly once: Group ID, Subgroup ID, Object ID,
1585 // Priority, Extensions, Object Payload Length.
1586 let probe = Self {
1587 serialization_flags,
1588 group_id: None,
1589 subgroup_id: None,
1590 object_id: None,
1591 publisher_priority: None,
1592 extensions: None,
1593 payload_length: VarInt::from_usize(0),
1594 };
1595
1596 let group_id = if probe.has_group_id() { Some(VarInt::decode(buf)?) } else { None };
1597 let subgroup_id = if probe.has_subgroup_id() { Some(VarInt::decode(buf)?) } else { None };
1598 let object_id = if probe.has_object_id() { Some(VarInt::decode(buf)?) } else { None };
1599 let publisher_priority = if probe.has_priority() {
1600 if buf.remaining() < 1 {
1601 return Err(CodecError::UnexpectedEnd);
1602 }
1603 Some(buf.get_u8())
1604 } else {
1605 None
1606 };
1607 let extensions = if probe.has_extensions() {
1608 let ext_len = VarInt::decode(buf)?.into_inner() as usize;
1609 Some(crate::types::read_bytes(buf, ext_len)?)
1610 } else {
1611 None
1612 };
1613 let payload_length = VarInt::decode(buf)?;
1614
1615 Ok(Self {
1616 serialization_flags,
1617 group_id,
1618 subgroup_id,
1619 object_id,
1620 publisher_priority,
1621 extensions,
1622 payload_length,
1623 })
1624 }
1625}
1626
1627/// One fetch object's Location and priority, with every field the flags left
1628/// off the wire filled in from the Object before it.
1629#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1630pub struct FetchObjectLocation {
1631 /// Resolved absolute Group ID.
1632 pub group_id: u64,
1633 /// Resolved absolute Subgroup ID, and `None` for an Object whose Forwarding
1634 /// Preference is Datagram — draft-16 Section 10.2.1 omits the Subgroup ID
1635 /// for those, and Section 10.4.4.1 marks them with bit 0x40.
1636 pub subgroup_id: Option<u64>,
1637 /// Resolved absolute Object ID.
1638 pub object_id: u64,
1639 /// Publisher Priority, and `None` when no Object on the stream so far has
1640 /// stated one.
1641 ///
1642 /// That happens only after a [`FetchEndOfRange`] marker, which carries no
1643 /// Priority of its own and leaves the running value alone; any other first
1644 /// Object with no Priority is refused by [`FetchObjectReader::resolve`].
1645 /// Draft-16 Section 11.1.1.1 supplies the fallback for the gap — a
1646 /// subscription has Publisher Priority 128 when DEFAULT PUBLISHER PRIORITY
1647 /// is omitted — and it is left to the caller rather than substituted here,
1648 /// so that *the stream never said* stays distinguishable from "the stream
1649 /// said 128".
1650 pub publisher_priority: Option<u8>,
1651 /// The Table 4 marker this Object is, or `None` for an ordinary Object.
1652 ///
1653 /// When set, this Location is the far end of a span: Section 10.4.4.2 reads
1654 /// every Location between the previously serialized Object, if any, and
1655 /// this one — inclusive — as non-existent or unknown.
1656 pub end_of_range: Option<FetchEndOfRange>,
1657}
1658
1659/// Stateful resolver for the Objects on one draft-16 fetch stream.
1660///
1661/// Draft-16 Section 10.4.4.1 lets an Object leave its Group ID, Subgroup ID,
1662/// Object ID or Priority off the wire and take the value from the Object before
1663/// it, so the fields of a fetch object are only meaningful in stream order.
1664/// This carries that running state; [`FetchObjectHeader`] carries only what the
1665/// bytes said.
1666///
1667/// Unlike a subgroup stream, nothing here is a delta: a field that is on the
1668/// wire replaces the running value outright, and a field that is absent either
1669/// repeats it or steps it by one, as Table 5 and Table 6 say.
1670#[derive(Debug, Clone, Default)]
1671pub struct FetchObjectReader {
1672 group_id: Option<u64>,
1673 subgroup_id: Option<u64>,
1674 object_id: Option<u64>,
1675 publisher_priority: Option<u8>,
1676 /// The Prior gap conditions the stream's Objects show.
1677 gaps: FetchGaps,
1678}
1679
1680impl FetchObjectReader {
1681 /// A reader positioned before the first Object of a fetch stream, with no
1682 /// prior Object to inherit from.
1683 pub fn new() -> Self {
1684 Self::default()
1685 }
1686
1687 /// Resolve one Object's Location against the Object before it, advancing
1688 /// the running state.
1689 ///
1690 /// Errors with [`CodecError::InvalidField`] when the first Object of the
1691 /// stream leaves out a Group ID or an Object ID, or names the prior
1692 /// Object's Subgroup ID: draft-16 Section 10.4.4.1 makes a first Object
1693 /// that "uses a flag that references fields in the prior Object" a
1694 /// PROTOCOL_VIOLATION, and for these three fields there is no value to
1695 /// produce even if it were not. An absent Priority is the same violation by
1696 /// the draft's letter and is reported by
1697 /// [`FetchObjectHeader::references_prior_object`], but it is resolvable —
1698 /// the Location simply carries no priority — so it is not refused here.
1699 ///
1700 /// Errors with [`CodecError::InvalidField`] when a Subgroup ID one greater
1701 /// than the prior Object's would not fit a varint.
1702 ///
1703 /// A [`FetchEndOfRange`] marker takes part in the running state like any
1704 /// other Object: Section 10.4.4.2 gives it a Location, and it is a
1705 /// "serialized Object" in the same section's words, so the Object after it
1706 /// inherits its Group ID and steps from its Object ID.
1707 ///
1708 /// Refuses, with [`CodecError::MalformedTrack`], the Prior gap conditions an
1709 /// Object shows alone or after the Objects before it on the stream, as
1710 /// [`FetchGaps`] does, and an extension block that does not parse as
1711 /// [`crate::object_properties::parse_object_properties`] does.
1712 pub fn resolve(
1713 &mut self,
1714 header: &FetchObjectHeader,
1715 ) -> Result<FetchObjectLocation, CodecError> {
1716 let at = self.resolve_frame(header)?;
1717 if at.end_of_range.is_none() {
1718 self.gaps.check(
1719 at.group_id,
1720 at.object_id,
1721 header.extensions.as_deref().unwrap_or_default(),
1722 )?;
1723 }
1724 Ok(at)
1725 }
1726
1727 /// [`Self::resolve`] without the Prior gap conditions and without reading
1728 /// inside the extension block, for the facade's frame and meta reads, which
1729 /// a forwarder uses.
1730 pub(crate) fn resolve_frame(
1731 &mut self,
1732 header: &FetchObjectHeader,
1733 ) -> Result<FetchObjectLocation, CodecError> {
1734 let group_id = match header.group_id {
1735 Some(v) => v.into_inner(),
1736 None => self.group_id.ok_or(CodecError::InvalidField)?,
1737 };
1738
1739 let subgroup_id = if header.is_datagram() {
1740 None
1741 } else {
1742 match header.subgroup_mode() {
1743 FetchSubgroupMode::Zero => Some(0),
1744 FetchSubgroupMode::SameAsPrior => {
1745 Some(self.subgroup_id.ok_or(CodecError::InvalidField)?)
1746 }
1747 FetchSubgroupMode::PriorPlusOne => {
1748 let prior = self.subgroup_id.ok_or(CodecError::InvalidField)?;
1749 let next = prior.checked_add(1).ok_or(CodecError::InvalidField)?;
1750 VarInt::from_u64(next).map_err(|_| CodecError::InvalidField)?;
1751 Some(next)
1752 }
1753 FetchSubgroupMode::Present => {
1754 Some(header.subgroup_id.ok_or(CodecError::InvalidField)?.into_inner())
1755 }
1756 }
1757 };
1758
1759 let object_id = match header.object_id {
1760 Some(v) => v.into_inner(),
1761 None => {
1762 let prior = self.object_id.ok_or(CodecError::InvalidField)?;
1763 let next = prior.checked_add(1).ok_or(CodecError::InvalidField)?;
1764 VarInt::from_u64(next).map_err(|_| CodecError::InvalidField)?;
1765 next
1766 }
1767 };
1768
1769 // An Object that states no Priority leaves the running one alone, which
1770 // is what "Priority is the prior Object's Priority" asks for and also
1771 // what an End of Range marker — which has no Priority of its own —
1772 // needs.
1773 if let Some(priority) = header.publisher_priority {
1774 self.publisher_priority = Some(priority);
1775 }
1776 self.group_id = Some(group_id);
1777 // A Datagram-forwarded Object has no Subgroup ID at all, so it leaves no
1778 // "prior Object's Subgroup ID" behind: the next Object naming one is
1779 // refused rather than reaching past it to the Object before.
1780 self.subgroup_id = subgroup_id;
1781 self.object_id = Some(object_id);
1782
1783 Ok(FetchObjectLocation {
1784 group_id,
1785 subgroup_id,
1786 object_id,
1787 publisher_priority: self.publisher_priority,
1788 end_of_range: header.end_of_range(),
1789 })
1790 }
1791}
1792
1793/// Re-encodes resolved fetch Objects onto one FETCH stream.
1794///
1795/// The exact inverse of [`FetchObjectReader`], and it exists for one caller:
1796/// something that has read a stream and is writing a different stream from the
1797/// same Objects. Draft-16 Section 10.4.4.1 lets an Object leave out its Group
1798/// ID, Object ID, Subgroup ID and Priority and take the prior Object's, so
1799/// removing an Object changes what the Objects after it are read against: a
1800/// field the survivor left off has to appear, and a flag bit with it.
1801///
1802/// Every field draft-16 puts on the wire here is the absolute value rather than
1803/// a difference — the deltas arrive at draft-18. What is stateful is the
1804/// *omission*, and that is enough to make removal a re-encode.
1805///
1806/// # Why this is not a general encoder
1807///
1808/// Every Object it writes came off a stream, so the caller holds the Object's
1809/// own [`FetchObjectHeader`] beside its resolved
1810/// [`FetchObjectLocation`]. The header is used as the preference: wherever the
1811/// original shape still says the same thing against the new predecessor it is
1812/// kept, so a stream with nothing removed is reproduced byte for byte.
1813#[derive(Debug, Clone, Default)]
1814pub struct FetchObjectWriter {
1815 group_id: Option<u64>,
1816 subgroup_id: Option<u64>,
1817 object_id: Option<u64>,
1818 publisher_priority: Option<u8>,
1819}
1820
1821impl FetchObjectWriter {
1822 /// A writer positioned before the first Object of a fetch stream, with no
1823 /// prior Object for anything to be written against.
1824 pub fn new() -> Self {
1825 Self::default()
1826 }
1827
1828 /// The header that encodes `location` against everything written so far,
1829 /// starting from the shape `original` arrived in.
1830 ///
1831 /// Does not advance the writer — [`Self::write_object_header`] is the call
1832 /// that does both.
1833 ///
1834 /// # Errors
1835 ///
1836 /// [`CodecError::InvalidField`] for an Object with neither a Subgroup ID
1837 /// nor the Datagram bit, and where a value does not fit a variable-length
1838 /// integer.
1839 pub fn header_for(
1840 &self,
1841 original: &FetchObjectHeader,
1842 location: &FetchObjectLocation,
1843 ) -> Result<FetchObjectHeader, CodecError> {
1844 // A marker's Group ID and Object ID are on the wire by definition —
1845 // `has_group_id` and `has_object_id` are true for both marker values —
1846 // and it carries no Subgroup ID, Priority or Extensions.
1847 if original.end_of_range().is_some() {
1848 return Ok(FetchObjectHeader {
1849 serialization_flags: original.serialization_flags,
1850 group_id: Some(VarInt::from_u64(location.group_id)?),
1851 subgroup_id: None,
1852 object_id: Some(VarInt::from_u64(location.object_id)?),
1853 publisher_priority: None,
1854 extensions: None,
1855 payload_length: original.payload_length,
1856 });
1857 }
1858
1859 let group_id = if !original.has_group_id() && self.group_id == Some(location.group_id) {
1860 None
1861 } else {
1862 Some(VarInt::from_u64(location.group_id)?)
1863 };
1864 let object_id = if !original.has_object_id()
1865 && self.object_id.and_then(|p| p.checked_add(1)) == Some(location.object_id)
1866 {
1867 None
1868 } else {
1869 Some(VarInt::from_u64(location.object_id)?)
1870 };
1871 let (subgroup_mode, subgroup_id) = self.subgroup_field(original, location)?;
1872 let publisher_priority = self.priority_field(original, location);
1873
1874 let flags = original.flags();
1875 let mut new_flags = subgroup_mode;
1876 if flags & 0x40 != 0 {
1877 new_flags |= 0x40;
1878 }
1879 if group_id.is_some() {
1880 new_flags |= 0x08;
1881 }
1882 if object_id.is_some() {
1883 new_flags |= 0x04;
1884 }
1885 if publisher_priority.is_some() {
1886 new_flags |= 0x10;
1887 }
1888 if original.has_extensions() {
1889 new_flags |= 0x20;
1890 }
1891
1892 Ok(FetchObjectHeader {
1893 serialization_flags: VarInt::from_u64(new_flags)?,
1894 group_id,
1895 subgroup_id,
1896 object_id,
1897 publisher_priority,
1898 extensions: original.extensions.clone(),
1899 payload_length: original.payload_length,
1900 })
1901 }
1902
1903 /// The Subgroup ID mode bits and the explicit field, if one is needed.
1904 ///
1905 /// The Object's own mode is tried first, so a run that inherited its
1906 /// Subgroup ID keeps inheriting it and its bytes do not move.
1907 fn subgroup_field(
1908 &self,
1909 original: &FetchObjectHeader,
1910 location: &FetchObjectLocation,
1911 ) -> Result<(u64, Option<VarInt>), CodecError> {
1912 // With the Datagram bit set the two low bits say nothing and no field
1913 // is on the wire, so the Object's own bits are carried across.
1914 if original.is_datagram() {
1915 return Ok((original.flags() & 0x03, None));
1916 }
1917
1918 let subgroup_id = location.subgroup_id.ok_or(CodecError::InvalidField)?;
1919 let inherits = self.subgroup_id == Some(subgroup_id);
1920 let successor = self.subgroup_id.is_some_and(|p| p.checked_add(1) == Some(subgroup_id));
1921
1922 let kept = match original.subgroup_mode() {
1923 FetchSubgroupMode::Zero if subgroup_id == 0 => Some((0x00, None)),
1924 FetchSubgroupMode::SameAsPrior if inherits => Some((0x01, None)),
1925 FetchSubgroupMode::PriorPlusOne if successor => Some((0x02, None)),
1926 FetchSubgroupMode::Present => Some((0x03, Some(subgroup_id))),
1927 _ => None,
1928 };
1929 let (mode, explicit) = match kept {
1930 Some(pair) => pair,
1931 None if subgroup_id == 0 => (0x00, None),
1932 None if inherits => (0x01, None),
1933 None if successor => (0x02, None),
1934 None => (0x03, Some(subgroup_id)),
1935 };
1936 Ok((mode, explicit.map(VarInt::from_u64).transpose()?))
1937 }
1938
1939 /// The Publisher Priority field, or `None` when the running one already
1940 /// says it.
1941 ///
1942 /// `location.publisher_priority` is the Priority in force rather than one
1943 /// this Object stated, so the comparison is against the Priority in force
1944 /// on the stream being written. Where the Object that stated it was the one
1945 /// removed, the two differ and this Object states it instead.
1946 fn priority_field(
1947 &self,
1948 original: &FetchObjectHeader,
1949 location: &FetchObjectLocation,
1950 ) -> Option<u8> {
1951 if original.has_priority() || self.publisher_priority != location.publisher_priority {
1952 return location.publisher_priority;
1953 }
1954 None
1955 }
1956
1957 /// Encode `location` against everything written so far and advance.
1958 ///
1959 /// Writes the header only. The payload is `original.payload_length` bytes
1960 /// and is the caller's to copy, unchanged.
1961 ///
1962 /// # Errors
1963 ///
1964 /// [`CodecError::InvalidField`] for an Object with no encoding; the writer
1965 /// is left untouched when this happens.
1966 pub fn write_object_header(
1967 &mut self,
1968 original: &FetchObjectHeader,
1969 location: &FetchObjectLocation,
1970 out: &mut impl BufMut,
1971 ) -> Result<FetchObjectHeader, CodecError> {
1972 let header = self.header_for(original, location)?;
1973 header.encode(out)?;
1974 self.advance(&header, location);
1975 Ok(header)
1976 }
1977
1978 /// Record what was written as the predecessor of whatever comes next.
1979 ///
1980 /// Mirrors [`FetchObjectReader::resolve`] exactly, including the two places
1981 /// draft-16 differs from the drafts after it: a Datagram-forwarded Object
1982 /// clears the running Subgroup ID rather than leaving it standing, and an
1983 /// Object that states no Priority leaves the running one alone.
1984 ///
1985 /// Public because a re-emitting caller has a second way of putting a frame
1986 /// on the wire: when the framing it arrived in still encodes the same
1987 /// meaning against the frame before it, its own bytes are forwarded
1988 /// untouched — no header is produced and nothing is copied. The writer
1989 /// still has to move, or the frame after it is encoded against a
1990 /// predecessor one frame stale. `written` is then the Object's own header, which is
1991 /// what was put on the wire.
1992 pub fn advance(&mut self, written: &FetchObjectHeader, location: &FetchObjectLocation) {
1993 if let Some(priority) = written.publisher_priority {
1994 self.publisher_priority = Some(priority);
1995 }
1996 self.group_id = Some(location.group_id);
1997 self.subgroup_id = location.subgroup_id;
1998 self.object_id = Some(location.object_id);
1999 }
2000}
2001
2002#[cfg(test)]
2003mod tests {
2004 use super::*;
2005
2006 /// Canonically encoded subgroup stream vectors from
2007 /// `test-vectors/transport/draft16/codec/data-streams/subgroup.json`.
2008 /// `subgroup-explicit-subgroup-id` is omitted: it encodes group_id 100 as a
2009 /// two-byte varint, which does not survive a minimal-width re-encode.
2010 const VECTORS: &[&str] = &[
2011 // subgroup-single-object
2012 "100100800004deadbeef",
2013 // subgroup-two-objects
2014 "100100800004deadbeef0002cafe",
2015 // subgroup-no-priority
2016 "3001000004deadbeef",
2017 // subgroup-with-extensions
2018 "11010080000004deadbeef",
2019 // subgroup-nonempty-extensions
2020 "1101058000023c0104deadbeef",
2021 // subgroup-status-end-of-group
2022 "100105800004deadbeef000003",
2023 // subgroup-status-end-of-track
2024 "10010a800004deadbeef000004",
2025 // subgroup-extensions-two-objects-empty
2026 "11010080000004deadbeef000002cafe",
2027 // subgroup-extensions-two-objects-nonempty
2028 "1101058000023c0204deadbeef00023c0202cafe",
2029 // subgroup-end-of-group
2030 "180105800004deadbeef",
2031 // subgroup-id-first-object
2032 "120103800504deadbeef",
2033 ];
2034
2035 fn vi(v: u64) -> VarInt {
2036 VarInt::from_u64(v).unwrap()
2037 }
2038
2039 fn hex(s: &str) -> Vec<u8> {
2040 (0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap()).collect()
2041 }
2042
2043 /// Decode a whole subgroup stream: the header, then every object up to
2044 /// the end of the buffer.
2045 fn decode_all(bytes: &[u8]) -> (SubgroupHeader, Vec<SubgroupObject>) {
2046 let mut cursor = bytes;
2047 let header = SubgroupHeader::decode(&mut cursor)
2048 .unwrap_or_else(|e| panic!("header decode failed: {e:?}"));
2049 let mut reader = SubgroupObjectReader::new(&header);
2050 let mut objects = Vec::new();
2051 while cursor.has_remaining() {
2052 objects.push(
2053 reader
2054 .read_object(&mut cursor)
2055 .unwrap_or_else(|e| panic!("object {} decode failed: {e:?}", objects.len())),
2056 );
2057 }
2058 (header, objects)
2059 }
2060
2061 fn encode_all(header: &SubgroupHeader, objects: &[SubgroupObject]) -> Vec<u8> {
2062 let mut buf = Vec::new();
2063 header.encode(&mut buf);
2064 let mut writer = SubgroupObjectReader::new(header);
2065 for o in objects {
2066 writer.write_object(o, &mut buf).unwrap_or_else(|e| panic!("write failed: {e:?}"));
2067 }
2068 buf
2069 }
2070
2071 fn object(id: u64, extensions: Vec<u8>, payload: Vec<u8>) -> SubgroupObject {
2072 SubgroupObject {
2073 object_id: vi(id),
2074 extension_headers: extensions,
2075 payload_length: vi(payload.len() as u64),
2076 object_status: None,
2077 payload,
2078 }
2079 }
2080
2081 // ── Object ID deltas ────────────────────────────────────
2082
2083 #[test]
2084 fn two_objects_with_extensions_have_distinct_ids() {
2085 // Vector `subgroup-extensions-two-objects-empty`: two objects, each
2086 // carrying an empty extensions block and a delta of 0. The delta is
2087 // biased by one whether or not the extensions bit is set, so the IDs
2088 // are 0 and 1 — not 0 and 0.
2089 let bytes = hex("11010080000004deadbeef000002cafe");
2090 let (header, objects) = decode_all(&bytes);
2091 assert!(header.has_extensions());
2092 assert_eq!(objects.len(), 2);
2093 assert_eq!(objects[0].object_id.into_inner(), 0);
2094 assert_eq!(objects[1].object_id.into_inner(), 1);
2095 assert_eq!(objects[0].payload, hex("deadbeef"));
2096 assert_eq!(objects[1].payload, hex("cafe"));
2097 assert!(objects.iter().all(|o| o.extension_headers.is_empty()));
2098 }
2099
2100 #[test]
2101 fn deltas_resolve_sparse_ids() {
2102 let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
2103 let objects: Vec<_> =
2104 [3u64, 4, 40].iter().map(|&id| object(id, vec![], vec![0xAA, id as u8])).collect();
2105 let (_, decoded) = decode_all(&encode_all(&header, &objects));
2106 let ids: Vec<u64> = decoded.iter().map(|o| o.object_id.into_inner()).collect();
2107 assert_eq!(ids, vec![3, 4, 40]);
2108 }
2109
2110 #[test]
2111 fn write_rejects_non_increasing_ids() {
2112 let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
2113 let mut writer = SubgroupObjectReader::new(&header);
2114 let mut buf = Vec::new();
2115 writer.write_object(&object(7, vec![], vec![0x01]), &mut buf).unwrap();
2116 for id in [7u64, 6, 0] {
2117 let err = writer.write_object(&object(id, vec![], vec![0x01]), &mut buf).unwrap_err();
2118 assert!(matches!(err, CodecError::InvalidField), "id {id} gave {err:?}");
2119 }
2120 }
2121
2122 #[test]
2123 fn eliding_an_object_renumbers_its_successor() {
2124 let header = SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap();
2125 let all: Vec<_> = (0..5u64).map(|id| object(id, vec![], vec![id as u8])).collect();
2126 for elided in 0..5u64 {
2127 let kept: Vec<_> =
2128 all.iter().filter(|o| o.object_id.into_inner() != elided).cloned().collect();
2129 let (_, decoded) = decode_all(&encode_all(&header, &kept));
2130 let ids: Vec<u64> = decoded.iter().map(|o| o.object_id.into_inner()).collect();
2131 let expected: Vec<u64> = (0..5u64).filter(|&i| i != elided).collect();
2132 assert_eq!(ids, expected, "eliding object {elided}");
2133 }
2134 }
2135
2136 // ── Extension blocks ──────────────────────────────
2137
2138 #[test]
2139 fn extensions_blob_excludes_its_length_prefix() {
2140 // Vector `subgroup-extensions-two-objects-nonempty`: each
2141 // object carries a two-byte block, so the blob is those two bytes
2142 // with the `02` length prefix stripped.
2143 let bytes = hex("1101058000023c0204deadbeef00023c0202cafe");
2144 let (_, objects) = decode_all(&bytes);
2145 assert_eq!(objects.len(), 2);
2146 assert_eq!(objects[0].object_id.into_inner(), 0);
2147 assert_eq!(objects[1].object_id.into_inner(), 1);
2148 assert_eq!(objects[0].extension_headers, hex("3c02"));
2149 assert_eq!(objects[1].extension_headers, hex("3c02"));
2150 assert_eq!(objects[0].payload, hex("deadbeef"));
2151 assert_eq!(objects[1].payload, hex("cafe"));
2152 }
2153
2154 /// Read every object on a subgroup stream, returning the first error.
2155 fn first_error(stream: &str) -> Option<CodecError> {
2156 let bytes = hex(stream);
2157 let mut cursor = &bytes[..];
2158 let header = SubgroupHeader::decode(&mut cursor).unwrap();
2159 let mut reader = SubgroupObjectReader::new(&header);
2160 while cursor.has_remaining() {
2161 if let Err(e) = reader.read_object(&mut cursor) {
2162 return Some(e);
2163 }
2164 }
2165 None
2166 }
2167
2168 #[test]
2169 fn prior_gaps_are_checked_per_object_and_along_the_stream() {
2170 let malformed =
2171 |stream| matches!(first_error(stream), Some(CodecError::MalformedTrack { .. }));
2172 // Section 11.3: "An Object contains more than one instance of Prior
2173 // Group ID Gap".
2174 assert!(malformed("1101058000043c01000104deadbeef"));
2175 // Section 11.4: Object 2's gap of 2 covers Object 0, received earlier,
2176 // which draft-16 Section 2.1 allows: an Object known to exist "can
2177 // transition to not existing".
2178 assert_eq!(first_error("11010580000004deadbeef01023e0202cafe"), None);
2179 // Object 3's gap of 2 covers Objects 1 and 2, neither received.
2180 assert_eq!(first_error("11010580000004deadbeef02023e0202cafe"), None);
2181 // Equal Prior Group ID Gap values in one Group are fine.
2182 assert_eq!(first_error("1101058000023c0204deadbeef00023c0202cafe"), None);
2183 }
2184
2185 #[test]
2186 fn status_object_carries_its_extensions_block() {
2187 // Section 10.2.1.2 forbids extension headers on a non-Normal status, so
2188 // the whole-object reader refuses the frame and the meta reader frames it.
2189 let bytes = hex("1101058000023c010003");
2190 let mut cursor = &bytes[..];
2191 let header = SubgroupHeader::decode(&mut cursor).unwrap();
2192 let mut reader = SubgroupObjectReader::new(&header);
2193 assert!(matches!(reader.read_object(&mut &cursor[..]), Err(CodecError::InvalidField)));
2194 let meta = SubgroupObjectReader::new(&header).read_object_meta(&mut &cursor[..]).unwrap();
2195 assert_eq!(meta.extension_headers_len, 2);
2196 assert_eq!(meta.status, Some(3));
2197 }
2198
2199 // ── Object status ───────────────────────────────────────
2200
2201 /// A subgroup header with neither extensions nor an explicit subgroup ID,
2202 /// so an object on this stream is just `delta, payload_length, [status]`.
2203 fn plain_header() -> SubgroupHeader {
2204 SubgroupHeader::decode(&mut &hex("100100800004deadbeef")[..]).unwrap()
2205 }
2206
2207 /// A status object on a [`plain_header`] stream: object 0, empty payload,
2208 /// `code` as the status. Every code used here is one wire byte.
2209 fn subgroup_status_body(code: u64) -> Vec<u8> {
2210 let mut buf = vec![0x00, 0x00];
2211 VarInt::from_u64(code).unwrap().encode(&mut buf);
2212 buf
2213 }
2214
2215 fn status_object(status: Option<ObjectStatus>) -> SubgroupObject {
2216 SubgroupObject {
2217 object_id: vi(0),
2218 extension_headers: vec![],
2219 payload_length: vi(0),
2220 object_status: status,
2221 payload: vec![],
2222 }
2223 }
2224
2225 /// A status datagram carrying `code`. Type 0x20 sets the status flag and
2226 /// leaves the object-id (0x04), default-priority (0x08) and extensions
2227 /// (0x01) flags clear, so the layout is `type, track_alias, group_id,
2228 /// object_id, priority, status`.
2229 fn datagram_status_bytes(code: u64) -> Vec<u8> {
2230 let mut buf = vec![0x20, 0x01, 0x02, 0x03, 0x80];
2231 VarInt::from_u64(code).unwrap().encode(&mut buf);
2232 buf
2233 }
2234
2235 fn status_datagram(status: Option<ObjectStatus>) -> DatagramHeader {
2236 DatagramHeader {
2237 datagram_type: 0x20,
2238 track_alias: vi(1),
2239 group_id: vi(2),
2240 object_id: vi(3),
2241 publisher_priority: Some(0x80),
2242 extension_headers: vec![],
2243 object_status: status,
2244 }
2245 }
2246
2247 /// Every status the type can hold reaches the wire as its own code and
2248 /// comes back unchanged, through both subgroup readers and the datagram.
2249 ///
2250 /// Fails if `write_object` writes Normal in place of the object's own status.
2251 #[test]
2252 fn assigned_statuses_round_trip() {
2253 let header = plain_header();
2254 for &status in ObjectStatus::ALL {
2255 let mut bytes = Vec::new();
2256 SubgroupObjectReader::new(&header)
2257 .write_object(&status_object(Some(status)), &mut bytes)
2258 .unwrap();
2259 assert_eq!(
2260 bytes,
2261 subgroup_status_body(status.as_u64()),
2262 "{status:?} on the subgroup wire"
2263 );
2264
2265 let decoded = SubgroupObjectReader::new(&header)
2266 .read_object(&mut &bytes[..])
2267 .unwrap_or_else(|e| panic!("{status:?} was written and then refused: {e:?}"));
2268 assert_eq!(decoded.object_status, Some(status), "{status:?} through read_object");
2269
2270 let meta = SubgroupObjectReader::new(&header)
2271 .read_object_meta(&mut &bytes[..])
2272 .unwrap_or_else(|e| panic!("{status:?} was written and then refused: {e:?}"));
2273 assert_eq!(meta.status, Some(status.as_u64()), "{status:?} through read_object_meta");
2274
2275 let datagram = status_datagram(Some(status));
2276 let mut bytes = Vec::new();
2277 datagram.encode(&mut bytes);
2278 assert_eq!(
2279 bytes,
2280 datagram_status_bytes(status.as_u64()),
2281 "{status:?} on the datagram wire"
2282 );
2283 let decoded = DatagramHeader::decode(&mut &bytes[..]).unwrap_or_else(|e| {
2284 panic!("{status:?} datagram was written and then refused: {e:?}")
2285 });
2286 assert_eq!(decoded, datagram, "{status:?} datagram round trip");
2287 }
2288 }
2289
2290 /// A datagram whose type byte sets the status flag always carries a status
2291 /// field, because the flag is what puts the field on the wire — an unset
2292 /// `object_status` writes Normal rather than nothing.
2293 ///
2294 /// Fails if a datagram with the status flag and no `object_status` stops
2295 /// before its status field.
2296 #[test]
2297 fn a_status_datagram_without_a_status_encodes_normal() {
2298 let mut bytes = Vec::new();
2299 status_datagram(None).encode(&mut bytes);
2300 assert_eq!(bytes, datagram_status_bytes(ObjectStatus::Normal.as_u64()));
2301 let decoded = DatagramHeader::decode(&mut &bytes[..])
2302 .unwrap_or_else(|e| panic!("own output refused: {e:?}"));
2303 assert_eq!(decoded.object_status, Some(ObjectStatus::Normal));
2304 }
2305
2306 /// The codes this draft's decoders accept are exactly the codes its
2307 /// encoders can emit.
2308 ///
2309 /// The sweep covers `0x00..=0x3f`, the whole one-byte varint range, so it
2310 /// contains every code draft-16 assigns, the gap inside that range (0x2)
2311 /// and the code draft-16 dropped (0x1, Object Does Not Exist). The
2312 /// expected set is read from [`ObjectStatus::ALL`] rather than written out
2313 /// here, so reassigning a code moves both halves of the test at once.
2314 ///
2315 /// Fails if `ObjectStatus::from_u64` accepts 0x1, the code draft-15
2316 /// assigned to Object Does Not Exist and draft-16 dropped.
2317 #[test]
2318 fn the_wire_accepts_exactly_what_the_type_can_hold() {
2319 let header = plain_header();
2320 for code in 0x00u64..=0x3f {
2321 let assigned = ObjectStatus::ALL.iter().any(|s| s.as_u64() == code);
2322 let body = subgroup_status_body(code);
2323
2324 let read = SubgroupObjectReader::new(&header).read_object(&mut &body[..]);
2325 assert_eq!(
2326 read.is_ok(),
2327 assigned,
2328 "subgroup read_object on status {code:#x}: {read:?}"
2329 );
2330
2331 let meta = SubgroupObjectReader::new(&header).read_object_meta(&mut &body[..]);
2332 assert_eq!(
2333 meta.is_ok(),
2334 assigned,
2335 "subgroup read_object_meta on status {code:#x}: {meta:?}"
2336 );
2337
2338 let datagram = DatagramHeader::decode(&mut &datagram_status_bytes(code)[..]);
2339 assert_eq!(
2340 datagram.is_ok(),
2341 assigned,
2342 "status datagram on status {code:#x}: {datagram:?}"
2343 );
2344
2345 // The other direction: an accepted code is one the encoders can
2346 // reach, and they reach it with exactly these bytes.
2347 if assigned {
2348 let status = ObjectStatus::from_u64(code).unwrap();
2349 let mut bytes = Vec::new();
2350 SubgroupObjectReader::new(&header)
2351 .write_object(&status_object(Some(status)), &mut bytes)
2352 .unwrap();
2353 assert_eq!(bytes, body, "write_object on status {code:#x}");
2354 let mut bytes = Vec::new();
2355 status_datagram(Some(status)).encode(&mut bytes);
2356 assert_eq!(
2357 bytes,
2358 datagram_status_bytes(code),
2359 "datagram encode on status {code:#x}"
2360 );
2361 }
2362 }
2363 }
2364
2365 // ── Re-encoding ─────────────────────────────────────────
2366
2367 #[test]
2368 fn vectors_re_encode_byte_identically() {
2369 for vector in VECTORS {
2370 let bytes = hex(vector);
2371 let (header, objects) = decode_all(&bytes);
2372 assert_eq!(encode_all(&header, &objects), bytes, "[{vector}] re-encode");
2373 }
2374 }
2375
2376 // ── Payload-free framing ────────────────────────────────
2377
2378 #[test]
2379 fn meta_matches_read_object() {
2380 for vector in VECTORS {
2381 let bytes = hex(vector);
2382 let mut cursor = &bytes[..];
2383 let header = SubgroupHeader::decode(&mut cursor).unwrap();
2384 let mut full_reader = SubgroupObjectReader::new(&header);
2385 let mut meta_reader = SubgroupObjectReader::new(&header);
2386 let mut full_cursor = cursor;
2387 let mut meta_cursor = cursor;
2388 while meta_cursor.has_remaining() {
2389 let before = meta_cursor.remaining();
2390 let object = full_reader.read_object(&mut full_cursor).unwrap();
2391 let meta = meta_reader.read_object_meta(&mut meta_cursor).unwrap();
2392 assert_eq!(meta.object_id, object.object_id.into_inner(), "[{vector}]");
2393 assert_eq!(
2394 meta.extension_headers_len,
2395 object.extension_headers.len() as u64,
2396 "[{vector}]"
2397 );
2398 assert_eq!(meta.payload_length, object.payload_length.into_inner(), "[{vector}]");
2399 assert_eq!(
2400 meta.status,
2401 object.object_status.map(ObjectStatus::as_u64),
2402 "[{vector}]"
2403 );
2404 assert_eq!(meta.wire_len, (before - meta_cursor.remaining()) as u64, "[{vector}]");
2405 assert_eq!(full_cursor.remaining(), meta_cursor.remaining(), "[{vector}]");
2406 }
2407 }
2408 }
2409
2410 #[test]
2411 fn short_buffers_report_unexpected_end() {
2412 let bytes = hex("1101058000023c0204deadbeef00023c0202cafe");
2413 let mut cursor = &bytes[..];
2414 let header = SubgroupHeader::decode(&mut cursor).unwrap();
2415 let objects_start = bytes.len() - cursor.len();
2416 for cut in objects_start..bytes.len() {
2417 let mut reader = SubgroupObjectReader::new(&header);
2418 let mut meta_reader = SubgroupObjectReader::new(&header);
2419 let mut cursor = &bytes[objects_start..cut];
2420 let mut meta_cursor = cursor;
2421 while cursor.has_remaining() {
2422 if let Err(err) = reader.read_object(&mut cursor) {
2423 assert!(
2424 matches!(err, CodecError::UnexpectedEnd | CodecError::VarInt(_)),
2425 "cut {cut} gave {err:?}"
2426 );
2427 break;
2428 }
2429 }
2430 while meta_cursor.has_remaining() {
2431 if let Err(err) = meta_reader.read_object_meta(&mut meta_cursor) {
2432 assert!(
2433 matches!(err, CodecError::UnexpectedEnd | CodecError::VarInt(_)),
2434 "cut {cut} gave {err:?}"
2435 );
2436 break;
2437 }
2438 }
2439 }
2440 }
2441}
2442
2443#[cfg(test)]
2444mod datagram_gap_tests {
2445 use super::*;
2446
2447 fn datagram(group: usize, object: usize, block: &[u8]) -> Vec<u8> {
2448 let mut bytes = Vec::new();
2449 DatagramHeader {
2450 datagram_type: 0x01,
2451 track_alias: VarInt::from_usize(1),
2452 group_id: VarInt::from_usize(group),
2453 object_id: VarInt::from_usize(object),
2454 publisher_priority: Some(0x80),
2455 extension_headers: block.to_vec(),
2456 object_status: None,
2457 }
2458 .encode(&mut bytes);
2459 bytes.push(0xAA);
2460 bytes
2461 }
2462
2463 fn is_malformed_track(bytes: &[u8]) -> bool {
2464 matches!(
2465 DatagramHeader::decode_object(&mut &bytes[..]),
2466 Err(CodecError::MalformedTrack { .. })
2467 )
2468 }
2469
2470 #[test]
2471 fn decode_object_refuses_the_gap_conditions_one_datagram_shows() {
2472 for bytes in [
2473 datagram(1, 0, &[0x3C, 2]),
2474 datagram(5, 0, &[0x3E, 1]),
2475 datagram(5, 5, &[0x3C, 1, 0x00, 1]),
2476 ] {
2477 assert!(is_malformed_track(&bytes));
2478 assert!(DatagramHeader::decode(&mut &bytes[..]).is_ok());
2479 }
2480 let (_, payload) =
2481 DatagramHeader::decode_object(&mut &datagram(2, 1, &[0x3C, 2])[..]).unwrap();
2482 assert_eq!(payload, vec![0xAA]);
2483 }
2484
2485 #[test]
2486 fn decode_object_refuses_a_payload_after_a_status_datagram() {
2487 let mut wire = Vec::new();
2488 DatagramHeader {
2489 datagram_type: 0x20,
2490 track_alias: VarInt::from_usize(1),
2491 group_id: VarInt::from_usize(0),
2492 object_id: VarInt::from_usize(0),
2493 publisher_priority: Some(0x80),
2494 extension_headers: Vec::new(),
2495 object_status: Some(ObjectStatus::Normal),
2496 }
2497 .encode(&mut wire);
2498 let (_, payload) = DatagramHeader::decode_object(&mut &wire[..]).unwrap();
2499 assert!(payload.is_empty());
2500 wire.push(0xAA);
2501 assert!(matches!(
2502 DatagramHeader::decode_object(&mut &wire[..]),
2503 Err(CodecError::PayloadNotPermitted { len: 1, .. })
2504 ));
2505 }
2506}