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