moqtap_codec/draft19/
data_stream.rs

1//! Draft-19 data stream header encoding and decoding.
2//!
3//! Byte-for-byte identical to draft-18; only Object Property scope and the
4//! object-status payload rule changed semantically, not the wire layout.
5//!
6//! Subgroup header type byte: bit 4 must be set; valid type ranges are
7//! 0x10..0x1F, 0x30..0x3F, 0x50..0x5F, 0x70..0x7F.
8//!   - bit 0 (0x01): PROPERTIES
9//!   - bits 1-2 (0x06): SUBGROUP_ID_MODE (0=zero, 1=first_obj, 2=explicit, 3=reserved)
10//!   - bit 3 (0x08): END_OF_GROUP
11//!   - bit 5 (0x20): DEFAULT_PRIORITY (no priority byte)
12//!   - bit 6 (0x40): FIRST_OBJECT
13//!
14//! Datagram type byte: 0b00X0XXXX (bit 4 always 0)
15//!   - bit 0 (0x01): PROPERTIES
16//!   - bit 1 (0x02): END_OF_GROUP
17//!   - bit 2 (0x04): ZERO_OBJECT_ID (object_id=0, field omitted)
18//!   - bit 3 (0x08): DEFAULT_PRIORITY (no priority byte)
19//!   - bit 5 (0x20): STATUS (status byte replaces payload)
20//!
21//! Fetch header: stream type 0x05 + request_id. Fetch objects use
22//! delta-encoded Group ID and Object ID (the first object's deltas are
23//! interpreted as absolute values).
24
25use bytes::{Buf, BufMut};
26
27use crate::error::CodecError;
28use crate::varint::VarInt;
29
30// ── Subgroup ──────────────────────────────────────────────────
31
32const SUBGROUP_PROPERTIES_BIT: u8 = 0x01;
33const SUBGROUP_ID_MODE_MASK: u8 = 0x06;
34const SUBGROUP_END_OF_GROUP_BIT: u8 = 0x08;
35const SUBGROUP_BASE_BIT: u8 = 0x10;
36const SUBGROUP_DEFAULT_PRIORITY_BIT: u8 = 0x20;
37const SUBGROUP_FIRST_OBJECT_BIT: u8 = 0x40;
38
39#[derive(Debug, Clone)]
40pub struct SubgroupHeader {
41    pub header_type: u8,
42    pub track_alias: VarInt,
43    pub group_id: VarInt,
44    pub subgroup_id: VarInt,
45    pub publisher_priority: Option<u8>,
46}
47
48impl SubgroupHeader {
49    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
50        if buf.remaining() < 1 {
51            return Err(CodecError::UnexpectedEnd);
52        }
53        let header_type = buf.get_u8();
54
55        // Validate: bit 4 must be set
56        if header_type & SUBGROUP_BASE_BIT == 0 {
57            return Err(CodecError::InvalidField);
58        }
59
60        let track_alias = VarInt::decode(buf)?;
61        let group_id = VarInt::decode(buf)?;
62
63        let subgroup_id_mode = (header_type & SUBGROUP_ID_MODE_MASK) >> 1;
64        let subgroup_id = match subgroup_id_mode {
65            0 => VarInt::from_u64(0).unwrap(),
66            2 => VarInt::decode(buf)?,
67            // Modes 1 and 3: mode 1 = first object's ID (resolved later),
68            // mode 3 = reserved. Store 0 for now.
69            _ => VarInt::from_u64(0).unwrap(),
70        };
71
72        let publisher_priority = if header_type & SUBGROUP_DEFAULT_PRIORITY_BIT == 0 {
73            if buf.remaining() < 1 {
74                return Err(CodecError::UnexpectedEnd);
75            }
76            Some(buf.get_u8())
77        } else {
78            None
79        };
80
81        Ok(SubgroupHeader { header_type, track_alias, group_id, subgroup_id, publisher_priority })
82    }
83
84    pub fn encode(&self, buf: &mut impl BufMut) {
85        buf.put_u8(self.header_type);
86        self.track_alias.encode(buf);
87        self.group_id.encode(buf);
88
89        let subgroup_id_mode = (self.header_type & SUBGROUP_ID_MODE_MASK) >> 1;
90        if subgroup_id_mode == 2 {
91            self.subgroup_id.encode(buf);
92        }
93
94        if self.header_type & SUBGROUP_DEFAULT_PRIORITY_BIT == 0 {
95            buf.put_u8(self.publisher_priority.unwrap_or(128));
96        }
97    }
98
99    pub fn has_properties(&self) -> bool {
100        self.header_type & SUBGROUP_PROPERTIES_BIT != 0
101    }
102
103    pub fn is_end_of_group(&self) -> bool {
104        self.header_type & SUBGROUP_END_OF_GROUP_BIT != 0
105    }
106
107    /// `true` when the FIRST_OBJECT bit (0x40) is set, signaling the first
108    /// object on this stream is the original publisher's first object in the
109    /// subgroup. Added in draft-18.
110    pub fn is_first_object(&self) -> bool {
111        self.header_type & SUBGROUP_FIRST_OBJECT_BIT != 0
112    }
113}
114
115// ── Subgroup objects (stateful) ───────────────────────────────
116
117/// One object within a draft-18 subgroup stream. Object IDs are
118/// delta-encoded; whether a per-object "properties" block (the draft-18
119/// equivalent of extension headers) is present depends on the PROPERTIES
120/// bit on the enclosing [`SubgroupHeader`]. Use [`SubgroupObjectReader`]
121/// to encode/decode.
122#[derive(Debug, Clone)]
123pub struct SubgroupObject {
124    pub object_id: VarInt,
125    /// Raw properties bytes (empty unless the subgroup header sets the
126    /// PROPERTIES bit). When present, holds the `ext_count` varint
127    /// followed by each property's `key`, `vlen`, and value.
128    pub extension_headers: Vec<u8>,
129    pub payload_length: VarInt,
130    pub object_status: Option<VarInt>,
131    pub payload: Vec<u8>,
132}
133
134#[derive(Debug, Clone)]
135pub struct SubgroupObjectReader {
136    extensions_present: bool,
137    prev_object_id: Option<u64>,
138}
139
140impl SubgroupObjectReader {
141    pub fn new(header: &SubgroupHeader) -> Self {
142        Self { extensions_present: header.has_properties(), prev_object_id: None }
143    }
144
145    pub fn read_object(&mut self, buf: &mut impl Buf) -> Result<SubgroupObject, CodecError> {
146        let delta = VarInt::decode(buf)?.into_inner();
147        let object_id_val = match self.prev_object_id {
148            None => delta,
149            Some(prev) => prev
150                .checked_add(1)
151                .and_then(|v| v.checked_add(delta))
152                .ok_or(CodecError::InvalidField)?,
153        };
154        self.prev_object_id = Some(object_id_val);
155        let object_id = VarInt::from_u64(object_id_val).map_err(|_| CodecError::InvalidField)?;
156
157        let extension_headers = if self.extensions_present {
158            let mut out: Vec<u8> = Vec::new();
159            let ext_count = VarInt::decode(buf)?;
160            ext_count.encode(&mut out);
161            let count = ext_count.into_inner();
162            for _ in 0..count {
163                let key = VarInt::decode(buf)?;
164                let vlen = VarInt::decode(buf)?;
165                let vlen_usize = vlen.into_inner() as usize;
166                if buf.remaining() < vlen_usize {
167                    return Err(CodecError::UnexpectedEnd);
168                }
169                key.encode(&mut out);
170                vlen.encode(&mut out);
171                let value = buf.copy_to_bytes(vlen_usize);
172                out.extend_from_slice(&value);
173            }
174            out
175        } else {
176            Vec::new()
177        };
178
179        let payload_length_vi = VarInt::decode(buf)?;
180        let payload_length_val = payload_length_vi.into_inner() as usize;
181        let (object_status, payload) = if payload_length_val == 0 {
182            let status = VarInt::decode(buf)?;
183            (Some(status), Vec::new())
184        } else {
185            let payload = crate::types::read_bytes(buf, payload_length_val)?;
186            (None, payload)
187        };
188
189        Ok(SubgroupObject {
190            object_id,
191            extension_headers,
192            payload_length: payload_length_vi,
193            object_status,
194            payload,
195        })
196    }
197
198    pub fn write_object(
199        &mut self,
200        object: &SubgroupObject,
201        buf: &mut impl BufMut,
202    ) -> Result<(), CodecError> {
203        let oid = object.object_id.into_inner();
204        let delta = match self.prev_object_id {
205            None => oid,
206            Some(prev) => oid
207                .checked_sub(prev)
208                .and_then(|v| v.checked_sub(1))
209                .ok_or(CodecError::InvalidField)?,
210        };
211        VarInt::from_u64(delta).map_err(|_| CodecError::InvalidField)?.encode(buf);
212        if self.extensions_present {
213            buf.put_slice(&object.extension_headers);
214        }
215        object.payload_length.encode(buf);
216        if object.payload_length.into_inner() == 0 {
217            if let Some(s) = &object.object_status {
218                s.encode(buf);
219            } else {
220                VarInt::from_u64(0).unwrap().encode(buf);
221            }
222        } else {
223            buf.put_slice(&object.payload);
224        }
225        self.prev_object_id = Some(oid);
226        Ok(())
227    }
228}
229
230// ── Datagram ──────────────────────────────────────────────────
231
232const DATAGRAM_PROPERTIES_BIT: u8 = 0x01;
233const DATAGRAM_END_OF_GROUP_BIT: u8 = 0x02;
234const DATAGRAM_ZERO_OBJECT_ID_BIT: u8 = 0x04;
235const DATAGRAM_DEFAULT_PRIORITY_BIT: u8 = 0x08;
236const DATAGRAM_STATUS_BIT: u8 = 0x20;
237
238#[derive(Debug, Clone)]
239pub struct DatagramHeader {
240    pub datagram_type: u8,
241    pub track_alias: VarInt,
242    pub group_id: VarInt,
243    pub object_id: VarInt,
244    pub publisher_priority: Option<u8>,
245    pub object_status: Option<u8>,
246}
247
248impl DatagramHeader {
249    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
250        if buf.remaining() < 1 {
251            return Err(CodecError::UnexpectedEnd);
252        }
253        let datagram_type = buf.get_u8();
254
255        let track_alias = VarInt::decode(buf)?;
256        let group_id = VarInt::decode(buf)?;
257
258        let object_id = if datagram_type & DATAGRAM_ZERO_OBJECT_ID_BIT != 0 {
259            VarInt::from_usize(0)
260        } else {
261            VarInt::decode(buf)?
262        };
263
264        let publisher_priority = if datagram_type & DATAGRAM_DEFAULT_PRIORITY_BIT == 0 {
265            if buf.remaining() < 1 {
266                return Err(CodecError::UnexpectedEnd);
267            }
268            Some(buf.get_u8())
269        } else {
270            None
271        };
272
273        // Skip properties if present
274        if datagram_type & DATAGRAM_PROPERTIES_BIT != 0 {
275            let props_len = VarInt::decode(buf)?.into_inner() as usize;
276            if buf.remaining() < props_len {
277                return Err(CodecError::UnexpectedEnd);
278            }
279            buf.advance(props_len);
280        }
281
282        let object_status = if datagram_type & DATAGRAM_STATUS_BIT != 0 {
283            if buf.remaining() < 1 {
284                return Err(CodecError::UnexpectedEnd);
285            }
286            Some(buf.get_u8())
287        } else {
288            None
289        };
290
291        Ok(DatagramHeader {
292            datagram_type,
293            track_alias,
294            group_id,
295            object_id,
296            publisher_priority,
297            object_status,
298        })
299    }
300
301    pub fn encode(&self, buf: &mut impl BufMut) {
302        buf.put_u8(self.datagram_type);
303        self.track_alias.encode(buf);
304        self.group_id.encode(buf);
305
306        if self.datagram_type & DATAGRAM_ZERO_OBJECT_ID_BIT == 0 {
307            self.object_id.encode(buf);
308        }
309
310        if self.datagram_type & DATAGRAM_DEFAULT_PRIORITY_BIT == 0 {
311            buf.put_u8(self.publisher_priority.unwrap_or(128));
312        }
313
314        if self.datagram_type & DATAGRAM_STATUS_BIT != 0 {
315            buf.put_u8(self.object_status.unwrap_or(0));
316        }
317    }
318
319    pub fn is_end_of_group(&self) -> bool {
320        self.datagram_type & DATAGRAM_END_OF_GROUP_BIT != 0
321    }
322
323    pub fn has_status(&self) -> bool {
324        self.datagram_type & DATAGRAM_STATUS_BIT != 0
325    }
326}
327
328// ── Fetch Header ──────────────────────────────────────────────
329
330const FETCH_STREAM_TYPE: u64 = 0x05;
331
332#[derive(Debug, Clone)]
333pub struct FetchHeader {
334    pub request_id: VarInt,
335}
336
337impl FetchHeader {
338    pub fn decode(buf: &mut impl Buf) -> Result<Self, CodecError> {
339        let stream_type = VarInt::decode(buf)?.into_inner();
340        if stream_type != FETCH_STREAM_TYPE {
341            return Err(CodecError::InvalidField);
342        }
343        let request_id = VarInt::decode(buf)?;
344        Ok(FetchHeader { request_id })
345    }
346
347    pub fn encode(&self, buf: &mut impl BufMut) {
348        VarInt::from_usize(FETCH_STREAM_TYPE as usize).encode(buf);
349        self.request_id.encode(buf);
350    }
351}
moqtap_codec/draft19/data_stream.rs

moqtap_codec/draft19/
data_stream.rs
