Skip to main content

moqtap_codec/
data_dispatch.rs

1//! Draft-neutral object framing for MoQT data streams.
2//!
3//! [`AnySubgroupObjectReader`],
4//! [`AnySubgroupObjectWriter`],
5//! [`AnyFetchObjectReader`] and
6//! [`AnyFetchObjectWriter`]
7//! present one API over drafts' object encodings. The values they
8//! produce — `AnySubgroupObject`, `AnySubgroupObjectMeta`, `AnyFetchObject`,
9//! `AnyFetchObjectMeta` — are plain structs of primitives, so a caller can
10//! address objects without naming a `draftNN` type. All of them are also
11//! re-exported from [`crate::dispatch`].
12//!
13//! Objects on drafts 07-13 are standalone: absolute Object IDs, and (on
14//! drafts 11-13) an extension block whose presence is fixed by the stream
15//! type. Drafts 14-23 delta-encode Object IDs against the previous object on
16//! the stream. Both are constructed from the stream's header and read one
17//! object at a time, so the difference stays inside this module.
18//!
19//! Fetch streams split the same way, at a different draft. Through draft-14 a
20//! fetch object spells out its Group ID, Subgroup ID, Object ID and Publisher
21//! Priority on every object, so each one stands alone. Drafts 15-23 put a
22//! Serialization Flags field first and let it leave any of those four off the
23//! wire, meaning "the prior object's" — and from draft-18 the two ID fields
24//! that remain are differences rather than values. So a fetch object on those
25//! drafts is only meaningful in stream order, and
26//! [`AnyFetchObjectReader`]
27//! carries the running state that resolves it. The
28//! values it produces are absolute on every draft.
29//!
30//! # Partial buffers
31//!
32//! Reader state after an error is unspecified. A caller that may be handed an
33//! incomplete object clones the reader, decodes against the clone, and
34//! overwrites the real reader only once the decode succeeds.
35
36use bytes::{Buf, BufMut};
37
38use crate::dispatch::{AnyFetchHeader, AnySubgroupHeader};
39use crate::error::CodecError;
40use crate::varint::VarInt;
41use crate::version::DraftVersion;
42
43// ── Draft-neutral object values ─────────────────────────────
44
45/// One object read from a subgroup data stream, normalised across drafts.
46///
47/// Field semantics are identical on every draft 07-23; the per-draft wire
48/// differences (absolute vs delta object IDs, count- vs length-prefixed
49/// extension blocks, typed vs raw status codes) are resolved by
50/// [`AnySubgroupObjectReader`] before this value is produced.
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct AnySubgroupObject {
53    /// Absolute Object ID. Already resolved from delta encoding on drafts
54    /// 14-23; copied verbatim on drafts 07-13.
55    pub object_id: u64,
56    /// The extension-header (draft-17+: "property") block's contents,
57    /// excluding any length or count prefix. Empty when the draft has no
58    /// extension block, when the enclosing header's extensions bit is clear,
59    /// or when the block is present but zero-length.
60    ///
61    /// [`crate::object_properties::parse_object_properties`] reads it as a
62    /// list. A verbatim copy of the wire bytes on every draft except draft-08.
63    /// Draft-08's block is count-prefixed with no byte length, so its contents
64    /// can only be delimited by parsing each extension; the blob is therefore
65    /// re-serialized from that parsed form, which re-encodes every varint
66    /// minimally. A draft-08 extension whose value arrived as a legal
67    /// non-minimal varint is semantically but not byte-identically preserved.
68    pub extension_headers: Vec<u8>,
69    /// Number of extensions in `extension_headers`. `Some` only on draft-08,
70    /// whose extension block is count-prefixed rather than
71    /// byte-length-prefixed, so the count cannot be recovered from the blob
72    /// without re-parsing it. `None` on every other draft.
73    pub extension_count: Option<u64>,
74    /// Object Status as the raw wire code, present only when the payload
75    /// length is zero. `None` means a non-empty payload followed and the
76    /// status is implicitly Normal.
77    ///
78    /// Kept as a raw code rather than a typed enum because the assigned set
79    /// changes across drafts and this value crosses drafts — a relay reads a
80    /// status on one and writes it on another, where the same number may mean
81    /// something else or nothing at all. The code is nonetheless always one
82    /// the *source* draft assigns: every draft's decoder refuses an unassigned
83    /// status, so this field never carries a value its draft's Object Status
84    /// section forbids.
85    pub status: Option<u64>,
86    /// Object payload. Empty when `status` is `Some`.
87    pub payload: Vec<u8>,
88}
89
90/// The framing of one subgroup object, without its payload.
91///
92/// Produced by [`AnySubgroupObjectReader::read_object_meta`] for callers that
93/// forward an object's bytes verbatim and never inspect the payload.
94#[derive(Debug, Clone, Copy, PartialEq, Eq)]
95pub struct AnySubgroupObjectMeta {
96    /// Absolute Object ID, resolved as for [`AnySubgroupObject::object_id`].
97    pub object_id: u64,
98    /// Declared payload length in bytes. Zero when `status` is `Some`.
99    pub payload_length: u64,
100    /// Object Status wire code, as for [`AnySubgroupObject::status`].
101    pub status: Option<u64>,
102    /// Byte length of the extension/property block's contents, excluding its
103    /// prefix. Always the length of the blob
104    /// [`AnySubgroupObject::extension_headers`] would carry, which on draft-08
105    /// is a re-serialized copy rather than the wire bytes.
106    pub extension_headers_len: u64,
107    /// Total bytes this object occupies on the wire, prefix fields included.
108    /// Equals the number of bytes the reader consumed.
109    pub wire_len: u64,
110}
111
112/// What a fetch stream indicator asserts: an End of Range about the Locations
113/// it covers, or (draft-23 on) the end of a Group or of the Track.
114///
115/// Drafts 16-23 let a fetch stream state that a run of Objects was not
116/// serialized instead of sending them: one frame names the Location that ends
117/// the run, and every Location from the previously serialized Object up to and
118/// including that one is covered. The indicators are the same frame shape with
119/// different claims behind it, and a subscriber may cache the first as a
120/// settled gap while it must not cache the others, so they are carried apart
121/// rather than merged.
122///
123/// Never produced on drafts 07-15, which have no such frame.
124#[derive(Debug, Clone, Copy, PartialEq, Eq)]
125pub enum AnyFetchEndOfRange {
126    /// The covered Objects do not exist.
127    NonExistent,
128    /// The covered Objects' status is unknown to the publisher.
129    Unknown,
130    /// The covered Objects timed out: the relay abandoned them when its
131    /// `FILL_TIMEOUT` budget ran out.
132    ///
133    /// **Draft-20 and later only.** Drafts 16 through 19 have two indicators
134    /// and report the same Objects as [`Self::Unknown`], so this variant is
135    /// never produced on them — which makes its presence a fact about the
136    /// stream's draft as much as about the frame, and is why a receiver cannot
137    /// treat the two as interchangeable.
138    TimedOut,
139    /// The covered Objects are Gone: the original publisher cannot deliver
140    /// them and does not know whether they exist.
141    ///
142    /// **Draft-23 and later only** (draft-23 Section 11.4.1, Table 8, 0x40C).
143    Gone,
144    /// End of Group (draft-23 Table 8, 0x80). Not a range: it carries no
145    /// Location and says the Object before it is the last in its Group. The
146    /// frame's Location is that Object's.
147    ///
148    /// **Draft-23 and later only.**
149    EndOfGroup,
150    /// End of Track (draft-23 Table 8, 0x81), which also ends the Group. Not a
151    /// range, as for [`Self::EndOfGroup`].
152    ///
153    /// **Draft-23 and later only.**
154    EndOfTrack,
155}
156
157/// The order a fetch response's groups arrive in.
158///
159/// Drafts 18-23 encode an Object's Group ID as a difference from the
160/// previous Object's, and the direction that difference moves is the fetch's
161/// Group Order — which is settled by the control exchange that opened the
162/// fetch and never appears on the data stream. A reader therefore has to be
163/// told, and telling it wrong does not fail to parse: every Object decodes
164/// under a Group ID walking the wrong way.
165///
166/// Ignored on drafts 07-17, whose fetch objects state their Group ID outright.
167#[derive(Debug, Clone, Copy, PartialEq, Eq)]
168pub enum AnyFetchGroupOrder {
169    /// Group IDs increase along the stream; a difference is added.
170    Ascending,
171    /// Group IDs decrease along the stream; a difference is subtracted.
172    Descending,
173}
174
175/// One frame read from a fetch data stream, normalised across drafts.
176///
177/// Usually an object. On drafts 16-23 it may instead be an End of Range
178/// indicator, which carries a Location and no content — [`Self::end_of_range`]
179/// is what tells the two apart, and it is `None` for every object.
180#[derive(Debug, Clone, PartialEq, Eq)]
181pub struct AnyFetchObject {
182    /// Absolute Group ID. Already resolved against the objects before it on
183    /// drafts 15-23, whose fetch objects may omit the field or (on drafts
184    /// 18-23) encode it as a difference; copied verbatim on drafts 07-14.
185    pub group_id: u64,
186    /// Absolute Subgroup ID, resolved as [`Self::group_id`] is. Zero and
187    /// meaningless when [`Self::has_subgroup_id`] is `false`.
188    pub subgroup_id: u64,
189    /// Whether this frame has a Subgroup ID at all.
190    ///
191    /// `true` on every draft 07-15, and for every End of Range indicator's
192    /// predecessor. `false` in two cases drafts 16-23 add: an object whose
193    /// Forwarding Preference is Datagram, which has no Subgroup ID anywhere in
194    /// its framing, and an End of Range indicator, whose Location is a Group
195    /// and Object ID only.
196    ///
197    /// Kept beside `subgroup_id` rather than folded into it because a relay
198    /// that writes this object onto another stream must not invent a Subgroup
199    /// ID of zero for an object that has none: on the receiving draft zero is a
200    /// real subgroup.
201    pub has_subgroup_id: bool,
202    /// Absolute Object ID, resolved as [`Self::group_id`] is.
203    pub object_id: u64,
204    /// Publisher Priority in force for this frame.
205    ///
206    /// Drafts 15-23 let an object omit the field and take the previous
207    /// object's, and an End of Range indicator never carries one. Where
208    /// nothing on the stream has stated a priority, this is 128 — the value
209    /// every draft 15-23 gives a subscription whose Default Publisher Priority
210    /// property is omitted (draft-19 Section 12.4).
211    pub publisher_priority: u8,
212    /// Extension/property block contents, excluding its prefix. Same
213    /// convention as [`AnySubgroupObject::extension_headers`].
214    pub extension_headers: Vec<u8>,
215    /// Number of extensions; `Some` only on draft-08. Same convention as
216    /// [`AnySubgroupObject::extension_count`].
217    pub extension_count: Option<u64>,
218    /// Object Status wire code, present only when the payload is empty.
219    ///
220    /// Always `None` on drafts 16-23: those drafts removed the field from
221    /// fetch objects entirely, stating that Object Status "is only present in
222    /// objects that are delivered via a SUBSCRIPTION, and is absent in Objects
223    /// delivered via a FETCH" (draft-19 Section 11.2.1.1). A zero-length fetch
224    /// object there is an object with no bytes, not a status object.
225    pub status: Option<u64>,
226    /// Which End of Range indicator this frame is, or `None` for an object.
227    ///
228    /// An indicator has a Location and a payload length and nothing else: its
229    /// `payload`, `extension_headers` and `status` are empty, its
230    /// [`Self::has_subgroup_id`] is `false`, and its `publisher_priority` is
231    /// whatever was in force rather than anything it stated.
232    pub end_of_range: Option<AnyFetchEndOfRange>,
233    /// Object payload. Empty when `status` is `Some`.
234    pub payload: Vec<u8>,
235}
236
237/// The framing of one fetch frame, without its payload.
238///
239/// Every field carries the meaning it does on [`AnyFetchObject`].
240#[derive(Debug, Clone, Copy, PartialEq, Eq)]
241pub struct AnyFetchObjectMeta {
242    /// Absolute Group ID.
243    pub group_id: u64,
244    /// Absolute Subgroup ID. Meaningless when `has_subgroup_id` is `false`.
245    pub subgroup_id: u64,
246    /// Whether this frame has a Subgroup ID at all; see
247    /// [`AnyFetchObject::has_subgroup_id`].
248    pub has_subgroup_id: bool,
249    /// Absolute Object ID.
250    pub object_id: u64,
251    /// Publisher Priority in force for this frame; see
252    /// [`AnyFetchObject::publisher_priority`].
253    pub publisher_priority: u8,
254    /// Declared payload length in bytes.
255    pub payload_length: u64,
256    /// Object Status wire code; always `None` on drafts 16-23.
257    pub status: Option<u64>,
258    /// Which End of Range indicator this frame is, or `None` for an object.
259    pub end_of_range: Option<AnyFetchEndOfRange>,
260    /// Byte length of the extension block's contents, excluding its prefix.
261    pub extension_headers_len: u64,
262    /// Total bytes this object occupies on the wire.
263    pub wire_len: u64,
264}
265
266/// The Publisher Priority a fetch frame that states none is read under.
267///
268/// Drafts 16-23 let an object leave the field off the wire and take the
269/// previous object's, and an End of Range indicator carries none at all, so a
270/// stream can reach a frame with no priority ever having been stated. Every one
271/// of those drafts fixes the same fallback for a subscription that never stated
272/// one — draft-19 Section 12.4: "If omitted, the Default Publisher Priority is
273/// 128" — and that is what is reported here.
274///
275/// Draft-15 needs no such fallback: its own reader refuses an object that
276/// inherits a priority with nothing to inherit from, and it has no End of Range
277/// frame, so every draft-15 fetch object has a priority the stream stated.
278///
279/// An End of Range indicator reports the Priority still in force from the last
280/// Object before it, and this constant only when no Object has preceded it.
281/// None of the four drafts decides that. Drafts 17, 18 and 19 say what the
282/// *next* Object inherits — draft-19 Section 11.4.4.2: "Prior Priority: The
283/// Priority from the last actual Object before the End of Range indicator" —
284/// draft-16 does not say even that, and all four agree only that a marker
285/// carries no Priority field. So the answer is chosen here, once, for the four
286/// of them: [`AnyFetchObject::publisher_priority`] is a `u8` with no way to
287/// report "none", and the value in force is one the stream did state, where
288/// this constant would be one it never did.
289#[cfg(any(
290    feature = "draft16",
291    feature = "draft17",
292    feature = "draft18",
293    feature = "draft19",
294    feature = "draft20",
295    feature = "draft21",
296    feature = "draft22",
297    feature = "draft23"
298))]
299const DEFAULT_PUBLISHER_PRIORITY: u8 = 128;
300
301/// Conversions shared by the per-draft glue below. Unused when no draft
302/// feature is enabled.
303#[allow(dead_code)]
304mod conv {
305    use super::{AnySubgroupObject, Buf, CodecError};
306    use crate::varint::VarInt;
307
308    /// Advance `buf` past `len` bytes without copying them.
309    pub fn skip(buf: &mut impl Buf, len: u64) -> Result<(), CodecError> {
310        let len = usize::try_from(len).map_err(|_| CodecError::UnexpectedEnd)?;
311        if buf.remaining() < len {
312            return Err(CodecError::UnexpectedEnd);
313        }
314        buf.advance(len);
315        Ok(())
316    }
317
318    /// Copy `len` bytes out of `buf`.
319    pub fn take(buf: &mut impl Buf, len: u64) -> Result<Vec<u8>, CodecError> {
320        let len = usize::try_from(len).map_err(|_| CodecError::UnexpectedEnd)?;
321        crate::types::read_bytes(buf, len)
322    }
323
324    /// Wrap a value that must fit the varint range.
325    pub fn varint(v: u64) -> Result<VarInt, CodecError> {
326        VarInt::from_u64(v).map_err(|_| CodecError::InvalidField)
327    }
328
329    /// The status code to encode for `object`, or `None` when a payload
330    /// follows instead. An empty payload always carries a status on the
331    /// wire, so a missing one defaults to Normal.
332    ///
333    /// Status `0` with a payload is not a contradiction and is not refused.
334    /// Every draft from 07 to 20 assigns `0x0` to Normal, and every one of them
335    /// encodes a payload-bearing object by leaving the status field off — the
336    /// status is Normal precisely because bytes follow. Saying so explicitly
337    /// asks for the same frame as leaving it out, so both answer `None`, and
338    /// the byte written is identical either way.
339    ///
340    /// Any other status with a payload is refused: those are the statuses whose
341    /// wire form is a status code standing where the payload would be, so there
342    /// is no frame that carries both.
343    pub fn status_to_write(object: &AnySubgroupObject) -> Result<Option<u64>, CodecError> {
344        match (object.status, object.payload.is_empty()) {
345            (Some(0), false) => Ok(None),
346            (Some(_), false) => Err(CodecError::InvalidField),
347            (Some(code), true) => Ok(Some(code)),
348            (None, true) => Ok(Some(0)),
349            (None, false) => Ok(None),
350        }
351    }
352}
353
354// ── Per-draft glue ──────────────────────────────────────────
355
356/// Generates the conversion glue between one draft's standalone
357/// `ObjectHeader` and the draft-neutral object types.
358///
359/// The leading keyword selects the draft's extension-block shape: absent
360/// (draft-07), count-prefixed (draft-08), byte-length-prefixed (drafts
361/// 09/10), or byte-length-prefixed and gated on the stream type (drafts
362/// 11-13).
363macro_rules! legacy_subgroup_glue {
364    (no_extensions $name:ident, $feat:literal, $draft:ident) => {
365        #[cfg(feature = $feat)]
366        mod $name {
367            use super::conv;
368            use super::{AnySubgroupObject, AnySubgroupObjectMeta};
369            use crate::error::CodecError;
370            use crate::$draft::data_stream::ObjectHeader;
371            use crate::$draft::types::ObjectStatus;
372            use bytes::{Buf, BufMut};
373
374            pub fn read_object(buf: &mut impl Buf) -> Result<AnySubgroupObject, CodecError> {
375                let header = ObjectHeader::decode(buf)?;
376                let payload_length = header.payload_length.into_inner();
377                let (status, payload) = if payload_length == 0 {
378                    (Some(header.object_status as u64), Vec::new())
379                } else {
380                    (None, conv::take(buf, payload_length)?)
381                };
382                Ok(AnySubgroupObject {
383                    object_id: header.object_id.into_inner(),
384                    extension_headers: Vec::new(),
385                    extension_count: None,
386                    status,
387                    payload,
388                })
389            }
390
391            pub fn read_object_meta(
392                buf: &mut impl Buf,
393            ) -> Result<AnySubgroupObjectMeta, CodecError> {
394                let start = buf.remaining();
395                let header = ObjectHeader::decode(buf)?;
396                let payload_length = header.payload_length.into_inner();
397                let status = if payload_length == 0 {
398                    Some(header.object_status as u64)
399                } else {
400                    conv::skip(buf, payload_length)?;
401                    None
402                };
403                Ok(AnySubgroupObjectMeta {
404                    object_id: header.object_id.into_inner(),
405                    payload_length,
406                    status,
407                    extension_headers_len: 0,
408                    wire_len: (start - buf.remaining()) as u64,
409                })
410            }
411
412            pub fn write_object(
413                object: &AnySubgroupObject,
414                buf: &mut impl BufMut,
415            ) -> Result<(), CodecError> {
416                if !object.extension_headers.is_empty() {
417                    return Err(CodecError::InvalidField);
418                }
419                let object_status = match conv::status_to_write(object)? {
420                    Some(code) => ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?,
421                    None => ObjectStatus::Normal,
422                };
423                ObjectHeader {
424                    object_id: conv::varint(object.object_id)?,
425                    payload_length: conv::varint(object.payload.len() as u64)?,
426                    object_status,
427                }
428                .encode(buf);
429                buf.put_slice(&object.payload);
430                Ok(())
431            }
432        }
433    };
434
435    (count_extensions $name:ident, $feat:literal, $draft:ident) => {
436        #[cfg(feature = $feat)]
437        mod $name {
438            use super::conv;
439            use super::{AnySubgroupObject, AnySubgroupObjectMeta};
440            use crate::error::CodecError;
441            use crate::$draft::data_stream::ObjectHeader;
442            use crate::$draft::types::ObjectStatus;
443            use bytes::{Buf, BufMut};
444
445            pub fn read_object(buf: &mut impl Buf) -> Result<AnySubgroupObject, CodecError> {
446                let header = ObjectHeader::decode(buf)?;
447                let payload_length = header.payload_length.into_inner();
448                let (status, payload) = if payload_length == 0 {
449                    (Some(header.object_status as u64), Vec::new())
450                } else {
451                    (None, conv::take(buf, payload_length)?)
452                };
453                Ok(AnySubgroupObject {
454                    object_id: header.object_id.into_inner(),
455                    extension_headers: header.extensions,
456                    extension_count: Some(header.extension_count.into_inner()),
457                    status,
458                    payload,
459                })
460            }
461
462            pub fn read_object_meta(
463                buf: &mut impl Buf,
464            ) -> Result<AnySubgroupObjectMeta, CodecError> {
465                let start = buf.remaining();
466                let header = ObjectHeader::decode(buf)?;
467                let payload_length = header.payload_length.into_inner();
468                let status = if payload_length == 0 {
469                    Some(header.object_status as u64)
470                } else {
471                    conv::skip(buf, payload_length)?;
472                    None
473                };
474                Ok(AnySubgroupObjectMeta {
475                    object_id: header.object_id.into_inner(),
476                    payload_length,
477                    status,
478                    extension_headers_len: header.extensions.len() as u64,
479                    wire_len: (start - buf.remaining()) as u64,
480                })
481            }
482
483            pub fn write_object(
484                object: &AnySubgroupObject,
485                buf: &mut impl BufMut,
486            ) -> Result<(), CodecError> {
487                let extension_count = match object.extension_count {
488                    Some(count) => count,
489                    None if object.extension_headers.is_empty() => 0,
490                    None => return Err(CodecError::InvalidField),
491                };
492                let object_status = match conv::status_to_write(object)? {
493                    Some(code) => ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?,
494                    None => ObjectStatus::Normal,
495                };
496                ObjectHeader {
497                    object_id: conv::varint(object.object_id)?,
498                    extension_count: conv::varint(extension_count)?,
499                    extensions: object.extension_headers.clone(),
500                    payload_length: conv::varint(object.payload.len() as u64)?,
501                    object_status,
502                }
503                .encode(buf);
504                buf.put_slice(&object.payload);
505                Ok(())
506            }
507        }
508    };
509
510    (length_extensions $name:ident, $feat:literal, $draft:ident, $version:ident) => {
511        #[cfg(feature = $feat)]
512        mod $name {
513            use super::conv;
514            use super::{AnySubgroupObject, AnySubgroupObjectMeta};
515            use crate::error::CodecError;
516            use crate::$draft::data_stream::ObjectHeader;
517            use crate::$draft::types::ObjectStatus;
518            use bytes::{Buf, BufMut};
519
520            pub fn read_object(buf: &mut impl Buf) -> Result<AnySubgroupObject, CodecError> {
521                let header = ObjectHeader::decode(buf)?;
522                let payload_length = header.payload_length.into_inner();
523                let (status, payload) = if payload_length == 0 {
524                    (Some(header.object_status as u64), Vec::new())
525                } else {
526                    (None, conv::take(buf, payload_length)?)
527                };
528                crate::object_properties::parse_object_properties(
529                    crate::version::DraftVersion::$version,
530                    &header.extensions,
531                )?;
532                Ok(AnySubgroupObject {
533                    object_id: header.object_id.into_inner(),
534                    extension_headers: header.extensions,
535                    extension_count: None,
536                    status,
537                    payload,
538                })
539            }
540
541            pub fn read_object_meta(
542                buf: &mut impl Buf,
543            ) -> Result<AnySubgroupObjectMeta, CodecError> {
544                let start = buf.remaining();
545                let header = ObjectHeader::decode(buf)?;
546                let payload_length = header.payload_length.into_inner();
547                let status = if payload_length == 0 {
548                    Some(header.object_status as u64)
549                } else {
550                    conv::skip(buf, payload_length)?;
551                    None
552                };
553                Ok(AnySubgroupObjectMeta {
554                    object_id: header.object_id.into_inner(),
555                    payload_length,
556                    status,
557                    extension_headers_len: header.extension_headers_length.into_inner(),
558                    wire_len: (start - buf.remaining()) as u64,
559                })
560            }
561
562            pub fn write_object(
563                object: &AnySubgroupObject,
564                buf: &mut impl BufMut,
565            ) -> Result<(), CodecError> {
566                let object_status = match conv::status_to_write(object)? {
567                    Some(code) => ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?,
568                    None => ObjectStatus::Normal,
569                };
570                ObjectHeader {
571                    object_id: conv::varint(object.object_id)?,
572                    extension_headers_length: conv::varint(object.extension_headers.len() as u64)?,
573                    extensions: object.extension_headers.clone(),
574                    payload_length: conv::varint(object.payload.len() as u64)?,
575                    object_status,
576                }
577                .encode(buf);
578                buf.put_slice(&object.payload);
579                Ok(())
580            }
581        }
582    };
583
584    (gated_extensions $name:ident, $feat:literal, $draft:ident) => {
585        #[cfg(feature = $feat)]
586        mod $name {
587            use super::conv;
588            use super::{AnySubgroupObject, AnySubgroupObjectMeta};
589            use crate::error::CodecError;
590            use crate::$draft::data_stream::{ObjectHeader, SubgroupGaps};
591            use crate::$draft::types::ObjectStatus;
592            use bytes::{Buf, BufMut};
593
594            pub fn read_object(
595                extensions: bool,
596                gaps: &mut SubgroupGaps,
597                buf: &mut impl Buf,
598            ) -> Result<AnySubgroupObject, CodecError> {
599                let header = ObjectHeader::decode_with_extensions(extensions, buf)?;
600                let payload_length = header.payload_length.into_inner();
601                let (status, payload) = if payload_length == 0 {
602                    (Some(header.object_status as u64), Vec::new())
603                } else {
604                    (None, conv::take(buf, payload_length)?)
605                };
606                gaps.check(&header)?;
607                Ok(AnySubgroupObject {
608                    object_id: header.object_id.into_inner(),
609                    extension_headers: header.extensions,
610                    extension_count: None,
611                    status,
612                    payload,
613                })
614            }
615
616            pub fn read_object_meta(
617                extensions: bool,
618                buf: &mut impl Buf,
619            ) -> Result<AnySubgroupObjectMeta, CodecError> {
620                let start = buf.remaining();
621                let header = ObjectHeader::decode_with_extensions(extensions, buf)?;
622                let payload_length = header.payload_length.into_inner();
623                let status = if payload_length == 0 {
624                    Some(header.object_status as u64)
625                } else {
626                    conv::skip(buf, payload_length)?;
627                    None
628                };
629                Ok(AnySubgroupObjectMeta {
630                    object_id: header.object_id.into_inner(),
631                    payload_length,
632                    status,
633                    extension_headers_len: header.extension_headers_length.into_inner(),
634                    wire_len: (start - buf.remaining()) as u64,
635                })
636            }
637
638            pub fn write_object(
639                extensions: bool,
640                object: &AnySubgroupObject,
641                buf: &mut impl BufMut,
642            ) -> Result<(), CodecError> {
643                if !extensions && !object.extension_headers.is_empty() {
644                    return Err(CodecError::InvalidField);
645                }
646                let object_status = match conv::status_to_write(object)? {
647                    Some(code) => ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?,
648                    None => ObjectStatus::Normal,
649                };
650                ObjectHeader {
651                    object_id: conv::varint(object.object_id)?,
652                    extension_headers_length: conv::varint(object.extension_headers.len() as u64)?,
653                    extensions: object.extension_headers.clone(),
654                    payload_length: conv::varint(object.payload.len() as u64)?,
655                    object_status,
656                }
657                .encode_with_extensions(extensions, buf);
658                buf.put_slice(&object.payload);
659                Ok(())
660            }
661        }
662    };
663}
664
665legacy_subgroup_glue!(no_extensions sg07, "draft07", draft07);
666legacy_subgroup_glue!(count_extensions sg08, "draft08", draft08);
667legacy_subgroup_glue!(length_extensions sg09, "draft09", draft09, Draft09);
668legacy_subgroup_glue!(length_extensions sg10, "draft10", draft10, Draft10);
669legacy_subgroup_glue!(gated_extensions sg11, "draft11", draft11);
670legacy_subgroup_glue!(gated_extensions sg12, "draft12", draft12);
671legacy_subgroup_glue!(gated_extensions sg13, "draft13", draft13);
672
673/// Generates the conversion glue between one draft's stateful
674/// `SubgroupObjectReader` and the draft-neutral object types.
675///
676/// Every draft from 14 on carries a typed `ObjectStatus`, so the leading
677/// keyword selects how the payload length reaches the wire instead: draft-14
678/// derives it from the payload, while drafts 15-23 carry an explicit
679/// payload-length field, which this glue always sets from the payload.
680///
681/// Both arms funnel the draft-neutral `AnySubgroupObject`, whose status is a
682/// bare `u64`, through the target draft's `ObjectStatus::from_u64`. That is
683/// the one place a status code the target draft does not assign can still be
684/// offered to an encoder at run time — relaying an object between drafts, for
685/// instance — and it is refused there with `CodecError::InvalidField`.
686macro_rules! modern_subgroup_glue {
687    (derived_length $name:ident, $feat:literal, $draft:ident) => {
688        #[cfg(feature = $feat)]
689        mod $name {
690            use super::conv;
691            use super::{AnySubgroupObject, AnySubgroupObjectMeta};
692            use crate::error::CodecError;
693            use crate::$draft::data_stream::{SubgroupObject, SubgroupObjectReader};
694            use crate::$draft::types::ObjectStatus;
695            use bytes::{Buf, BufMut};
696
697            pub fn read_object(
698                reader: &mut SubgroupObjectReader,
699                buf: &mut impl Buf,
700            ) -> Result<AnySubgroupObject, CodecError> {
701                let object = reader.read_object(buf)?;
702                Ok(AnySubgroupObject {
703                    object_id: object.object_id.into_inner(),
704                    extension_headers: object.extension_headers,
705                    extension_count: None,
706                    status: object.status.map(ObjectStatus::as_u64),
707                    payload: object.payload,
708                })
709            }
710
711            pub fn read_object_meta(
712                reader: &mut SubgroupObjectReader,
713                buf: &mut impl Buf,
714            ) -> Result<AnySubgroupObjectMeta, CodecError> {
715                let meta = reader.read_object_meta(buf)?;
716                Ok(AnySubgroupObjectMeta {
717                    object_id: meta.object_id,
718                    payload_length: meta.payload_length,
719                    status: meta.status,
720                    extension_headers_len: meta.extension_headers_len,
721                    wire_len: meta.wire_len,
722                })
723            }
724
725            pub fn write_object(
726                writer: &mut SubgroupObjectReader,
727                object: &AnySubgroupObject,
728                buf: &mut impl BufMut,
729            ) -> Result<(), CodecError> {
730                let status = match conv::status_to_write(object)? {
731                    Some(code) => {
732                        Some(ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?)
733                    }
734                    None => None,
735                };
736                writer.write_object(
737                    &SubgroupObject {
738                        object_id: conv::varint(object.object_id)?,
739                        extension_headers: object.extension_headers.clone(),
740                        status,
741                        payload: object.payload.clone(),
742                    },
743                    buf,
744                )
745            }
746        }
747    };
748
749    (explicit_length $name:ident, $feat:literal, $draft:ident) => {
750        #[cfg(feature = $feat)]
751        mod $name {
752            use super::conv;
753            use super::{AnySubgroupObject, AnySubgroupObjectMeta};
754            use crate::error::CodecError;
755            use crate::$draft::data_stream::{SubgroupObject, SubgroupObjectReader};
756            use crate::$draft::types::ObjectStatus;
757            use bytes::{Buf, BufMut};
758
759            pub fn read_object(
760                reader: &mut SubgroupObjectReader,
761                buf: &mut impl Buf,
762            ) -> Result<AnySubgroupObject, CodecError> {
763                let object = reader.read_object(buf)?;
764                Ok(AnySubgroupObject {
765                    object_id: object.object_id.into_inner(),
766                    extension_headers: object.extension_headers,
767                    extension_count: None,
768                    status: object.object_status.map(ObjectStatus::as_u64),
769                    payload: object.payload,
770                })
771            }
772
773            pub fn read_object_meta(
774                reader: &mut SubgroupObjectReader,
775                buf: &mut impl Buf,
776            ) -> Result<AnySubgroupObjectMeta, CodecError> {
777                let meta = reader.read_object_meta(buf)?;
778                Ok(AnySubgroupObjectMeta {
779                    object_id: meta.object_id,
780                    payload_length: meta.payload_length,
781                    status: meta.status,
782                    extension_headers_len: meta.extension_headers_len,
783                    wire_len: meta.wire_len,
784                })
785            }
786
787            pub fn write_object(
788                writer: &mut SubgroupObjectReader,
789                object: &AnySubgroupObject,
790                buf: &mut impl BufMut,
791            ) -> Result<(), CodecError> {
792                let object_status = match conv::status_to_write(object)? {
793                    Some(code) => {
794                        Some(ObjectStatus::from_u64(code).ok_or(CodecError::InvalidField)?)
795                    }
796                    None => None,
797                };
798                writer.write_object(
799                    &SubgroupObject {
800                        object_id: conv::varint(object.object_id)?,
801                        extension_headers: object.extension_headers.clone(),
802                        payload_length: conv::varint(object.payload.len() as u64)?,
803                        object_status,
804                        payload: object.payload.clone(),
805                    },
806                    buf,
807                )
808            }
809        }
810    };
811}
812
813modern_subgroup_glue!(derived_length sg14, "draft14", draft14);
814modern_subgroup_glue!(explicit_length sg15, "draft15", draft15);
815modern_subgroup_glue!(explicit_length sg16, "draft16", draft16);
816modern_subgroup_glue!(explicit_length sg17, "draft17", draft17);
817modern_subgroup_glue!(explicit_length sg18, "draft18", draft18);
818modern_subgroup_glue!(explicit_length sg19, "draft19", draft19);
819modern_subgroup_glue!(explicit_length sg20, "draft20", draft20);
820modern_subgroup_glue!(explicit_length sg21, "draft21", draft21);
821modern_subgroup_glue!(explicit_length sg22, "draft22", draft22);
822modern_subgroup_glue!(explicit_length sg23, "draft23", draft23);
823
824/// Generates the conversion glue for one draft's fetch objects.
825///
826/// The leading keyword selects the extension-block shape, as for
827/// `legacy_subgroup_glue!`. Unlike subgroup objects, the fetch extension
828/// block is unconditional on drafts 09-13 — it is never gated on the stream
829/// type.
830macro_rules! fetch_glue {
831    (no_extensions $name:ident, $feat:literal, $draft:ident) => {
832        #[cfg(feature = $feat)]
833        mod $name {
834            use super::conv;
835            use super::{AnyFetchObject, AnyFetchObjectMeta};
836            use crate::error::CodecError;
837            use crate::$draft::data_stream::FetchObjectHeader;
838            use bytes::Buf;
839
840            pub fn read_object(buf: &mut impl Buf) -> Result<AnyFetchObject, CodecError> {
841                let header = FetchObjectHeader::decode(buf)?;
842                let payload_length = header.payload_length.into_inner();
843                let (status, payload) = if payload_length == 0 {
844                    (Some(header.object_status as u64), Vec::new())
845                } else {
846                    (None, conv::take(buf, payload_length)?)
847                };
848                Ok(AnyFetchObject {
849                    group_id: header.group_id.into_inner(),
850                    subgroup_id: header.subgroup_id.into_inner(),
851                    has_subgroup_id: true,
852                    object_id: header.object_id.into_inner(),
853                    publisher_priority: header.publisher_priority,
854                    extension_headers: Vec::new(),
855                    extension_count: None,
856                    status,
857                    end_of_range: None,
858                    payload,
859                })
860            }
861
862            pub fn read_object_meta(buf: &mut impl Buf) -> Result<AnyFetchObjectMeta, CodecError> {
863                let start = buf.remaining();
864                let header = FetchObjectHeader::decode(buf)?;
865                let payload_length = header.payload_length.into_inner();
866                let status = if payload_length == 0 {
867                    Some(header.object_status as u64)
868                } else {
869                    conv::skip(buf, payload_length)?;
870                    None
871                };
872                Ok(AnyFetchObjectMeta {
873                    group_id: header.group_id.into_inner(),
874                    subgroup_id: header.subgroup_id.into_inner(),
875                    has_subgroup_id: true,
876                    object_id: header.object_id.into_inner(),
877                    publisher_priority: header.publisher_priority,
878                    payload_length,
879                    status,
880                    end_of_range: None,
881                    extension_headers_len: 0,
882                    wire_len: (start - buf.remaining()) as u64,
883                })
884            }
885        }
886    };
887
888    (count_extensions $name:ident, $feat:literal, $draft:ident) => {
889        #[cfg(feature = $feat)]
890        mod $name {
891            use super::conv;
892            use super::{AnyFetchObject, AnyFetchObjectMeta};
893            use crate::error::CodecError;
894            use crate::$draft::data_stream::FetchObjectHeader;
895            use bytes::Buf;
896
897            pub fn read_object(buf: &mut impl Buf) -> Result<AnyFetchObject, CodecError> {
898                let header = FetchObjectHeader::decode(buf)?;
899                let payload_length = header.payload_length.into_inner();
900                let (status, payload) = if payload_length == 0 {
901                    (Some(header.object_status as u64), Vec::new())
902                } else {
903                    (None, conv::take(buf, payload_length)?)
904                };
905                Ok(AnyFetchObject {
906                    group_id: header.group_id.into_inner(),
907                    subgroup_id: header.subgroup_id.into_inner(),
908                    has_subgroup_id: true,
909                    object_id: header.object_id.into_inner(),
910                    publisher_priority: header.publisher_priority,
911                    extension_headers: header.extensions,
912                    extension_count: Some(header.extension_count.into_inner()),
913                    status,
914                    end_of_range: None,
915                    payload,
916                })
917            }
918
919            pub fn read_object_meta(buf: &mut impl Buf) -> Result<AnyFetchObjectMeta, CodecError> {
920                let start = buf.remaining();
921                let header = FetchObjectHeader::decode(buf)?;
922                let payload_length = header.payload_length.into_inner();
923                let status = if payload_length == 0 {
924                    Some(header.object_status as u64)
925                } else {
926                    conv::skip(buf, payload_length)?;
927                    None
928                };
929                Ok(AnyFetchObjectMeta {
930                    group_id: header.group_id.into_inner(),
931                    subgroup_id: header.subgroup_id.into_inner(),
932                    has_subgroup_id: true,
933                    object_id: header.object_id.into_inner(),
934                    publisher_priority: header.publisher_priority,
935                    payload_length,
936                    status,
937                    end_of_range: None,
938                    extension_headers_len: header.extensions.len() as u64,
939                    wire_len: (start - buf.remaining()) as u64,
940                })
941            }
942        }
943    };
944
945    (length_extensions $name:ident, $feat:literal, $draft:ident) => {
946        #[cfg(feature = $feat)]
947        mod $name {
948            use super::conv;
949            use super::{AnyFetchObject, AnyFetchObjectMeta};
950            use crate::error::CodecError;
951            use crate::$draft::data_stream::FetchObjectHeader;
952            use bytes::Buf;
953
954            pub fn read_object(buf: &mut impl Buf) -> Result<AnyFetchObject, CodecError> {
955                let header = FetchObjectHeader::decode(buf)?;
956                let payload_length = header.payload_length.into_inner();
957                let (status, payload) = if payload_length == 0 {
958                    (Some(header.object_status as u64), Vec::new())
959                } else {
960                    (None, conv::take(buf, payload_length)?)
961                };
962                Ok(AnyFetchObject {
963                    group_id: header.group_id.into_inner(),
964                    subgroup_id: header.subgroup_id.into_inner(),
965                    has_subgroup_id: true,
966                    object_id: header.object_id.into_inner(),
967                    publisher_priority: header.publisher_priority,
968                    extension_headers: header.extensions,
969                    extension_count: None,
970                    status,
971                    end_of_range: None,
972                    payload,
973                })
974            }
975
976            pub fn read_object_meta(buf: &mut impl Buf) -> Result<AnyFetchObjectMeta, CodecError> {
977                let start = buf.remaining();
978                let header = FetchObjectHeader::decode(buf)?;
979                let payload_length = header.payload_length.into_inner();
980                let status = if payload_length == 0 {
981                    Some(header.object_status as u64)
982                } else {
983                    conv::skip(buf, payload_length)?;
984                    None
985                };
986                Ok(AnyFetchObjectMeta {
987                    group_id: header.group_id.into_inner(),
988                    subgroup_id: header.subgroup_id.into_inner(),
989                    has_subgroup_id: true,
990                    object_id: header.object_id.into_inner(),
991                    publisher_priority: header.publisher_priority,
992                    payload_length,
993                    status,
994                    end_of_range: None,
995                    extension_headers_len: header.extension_headers_length.into_inner(),
996                    wire_len: (start - buf.remaining()) as u64,
997                })
998            }
999        }
1000    };
1001}
1002
1003fetch_glue!(no_extensions fo07, "draft07", draft07);
1004fetch_glue!(count_extensions fo08, "draft08", draft08);
1005fetch_glue!(length_extensions fo09, "draft09", draft09);
1006fetch_glue!(length_extensions fo10, "draft10", draft10);
1007fetch_glue!(length_extensions fo11, "draft11", draft11);
1008fetch_glue!(length_extensions fo12, "draft12", draft12);
1009fetch_glue!(length_extensions fo13, "draft13", draft13);
1010
1011#[cfg(feature = "draft14")]
1012mod fo14 {
1013    use super::{AnyFetchObject, AnyFetchObjectMeta};
1014    use crate::draft14::data_stream::FetchObject;
1015    use crate::draft14::types::ObjectStatus;
1016    use crate::error::CodecError;
1017    use bytes::Buf;
1018
1019    pub fn read_object(buf: &mut impl Buf) -> Result<AnyFetchObject, CodecError> {
1020        let object = FetchObject::decode(buf)?;
1021        Ok(AnyFetchObject {
1022            group_id: object.group_id.into_inner(),
1023            subgroup_id: object.subgroup_id.into_inner(),
1024            has_subgroup_id: true,
1025            object_id: object.object_id.into_inner(),
1026            publisher_priority: object.publisher_priority,
1027            extension_headers: object.extension_headers,
1028            extension_count: None,
1029            status: object.status.map(ObjectStatus::as_u64),
1030            end_of_range: None,
1031            payload: object.payload,
1032        })
1033    }
1034
1035    pub fn read_object_meta(buf: &mut impl Buf) -> Result<AnyFetchObjectMeta, CodecError> {
1036        let meta = FetchObject::decode_meta(buf)?;
1037        Ok(AnyFetchObjectMeta {
1038            group_id: meta.group_id,
1039            subgroup_id: meta.subgroup_id,
1040            has_subgroup_id: true,
1041            object_id: meta.object_id,
1042            publisher_priority: meta.publisher_priority,
1043            payload_length: meta.payload_length,
1044            status: meta.status,
1045            end_of_range: None,
1046            extension_headers_len: meta.extension_headers_len,
1047            wire_len: meta.wire_len,
1048        })
1049    }
1050}
1051
1052#[cfg(feature = "draft15")]
1053mod fo15 {
1054    use super::{conv, AnyFetchObject, AnyFetchObjectMeta};
1055    use crate::draft15::data_stream::FetchObjectReader;
1056    use crate::draft15::types::ObjectStatus;
1057    use crate::error::CodecError;
1058    use bytes::Buf;
1059
1060    pub fn read_object(
1061        reader: &mut FetchObjectReader,
1062        buf: &mut impl Buf,
1063    ) -> Result<AnyFetchObject, CodecError> {
1064        let header = reader.read_object_header(buf)?;
1065        // The draft-15 reader stops at the payload length so a caller can
1066        // forward the bytes without copying them; the draft-neutral object
1067        // holds the payload, so the copy happens here instead.
1068        let payload = conv::take(buf, header.payload_length.into_inner())?;
1069        Ok(AnyFetchObject {
1070            group_id: header.group_id.into_inner(),
1071            subgroup_id: header.subgroup_id.into_inner(),
1072            has_subgroup_id: true,
1073            object_id: header.object_id.into_inner(),
1074            publisher_priority: header.publisher_priority,
1075            extension_headers: header.extension_headers,
1076            extension_count: None,
1077            // Already `Some` only for a zero-length object, which is the
1078            // draft-neutral convention too.
1079            status: header.object_status.map(ObjectStatus::as_u64),
1080            end_of_range: None,
1081            payload,
1082        })
1083    }
1084
1085    pub fn read_object_frame(
1086        reader: &mut FetchObjectReader,
1087        buf: &mut impl Buf,
1088    ) -> Result<super::AnyFetchFrame, CodecError> {
1089        let start = buf.remaining();
1090        let header = reader.read_frame_header(buf)?;
1091        let payload_length = header.payload_length.into_inner();
1092        conv::skip(buf, payload_length)?;
1093        let meta = AnyFetchObjectMeta {
1094            group_id: header.group_id.into_inner(),
1095            subgroup_id: header.subgroup_id.into_inner(),
1096            has_subgroup_id: true,
1097            object_id: header.object_id.into_inner(),
1098            publisher_priority: header.publisher_priority,
1099            payload_length,
1100            status: header.object_status.map(ObjectStatus::as_u64),
1101            end_of_range: None,
1102            extension_headers_len: header.extension_headers.len() as u64,
1103            wire_len: (start - buf.remaining()) as u64,
1104        };
1105        Ok(super::AnyFetchFrame {
1106            meta,
1107            draft: crate::version::DraftVersion::Draft15,
1108            shape: super::FetchFrameShape::Draft15(header),
1109        })
1110    }
1111
1112    pub fn read_object_meta(
1113        reader: &mut FetchObjectReader,
1114        buf: &mut impl Buf,
1115    ) -> Result<AnyFetchObjectMeta, CodecError> {
1116        read_object_frame(reader, buf).map(|frame| frame.meta)
1117    }
1118}
1119
1120#[cfg(feature = "draft16")]
1121mod fo16 {
1122    use super::DEFAULT_PUBLISHER_PRIORITY;
1123    use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1124    use crate::draft16::data_stream::{
1125        FetchEndOfRange, FetchObjectHeader, FetchObjectLocation, FetchObjectReader,
1126    };
1127    use crate::error::CodecError;
1128    use bytes::Buf;
1129
1130    /// Draft-16 keeps the frame and its resolved Location apart, and both are
1131    /// carried out of here: the Location fills the draft-neutral value, and the
1132    /// pair of them is what re-encoding this frame later takes.
1133    fn parts(
1134        reader: &mut FetchObjectReader,
1135        buf: &mut impl Buf,
1136        checked: bool,
1137    ) -> Result<(FetchObjectHeader, FetchObjectLocation), CodecError> {
1138        let header = FetchObjectHeader::decode(buf)?;
1139        let location =
1140            if checked { reader.resolve(&header)? } else { reader.resolve_frame(&header)? };
1141        Ok((header, location))
1142    }
1143
1144    fn resolved(location: &FetchObjectLocation) -> super::Resolved {
1145        let end_of_range = location.end_of_range;
1146        super::Resolved {
1147            group_id: location.group_id,
1148            // Draft-16 Section 10.4.4.2 gives an End of Range indicator a Group
1149            // ID and an Object ID and says "Subgroup ID, Priority and Extensions
1150            // are not present". Its per-draft resolver still reads the two low
1151            // flag bits of a marker as Subgroup ID mode zero and answers zero,
1152            // which is a real Subgroup ID; the draft-neutral value says the
1153            // marker has none, as drafts 17-23 do.
1154            subgroup_id: location.subgroup_id.filter(|_| end_of_range.is_none()),
1155            object_id: location.object_id,
1156            publisher_priority: location.publisher_priority.unwrap_or(DEFAULT_PUBLISHER_PRIORITY),
1157            end_of_range: end_of_range.map(|r| match r {
1158                FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1159                FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1160            }),
1161        }
1162    }
1163
1164    pub fn read_object(
1165        reader: &mut FetchObjectReader,
1166        buf: &mut impl Buf,
1167    ) -> Result<AnyFetchObject, CodecError> {
1168        let (header, location) = parts(reader, buf, true)?;
1169        let payload = conv::take(buf, header.payload_length.into_inner())?;
1170        Ok(resolved(&location).into_object(header.extensions.unwrap_or_default(), payload))
1171    }
1172
1173    pub fn read_object_frame(
1174        reader: &mut FetchObjectReader,
1175        buf: &mut impl Buf,
1176    ) -> Result<super::AnyFetchFrame, CodecError> {
1177        let start = buf.remaining();
1178        let (header, location) = parts(reader, buf, false)?;
1179        let payload_length = header.payload_length.into_inner();
1180        conv::skip(buf, payload_length)?;
1181        let meta = resolved(&location).into_meta(
1182            header.extensions.as_ref().map_or(0, |e| e.len() as u64),
1183            payload_length,
1184            (start - buf.remaining()) as u64,
1185        );
1186        Ok(super::AnyFetchFrame {
1187            meta,
1188            draft: crate::version::DraftVersion::Draft16,
1189            shape: super::FetchFrameShape::Draft16(header, location),
1190        })
1191    }
1192
1193    pub fn read_object_meta(
1194        reader: &mut FetchObjectReader,
1195        buf: &mut impl Buf,
1196    ) -> Result<AnyFetchObjectMeta, CodecError> {
1197        read_object_frame(reader, buf).map(|frame| frame.meta)
1198    }
1199}
1200
1201#[cfg(feature = "draft17")]
1202mod fo17 {
1203    use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1204    use crate::draft17::data_stream::{EndOfRange, FetchObject, FetchObjectReader};
1205    use crate::error::CodecError;
1206    use bytes::Buf;
1207
1208    fn resolved(object: &FetchObject) -> super::Resolved {
1209        super::Resolved {
1210            group_id: object.group_id,
1211            subgroup_id: object.subgroup_id,
1212            object_id: object.object_id,
1213            publisher_priority: object
1214                .publisher_priority
1215                .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1216            end_of_range: object.header.end_of_range().map(|r| match r {
1217                EndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1218                EndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1219            }),
1220        }
1221    }
1222
1223    pub fn read_object(
1224        reader: &mut FetchObjectReader,
1225        buf: &mut impl Buf,
1226    ) -> Result<AnyFetchObject, CodecError> {
1227        let object = reader.read_object_header(buf)?;
1228        let resolved = resolved(&object);
1229        let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1230        Ok(resolved.into_object(object.header.properties, payload))
1231    }
1232
1233    pub fn read_object_frame(
1234        reader: &mut FetchObjectReader,
1235        buf: &mut impl Buf,
1236    ) -> Result<super::AnyFetchFrame, CodecError> {
1237        let start = buf.remaining();
1238        let object = reader.read_frame_header(buf)?;
1239        let resolved = resolved(&object);
1240        let payload_length = object.header.payload_length.into_inner();
1241        conv::skip(buf, payload_length)?;
1242        let meta = resolved.into_meta(
1243            object.header.properties.len() as u64,
1244            payload_length,
1245            (start - buf.remaining()) as u64,
1246        );
1247        Ok(super::AnyFetchFrame {
1248            meta,
1249            draft: crate::version::DraftVersion::Draft17,
1250            shape: super::FetchFrameShape::Draft17(object),
1251        })
1252    }
1253
1254    pub fn read_object_meta(
1255        reader: &mut FetchObjectReader,
1256        buf: &mut impl Buf,
1257    ) -> Result<AnyFetchObjectMeta, CodecError> {
1258        read_object_frame(reader, buf).map(|frame| frame.meta)
1259    }
1260}
1261
1262#[cfg(feature = "draft18")]
1263mod fo18 {
1264    use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1265    use crate::draft18::data_stream::{EndOfRange, FetchObject, FetchObjectReader};
1266    use crate::error::CodecError;
1267    use bytes::Buf;
1268
1269    fn resolved(object: &FetchObject) -> super::Resolved {
1270        super::Resolved {
1271            group_id: object.group_id,
1272            subgroup_id: object.subgroup_id,
1273            object_id: object.object_id,
1274            publisher_priority: object
1275                .publisher_priority
1276                .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1277            end_of_range: object.header.end_of_range().map(|r| match r {
1278                EndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1279                EndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1280            }),
1281        }
1282    }
1283
1284    pub fn read_object(
1285        reader: &mut FetchObjectReader,
1286        buf: &mut impl Buf,
1287    ) -> Result<AnyFetchObject, CodecError> {
1288        let object = reader.read_object_header(buf)?;
1289        let resolved = resolved(&object);
1290        let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1291        Ok(resolved.into_object(object.header.properties, payload))
1292    }
1293
1294    pub fn read_object_frame(
1295        reader: &mut FetchObjectReader,
1296        buf: &mut impl Buf,
1297    ) -> Result<super::AnyFetchFrame, CodecError> {
1298        let start = buf.remaining();
1299        let object = reader.read_frame_header(buf)?;
1300        let resolved = resolved(&object);
1301        let payload_length = object.header.payload_length.into_inner();
1302        conv::skip(buf, payload_length)?;
1303        let meta = resolved.into_meta(
1304            object.header.properties.len() as u64,
1305            payload_length,
1306            (start - buf.remaining()) as u64,
1307        );
1308        Ok(super::AnyFetchFrame {
1309            meta,
1310            draft: crate::version::DraftVersion::Draft18,
1311            shape: super::FetchFrameShape::Draft18(object),
1312        })
1313    }
1314
1315    pub fn read_object_meta(
1316        reader: &mut FetchObjectReader,
1317        buf: &mut impl Buf,
1318    ) -> Result<AnyFetchObjectMeta, CodecError> {
1319        read_object_frame(reader, buf).map(|frame| frame.meta)
1320    }
1321}
1322
1323#[cfg(feature = "draft19")]
1324mod fo19 {
1325    use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1326    use crate::draft19::data_stream::{FetchEndOfRange, FetchObject, FetchObjectReader};
1327    use crate::error::CodecError;
1328    use bytes::Buf;
1329
1330    fn resolved(object: &FetchObject) -> super::Resolved {
1331        super::Resolved {
1332            group_id: object.group_id,
1333            subgroup_id: object.subgroup_id,
1334            object_id: object.object_id,
1335            publisher_priority: object
1336                .publisher_priority
1337                .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1338            end_of_range: object.header.end_of_range().map(|r| match r {
1339                FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1340                FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1341            }),
1342        }
1343    }
1344
1345    pub fn read_object(
1346        reader: &mut FetchObjectReader,
1347        buf: &mut impl Buf,
1348    ) -> Result<AnyFetchObject, CodecError> {
1349        let object = reader.read_object_header(buf)?;
1350        let resolved = resolved(&object);
1351        let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1352        Ok(resolved.into_object(object.header.properties.unwrap_or_default(), payload))
1353    }
1354
1355    pub fn read_object_frame(
1356        reader: &mut FetchObjectReader,
1357        buf: &mut impl Buf,
1358    ) -> Result<super::AnyFetchFrame, CodecError> {
1359        let start = buf.remaining();
1360        let object = reader.read_frame_header(buf)?;
1361        let resolved = resolved(&object);
1362        let payload_length = object.header.payload_length.into_inner();
1363        conv::skip(buf, payload_length)?;
1364        let meta = resolved.into_meta(
1365            object.header.properties.as_ref().map_or(0, |p| p.len() as u64),
1366            payload_length,
1367            (start - buf.remaining()) as u64,
1368        );
1369        Ok(super::AnyFetchFrame {
1370            meta,
1371            draft: crate::version::DraftVersion::Draft19,
1372            shape: super::FetchFrameShape::Draft19(object),
1373        })
1374    }
1375
1376    pub fn read_object_meta(
1377        reader: &mut FetchObjectReader,
1378        buf: &mut impl Buf,
1379    ) -> Result<AnyFetchObjectMeta, CodecError> {
1380        read_object_frame(reader, buf).map(|frame| frame.meta)
1381    }
1382}
1383
1384#[cfg(feature = "draft20")]
1385mod fo20 {
1386    use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1387    use crate::draft20::data_stream::{FetchEndOfRange, FetchObject, FetchObjectReader};
1388    use crate::error::CodecError;
1389    use bytes::Buf;
1390
1391    fn resolved(object: &FetchObject) -> super::Resolved {
1392        super::Resolved {
1393            group_id: object.group_id,
1394            subgroup_id: object.subgroup_id,
1395            object_id: object.object_id,
1396            publisher_priority: object
1397                .publisher_priority
1398                .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1399            end_of_range: object.header.end_of_range().map(|r| match r {
1400                FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1401                FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1402                // Draft-20's third marker, Table 7's 0x20C. Only this draft's
1403                // arm can produce it.
1404                FetchEndOfRange::TimedOut => AnyFetchEndOfRange::TimedOut,
1405            }),
1406        }
1407    }
1408
1409    pub fn read_object(
1410        reader: &mut FetchObjectReader,
1411        buf: &mut impl Buf,
1412    ) -> Result<AnyFetchObject, CodecError> {
1413        let object = reader.read_object_header(buf)?;
1414        let resolved = resolved(&object);
1415        let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1416        Ok(resolved.into_object(object.header.properties.unwrap_or_default(), payload))
1417    }
1418
1419    pub fn read_object_frame(
1420        reader: &mut FetchObjectReader,
1421        buf: &mut impl Buf,
1422    ) -> Result<super::AnyFetchFrame, CodecError> {
1423        let start = buf.remaining();
1424        let object = reader.read_frame_header(buf)?;
1425        let resolved = resolved(&object);
1426        let payload_length = object.header.payload_length.into_inner();
1427        conv::skip(buf, payload_length)?;
1428        let meta = resolved.into_meta(
1429            object.header.properties.as_ref().map_or(0, |p| p.len() as u64),
1430            payload_length,
1431            (start - buf.remaining()) as u64,
1432        );
1433        Ok(super::AnyFetchFrame {
1434            meta,
1435            draft: crate::version::DraftVersion::Draft20,
1436            shape: super::FetchFrameShape::Draft20(object),
1437        })
1438    }
1439
1440    pub fn read_object_meta(
1441        reader: &mut FetchObjectReader,
1442        buf: &mut impl Buf,
1443    ) -> Result<AnyFetchObjectMeta, CodecError> {
1444        read_object_frame(reader, buf).map(|frame| frame.meta)
1445    }
1446}
1447#[cfg(feature = "draft21")]
1448mod fo21 {
1449    use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1450    use crate::draft21::data_stream::{FetchEndOfRange, FetchObject, FetchObjectReader};
1451    use crate::error::CodecError;
1452    use bytes::Buf;
1453
1454    fn resolved(object: &FetchObject) -> super::Resolved {
1455        super::Resolved {
1456            group_id: object.group_id,
1457            subgroup_id: object.subgroup_id,
1458            object_id: object.object_id,
1459            publisher_priority: object
1460                .publisher_priority
1461                .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1462            end_of_range: object.header.end_of_range().map(|r| match r {
1463                FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1464                FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1465                // Draft-21's third marker, Table 7's 0x20C. Only this draft's
1466                // arm can produce it.
1467                FetchEndOfRange::TimedOut => AnyFetchEndOfRange::TimedOut,
1468            }),
1469        }
1470    }
1471
1472    pub fn read_object(
1473        reader: &mut FetchObjectReader,
1474        buf: &mut impl Buf,
1475    ) -> Result<AnyFetchObject, CodecError> {
1476        let object = reader.read_object_header(buf)?;
1477        let resolved = resolved(&object);
1478        let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1479        Ok(resolved.into_object(object.header.properties.unwrap_or_default(), payload))
1480    }
1481
1482    pub fn read_object_frame(
1483        reader: &mut FetchObjectReader,
1484        buf: &mut impl Buf,
1485    ) -> Result<super::AnyFetchFrame, CodecError> {
1486        let start = buf.remaining();
1487        let object = reader.read_frame_header(buf)?;
1488        let resolved = resolved(&object);
1489        let payload_length = object.header.payload_length.into_inner();
1490        conv::skip(buf, payload_length)?;
1491        let meta = resolved.into_meta(
1492            object.header.properties.as_ref().map_or(0, |p| p.len() as u64),
1493            payload_length,
1494            (start - buf.remaining()) as u64,
1495        );
1496        Ok(super::AnyFetchFrame {
1497            meta,
1498            draft: crate::version::DraftVersion::Draft21,
1499            shape: super::FetchFrameShape::Draft21(object),
1500        })
1501    }
1502
1503    pub fn read_object_meta(
1504        reader: &mut FetchObjectReader,
1505        buf: &mut impl Buf,
1506    ) -> Result<AnyFetchObjectMeta, CodecError> {
1507        read_object_frame(reader, buf).map(|frame| frame.meta)
1508    }
1509}
1510
1511#[cfg(feature = "draft22")]
1512mod fo22 {
1513    use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1514    use crate::draft22::data_stream::{FetchEndOfRange, FetchObject, FetchObjectReader};
1515    use crate::error::CodecError;
1516    use bytes::Buf;
1517
1518    fn resolved(object: &FetchObject) -> super::Resolved {
1519        super::Resolved {
1520            group_id: object.group_id,
1521            subgroup_id: object.subgroup_id,
1522            object_id: object.object_id,
1523            publisher_priority: object
1524                .publisher_priority
1525                .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1526            end_of_range: object.header.end_of_range().map(|r| match r {
1527                FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1528                FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1529                // Draft-22's third marker, Table 8's 0x20C. Only this draft's
1530                // arm can produce it.
1531                FetchEndOfRange::TimedOut => AnyFetchEndOfRange::TimedOut,
1532            }),
1533        }
1534    }
1535
1536    pub fn read_object(
1537        reader: &mut FetchObjectReader,
1538        buf: &mut impl Buf,
1539    ) -> Result<AnyFetchObject, CodecError> {
1540        let object = reader.read_object_header(buf)?;
1541        let resolved = resolved(&object);
1542        let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1543        Ok(resolved.into_object(object.header.properties.unwrap_or_default(), payload))
1544    }
1545
1546    pub fn read_object_frame(
1547        reader: &mut FetchObjectReader,
1548        buf: &mut impl Buf,
1549    ) -> Result<super::AnyFetchFrame, CodecError> {
1550        let start = buf.remaining();
1551        let object = reader.read_frame_header(buf)?;
1552        let resolved = resolved(&object);
1553        let payload_length = object.header.payload_length.into_inner();
1554        conv::skip(buf, payload_length)?;
1555        let meta = resolved.into_meta(
1556            object.header.properties.as_ref().map_or(0, |p| p.len() as u64),
1557            payload_length,
1558            (start - buf.remaining()) as u64,
1559        );
1560        Ok(super::AnyFetchFrame {
1561            meta,
1562            draft: crate::version::DraftVersion::Draft22,
1563            shape: super::FetchFrameShape::Draft22(object),
1564        })
1565    }
1566
1567    pub fn read_object_meta(
1568        reader: &mut FetchObjectReader,
1569        buf: &mut impl Buf,
1570    ) -> Result<AnyFetchObjectMeta, CodecError> {
1571        read_object_frame(reader, buf).map(|frame| frame.meta)
1572    }
1573}
1574#[cfg(feature = "draft23")]
1575mod fo23 {
1576    use super::{conv, AnyFetchEndOfRange, AnyFetchObject, AnyFetchObjectMeta};
1577    use crate::draft23::data_stream::{
1578        FetchEndMarker, FetchEndOfRange, FetchObject, FetchObjectReader,
1579    };
1580    use crate::error::CodecError;
1581    use bytes::Buf;
1582
1583    fn resolved(object: &FetchObject) -> super::Resolved {
1584        let range = object.header.end_of_range().map(|r| match r {
1585            FetchEndOfRange::NonExistent => AnyFetchEndOfRange::NonExistent,
1586            FetchEndOfRange::Unknown => AnyFetchEndOfRange::Unknown,
1587            FetchEndOfRange::TimedOut => AnyFetchEndOfRange::TimedOut,
1588            FetchEndOfRange::Gone => AnyFetchEndOfRange::Gone,
1589        });
1590        let marker = object.header.end_marker().map(|m| match m {
1591            FetchEndMarker::EndOfGroup => AnyFetchEndOfRange::EndOfGroup,
1592            FetchEndMarker::EndOfTrack => AnyFetchEndOfRange::EndOfTrack,
1593        });
1594        super::Resolved {
1595            group_id: object.group_id,
1596            subgroup_id: object.subgroup_id,
1597            object_id: object.object_id,
1598            publisher_priority: object
1599                .publisher_priority
1600                .unwrap_or(super::DEFAULT_PUBLISHER_PRIORITY),
1601            end_of_range: range.or(marker),
1602        }
1603    }
1604
1605    pub fn read_object(
1606        reader: &mut FetchObjectReader,
1607        buf: &mut impl Buf,
1608    ) -> Result<AnyFetchObject, CodecError> {
1609        let object = reader.read_object_header(buf)?;
1610        let resolved = resolved(&object);
1611        let payload = conv::take(buf, object.header.payload_length.into_inner())?;
1612        Ok(resolved.into_object(object.header.properties.unwrap_or_default(), payload))
1613    }
1614
1615    pub fn read_object_frame(
1616        reader: &mut FetchObjectReader,
1617        buf: &mut impl Buf,
1618    ) -> Result<super::AnyFetchFrame, CodecError> {
1619        let start = buf.remaining();
1620        let object = reader.read_frame_header(buf)?;
1621        let resolved = resolved(&object);
1622        let payload_length = object.header.payload_length.into_inner();
1623        conv::skip(buf, payload_length)?;
1624        let meta = resolved.into_meta(
1625            object.header.properties.as_ref().map_or(0, |p| p.len() as u64),
1626            payload_length,
1627            (start - buf.remaining()) as u64,
1628        );
1629        Ok(super::AnyFetchFrame {
1630            meta,
1631            draft: crate::version::DraftVersion::Draft23,
1632            shape: super::FetchFrameShape::Draft23(object),
1633        })
1634    }
1635
1636    pub fn read_object_meta(
1637        reader: &mut FetchObjectReader,
1638        buf: &mut impl Buf,
1639    ) -> Result<AnyFetchObjectMeta, CodecError> {
1640        read_object_frame(reader, buf).map(|frame| frame.meta)
1641    }
1642}
1643
1644/// A drafts-16-to-19 fetch frame's identity once the fields its Serialization
1645/// Flags left off the wire have been filled in.
1646///
1647/// The four drafts read those flags differently enough to need a resolver
1648/// each, but they all end up saying the same five things, and turning that into
1649/// a draft-neutral value is the same work every time. `subgroup_id` is `None`
1650/// for the frames that have none at all rather than zero, which is a real
1651/// Subgroup ID; see [`AnyFetchObject::has_subgroup_id`].
1652#[cfg(any(
1653    feature = "draft16",
1654    feature = "draft17",
1655    feature = "draft18",
1656    feature = "draft19",
1657    feature = "draft20",
1658    feature = "draft21",
1659    feature = "draft22",
1660    feature = "draft23"
1661))]
1662struct Resolved {
1663    group_id: u64,
1664    subgroup_id: Option<u64>,
1665    object_id: u64,
1666    publisher_priority: u8,
1667    end_of_range: Option<AnyFetchEndOfRange>,
1668}
1669
1670#[cfg(any(
1671    feature = "draft16",
1672    feature = "draft17",
1673    feature = "draft18",
1674    feature = "draft19",
1675    feature = "draft20",
1676    feature = "draft21",
1677    feature = "draft22",
1678    feature = "draft23"
1679))]
1680impl Resolved {
1681    fn into_object(self, extension_headers: Vec<u8>, payload: Vec<u8>) -> AnyFetchObject {
1682        AnyFetchObject {
1683            group_id: self.group_id,
1684            subgroup_id: self.subgroup_id.unwrap_or(0),
1685            has_subgroup_id: self.subgroup_id.is_some(),
1686            object_id: self.object_id,
1687            publisher_priority: self.publisher_priority,
1688            extension_headers,
1689            extension_count: None,
1690            // Drafts 16-23 removed the Object Status field from fetch objects;
1691            // a zero-length payload here carries no code to report.
1692            status: None,
1693            end_of_range: self.end_of_range,
1694            payload,
1695        }
1696    }
1697
1698    fn into_meta(
1699        self,
1700        extension_headers_len: u64,
1701        payload_length: u64,
1702        wire_len: u64,
1703    ) -> AnyFetchObjectMeta {
1704        AnyFetchObjectMeta {
1705            group_id: self.group_id,
1706            subgroup_id: self.subgroup_id.unwrap_or(0),
1707            has_subgroup_id: self.subgroup_id.is_some(),
1708            object_id: self.object_id,
1709            publisher_priority: self.publisher_priority,
1710            payload_length,
1711            status: None,
1712            end_of_range: self.end_of_range,
1713            extension_headers_len,
1714            wire_len,
1715        }
1716    }
1717}
1718
1719// ── Subgroup object reader ──────────────────────────────────
1720
1721/// Per-draft reader state. Drafts 07-10 need none, drafts 11-13 need the
1722/// stream type's extensions-present flag and the Prior Group ID Gap values the
1723/// stream's Objects carried, drafts 14-23 own a stateful per-draft reader that
1724/// tracks the Object ID delta.
1725#[derive(Debug, Clone)]
1726enum SubgroupReaderState {
1727    #[cfg(feature = "draft07")]
1728    Draft07,
1729    #[cfg(feature = "draft08")]
1730    Draft08,
1731    #[cfg(feature = "draft09")]
1732    Draft09,
1733    #[cfg(feature = "draft10")]
1734    Draft10,
1735    #[cfg(feature = "draft11")]
1736    Draft11 { extensions: bool, gaps: crate::draft11::data_stream::SubgroupGaps },
1737    #[cfg(feature = "draft12")]
1738    Draft12 { extensions: bool, gaps: crate::draft12::data_stream::SubgroupGaps },
1739    #[cfg(feature = "draft13")]
1740    Draft13 { extensions: bool, gaps: crate::draft13::data_stream::SubgroupGaps },
1741    #[cfg(feature = "draft14")]
1742    Draft14(crate::draft14::data_stream::SubgroupObjectReader),
1743    #[cfg(feature = "draft15")]
1744    Draft15(crate::draft15::data_stream::SubgroupObjectReader),
1745    #[cfg(feature = "draft16")]
1746    Draft16(crate::draft16::data_stream::SubgroupObjectReader),
1747    #[cfg(feature = "draft17")]
1748    Draft17(crate::draft17::data_stream::SubgroupObjectReader),
1749    #[cfg(feature = "draft18")]
1750    Draft18(crate::draft18::data_stream::SubgroupObjectReader),
1751    #[cfg(feature = "draft19")]
1752    Draft19(crate::draft19::data_stream::SubgroupObjectReader),
1753    #[cfg(feature = "draft20")]
1754    Draft20(crate::draft20::data_stream::SubgroupObjectReader),
1755    #[cfg(feature = "draft21")]
1756    Draft21(crate::draft21::data_stream::SubgroupObjectReader),
1757    #[cfg(feature = "draft22")]
1758    Draft22(crate::draft22::data_stream::SubgroupObjectReader),
1759    #[cfg(feature = "draft23")]
1760    Draft23(crate::draft23::data_stream::SubgroupObjectReader),
1761}
1762
1763/// Stateful reader for the objects on a subgroup data stream, for any
1764/// enabled draft.
1765///
1766/// Drafts 07-13 encode absolute object IDs and need no state, drafts 14-23
1767/// delta-encode them against the previous object. This reader presents both
1768/// as the same API: construct it from the stream's header, then call
1769/// [`read_object`](Self::read_object) once per object.
1770///
1771/// The reader is [`Clone`] specifically so callers can probe a partial buffer
1772/// against a copy and commit only on success; see the module docs.
1773#[derive(Debug, Clone)]
1774pub struct AnySubgroupObjectReader {
1775    state: SubgroupReaderState,
1776}
1777
1778impl AnySubgroupObjectReader {
1779    /// Create a reader seeded from the stream's subgroup header.
1780    ///
1781    /// Returns [`CodecError::UnsupportedDraft`] when the header's draft is
1782    /// not compiled in, and [`CodecError::InvalidField`] when the header's
1783    /// stream type is not a subgroup type.
1784    #[allow(unused_variables, unreachable_code)]
1785    pub fn new(header: &AnySubgroupHeader) -> Result<Self, CodecError> {
1786        let state = match header {
1787            #[cfg(feature = "draft07")]
1788            AnySubgroupHeader::Draft07(_) => SubgroupReaderState::Draft07,
1789            #[cfg(feature = "draft08")]
1790            AnySubgroupHeader::Draft08(_) => SubgroupReaderState::Draft08,
1791            #[cfg(feature = "draft09")]
1792            AnySubgroupHeader::Draft09(_) => SubgroupReaderState::Draft09,
1793            #[cfg(feature = "draft10")]
1794            AnySubgroupHeader::Draft10(_) => SubgroupReaderState::Draft10,
1795            #[cfg(feature = "draft11")]
1796            AnySubgroupHeader::Draft11(h) => SubgroupReaderState::Draft11 {
1797                extensions: subgroup_extensions_11(h)?,
1798                gaps: crate::draft11::data_stream::SubgroupGaps::new(h),
1799            },
1800            #[cfg(feature = "draft12")]
1801            AnySubgroupHeader::Draft12(h) => SubgroupReaderState::Draft12 {
1802                extensions: subgroup_extensions_12(h)?,
1803                gaps: crate::draft12::data_stream::SubgroupGaps::new(h),
1804            },
1805            #[cfg(feature = "draft13")]
1806            AnySubgroupHeader::Draft13(h) => SubgroupReaderState::Draft13 {
1807                extensions: subgroup_extensions_13(h)?,
1808                gaps: crate::draft13::data_stream::SubgroupGaps::new(h),
1809            },
1810            #[cfg(feature = "draft14")]
1811            AnySubgroupHeader::Draft14(h) => SubgroupReaderState::Draft14(
1812                crate::draft14::data_stream::SubgroupObjectReader::new(h),
1813            ),
1814            #[cfg(feature = "draft15")]
1815            AnySubgroupHeader::Draft15(h) => SubgroupReaderState::Draft15(
1816                crate::draft15::data_stream::SubgroupObjectReader::new(h),
1817            ),
1818            #[cfg(feature = "draft16")]
1819            AnySubgroupHeader::Draft16(h) => SubgroupReaderState::Draft16(
1820                crate::draft16::data_stream::SubgroupObjectReader::new(h),
1821            ),
1822            #[cfg(feature = "draft17")]
1823            AnySubgroupHeader::Draft17(h) => SubgroupReaderState::Draft17(
1824                crate::draft17::data_stream::SubgroupObjectReader::new(h),
1825            ),
1826            #[cfg(feature = "draft18")]
1827            AnySubgroupHeader::Draft18(h) => SubgroupReaderState::Draft18(
1828                crate::draft18::data_stream::SubgroupObjectReader::new(h),
1829            ),
1830            #[cfg(feature = "draft19")]
1831            AnySubgroupHeader::Draft19(h) => SubgroupReaderState::Draft19(
1832                crate::draft19::data_stream::SubgroupObjectReader::new(h),
1833            ),
1834            #[cfg(feature = "draft20")]
1835            AnySubgroupHeader::Draft20(h) => SubgroupReaderState::Draft20(
1836                crate::draft20::data_stream::SubgroupObjectReader::new(h),
1837            ),
1838            #[cfg(feature = "draft21")]
1839            AnySubgroupHeader::Draft21(h) => SubgroupReaderState::Draft21(
1840                crate::draft21::data_stream::SubgroupObjectReader::new(h),
1841            ),
1842            #[cfg(feature = "draft22")]
1843            AnySubgroupHeader::Draft22(h) => SubgroupReaderState::Draft22(
1844                crate::draft22::data_stream::SubgroupObjectReader::new(h),
1845            ),
1846            #[cfg(feature = "draft23")]
1847            AnySubgroupHeader::Draft23(h) => SubgroupReaderState::Draft23(
1848                crate::draft23::data_stream::SubgroupObjectReader::new(h),
1849            ),
1850            #[allow(unreachable_patterns)]
1851            _ => {
1852                return Err(CodecError::UnsupportedDraft(format!(
1853                    "draft {:?} not enabled via feature flag",
1854                    header.draft()
1855                )));
1856            }
1857        };
1858        Ok(Self { state })
1859    }
1860
1861    /// The draft this reader decodes.
1862    #[allow(unreachable_code)]
1863    pub fn draft(&self) -> DraftVersion {
1864        match &self.state {
1865            #[cfg(feature = "draft07")]
1866            SubgroupReaderState::Draft07 => DraftVersion::Draft07,
1867            #[cfg(feature = "draft08")]
1868            SubgroupReaderState::Draft08 => DraftVersion::Draft08,
1869            #[cfg(feature = "draft09")]
1870            SubgroupReaderState::Draft09 => DraftVersion::Draft09,
1871            #[cfg(feature = "draft10")]
1872            SubgroupReaderState::Draft10 => DraftVersion::Draft10,
1873            #[cfg(feature = "draft11")]
1874            SubgroupReaderState::Draft11 { .. } => DraftVersion::Draft11,
1875            #[cfg(feature = "draft12")]
1876            SubgroupReaderState::Draft12 { .. } => DraftVersion::Draft12,
1877            #[cfg(feature = "draft13")]
1878            SubgroupReaderState::Draft13 { .. } => DraftVersion::Draft13,
1879            #[cfg(feature = "draft14")]
1880            SubgroupReaderState::Draft14(_) => DraftVersion::Draft14,
1881            #[cfg(feature = "draft15")]
1882            SubgroupReaderState::Draft15(_) => DraftVersion::Draft15,
1883            #[cfg(feature = "draft16")]
1884            SubgroupReaderState::Draft16(_) => DraftVersion::Draft16,
1885            #[cfg(feature = "draft17")]
1886            SubgroupReaderState::Draft17(_) => DraftVersion::Draft17,
1887            #[cfg(feature = "draft18")]
1888            SubgroupReaderState::Draft18(_) => DraftVersion::Draft18,
1889            #[cfg(feature = "draft19")]
1890            SubgroupReaderState::Draft19(_) => DraftVersion::Draft19,
1891            #[cfg(feature = "draft20")]
1892            SubgroupReaderState::Draft20(_) => DraftVersion::Draft20,
1893            #[cfg(feature = "draft21")]
1894            SubgroupReaderState::Draft21(_) => DraftVersion::Draft21,
1895            #[cfg(feature = "draft22")]
1896            SubgroupReaderState::Draft22(_) => DraftVersion::Draft22,
1897            #[cfg(feature = "draft23")]
1898            SubgroupReaderState::Draft23(_) => DraftVersion::Draft23,
1899            #[allow(unreachable_patterns)]
1900            _ => unreachable!("AnySubgroupObjectReader has no enabled variants"),
1901        }
1902    }
1903
1904    /// Decode the next object, including its payload.
1905    ///
1906    /// Returns [`CodecError::UnexpectedEnd`] when `buf` holds only part of an
1907    /// object; the reader's state is unspecified after such an error, so
1908    /// callers that may be fed partial buffers must probe against a clone.
1909    ///
1910    /// Refuses a property block that does not parse, as
1911    /// [`crate::object_properties::parse_object_properties`] does, and, as
1912    /// [`CodecError::MalformedTrack`], the Prior Group ID Gap conditions one
1913    /// stream shows on drafts 11-23 and the Prior Object ID Gap ones on drafts
1914    /// 14-23, as each draft's subgroup reader does;
1915    /// [`read_object_meta`](Self::read_object_meta) does not read inside it.
1916    #[allow(unused_variables, unreachable_code)]
1917    pub fn read_object(&mut self, buf: &mut impl Buf) -> Result<AnySubgroupObject, CodecError> {
1918        match &mut self.state {
1919            #[cfg(feature = "draft07")]
1920            SubgroupReaderState::Draft07 => sg07::read_object(buf),
1921            #[cfg(feature = "draft08")]
1922            SubgroupReaderState::Draft08 => sg08::read_object(buf),
1923            #[cfg(feature = "draft09")]
1924            SubgroupReaderState::Draft09 => sg09::read_object(buf),
1925            #[cfg(feature = "draft10")]
1926            SubgroupReaderState::Draft10 => sg10::read_object(buf),
1927            #[cfg(feature = "draft11")]
1928            SubgroupReaderState::Draft11 { extensions, gaps } => {
1929                sg11::read_object(*extensions, gaps, buf)
1930            }
1931            #[cfg(feature = "draft12")]
1932            SubgroupReaderState::Draft12 { extensions, gaps } => {
1933                sg12::read_object(*extensions, gaps, buf)
1934            }
1935            #[cfg(feature = "draft13")]
1936            SubgroupReaderState::Draft13 { extensions, gaps } => {
1937                sg13::read_object(*extensions, gaps, buf)
1938            }
1939            #[cfg(feature = "draft14")]
1940            SubgroupReaderState::Draft14(inner) => sg14::read_object(inner, buf),
1941            #[cfg(feature = "draft15")]
1942            SubgroupReaderState::Draft15(inner) => sg15::read_object(inner, buf),
1943            #[cfg(feature = "draft16")]
1944            SubgroupReaderState::Draft16(inner) => sg16::read_object(inner, buf),
1945            #[cfg(feature = "draft17")]
1946            SubgroupReaderState::Draft17(inner) => sg17::read_object(inner, buf),
1947            #[cfg(feature = "draft18")]
1948            SubgroupReaderState::Draft18(inner) => sg18::read_object(inner, buf),
1949            #[cfg(feature = "draft19")]
1950            SubgroupReaderState::Draft19(inner) => sg19::read_object(inner, buf),
1951            #[cfg(feature = "draft20")]
1952            SubgroupReaderState::Draft20(inner) => sg20::read_object(inner, buf),
1953            #[cfg(feature = "draft21")]
1954            SubgroupReaderState::Draft21(inner) => sg21::read_object(inner, buf),
1955            #[cfg(feature = "draft22")]
1956            SubgroupReaderState::Draft22(inner) => sg22::read_object(inner, buf),
1957            #[cfg(feature = "draft23")]
1958            SubgroupReaderState::Draft23(inner) => sg23::read_object(inner, buf),
1959            #[allow(unreachable_patterns)]
1960            _ => unreachable!("AnySubgroupObjectReader has no enabled variants"),
1961        }
1962    }
1963
1964    /// Decode the next object's framing without copying its payload.
1965    ///
1966    /// Advances `buf` past the whole object exactly as
1967    /// [`read_object`](Self::read_object) does, but returns only scalars.
1968    /// This is the path a relay uses when it forwards the object's bytes
1969    /// verbatim and never inspects the payload.
1970    #[allow(unused_variables, unreachable_code)]
1971    pub fn read_object_meta(
1972        &mut self,
1973        buf: &mut impl Buf,
1974    ) -> Result<AnySubgroupObjectMeta, CodecError> {
1975        match &mut self.state {
1976            #[cfg(feature = "draft07")]
1977            SubgroupReaderState::Draft07 => sg07::read_object_meta(buf),
1978            #[cfg(feature = "draft08")]
1979            SubgroupReaderState::Draft08 => sg08::read_object_meta(buf),
1980            #[cfg(feature = "draft09")]
1981            SubgroupReaderState::Draft09 => sg09::read_object_meta(buf),
1982            #[cfg(feature = "draft10")]
1983            SubgroupReaderState::Draft10 => sg10::read_object_meta(buf),
1984            #[cfg(feature = "draft11")]
1985            SubgroupReaderState::Draft11 { extensions, .. } => {
1986                sg11::read_object_meta(*extensions, buf)
1987            }
1988            #[cfg(feature = "draft12")]
1989            SubgroupReaderState::Draft12 { extensions, .. } => {
1990                sg12::read_object_meta(*extensions, buf)
1991            }
1992            #[cfg(feature = "draft13")]
1993            SubgroupReaderState::Draft13 { extensions, .. } => {
1994                sg13::read_object_meta(*extensions, buf)
1995            }
1996            #[cfg(feature = "draft14")]
1997            SubgroupReaderState::Draft14(inner) => sg14::read_object_meta(inner, buf),
1998            #[cfg(feature = "draft15")]
1999            SubgroupReaderState::Draft15(inner) => sg15::read_object_meta(inner, buf),
2000            #[cfg(feature = "draft16")]
2001            SubgroupReaderState::Draft16(inner) => sg16::read_object_meta(inner, buf),
2002            #[cfg(feature = "draft17")]
2003            SubgroupReaderState::Draft17(inner) => sg17::read_object_meta(inner, buf),
2004            #[cfg(feature = "draft18")]
2005            SubgroupReaderState::Draft18(inner) => sg18::read_object_meta(inner, buf),
2006            #[cfg(feature = "draft19")]
2007            SubgroupReaderState::Draft19(inner) => sg19::read_object_meta(inner, buf),
2008            #[cfg(feature = "draft20")]
2009            SubgroupReaderState::Draft20(inner) => sg20::read_object_meta(inner, buf),
2010            #[cfg(feature = "draft21")]
2011            SubgroupReaderState::Draft21(inner) => sg21::read_object_meta(inner, buf),
2012            #[cfg(feature = "draft22")]
2013            SubgroupReaderState::Draft22(inner) => sg22::read_object_meta(inner, buf),
2014            #[cfg(feature = "draft23")]
2015            SubgroupReaderState::Draft23(inner) => sg23::read_object_meta(inner, buf),
2016            #[allow(unreachable_patterns)]
2017            _ => unreachable!("AnySubgroupObjectReader has no enabled variants"),
2018        }
2019    }
2020}
2021
2022// ── Subgroup object writer ──────────────────────────────────
2023
2024/// Per-draft writer state. Mirrors [`SubgroupReaderState`]; drafts 14-23
2025/// reuse each draft's `SubgroupObjectReader`, which owns both directions of
2026/// the delta state. Drafts 07-13 encode absolute IDs, so nothing on the wire
2027/// forces them to increase and they carry a `prev_object_id` of their own —
2028/// see `advance_absolute_id`.
2029#[derive(Debug, Clone)]
2030enum SubgroupWriterState {
2031    #[cfg(feature = "draft07")]
2032    Draft07 { prev_object_id: Option<u64> },
2033    #[cfg(feature = "draft08")]
2034    Draft08 { prev_object_id: Option<u64> },
2035    #[cfg(feature = "draft09")]
2036    Draft09 { prev_object_id: Option<u64> },
2037    #[cfg(feature = "draft10")]
2038    Draft10 { prev_object_id: Option<u64> },
2039    #[cfg(feature = "draft11")]
2040    Draft11 { extensions: bool, prev_object_id: Option<u64> },
2041    #[cfg(feature = "draft12")]
2042    Draft12 { extensions: bool, prev_object_id: Option<u64> },
2043    #[cfg(feature = "draft13")]
2044    Draft13 { extensions: bool, prev_object_id: Option<u64> },
2045    #[cfg(feature = "draft14")]
2046    Draft14 { inner: crate::draft14::data_stream::SubgroupObjectReader, extensions: bool },
2047    #[cfg(feature = "draft15")]
2048    Draft15 { inner: crate::draft15::data_stream::SubgroupObjectReader, extensions: bool },
2049    #[cfg(feature = "draft16")]
2050    Draft16 { inner: crate::draft16::data_stream::SubgroupObjectReader, extensions: bool },
2051    #[cfg(feature = "draft17")]
2052    Draft17 { inner: crate::draft17::data_stream::SubgroupObjectReader, extensions: bool },
2053    #[cfg(feature = "draft18")]
2054    Draft18 { inner: crate::draft18::data_stream::SubgroupObjectReader, extensions: bool },
2055    #[cfg(feature = "draft19")]
2056    Draft19 { inner: crate::draft19::data_stream::SubgroupObjectReader, extensions: bool },
2057    #[cfg(feature = "draft20")]
2058    Draft20 { inner: crate::draft20::data_stream::SubgroupObjectReader, extensions: bool },
2059    #[cfg(feature = "draft21")]
2060    Draft21 { inner: crate::draft21::data_stream::SubgroupObjectReader, extensions: bool },
2061    #[cfg(feature = "draft22")]
2062    Draft22 { inner: crate::draft22::data_stream::SubgroupObjectReader, extensions: bool },
2063    #[cfg(feature = "draft23")]
2064    Draft23 { inner: crate::draft23::data_stream::SubgroupObjectReader, extensions: bool },
2065}
2066
2067/// Serializer for the objects on a subgroup data stream, for any enabled
2068/// draft.
2069///
2070/// Mirrors [`AnySubgroupObjectReader`]. On drafts 14-23 it tracks the
2071/// previous Object ID so successive writes produce correct deltas; on drafts
2072/// 07-13 object IDs are absolute and the same state only enforces that they
2073/// increase.
2074///
2075/// # Eliding objects
2076///
2077/// To remove an object from a stream, read it and then simply do not write
2078/// it. The writer's delta state advances only when
2079/// [`write_object`](Self::write_object) succeeds, so the next object written
2080/// re-derives its delta against the last *retained* object automatically.
2081/// See [`Self::write_object`] for the exact invariant.
2082#[derive(Debug, Clone)]
2083pub struct AnySubgroupObjectWriter {
2084    state: SubgroupWriterState,
2085}
2086
2087impl AnySubgroupObjectWriter {
2088    /// Create a writer for a stream with the given header.
2089    ///
2090    /// The header fixes the extension-presence and (on drafts 11-13) stream
2091    /// type used for every object written, exactly as it does for
2092    /// [`AnySubgroupObjectReader::new`].
2093    #[allow(unused_variables, unreachable_code)]
2094    pub fn new(header: &AnySubgroupHeader) -> Result<Self, CodecError> {
2095        let state = match header {
2096            #[cfg(feature = "draft07")]
2097            AnySubgroupHeader::Draft07(_) => SubgroupWriterState::Draft07 { prev_object_id: None },
2098            #[cfg(feature = "draft08")]
2099            AnySubgroupHeader::Draft08(_) => SubgroupWriterState::Draft08 { prev_object_id: None },
2100            #[cfg(feature = "draft09")]
2101            AnySubgroupHeader::Draft09(_) => SubgroupWriterState::Draft09 { prev_object_id: None },
2102            #[cfg(feature = "draft10")]
2103            AnySubgroupHeader::Draft10(_) => SubgroupWriterState::Draft10 { prev_object_id: None },
2104            #[cfg(feature = "draft11")]
2105            AnySubgroupHeader::Draft11(h) => SubgroupWriterState::Draft11 {
2106                extensions: subgroup_extensions_11(h)?,
2107                prev_object_id: None,
2108            },
2109            #[cfg(feature = "draft12")]
2110            AnySubgroupHeader::Draft12(h) => SubgroupWriterState::Draft12 {
2111                extensions: subgroup_extensions_12(h)?,
2112                prev_object_id: None,
2113            },
2114            #[cfg(feature = "draft13")]
2115            AnySubgroupHeader::Draft13(h) => SubgroupWriterState::Draft13 {
2116                extensions: subgroup_extensions_13(h)?,
2117                prev_object_id: None,
2118            },
2119            #[cfg(feature = "draft14")]
2120            AnySubgroupHeader::Draft14(h) => SubgroupWriterState::Draft14 {
2121                inner: crate::draft14::data_stream::SubgroupObjectReader::new(h),
2122                extensions: h.stream_type.extensions_present(),
2123            },
2124            #[cfg(feature = "draft15")]
2125            AnySubgroupHeader::Draft15(h) => SubgroupWriterState::Draft15 {
2126                inner: crate::draft15::data_stream::SubgroupObjectReader::new(h),
2127                extensions: h.has_extensions(),
2128            },
2129            #[cfg(feature = "draft16")]
2130            AnySubgroupHeader::Draft16(h) => SubgroupWriterState::Draft16 {
2131                inner: crate::draft16::data_stream::SubgroupObjectReader::new(h),
2132                extensions: h.has_extensions(),
2133            },
2134            #[cfg(feature = "draft17")]
2135            AnySubgroupHeader::Draft17(h) => SubgroupWriterState::Draft17 {
2136                inner: crate::draft17::data_stream::SubgroupObjectReader::new(h),
2137                extensions: h.has_properties(),
2138            },
2139            #[cfg(feature = "draft18")]
2140            AnySubgroupHeader::Draft18(h) => SubgroupWriterState::Draft18 {
2141                inner: crate::draft18::data_stream::SubgroupObjectReader::new(h),
2142                extensions: h.has_properties(),
2143            },
2144            #[cfg(feature = "draft19")]
2145            AnySubgroupHeader::Draft19(h) => SubgroupWriterState::Draft19 {
2146                inner: crate::draft19::data_stream::SubgroupObjectReader::new(h),
2147                extensions: h.has_properties(),
2148            },
2149            #[cfg(feature = "draft20")]
2150            AnySubgroupHeader::Draft20(h) => SubgroupWriterState::Draft20 {
2151                inner: crate::draft20::data_stream::SubgroupObjectReader::new(h),
2152                extensions: h.has_properties(),
2153            },
2154            #[cfg(feature = "draft21")]
2155            AnySubgroupHeader::Draft21(h) => SubgroupWriterState::Draft21 {
2156                inner: crate::draft21::data_stream::SubgroupObjectReader::new(h),
2157                extensions: h.has_properties(),
2158            },
2159            #[cfg(feature = "draft22")]
2160            AnySubgroupHeader::Draft22(h) => SubgroupWriterState::Draft22 {
2161                inner: crate::draft22::data_stream::SubgroupObjectReader::new(h),
2162                extensions: h.has_properties(),
2163            },
2164            #[cfg(feature = "draft23")]
2165            AnySubgroupHeader::Draft23(h) => SubgroupWriterState::Draft23 {
2166                inner: crate::draft23::data_stream::SubgroupObjectReader::new(h),
2167                extensions: h.has_properties(),
2168            },
2169            #[allow(unreachable_patterns)]
2170            _ => {
2171                return Err(CodecError::UnsupportedDraft(format!(
2172                    "draft {:?} not enabled via feature flag",
2173                    header.draft()
2174                )));
2175            }
2176        };
2177        Ok(Self { state })
2178    }
2179
2180    /// The draft this writer encodes.
2181    #[allow(unreachable_code)]
2182    pub fn draft(&self) -> DraftVersion {
2183        match &self.state {
2184            #[cfg(feature = "draft07")]
2185            SubgroupWriterState::Draft07 { .. } => DraftVersion::Draft07,
2186            #[cfg(feature = "draft08")]
2187            SubgroupWriterState::Draft08 { .. } => DraftVersion::Draft08,
2188            #[cfg(feature = "draft09")]
2189            SubgroupWriterState::Draft09 { .. } => DraftVersion::Draft09,
2190            #[cfg(feature = "draft10")]
2191            SubgroupWriterState::Draft10 { .. } => DraftVersion::Draft10,
2192            #[cfg(feature = "draft11")]
2193            SubgroupWriterState::Draft11 { .. } => DraftVersion::Draft11,
2194            #[cfg(feature = "draft12")]
2195            SubgroupWriterState::Draft12 { .. } => DraftVersion::Draft12,
2196            #[cfg(feature = "draft13")]
2197            SubgroupWriterState::Draft13 { .. } => DraftVersion::Draft13,
2198            #[cfg(feature = "draft14")]
2199            SubgroupWriterState::Draft14 { .. } => DraftVersion::Draft14,
2200            #[cfg(feature = "draft15")]
2201            SubgroupWriterState::Draft15 { .. } => DraftVersion::Draft15,
2202            #[cfg(feature = "draft16")]
2203            SubgroupWriterState::Draft16 { .. } => DraftVersion::Draft16,
2204            #[cfg(feature = "draft17")]
2205            SubgroupWriterState::Draft17 { .. } => DraftVersion::Draft17,
2206            #[cfg(feature = "draft18")]
2207            SubgroupWriterState::Draft18 { .. } => DraftVersion::Draft18,
2208            #[cfg(feature = "draft19")]
2209            SubgroupWriterState::Draft19 { .. } => DraftVersion::Draft19,
2210            #[cfg(feature = "draft20")]
2211            SubgroupWriterState::Draft20 { .. } => DraftVersion::Draft20,
2212            #[cfg(feature = "draft21")]
2213            SubgroupWriterState::Draft21 { .. } => DraftVersion::Draft21,
2214            #[cfg(feature = "draft22")]
2215            SubgroupWriterState::Draft22 { .. } => DraftVersion::Draft22,
2216            #[cfg(feature = "draft23")]
2217            SubgroupWriterState::Draft23 { .. } => DraftVersion::Draft23,
2218            #[allow(unreachable_patterns)]
2219            _ => unreachable!("AnySubgroupObjectWriter has no enabled variants"),
2220        }
2221    }
2222
2223    /// Encode one object, advancing the delta state.
2224    ///
2225    /// # Invariant
2226    ///
2227    /// Let a stream's objects decode to absolute IDs `a_0, a_1, .., a_n`.
2228    /// Feeding any strictly-increasing subsequence of those objects through
2229    /// one writer, in order, produces a byte stream that decodes back to
2230    /// exactly that subsequence of absolute IDs, on every draft 07-23.
2231    ///
2232    /// Concretely: dropping `a_2` from `0,1,2,3,4` yields a stream decoding
2233    /// to `0,1,3,4` — not `0,1,2,3`.
2234    ///
2235    /// # Errors
2236    ///
2237    /// [`CodecError::InvalidField`] when `object.object_id` is not strictly
2238    /// greater than the previously written object's ID (two objects on a
2239    /// subgroup stream can never share an ID, so no valid delta exists), when
2240    /// a computed delta or length exceeds the varint range, when the object
2241    /// carries extension bytes that a stream without an extension block
2242    /// cannot represent, or when a non-empty payload is paired with a status.
2243    ///
2244    /// Also [`CodecError::InvalidField`] when `object.status` holds a code the
2245    /// draft being written does not assign. [`AnySubgroupObject::status`] is a
2246    /// raw wire code because it crosses drafts, and the assigned set moves
2247    /// between them, so a status read off one draft's stream is not
2248    /// necessarily writable onto another's: forwarding a draft-15 Object Does
2249    /// Not Exist (0x1) onto a draft-16 or later stream is refused here rather
2250    /// than emitted as a byte the peer must close the session over.
2251    #[allow(unused_variables, unreachable_code)]
2252    pub fn write_object(
2253        &mut self,
2254        object: &AnySubgroupObject,
2255        buf: &mut impl BufMut,
2256    ) -> Result<(), CodecError> {
2257        match &mut self.state {
2258            #[cfg(feature = "draft07")]
2259            SubgroupWriterState::Draft07 { prev_object_id } => {
2260                advance_absolute_id(prev_object_id, object, |o| sg07::write_object(o, buf))
2261            }
2262            #[cfg(feature = "draft08")]
2263            SubgroupWriterState::Draft08 { prev_object_id } => {
2264                advance_absolute_id(prev_object_id, object, |o| sg08::write_object(o, buf))
2265            }
2266            #[cfg(feature = "draft09")]
2267            SubgroupWriterState::Draft09 { prev_object_id } => {
2268                advance_absolute_id(prev_object_id, object, |o| sg09::write_object(o, buf))
2269            }
2270            #[cfg(feature = "draft10")]
2271            SubgroupWriterState::Draft10 { prev_object_id } => {
2272                advance_absolute_id(prev_object_id, object, |o| sg10::write_object(o, buf))
2273            }
2274            #[cfg(feature = "draft11")]
2275            SubgroupWriterState::Draft11 { extensions, prev_object_id } => {
2276                let extensions = *extensions;
2277                advance_absolute_id(prev_object_id, object, |o| {
2278                    sg11::write_object(extensions, o, buf)
2279                })
2280            }
2281            #[cfg(feature = "draft12")]
2282            SubgroupWriterState::Draft12 { extensions, prev_object_id } => {
2283                let extensions = *extensions;
2284                advance_absolute_id(prev_object_id, object, |o| {
2285                    sg12::write_object(extensions, o, buf)
2286                })
2287            }
2288            #[cfg(feature = "draft13")]
2289            SubgroupWriterState::Draft13 { extensions, prev_object_id } => {
2290                let extensions = *extensions;
2291                advance_absolute_id(prev_object_id, object, |o| {
2292                    sg13::write_object(extensions, o, buf)
2293                })
2294            }
2295            #[cfg(feature = "draft14")]
2296            SubgroupWriterState::Draft14 { inner, extensions } => {
2297                reject_unrepresentable_extensions(*extensions, object)?;
2298                sg14::write_object(inner, object, buf)
2299            }
2300            #[cfg(feature = "draft15")]
2301            SubgroupWriterState::Draft15 { inner, extensions } => {
2302                reject_unrepresentable_extensions(*extensions, object)?;
2303                sg15::write_object(inner, object, buf)
2304            }
2305            #[cfg(feature = "draft16")]
2306            SubgroupWriterState::Draft16 { inner, extensions } => {
2307                reject_unrepresentable_extensions(*extensions, object)?;
2308                sg16::write_object(inner, object, buf)
2309            }
2310            #[cfg(feature = "draft17")]
2311            SubgroupWriterState::Draft17 { inner, extensions } => {
2312                reject_unrepresentable_extensions(*extensions, object)?;
2313                sg17::write_object(inner, object, buf)
2314            }
2315            #[cfg(feature = "draft18")]
2316            SubgroupWriterState::Draft18 { inner, extensions } => {
2317                reject_unrepresentable_extensions(*extensions, object)?;
2318                sg18::write_object(inner, object, buf)
2319            }
2320            #[cfg(feature = "draft19")]
2321            SubgroupWriterState::Draft19 { inner, extensions } => {
2322                reject_unrepresentable_extensions(*extensions, object)?;
2323                sg19::write_object(inner, object, buf)
2324            }
2325            #[cfg(feature = "draft20")]
2326            SubgroupWriterState::Draft20 { inner, extensions } => {
2327                reject_unrepresentable_extensions(*extensions, object)?;
2328                sg20::write_object(inner, object, buf)
2329            }
2330            #[cfg(feature = "draft21")]
2331            SubgroupWriterState::Draft21 { inner, extensions } => {
2332                reject_unrepresentable_extensions(*extensions, object)?;
2333                sg21::write_object(inner, object, buf)
2334            }
2335            #[cfg(feature = "draft22")]
2336            SubgroupWriterState::Draft22 { inner, extensions } => {
2337                reject_unrepresentable_extensions(*extensions, object)?;
2338                sg22::write_object(inner, object, buf)
2339            }
2340            #[cfg(feature = "draft23")]
2341            SubgroupWriterState::Draft23 { inner, extensions } => {
2342                reject_unrepresentable_extensions(*extensions, object)?;
2343                sg23::write_object(inner, object, buf)
2344            }
2345            #[allow(unreachable_patterns)]
2346            _ => unreachable!("AnySubgroupObjectWriter has no enabled variants"),
2347        }
2348    }
2349}
2350
2351// ── Re-emitting an object whose bytes are already known ─────
2352
2353/// What [`reemit_subgroup_object`] had to do.
2354#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2355pub enum Reemit {
2356    /// The bytes were copied unchanged.
2357    Verbatim,
2358    /// Only the leading Object ID field changed.
2359    Reencoded {
2360        /// Bytes the ID field occupied in `raw`.
2361        id_bytes_before: usize,
2362        /// Bytes it occupies in the output.
2363        id_bytes_after: usize,
2364    },
2365}
2366
2367/// Re-emit a subgroup object whose wire bytes are already known, adjusting
2368/// only this draft's encoding of its identity.
2369///
2370/// This is the whole of what removing an object from a subgroup stream
2371/// costs. Drafts 07-13 encode absolute Object IDs, so every survivor's
2372/// bytes are already correct and this copies `raw` unchanged after checking
2373/// that IDs still increase. Drafts 14-23 encode `id - prev - 1`, so the
2374/// leading varint is recomputed against `prev_forwarded`; when its minimal
2375/// encoding is byte-identical to the one in `raw` the bytes are still
2376/// copied unchanged. Everything after the ID field — extension block,
2377/// length, status, payload — is always copied verbatim.
2378///
2379/// After one object is re-emitted following an elided run, the writer's
2380/// cursor re-converges with the reader's, so every later object's original
2381/// bytes remain correct. An elide therefore costs at most one fix-up, not a
2382/// re-encode of the stream's tail.
2383///
2384/// `prev_forwarded` is the absolute Object ID of the last object actually
2385/// forwarded on this stream, or `None` when none has been.
2386///
2387/// # `raw` need not be a complete object
2388///
2389/// **Any prefix is legal provided the whole leading Object ID field is
2390/// present.** Everything after that field is copied byte-for-byte, however
2391/// many bytes there are, and **no length validation is performed** — this
2392/// function never reads the extension block, never reads the payload
2393/// length field and never compares it to `raw.len()`. It cannot: on drafts
2394/// 07-13 it does not decode past the ID at all, and on 14-20 it decodes
2395/// exactly one varint.
2396///
2397/// This is what lets a caller fix up the **first chunk of an oversized
2398/// object** — an object too large to buffer is forwarded in chunks, and only
2399/// the first one carries the ID field. A caller that cannot guarantee the ID
2400/// field is whole in the chunk it passes gets
2401/// [`CodecError::InvalidField`] rather than a silent truncation.
2402///
2403/// **Do not add a completeness check.** A `raw.len() >= wire_len` assertion
2404/// would look defensive, would pass every test that feeds it whole objects,
2405/// and would refuse the prefix this function exists to accept.
2406///
2407/// # Errors
2408///
2409/// [`CodecError::InvalidField`] when `object_id` is not strictly greater
2410/// than `prev_forwarded`, when the recomputed delta exceeds the varint
2411/// range, or when `raw` does not begin with a decodable varint — which
2412/// includes a `raw` too short to hold the whole leading varint.
2413///
2414/// # Examples
2415///
2416/// ```
2417/// use moqtap_codec::dispatch::{reemit_subgroup_object, Reemit};
2418/// use moqtap_codec::version::DraftVersion;
2419///
2420/// // A draft-19 object that was encoded as the successor of ID 4 —
2421/// // leading delta 0 — re-emitted after ID 3 was the last one forwarded.
2422/// let raw = [0x00, 0x02, 0xca, 0xfe];
2423/// let mut out = Vec::new();
2424/// let what = reemit_subgroup_object(DraftVersion::Draft19, Some(3), 5, &raw, &mut out).unwrap();
2425/// assert_eq!(what, Reemit::Reencoded { id_bytes_before: 1, id_bytes_after: 1 });
2426/// assert_eq!(out, [0x01, 0x02, 0xca, 0xfe]);
2427/// ```
2428pub fn reemit_subgroup_object(
2429    draft: DraftVersion,
2430    prev_forwarded: Option<u64>,
2431    object_id: u64,
2432    raw: &[u8],
2433    out: &mut impl BufMut,
2434) -> Result<Reemit, CodecError> {
2435    if matches!(prev_forwarded, Some(prev) if object_id <= prev) {
2436        return Err(CodecError::InvalidField);
2437    }
2438
2439    // Measure the ID field. Every draft 07-23 puts it first and nothing past
2440    // it is decoded, so `raw` may stop anywhere after it. Which varint measures
2441    // it depends on the draft: 17 replaced the RFC 9000 encoding with MoQT's.
2442    let mut cursor: &[u8] = raw;
2443    draft.decode_varint(&mut cursor).map_err(|_| CodecError::InvalidField)?;
2444    let id_bytes_before = raw.len() - cursor.len();
2445
2446    // Drafts 07-13 write the ID absolutely: nothing about that field depends
2447    // on which objects were forwarded, so the bytes already say the truth.
2448    if !delta_encodes_object_ids(draft) {
2449        out.put_slice(raw);
2450        return Ok(Reemit::Verbatim);
2451    }
2452
2453    let delta = match prev_forwarded {
2454        None => object_id,
2455        Some(prev) => object_id
2456            .checked_sub(prev)
2457            .and_then(|v| v.checked_sub(1))
2458            .ok_or(CodecError::InvalidField)?,
2459    };
2460
2461    // The MoQT encoding reaches the full 64-bit range, so a delta a draft-17+
2462    // peer can legitimately send is not an error there.
2463    let field = if draft.uses_moqt_varint() {
2464        VarInt::from_u64_moqt(delta)
2465    } else {
2466        VarInt::from_u64(delta).map_err(|_| CodecError::InvalidField)?
2467    };
2468
2469    // Nine bytes: the MoQT encoding is one longer than RFC 9000 at the top.
2470    let mut encoded = [0u8; 9];
2471    let mut slot: &mut [u8] = &mut encoded;
2472    draft.encode_varint(field, &mut slot);
2473    let id_bytes_after = 9 - slot.len();
2474    let encoded = &encoded[..id_bytes_after];
2475
2476    if encoded == &raw[..id_bytes_before] {
2477        out.put_slice(raw);
2478        return Ok(Reemit::Verbatim);
2479    }
2480
2481    out.put_slice(encoded);
2482    out.put_slice(&raw[id_bytes_before..]);
2483    Ok(Reemit::Reencoded { id_bytes_before, id_bytes_after })
2484}
2485
2486/// `true` on the drafts whose subgroup objects encode the Object ID as
2487/// `id - prev - 1` rather than absolutely.
2488///
2489/// Needs no `#[cfg]`: [`DraftVersion`] is not feature-gated, so this answers
2490/// for a draft whose codec is not compiled in.
2491fn delta_encodes_object_ids(draft: DraftVersion) -> bool {
2492    match draft {
2493        DraftVersion::Draft07
2494        | DraftVersion::Draft08
2495        | DraftVersion::Draft09
2496        | DraftVersion::Draft10
2497        | DraftVersion::Draft11
2498        | DraftVersion::Draft12
2499        | DraftVersion::Draft13 => false,
2500        DraftVersion::Draft14
2501        | DraftVersion::Draft15
2502        | DraftVersion::Draft16
2503        | DraftVersion::Draft17
2504        | DraftVersion::Draft18
2505        | DraftVersion::Draft19
2506        | DraftVersion::Draft20
2507        | DraftVersion::Draft21
2508        | DraftVersion::Draft22
2509        | DraftVersion::Draft23 => true,
2510    }
2511}
2512
2513/// Enforce the strictly-increasing Object ID rule on the drafts that encode
2514/// IDs absolutely.
2515///
2516/// Drafts 14-23 get this for free: their delta is `id - prev - 1`, so a
2517/// repeated or decreasing ID underflows and the per-draft writer rejects it.
2518/// Drafts 07-13 write the ID verbatim and would happily emit a stream no
2519/// publisher can produce, so the check lives here. As on the delta drafts, the
2520/// state advances only once the object is actually written, which is what makes
2521/// elision *read it and do not write it*.
2522#[cfg(any(
2523    feature = "draft07",
2524    feature = "draft08",
2525    feature = "draft09",
2526    feature = "draft10",
2527    feature = "draft11",
2528    feature = "draft12",
2529    feature = "draft13"
2530))]
2531fn advance_absolute_id(
2532    prev_object_id: &mut Option<u64>,
2533    object: &AnySubgroupObject,
2534    write: impl FnOnce(&AnySubgroupObject) -> Result<(), CodecError>,
2535) -> Result<(), CodecError> {
2536    if matches!(*prev_object_id, Some(prev) if object.object_id <= prev) {
2537        return Err(CodecError::InvalidField);
2538    }
2539    write(object)?;
2540    *prev_object_id = Some(object.object_id);
2541    Ok(())
2542}
2543
2544/// A stream whose header says objects carry no extension block cannot encode
2545/// one, so refuse rather than drop the bytes.
2546#[cfg(any(
2547    feature = "draft14",
2548    feature = "draft15",
2549    feature = "draft16",
2550    feature = "draft17",
2551    feature = "draft18",
2552    feature = "draft19",
2553    feature = "draft20",
2554    feature = "draft21",
2555    feature = "draft22",
2556    feature = "draft23"
2557))]
2558fn reject_unrepresentable_extensions(
2559    extensions: bool,
2560    object: &AnySubgroupObject,
2561) -> Result<(), CodecError> {
2562    if !extensions && !object.extension_headers.is_empty() {
2563        return Err(CodecError::InvalidField);
2564    }
2565    Ok(())
2566}
2567
2568// ── Fetch object reader ─────────────────────────────────────
2569
2570/// Per-draft fetch reader state. Fetch objects are self-describing on drafts
2571/// 07-14, so those variants carry none; drafts 15-23 let an object take fields
2572/// from the one before it, so each owns the running state that resolves them.
2573#[derive(Debug, Clone)]
2574enum FetchReaderState {
2575    #[cfg(feature = "draft07")]
2576    Draft07,
2577    #[cfg(feature = "draft08")]
2578    Draft08,
2579    #[cfg(feature = "draft09")]
2580    Draft09,
2581    #[cfg(feature = "draft10")]
2582    Draft10,
2583    #[cfg(feature = "draft11")]
2584    Draft11,
2585    #[cfg(feature = "draft12")]
2586    Draft12,
2587    #[cfg(feature = "draft13")]
2588    Draft13,
2589    #[cfg(feature = "draft14")]
2590    Draft14,
2591    #[cfg(feature = "draft15")]
2592    Draft15(crate::draft15::data_stream::FetchObjectReader),
2593    #[cfg(feature = "draft16")]
2594    Draft16(crate::draft16::data_stream::FetchObjectReader),
2595    #[cfg(feature = "draft17")]
2596    Draft17(crate::draft17::data_stream::FetchObjectReader),
2597    #[cfg(feature = "draft18")]
2598    Draft18(crate::draft18::data_stream::FetchObjectReader),
2599    #[cfg(feature = "draft19")]
2600    Draft19(crate::draft19::data_stream::FetchObjectReader),
2601    #[cfg(feature = "draft20")]
2602    Draft20(crate::draft20::data_stream::FetchObjectReader),
2603    #[cfg(feature = "draft21")]
2604    Draft21(crate::draft21::data_stream::FetchObjectReader),
2605    #[cfg(feature = "draft22")]
2606    Draft22(crate::draft22::data_stream::FetchObjectReader),
2607    #[cfg(feature = "draft23")]
2608    Draft23(crate::draft23::data_stream::FetchObjectReader),
2609}
2610
2611/// The Prior gap conditions one fetch stream shows, on drafts 11-14, whose
2612/// fetch objects decode one at a time. From draft-15 each draft's
2613/// `FetchObjectReader` applies them in `read_object_header` (`resolve` on
2614/// draft-16); drafts 07-10 define no Prior gap.
2615#[derive(Debug, Clone)]
2616enum FetchGapsState {
2617    None,
2618    #[cfg(feature = "draft11")]
2619    Draft11(crate::draft11::data_stream::FetchGaps),
2620    #[cfg(feature = "draft12")]
2621    Draft12(crate::draft12::data_stream::FetchGaps),
2622    #[cfg(feature = "draft13")]
2623    Draft13(crate::draft13::data_stream::FetchGaps),
2624    #[cfg(feature = "draft14")]
2625    Draft14(crate::draft14::data_stream::FetchGaps),
2626}
2627
2628impl FetchGapsState {
2629    fn for_draft(draft: DraftVersion) -> Self {
2630        match draft {
2631            #[cfg(feature = "draft11")]
2632            DraftVersion::Draft11 => Self::Draft11(crate::draft11::data_stream::FetchGaps::new()),
2633            #[cfg(feature = "draft12")]
2634            DraftVersion::Draft12 => Self::Draft12(crate::draft12::data_stream::FetchGaps::new()),
2635            #[cfg(feature = "draft13")]
2636            DraftVersion::Draft13 => Self::Draft13(crate::draft13::data_stream::FetchGaps::new()),
2637            #[cfg(feature = "draft14")]
2638            DraftVersion::Draft14 => Self::Draft14(crate::draft14::data_stream::FetchGaps::new()),
2639            #[allow(unreachable_patterns)]
2640            _ => Self::None,
2641        }
2642    }
2643
2644    #[allow(unused_variables)]
2645    fn check(&mut self, group_id: u64, object_id: u64, block: &[u8]) -> Result<(), CodecError> {
2646        match self {
2647            Self::None => Ok(()),
2648            #[cfg(feature = "draft11")]
2649            Self::Draft11(gaps) => gaps.check(group_id, object_id, block).map(|_| ()),
2650            #[cfg(feature = "draft12")]
2651            Self::Draft12(gaps) => gaps.check(group_id, object_id, block).map(|_| ()),
2652            #[cfg(feature = "draft13")]
2653            Self::Draft13(gaps) => gaps.check(group_id, object_id, block).map(|_| ()),
2654            #[cfg(feature = "draft14")]
2655            Self::Draft14(gaps) => gaps.check(group_id, object_id, block).map(|_| ()),
2656        }
2657    }
2658}
2659
2660/// Stateful reader for the frames on a fetch data stream, for any enabled
2661/// draft.
2662///
2663/// Fetch objects are self-describing on drafts 07-14 and this reader carries no
2664/// state there. From draft-15 a Serialization Flags field decides which of an
2665/// object's Group ID, Subgroup ID, Object ID and Priority reach the wire at
2666/// all, and every field it omits is the object before it on the stream —
2667/// repeated, or stepped by one, or (from draft-18) counted from by a
2668/// difference. This reader holds that running state, so the values it produces
2669/// are absolute on every draft.
2670///
2671/// One reader belongs to one stream. Every draft counts "the prior Object"
2672/// along a single stream, so sharing a reader between streams, or restarting
2673/// one mid-stream, resolves later frames onto the wrong group, subgroup, ID or
2674/// priority — usually without an error anywhere.
2675///
2676/// The reader is [`Clone`] specifically so callers can probe a partial buffer
2677/// against a copy and commit only on success; see the module docs.
2678///
2679/// # Frames that are not objects
2680///
2681/// Drafts 16-23 add End of Range indicators, which state that a run of Objects
2682/// was not serialized. They arrive through the same calls as objects and are
2683/// told apart by [`AnyFetchObject::end_of_range`].
2684#[derive(Debug, Clone)]
2685pub struct AnyFetchObjectReader {
2686    state: FetchReaderState,
2687    gaps: FetchGapsState,
2688}
2689
2690impl AnyFetchObjectReader {
2691    /// Create a reader from the stream's fetch header and the Group Order the
2692    /// fetch was opened with.
2693    ///
2694    /// The order matters only on drafts 18-23, where an Object's Group ID
2695    /// is a difference from the previous Object's and the order decides its
2696    /// sign. Nothing on the data stream carries it — the FETCH settles it — and
2697    /// it is an argument rather than a default because a descending stream read
2698    /// as ascending does not fail: it decodes, under Group IDs walking the wrong
2699    /// way, and neither this crate nor the caller can tell afterwards. Both
2700    /// readings are legal streams.
2701    ///
2702    /// [`AnyControlMessage::fetch_group_order`](crate::dispatch::AnyControlMessage::fetch_group_order)
2703    /// answers it from the FETCH, including the case where the message names no
2704    /// GROUP_ORDER — draft-19 Section 10.2.8: "If omitted from FETCH, the
2705    /// receiver uses Ascending (0x1)". On drafts 07-17 the argument is ignored.
2706    ///
2707    /// Returns [`CodecError::UnsupportedDraft`] for drafts not compiled in.
2708    #[allow(unused_variables, unreachable_code)]
2709    pub fn new(
2710        header: &AnyFetchHeader,
2711        group_order: AnyFetchGroupOrder,
2712    ) -> Result<Self, CodecError> {
2713        let state = match header {
2714            #[cfg(feature = "draft07")]
2715            AnyFetchHeader::Draft07(_) => FetchReaderState::Draft07,
2716            #[cfg(feature = "draft08")]
2717            AnyFetchHeader::Draft08(_) => FetchReaderState::Draft08,
2718            #[cfg(feature = "draft09")]
2719            AnyFetchHeader::Draft09(_) => FetchReaderState::Draft09,
2720            #[cfg(feature = "draft10")]
2721            AnyFetchHeader::Draft10(_) => FetchReaderState::Draft10,
2722            #[cfg(feature = "draft11")]
2723            AnyFetchHeader::Draft11(_) => FetchReaderState::Draft11,
2724            #[cfg(feature = "draft12")]
2725            AnyFetchHeader::Draft12(_) => FetchReaderState::Draft12,
2726            #[cfg(feature = "draft13")]
2727            AnyFetchHeader::Draft13(_) => FetchReaderState::Draft13,
2728            #[cfg(feature = "draft14")]
2729            AnyFetchHeader::Draft14(_) => FetchReaderState::Draft14,
2730            // The header carries only a request id on drafts 15-23, so nothing
2731            // about it seeds the reader; the first object does.
2732            #[cfg(feature = "draft15")]
2733            AnyFetchHeader::Draft15(_) => {
2734                FetchReaderState::Draft15(crate::draft15::data_stream::FetchObjectReader::new())
2735            }
2736            #[cfg(feature = "draft16")]
2737            AnyFetchHeader::Draft16(_) => {
2738                FetchReaderState::Draft16(crate::draft16::data_stream::FetchObjectReader::new())
2739            }
2740            #[cfg(feature = "draft17")]
2741            AnyFetchHeader::Draft17(_) => {
2742                FetchReaderState::Draft17(crate::draft17::data_stream::FetchObjectReader::new())
2743            }
2744            #[cfg(feature = "draft18")]
2745            AnyFetchHeader::Draft18(_) => FetchReaderState::Draft18(
2746                crate::draft18::data_stream::FetchObjectReader::new(match group_order {
2747                    AnyFetchGroupOrder::Ascending => {
2748                        crate::draft18::data_stream::GroupOrder::Ascending
2749                    }
2750                    AnyFetchGroupOrder::Descending => {
2751                        crate::draft18::data_stream::GroupOrder::Descending
2752                    }
2753                }),
2754            ),
2755            #[cfg(feature = "draft19")]
2756            AnyFetchHeader::Draft19(_) => FetchReaderState::Draft19(
2757                crate::draft19::data_stream::FetchObjectReader::new(match group_order {
2758                    AnyFetchGroupOrder::Ascending => {
2759                        crate::draft19::data_stream::GroupOrder::Ascending
2760                    }
2761                    AnyFetchGroupOrder::Descending => {
2762                        crate::draft19::data_stream::GroupOrder::Descending
2763                    }
2764                }),
2765            ),
2766            #[cfg(feature = "draft20")]
2767            AnyFetchHeader::Draft20(_) => FetchReaderState::Draft20(
2768                crate::draft20::data_stream::FetchObjectReader::new(match group_order {
2769                    AnyFetchGroupOrder::Ascending => {
2770                        crate::draft20::data_stream::GroupOrder::Ascending
2771                    }
2772                    AnyFetchGroupOrder::Descending => {
2773                        crate::draft20::data_stream::GroupOrder::Descending
2774                    }
2775                }),
2776            ),
2777            #[cfg(feature = "draft21")]
2778            AnyFetchHeader::Draft21(_) => FetchReaderState::Draft21(
2779                crate::draft21::data_stream::FetchObjectReader::new(match group_order {
2780                    AnyFetchGroupOrder::Ascending => {
2781                        crate::draft21::data_stream::GroupOrder::Ascending
2782                    }
2783                    AnyFetchGroupOrder::Descending => {
2784                        crate::draft21::data_stream::GroupOrder::Descending
2785                    }
2786                }),
2787            ),
2788            #[cfg(feature = "draft22")]
2789            AnyFetchHeader::Draft22(_) => FetchReaderState::Draft22(
2790                crate::draft22::data_stream::FetchObjectReader::new(match group_order {
2791                    AnyFetchGroupOrder::Ascending => {
2792                        crate::draft22::data_stream::GroupOrder::Ascending
2793                    }
2794                    AnyFetchGroupOrder::Descending => {
2795                        crate::draft22::data_stream::GroupOrder::Descending
2796                    }
2797                }),
2798            ),
2799            #[cfg(feature = "draft23")]
2800            AnyFetchHeader::Draft23(_) => FetchReaderState::Draft23(
2801                crate::draft23::data_stream::FetchObjectReader::new(match group_order {
2802                    AnyFetchGroupOrder::Ascending => {
2803                        crate::draft23::data_stream::GroupOrder::Ascending
2804                    }
2805                    AnyFetchGroupOrder::Descending => {
2806                        crate::draft23::data_stream::GroupOrder::Descending
2807                    }
2808                }),
2809            ),
2810            #[allow(unreachable_patterns)]
2811            _ => {
2812                return Err(CodecError::UnsupportedDraft(format!(
2813                    "draft {:?} not enabled via feature flag",
2814                    header.draft()
2815                )));
2816            }
2817        };
2818        let gaps = FetchGapsState::for_draft(header.draft());
2819        Ok(Self { state, gaps })
2820    }
2821
2822    /// The draft this reader decodes.
2823    #[allow(unreachable_code)]
2824    pub fn draft(&self) -> DraftVersion {
2825        match &self.state {
2826            #[cfg(feature = "draft07")]
2827            FetchReaderState::Draft07 => DraftVersion::Draft07,
2828            #[cfg(feature = "draft08")]
2829            FetchReaderState::Draft08 => DraftVersion::Draft08,
2830            #[cfg(feature = "draft09")]
2831            FetchReaderState::Draft09 => DraftVersion::Draft09,
2832            #[cfg(feature = "draft10")]
2833            FetchReaderState::Draft10 => DraftVersion::Draft10,
2834            #[cfg(feature = "draft11")]
2835            FetchReaderState::Draft11 => DraftVersion::Draft11,
2836            #[cfg(feature = "draft12")]
2837            FetchReaderState::Draft12 => DraftVersion::Draft12,
2838            #[cfg(feature = "draft13")]
2839            FetchReaderState::Draft13 => DraftVersion::Draft13,
2840            #[cfg(feature = "draft14")]
2841            FetchReaderState::Draft14 => DraftVersion::Draft14,
2842            #[cfg(feature = "draft15")]
2843            FetchReaderState::Draft15(_) => DraftVersion::Draft15,
2844            #[cfg(feature = "draft16")]
2845            FetchReaderState::Draft16(_) => DraftVersion::Draft16,
2846            #[cfg(feature = "draft17")]
2847            FetchReaderState::Draft17(_) => DraftVersion::Draft17,
2848            #[cfg(feature = "draft18")]
2849            FetchReaderState::Draft18(_) => DraftVersion::Draft18,
2850            #[cfg(feature = "draft19")]
2851            FetchReaderState::Draft19(_) => DraftVersion::Draft19,
2852            #[cfg(feature = "draft20")]
2853            FetchReaderState::Draft20(_) => DraftVersion::Draft20,
2854            #[cfg(feature = "draft21")]
2855            FetchReaderState::Draft21(_) => DraftVersion::Draft21,
2856            #[cfg(feature = "draft22")]
2857            FetchReaderState::Draft22(_) => DraftVersion::Draft22,
2858            #[cfg(feature = "draft23")]
2859            FetchReaderState::Draft23(_) => DraftVersion::Draft23,
2860            #[allow(unreachable_patterns)]
2861            _ => unreachable!("AnyFetchObjectReader has no enabled variants"),
2862        }
2863    }
2864
2865    /// Decode the next fetch frame, including its payload.
2866    ///
2867    /// Returns [`CodecError::UnexpectedEnd`] when `buf` holds only part of a
2868    /// frame; the reader's state is unspecified after such an error, so callers
2869    /// that may be fed partial buffers must probe against a clone.
2870    ///
2871    /// Returns [`CodecError::InvalidField`] on drafts 15-23 when a frame takes
2872    /// a field from an object before it that does not exist — the first frame
2873    /// of a stream doing so is a protocol violation on every one of those
2874    /// drafts — and when a resolved Group ID, Subgroup ID or Object ID would
2875    /// leave the 64-bit range.
2876    ///
2877    /// Refuses a property block that does not parse, as
2878    /// [`crate::object_properties::parse_object_properties`] does;
2879    /// [`read_object_frame`](Self::read_object_frame) does not read inside it.
2880    ///
2881    /// Refuses, with [`CodecError::MalformedTrack`], the Prior gap conditions
2882    /// the stream shows on drafts 11-23, as each draft's `FetchGaps` does. The
2883    /// reader has moved past the Object when it does. Neither
2884    /// [`read_object_frame`](Self::read_object_frame) nor
2885    /// [`read_object_meta`](Self::read_object_meta) applies them, so a
2886    /// forwarder reading with those passes the Objects on as they arrived.
2887    #[allow(unused_variables, unreachable_code)]
2888    pub fn read_object(&mut self, buf: &mut impl Buf) -> Result<AnyFetchObject, CodecError> {
2889        let read: Result<AnyFetchObject, CodecError> = match &mut self.state {
2890            #[cfg(feature = "draft07")]
2891            FetchReaderState::Draft07 => fo07::read_object(buf),
2892            #[cfg(feature = "draft08")]
2893            FetchReaderState::Draft08 => fo08::read_object(buf),
2894            #[cfg(feature = "draft09")]
2895            FetchReaderState::Draft09 => fo09::read_object(buf),
2896            #[cfg(feature = "draft10")]
2897            FetchReaderState::Draft10 => fo10::read_object(buf),
2898            #[cfg(feature = "draft11")]
2899            FetchReaderState::Draft11 => fo11::read_object(buf),
2900            #[cfg(feature = "draft12")]
2901            FetchReaderState::Draft12 => fo12::read_object(buf),
2902            #[cfg(feature = "draft13")]
2903            FetchReaderState::Draft13 => fo13::read_object(buf),
2904            #[cfg(feature = "draft14")]
2905            FetchReaderState::Draft14 => fo14::read_object(buf),
2906            #[cfg(feature = "draft15")]
2907            FetchReaderState::Draft15(inner) => fo15::read_object(inner, buf),
2908            #[cfg(feature = "draft16")]
2909            FetchReaderState::Draft16(inner) => fo16::read_object(inner, buf),
2910            #[cfg(feature = "draft17")]
2911            FetchReaderState::Draft17(inner) => fo17::read_object(inner, buf),
2912            #[cfg(feature = "draft18")]
2913            FetchReaderState::Draft18(inner) => fo18::read_object(inner, buf),
2914            #[cfg(feature = "draft19")]
2915            FetchReaderState::Draft19(inner) => fo19::read_object(inner, buf),
2916            #[cfg(feature = "draft20")]
2917            FetchReaderState::Draft20(inner) => fo20::read_object(inner, buf),
2918            #[cfg(feature = "draft21")]
2919            FetchReaderState::Draft21(inner) => fo21::read_object(inner, buf),
2920            #[cfg(feature = "draft22")]
2921            FetchReaderState::Draft22(inner) => fo22::read_object(inner, buf),
2922            #[cfg(feature = "draft23")]
2923            FetchReaderState::Draft23(inner) => fo23::read_object(inner, buf),
2924            #[allow(unreachable_patterns)]
2925            _ => unreachable!("AnyFetchObjectReader has no enabled variants"),
2926        };
2927        let object = read?;
2928        crate::object_properties::parse_object_properties(self.draft(), &object.extension_headers)?;
2929        if object.end_of_range.is_none() {
2930            self.gaps.check(object.group_id, object.object_id, &object.extension_headers)?;
2931        }
2932        Ok(object)
2933    }
2934
2935    /// Decode the next fetch frame, keeping what re-encoding it later takes.
2936    ///
2937    /// Advances `buf` and this reader exactly as
2938    /// [`read_object_meta`](Self::read_object_meta) does, and reports the same
2939    /// framing in [`AnyFetchFrame::meta`]. What it additionally keeps is the
2940    /// shape the frame arrived in, which is the whole of what
2941    /// [`AnyFetchObjectWriter::reemit_object`] needs to write the frame back
2942    /// out against a different predecessor.
2943    ///
2944    /// Costs nothing over `read_object_meta`, which is itself defined over
2945    /// this: the per-draft header it keeps is one the decode produced and
2946    /// dropped.
2947    #[allow(unused_variables, unreachable_code)]
2948    pub fn read_object_frame(&mut self, buf: &mut impl Buf) -> Result<AnyFetchFrame, CodecError> {
2949        match &mut self.state {
2950            #[cfg(feature = "draft07")]
2951            FetchReaderState::Draft07 => fo07::read_object_meta(buf)
2952                .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft07, meta)),
2953            #[cfg(feature = "draft08")]
2954            FetchReaderState::Draft08 => fo08::read_object_meta(buf)
2955                .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft08, meta)),
2956            #[cfg(feature = "draft09")]
2957            FetchReaderState::Draft09 => fo09::read_object_meta(buf)
2958                .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft09, meta)),
2959            #[cfg(feature = "draft10")]
2960            FetchReaderState::Draft10 => fo10::read_object_meta(buf)
2961                .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft10, meta)),
2962            #[cfg(feature = "draft11")]
2963            FetchReaderState::Draft11 => fo11::read_object_meta(buf)
2964                .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft11, meta)),
2965            #[cfg(feature = "draft12")]
2966            FetchReaderState::Draft12 => fo12::read_object_meta(buf)
2967                .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft12, meta)),
2968            #[cfg(feature = "draft13")]
2969            FetchReaderState::Draft13 => fo13::read_object_meta(buf)
2970                .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft13, meta)),
2971            #[cfg(feature = "draft14")]
2972            FetchReaderState::Draft14 => fo14::read_object_meta(buf)
2973                .map(|meta| AnyFetchFrame::absolute(DraftVersion::Draft14, meta)),
2974            #[cfg(feature = "draft15")]
2975            FetchReaderState::Draft15(inner) => fo15::read_object_frame(inner, buf),
2976            #[cfg(feature = "draft16")]
2977            FetchReaderState::Draft16(inner) => fo16::read_object_frame(inner, buf),
2978            #[cfg(feature = "draft17")]
2979            FetchReaderState::Draft17(inner) => fo17::read_object_frame(inner, buf),
2980            #[cfg(feature = "draft18")]
2981            FetchReaderState::Draft18(inner) => fo18::read_object_frame(inner, buf),
2982            #[cfg(feature = "draft19")]
2983            FetchReaderState::Draft19(inner) => fo19::read_object_frame(inner, buf),
2984            #[cfg(feature = "draft20")]
2985            FetchReaderState::Draft20(inner) => fo20::read_object_frame(inner, buf),
2986            #[cfg(feature = "draft21")]
2987            FetchReaderState::Draft21(inner) => fo21::read_object_frame(inner, buf),
2988            #[cfg(feature = "draft22")]
2989            FetchReaderState::Draft22(inner) => fo22::read_object_frame(inner, buf),
2990            #[cfg(feature = "draft23")]
2991            FetchReaderState::Draft23(inner) => fo23::read_object_frame(inner, buf),
2992            #[allow(unreachable_patterns)]
2993            _ => unreachable!("AnyFetchObjectReader has no enabled variants"),
2994        }
2995    }
2996
2997    /// Decode the next fetch frame's framing without copying its payload.
2998    ///
2999    /// Advances `buf` past the whole frame exactly as
3000    /// [`read_object`](Self::read_object) does, and advances the same reader
3001    /// state, so the two are interchangeable on one stream.
3002    #[allow(unused_variables, unreachable_code)]
3003    pub fn read_object_meta(
3004        &mut self,
3005        buf: &mut impl Buf,
3006    ) -> Result<AnyFetchObjectMeta, CodecError> {
3007        match &mut self.state {
3008            #[cfg(feature = "draft07")]
3009            FetchReaderState::Draft07 => fo07::read_object_meta(buf),
3010            #[cfg(feature = "draft08")]
3011            FetchReaderState::Draft08 => fo08::read_object_meta(buf),
3012            #[cfg(feature = "draft09")]
3013            FetchReaderState::Draft09 => fo09::read_object_meta(buf),
3014            #[cfg(feature = "draft10")]
3015            FetchReaderState::Draft10 => fo10::read_object_meta(buf),
3016            #[cfg(feature = "draft11")]
3017            FetchReaderState::Draft11 => fo11::read_object_meta(buf),
3018            #[cfg(feature = "draft12")]
3019            FetchReaderState::Draft12 => fo12::read_object_meta(buf),
3020            #[cfg(feature = "draft13")]
3021            FetchReaderState::Draft13 => fo13::read_object_meta(buf),
3022            #[cfg(feature = "draft14")]
3023            FetchReaderState::Draft14 => fo14::read_object_meta(buf),
3024            #[cfg(feature = "draft15")]
3025            FetchReaderState::Draft15(inner) => fo15::read_object_meta(inner, buf),
3026            #[cfg(feature = "draft16")]
3027            FetchReaderState::Draft16(inner) => fo16::read_object_meta(inner, buf),
3028            #[cfg(feature = "draft17")]
3029            FetchReaderState::Draft17(inner) => fo17::read_object_meta(inner, buf),
3030            #[cfg(feature = "draft18")]
3031            FetchReaderState::Draft18(inner) => fo18::read_object_meta(inner, buf),
3032            #[cfg(feature = "draft19")]
3033            FetchReaderState::Draft19(inner) => fo19::read_object_meta(inner, buf),
3034            #[cfg(feature = "draft20")]
3035            FetchReaderState::Draft20(inner) => fo20::read_object_meta(inner, buf),
3036            #[cfg(feature = "draft21")]
3037            FetchReaderState::Draft21(inner) => fo21::read_object_meta(inner, buf),
3038            #[cfg(feature = "draft22")]
3039            FetchReaderState::Draft22(inner) => fo22::read_object_meta(inner, buf),
3040            #[cfg(feature = "draft23")]
3041            FetchReaderState::Draft23(inner) => fo23::read_object_meta(inner, buf),
3042            #[allow(unreachable_patterns)]
3043            _ => unreachable!("AnyFetchObjectReader has no enabled variants"),
3044        }
3045    }
3046}
3047
3048// ── Carrying a fetch frame from a reader to a writer ────────
3049
3050/// Per-draft capture of the shape one fetch frame arrived in.
3051///
3052/// Drafts 07-14 keep nothing: every field of a fetch object is on their wire
3053/// outright, so the bytes say the same thing whatever precedes them. Drafts
3054/// 15-23 keep the frame's own header, and draft-16 the resolved Location
3055/// beside it, because those two are exactly what each draft's
3056/// `FetchObjectWriter` is handed.
3057#[derive(Debug, Clone)]
3058enum FetchFrameShape {
3059    /// A frame whose fields are all absolute.
3060    #[cfg(any(
3061        feature = "draft07",
3062        feature = "draft08",
3063        feature = "draft09",
3064        feature = "draft10",
3065        feature = "draft11",
3066        feature = "draft12",
3067        feature = "draft13",
3068        feature = "draft14"
3069    ))]
3070    Absolute,
3071    #[cfg(feature = "draft15")]
3072    Draft15(crate::draft15::data_stream::FetchObjectHeader),
3073    #[cfg(feature = "draft16")]
3074    Draft16(
3075        crate::draft16::data_stream::FetchObjectHeader,
3076        crate::draft16::data_stream::FetchObjectLocation,
3077    ),
3078    #[cfg(feature = "draft17")]
3079    Draft17(crate::draft17::data_stream::FetchObject),
3080    #[cfg(feature = "draft18")]
3081    Draft18(crate::draft18::data_stream::FetchObject),
3082    #[cfg(feature = "draft19")]
3083    Draft19(crate::draft19::data_stream::FetchObject),
3084    #[cfg(feature = "draft20")]
3085    Draft20(crate::draft20::data_stream::FetchObject),
3086    #[cfg(feature = "draft21")]
3087    Draft21(crate::draft21::data_stream::FetchObject),
3088    #[cfg(feature = "draft22")]
3089    Draft22(crate::draft22::data_stream::FetchObject),
3090    #[cfg(feature = "draft23")]
3091    Draft23(crate::draft23::data_stream::FetchObject),
3092}
3093
3094/// One fetch frame, in the form re-encoding it takes.
3095///
3096/// Produced by [`AnyFetchObjectReader::read_object_frame`] and consumed by
3097/// [`AnyFetchObjectWriter::reemit_object`]. It is one value rather than two
3098/// because a frame's resolved identity and the shape it arrived in are only
3099/// meaningful together: the first says what the frame *is*, the second is the
3100/// encoding a writer keeps wherever it still says the same thing, and that is
3101/// what reproduces an untouched stream byte for byte.
3102#[derive(Debug, Clone)]
3103pub struct AnyFetchFrame {
3104    /// The framing, exactly as [`AnyFetchObjectReader::read_object_meta`]
3105    /// reports it.
3106    pub meta: AnyFetchObjectMeta,
3107    draft: DraftVersion,
3108    shape: FetchFrameShape,
3109}
3110
3111impl AnyFetchFrame {
3112    /// The draft whose stream this frame was read off.
3113    ///
3114    /// A writer refuses a frame from any other draft rather than re-encoding
3115    /// it: the two would agree on the resolved Location and disagree on
3116    /// everything the flags mean.
3117    #[must_use]
3118    pub fn draft(&self) -> DraftVersion {
3119        self.draft
3120    }
3121
3122    /// A frame from one of the drafts that keeps nothing.
3123    #[cfg(any(
3124        feature = "draft07",
3125        feature = "draft08",
3126        feature = "draft09",
3127        feature = "draft10",
3128        feature = "draft11",
3129        feature = "draft12",
3130        feature = "draft13",
3131        feature = "draft14"
3132    ))]
3133    fn absolute(draft: DraftVersion, meta: AnyFetchObjectMeta) -> Self {
3134        Self { meta, draft, shape: FetchFrameShape::Absolute }
3135    }
3136}
3137
3138// ── Fetch object writer ─────────────────────────────────────
3139
3140/// What [`AnyFetchObjectWriter::reemit_object`] had to do.
3141///
3142/// The counterpart of [`Reemit`], and deliberately not the same type. A
3143/// subgroup object's fix-up rewrites one leading varint and always writes the
3144/// whole object out; a fetch frame's is a re-encode of the whole header, and
3145/// the case worth having a shape for is the one where no re-encode is owed and
3146/// nothing is written at all.
3147#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3148pub enum FetchReemit {
3149    /// The frame's own bytes still encode it against the frame now in front of
3150    /// it, so **nothing was written** and the caller forwards `raw` untouched.
3151    ///
3152    /// This is every frame on a stream nothing was removed from, which is why
3153    /// it writes nothing: a relay that copied each frame through an output
3154    /// buffer to discover it would copy every payload it forwards.
3155    Unchanged,
3156    /// New framing was needed. The whole frame — the new framing followed by
3157    /// every byte of `raw` behind the old — was written to `out`, and the
3158    /// caller forwards that instead of `raw`.
3159    Reframed {
3160        /// Bytes the framing occupied in `raw`.
3161        framing_bytes_before: usize,
3162        /// Bytes it occupies in the output.
3163        framing_bytes_after: usize,
3164    },
3165}
3166
3167/// Per-draft writer state, mirroring [`FetchReaderState`].
3168#[derive(Debug, Clone)]
3169enum FetchWriterState {
3170    /// Drafts 07-14, which write every field of a fetch object outright and
3171    /// have nothing to write one *against*. The draft is carried so that a
3172    /// frame from another one is still refused.
3173    #[cfg(any(
3174        feature = "draft07",
3175        feature = "draft08",
3176        feature = "draft09",
3177        feature = "draft10",
3178        feature = "draft11",
3179        feature = "draft12",
3180        feature = "draft13",
3181        feature = "draft14"
3182    ))]
3183    Absolute(DraftVersion),
3184    #[cfg(feature = "draft15")]
3185    Draft15(crate::draft15::data_stream::FetchObjectWriter),
3186    #[cfg(feature = "draft16")]
3187    Draft16(crate::draft16::data_stream::FetchObjectWriter),
3188    #[cfg(feature = "draft17")]
3189    Draft17(crate::draft17::data_stream::FetchObjectWriter),
3190    #[cfg(feature = "draft18")]
3191    Draft18(crate::draft18::data_stream::FetchObjectWriter),
3192    #[cfg(feature = "draft19")]
3193    Draft19(crate::draft19::data_stream::FetchObjectWriter),
3194    #[cfg(feature = "draft20")]
3195    Draft20(crate::draft20::data_stream::FetchObjectWriter),
3196    #[cfg(feature = "draft21")]
3197    Draft21(crate::draft21::data_stream::FetchObjectWriter),
3198    #[cfg(feature = "draft22")]
3199    Draft22(crate::draft22::data_stream::FetchObjectWriter),
3200    #[cfg(feature = "draft23")]
3201    Draft23(crate::draft23::data_stream::FetchObjectWriter),
3202}
3203
3204/// Re-emitter for the frames of a fetch data stream, for any enabled draft.
3205///
3206/// The inverse of [`AnyFetchObjectReader`], and it exists for one caller: a
3207/// relay reading one fetch stream and writing another from the same frames,
3208/// having removed some of them. Removing a frame changes what the frames
3209/// behind it are encoded *against*, and on drafts 15-23 nearly every field of
3210/// a fetch object is defined against the frame before it — draft-17
3211/// Section 10.4.4.1, Table 7: "Object ID is the prior Object's ID plus one" —
3212/// so a survivor following a removed run cannot keep its original bytes.
3213///
3214/// # How it is driven
3215///
3216/// One writer belongs to one stream, and [`reemit_object`](Self::reemit_object)
3217/// is called **for every frame the caller forwards**, in wire order, whether or
3218/// not anything has been removed yet. That call is what moves the writer, so a
3219/// forwarded frame it never saw leaves it a frame behind and re-encodes the
3220/// next survivor against the wrong predecessor. A frame the caller *elides* is
3221/// the one it is not called for — that is the whole of eliding.
3222///
3223/// # What it costs
3224///
3225/// Nothing on drafts 07-14, and on drafts 15-23 one header re-derivation per
3226/// frame, which allocates only when the answer differs from the bytes that
3227/// arrived. A stream with nothing removed from it therefore forwards every
3228/// frame's own bytes and copies no payload.
3229///
3230/// # Drafts 18-23 need the Group Order
3231///
3232/// Their Group ID is a difference whose sign the fetch's Group Order decides,
3233/// exactly as for [`AnyFetchObjectReader`], and it is settled on the control
3234/// plane rather than on the data stream. [`new`](Self::new) takes it for that
3235/// reason: the wrong order re-encodes without error onto groups walking the
3236/// wrong way.
3237#[derive(Debug, Clone)]
3238pub struct AnyFetchObjectWriter {
3239    state: FetchWriterState,
3240}
3241
3242impl AnyFetchObjectWriter {
3243    /// Create a writer for a stream with the given header, whose groups are
3244    /// written in `group_order`.
3245    ///
3246    /// See the type's own documentation for what the order is for and why it
3247    /// cannot be read off the stream. On drafts 07-17 the argument is ignored.
3248    ///
3249    /// Returns [`CodecError::UnsupportedDraft`] for drafts not compiled in.
3250    #[allow(unused_variables, unreachable_code)]
3251    pub fn new(
3252        header: &AnyFetchHeader,
3253        group_order: AnyFetchGroupOrder,
3254    ) -> Result<Self, CodecError> {
3255        let state = match header {
3256            #[cfg(feature = "draft07")]
3257            AnyFetchHeader::Draft07(_) => FetchWriterState::Absolute(DraftVersion::Draft07),
3258            #[cfg(feature = "draft08")]
3259            AnyFetchHeader::Draft08(_) => FetchWriterState::Absolute(DraftVersion::Draft08),
3260            #[cfg(feature = "draft09")]
3261            AnyFetchHeader::Draft09(_) => FetchWriterState::Absolute(DraftVersion::Draft09),
3262            #[cfg(feature = "draft10")]
3263            AnyFetchHeader::Draft10(_) => FetchWriterState::Absolute(DraftVersion::Draft10),
3264            #[cfg(feature = "draft11")]
3265            AnyFetchHeader::Draft11(_) => FetchWriterState::Absolute(DraftVersion::Draft11),
3266            #[cfg(feature = "draft12")]
3267            AnyFetchHeader::Draft12(_) => FetchWriterState::Absolute(DraftVersion::Draft12),
3268            #[cfg(feature = "draft13")]
3269            AnyFetchHeader::Draft13(_) => FetchWriterState::Absolute(DraftVersion::Draft13),
3270            #[cfg(feature = "draft14")]
3271            AnyFetchHeader::Draft14(_) => FetchWriterState::Absolute(DraftVersion::Draft14),
3272            #[cfg(feature = "draft15")]
3273            AnyFetchHeader::Draft15(_) => {
3274                FetchWriterState::Draft15(crate::draft15::data_stream::FetchObjectWriter::new())
3275            }
3276            #[cfg(feature = "draft16")]
3277            AnyFetchHeader::Draft16(_) => {
3278                FetchWriterState::Draft16(crate::draft16::data_stream::FetchObjectWriter::new())
3279            }
3280            #[cfg(feature = "draft17")]
3281            AnyFetchHeader::Draft17(_) => {
3282                FetchWriterState::Draft17(crate::draft17::data_stream::FetchObjectWriter::new())
3283            }
3284            #[cfg(feature = "draft18")]
3285            AnyFetchHeader::Draft18(_) => FetchWriterState::Draft18(
3286                crate::draft18::data_stream::FetchObjectWriter::new(match group_order {
3287                    AnyFetchGroupOrder::Ascending => {
3288                        crate::draft18::data_stream::GroupOrder::Ascending
3289                    }
3290                    AnyFetchGroupOrder::Descending => {
3291                        crate::draft18::data_stream::GroupOrder::Descending
3292                    }
3293                }),
3294            ),
3295            #[cfg(feature = "draft19")]
3296            AnyFetchHeader::Draft19(_) => FetchWriterState::Draft19(
3297                crate::draft19::data_stream::FetchObjectWriter::new(match group_order {
3298                    AnyFetchGroupOrder::Ascending => {
3299                        crate::draft19::data_stream::GroupOrder::Ascending
3300                    }
3301                    AnyFetchGroupOrder::Descending => {
3302                        crate::draft19::data_stream::GroupOrder::Descending
3303                    }
3304                }),
3305            ),
3306            #[cfg(feature = "draft20")]
3307            AnyFetchHeader::Draft20(_) => FetchWriterState::Draft20(
3308                crate::draft20::data_stream::FetchObjectWriter::new(match group_order {
3309                    AnyFetchGroupOrder::Ascending => {
3310                        crate::draft20::data_stream::GroupOrder::Ascending
3311                    }
3312                    AnyFetchGroupOrder::Descending => {
3313                        crate::draft20::data_stream::GroupOrder::Descending
3314                    }
3315                }),
3316            ),
3317            #[cfg(feature = "draft21")]
3318            AnyFetchHeader::Draft21(_) => FetchWriterState::Draft21(
3319                crate::draft21::data_stream::FetchObjectWriter::new(match group_order {
3320                    AnyFetchGroupOrder::Ascending => {
3321                        crate::draft21::data_stream::GroupOrder::Ascending
3322                    }
3323                    AnyFetchGroupOrder::Descending => {
3324                        crate::draft21::data_stream::GroupOrder::Descending
3325                    }
3326                }),
3327            ),
3328            #[cfg(feature = "draft22")]
3329            AnyFetchHeader::Draft22(_) => FetchWriterState::Draft22(
3330                crate::draft22::data_stream::FetchObjectWriter::new(match group_order {
3331                    AnyFetchGroupOrder::Ascending => {
3332                        crate::draft22::data_stream::GroupOrder::Ascending
3333                    }
3334                    AnyFetchGroupOrder::Descending => {
3335                        crate::draft22::data_stream::GroupOrder::Descending
3336                    }
3337                }),
3338            ),
3339            #[cfg(feature = "draft23")]
3340            AnyFetchHeader::Draft23(_) => FetchWriterState::Draft23(
3341                crate::draft23::data_stream::FetchObjectWriter::new(match group_order {
3342                    AnyFetchGroupOrder::Ascending => {
3343                        crate::draft23::data_stream::GroupOrder::Ascending
3344                    }
3345                    AnyFetchGroupOrder::Descending => {
3346                        crate::draft23::data_stream::GroupOrder::Descending
3347                    }
3348                }),
3349            ),
3350            #[allow(unreachable_patterns)]
3351            _ => {
3352                return Err(CodecError::UnsupportedDraft(format!(
3353                    "draft {:?} not enabled via feature flag",
3354                    header.draft()
3355                )));
3356            }
3357        };
3358        Ok(Self { state })
3359    }
3360
3361    /// The draft this writer encodes.
3362    #[must_use]
3363    #[allow(unreachable_code)]
3364    pub fn draft(&self) -> DraftVersion {
3365        match &self.state {
3366            #[cfg(any(
3367                feature = "draft07",
3368                feature = "draft08",
3369                feature = "draft09",
3370                feature = "draft10",
3371                feature = "draft11",
3372                feature = "draft12",
3373                feature = "draft13",
3374                feature = "draft14"
3375            ))]
3376            FetchWriterState::Absolute(draft) => *draft,
3377            #[cfg(feature = "draft15")]
3378            FetchWriterState::Draft15(_) => DraftVersion::Draft15,
3379            #[cfg(feature = "draft16")]
3380            FetchWriterState::Draft16(_) => DraftVersion::Draft16,
3381            #[cfg(feature = "draft17")]
3382            FetchWriterState::Draft17(_) => DraftVersion::Draft17,
3383            #[cfg(feature = "draft18")]
3384            FetchWriterState::Draft18(_) => DraftVersion::Draft18,
3385            #[cfg(feature = "draft19")]
3386            FetchWriterState::Draft19(_) => DraftVersion::Draft19,
3387            #[cfg(feature = "draft20")]
3388            FetchWriterState::Draft20(_) => DraftVersion::Draft20,
3389            #[cfg(feature = "draft21")]
3390            FetchWriterState::Draft21(_) => DraftVersion::Draft21,
3391            #[cfg(feature = "draft22")]
3392            FetchWriterState::Draft22(_) => DraftVersion::Draft22,
3393            #[cfg(feature = "draft23")]
3394            FetchWriterState::Draft23(_) => DraftVersion::Draft23,
3395            #[allow(unreachable_patterns)]
3396            _ => unreachable!("AnyFetchObjectWriter has no enabled variants"),
3397        }
3398    }
3399
3400    /// Re-emit one forwarded fetch frame, re-encoding its framing against the
3401    /// frames actually forwarded before it, and advance.
3402    ///
3403    /// `frame` came from [`AnyFetchObjectReader::read_object_frame`] on the
3404    /// stream being read; `raw` is that frame's wire bytes. The return value
3405    /// says which bytes to forward, and the two answers are not symmetric:
3406    /// [`FetchReemit::Unchanged`] writes nothing and means `raw` is still
3407    /// correct, while [`FetchReemit::Reframed`] has written the whole frame to
3408    /// `out` and `raw` must not also be forwarded.
3409    ///
3410    /// # `raw` need not be a complete frame
3411    ///
3412    /// **Any prefix is legal provided the whole framing is present** — the
3413    /// framing being `meta.wire_len - meta.payload_length` bytes, which is a
3414    /// number the frame already carries. Everything behind it is copied
3415    /// byte-for-byte, however many bytes there are, and no length validation is
3416    /// performed. That is what lets a caller fix up the first chunk of a frame
3417    /// too large to buffer, where only the first chunk carries the framing at
3418    /// all.
3419    ///
3420    /// # Errors
3421    ///
3422    /// [`CodecError::UnsupportedDraft`] when `frame` was read off another
3423    /// draft's stream.
3424    ///
3425    /// [`CodecError::InvalidField`] when `raw` is shorter than the framing the
3426    /// frame declares, and when the frame has no encoding against the
3427    /// predecessor now in front of it — a Group ID that moves against the
3428    /// Group Order, an Object ID that does not advance, and the arithmetic
3429    /// overflows. The writer is left where it was in that case, so a caller
3430    /// that gives up on one frame and carries on is not also one frame out.
3431    #[allow(unused_variables)]
3432    pub fn reemit_object(
3433        &mut self,
3434        frame: &AnyFetchFrame,
3435        raw: &[u8],
3436        out: &mut impl BufMut,
3437    ) -> Result<FetchReemit, CodecError> {
3438        if frame.draft != self.draft() {
3439            return Err(CodecError::UnsupportedDraft(format!(
3440                "a draft {:?} fetch frame cannot be written onto a draft {:?} stream",
3441                frame.draft,
3442                self.draft()
3443            )));
3444        }
3445
3446        let framing_len = frame.meta.wire_len.saturating_sub(frame.meta.payload_length);
3447        let framing_len = usize::try_from(framing_len).map_err(|_| CodecError::InvalidField)?;
3448        if framing_len > raw.len() {
3449            return Err(CodecError::InvalidField);
3450        }
3451        let (framing, rest) = raw.split_at(framing_len);
3452
3453        match (&mut self.state, &frame.shape) {
3454            // Nothing on these drafts' wire is written against anything, so
3455            // the frame's own bytes are correct wherever it lands.
3456            #[cfg(any(
3457                feature = "draft07",
3458                feature = "draft08",
3459                feature = "draft09",
3460                feature = "draft10",
3461                feature = "draft11",
3462                feature = "draft12",
3463                feature = "draft13",
3464                feature = "draft14"
3465            ))]
3466            (FetchWriterState::Absolute(_), FetchFrameShape::Absolute) => {
3467                Ok(FetchReemit::Unchanged)
3468            }
3469            #[cfg(feature = "draft15")]
3470            (FetchWriterState::Draft15(writer), FetchFrameShape::Draft15(original)) => {
3471                let reframed = writer.header_for(original)?;
3472                if reframed == *original {
3473                    writer.advance(original);
3474                    return Ok(FetchReemit::Unchanged);
3475                }
3476                let mut encoded = Vec::with_capacity(framing.len() + 16);
3477                reframed.encode(&mut encoded)?;
3478                writer.advance(&reframed);
3479                Ok(put_reframed(&encoded, framing.len(), rest, out))
3480            }
3481            #[cfg(feature = "draft16")]
3482            (FetchWriterState::Draft16(writer), FetchFrameShape::Draft16(original, location)) => {
3483                let reframed = writer.header_for(original, location)?;
3484                if reframed == *original {
3485                    writer.advance(original, location);
3486                    return Ok(FetchReemit::Unchanged);
3487                }
3488                let mut encoded = Vec::with_capacity(framing.len() + 16);
3489                reframed.encode(&mut encoded)?;
3490                writer.advance(&reframed, location);
3491                Ok(put_reframed(&encoded, framing.len(), rest, out))
3492            }
3493            #[cfg(feature = "draft17")]
3494            (FetchWriterState::Draft17(writer), FetchFrameShape::Draft17(original)) => {
3495                let reframed = writer.header_for(original)?;
3496                if reframed == original.header {
3497                    writer.advance(original);
3498                    return Ok(FetchReemit::Unchanged);
3499                }
3500                let mut encoded = Vec::with_capacity(framing.len() + 16);
3501                reframed.encode(&mut encoded)?;
3502                writer.advance(original);
3503                Ok(put_reframed(&encoded, framing.len(), rest, out))
3504            }
3505            #[cfg(feature = "draft18")]
3506            (FetchWriterState::Draft18(writer), FetchFrameShape::Draft18(original)) => {
3507                let reframed = writer.header_for(original)?;
3508                if reframed == original.header {
3509                    writer.advance(original);
3510                    return Ok(FetchReemit::Unchanged);
3511                }
3512                let mut encoded = Vec::with_capacity(framing.len() + 16);
3513                reframed.encode(&mut encoded)?;
3514                writer.advance(original);
3515                Ok(put_reframed(&encoded, framing.len(), rest, out))
3516            }
3517            #[cfg(feature = "draft19")]
3518            (FetchWriterState::Draft19(writer), FetchFrameShape::Draft19(original)) => {
3519                let reframed = writer.header_for(original)?;
3520                if reframed == original.header {
3521                    writer.advance(original);
3522                    return Ok(FetchReemit::Unchanged);
3523                }
3524                let mut encoded = Vec::with_capacity(framing.len() + 16);
3525                reframed.encode(&mut encoded)?;
3526                writer.advance(original);
3527                Ok(put_reframed(&encoded, framing.len(), rest, out))
3528            }
3529            #[cfg(feature = "draft20")]
3530            (FetchWriterState::Draft20(writer), FetchFrameShape::Draft20(original)) => {
3531                let reframed = writer.header_for(original)?;
3532                if reframed == original.header {
3533                    writer.advance(original);
3534                    return Ok(FetchReemit::Unchanged);
3535                }
3536                let mut encoded = Vec::with_capacity(framing.len() + 16);
3537                reframed.encode(&mut encoded)?;
3538                writer.advance(original);
3539                Ok(put_reframed(&encoded, framing.len(), rest, out))
3540            }
3541            #[cfg(feature = "draft21")]
3542            (FetchWriterState::Draft21(writer), FetchFrameShape::Draft21(original)) => {
3543                let reframed = writer.header_for(original)?;
3544                if reframed == original.header {
3545                    writer.advance(original);
3546                    return Ok(FetchReemit::Unchanged);
3547                }
3548                let mut encoded = Vec::with_capacity(framing.len() + 16);
3549                reframed.encode(&mut encoded)?;
3550                writer.advance(original);
3551                Ok(put_reframed(&encoded, framing.len(), rest, out))
3552            }
3553            #[cfg(feature = "draft22")]
3554            (FetchWriterState::Draft22(writer), FetchFrameShape::Draft22(original)) => {
3555                let reframed = writer.header_for(original)?;
3556                if reframed == original.header {
3557                    writer.advance(original);
3558                    return Ok(FetchReemit::Unchanged);
3559                }
3560                let mut encoded = Vec::with_capacity(framing.len() + 16);
3561                reframed.encode(&mut encoded)?;
3562                writer.advance(original);
3563                Ok(put_reframed(&encoded, framing.len(), rest, out))
3564            }
3565            #[cfg(feature = "draft23")]
3566            (FetchWriterState::Draft23(writer), FetchFrameShape::Draft23(original)) => {
3567                let reframed = writer.header_for(original)?;
3568                if reframed == original.header {
3569                    writer.advance(original);
3570                    return Ok(FetchReemit::Unchanged);
3571                }
3572                let mut encoded = Vec::with_capacity(framing.len() + 16);
3573                reframed.encode(&mut encoded)?;
3574                writer.advance(original);
3575                Ok(put_reframed(&encoded, framing.len(), rest, out))
3576            }
3577            // Unreachable: the drafts were compared before this match, and one
3578            // draft has one state and one shape.
3579            #[allow(unreachable_patterns)]
3580            _ => Err(CodecError::UnsupportedDraft(format!(
3581                "no fetch writer for draft {:?}",
3582                frame.draft
3583            ))),
3584        }
3585    }
3586}
3587
3588/// Write a re-encoded frame out: the new framing, then every byte that stood
3589/// behind the old one.
3590#[cfg(any(
3591    feature = "draft15",
3592    feature = "draft16",
3593    feature = "draft17",
3594    feature = "draft18",
3595    feature = "draft19",
3596    feature = "draft20",
3597    feature = "draft21",
3598    feature = "draft22",
3599    feature = "draft23"
3600))]
3601fn put_reframed(
3602    encoded: &[u8],
3603    framing_bytes_before: usize,
3604    rest: &[u8],
3605    out: &mut impl BufMut,
3606) -> FetchReemit {
3607    out.put_slice(encoded);
3608    out.put_slice(rest);
3609    FetchReemit::Reframed { framing_bytes_before, framing_bytes_after: encoded.len() }
3610}
3611
3612// ── Header helpers for the stream-type-gated drafts ─────────
3613
3614/// Generates the subgroup stream-type check drafts 11-13 share: the header's
3615/// stream type must be a subgroup type, and it decides whether objects carry
3616/// an extension block.
3617macro_rules! subgroup_extensions_fn {
3618    ($name:ident, $feat:literal, $draft:ident) => {
3619        #[cfg(feature = $feat)]
3620        fn $name(header: &crate::$draft::data_stream::SubgroupHeader) -> Result<bool, CodecError> {
3621            if !header.stream_type.is_subgroup() {
3622                return Err(CodecError::InvalidField);
3623            }
3624            Ok(header.stream_type.has_extensions())
3625        }
3626    };
3627}
3628
3629subgroup_extensions_fn!(subgroup_extensions_11, "draft11", draft11);
3630subgroup_extensions_fn!(subgroup_extensions_12, "draft12", draft12);
3631subgroup_extensions_fn!(subgroup_extensions_13, "draft13", draft13);