Skip to main content

arcly_stream/protocol/
rtp.rs

1//! Shared RTP/RTCP parsing and NAL-codec (de)packetization — RFC 3550 framing
2//! plus H.264 (RFC 6184) and H.265 (RFC 7798) payload formats.
3//!
4//! Gated behind the internal `_rtp` marker, pulled in by both [`rtsp`] and
5//! [`webrtc`]. The two transports differ only in how RTP packets reach the
6//! process (TCP-interleaved / UDP for RTSP, DTLS-SRTP for WebRTC); once a packet
7//! is in hand, reassembling NAL units into an Annex-B access unit is identical,
8//! so it lives here once.
9//!
10//! [`rtsp`]: crate::protocol::rtsp
11//! [`webrtc`]: crate::protocol::webrtc
12//!
13//! # What it does
14//!
15//! - [`RtpHeader::parse`] decodes the fixed RTP header (RFC 3550 §5.1), honoring
16//!   the CSRC count and the extension-header flag to locate the payload.
17//! - [`H264Depacketizer`] / [`H265Depacketizer`] turn a sequence of RTP payloads
18//!   into complete access units in Annex-B form. Each handles the three NALU
19//!   packetization modes for its codec — single NAL, aggregation (STAP-A type 24
20//!   / AP type 48), and fragmentation (FU-A type 28 / FU type 49) — emitting an
21//!   access unit when the marker bit is set or the timestamp advances.
22//! - [`RtpPacketizer`] performs the reverse for egress (e.g. WebRTC WHEP),
23//!   selecting the H.264 or H.265 payload format.
24//!
25//! # What it does not do
26//!
27//! Jitter-buffer reordering and loss concealment are the caller's concern — the
28//! depacketizer assumes in-order delivery (true for TCP-interleaved RTSP; for
29//! UDP/SRTP a small reorder buffer should sit in front of it). It reports a
30//! [`DepacketizeError::OutOfOrder`] gap so a handler can request a keyframe
31//! (PLI/FIR) rather than emit a corrupt access unit.
32
33use bytes::Bytes;
34
35/// Annex-B start code prefixed to every reassembled NAL unit.
36const ANNEXB_START: [u8; 4] = [0, 0, 0, 1];
37
38/// A parsed RTP fixed header (RFC 3550 §5.1).
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub struct RtpHeader {
41    /// Payload type (7 bits) — identifies the codec/format binding from SDP.
42    pub payload_type: u8,
43    /// Marker bit. For H.264 it flags the last packet of an access unit.
44    pub marker: bool,
45    /// 16-bit sequence number, increments by one per packet (wraps).
46    pub sequence: u16,
47    /// 32-bit media timestamp in the payload's clock (90 kHz for H.264 video).
48    pub timestamp: u32,
49    /// Synchronization source identifier.
50    pub ssrc: u32,
51    /// Byte offset at which the payload begins (past CSRCs and any extension).
52    pub payload_offset: usize,
53}
54
55impl RtpHeader {
56    /// Parse the fixed header from the front of `buf`, returning the header and
57    /// the payload offset. Returns `None` if `buf` is too short or the version
58    /// field is not 2.
59    pub fn parse(buf: &[u8]) -> Option<RtpHeader> {
60        use super::byteops::ByteReader;
61        let mut r = ByteReader::new(buf);
62        let b0 = r.u8()?;
63        if b0 >> 6 != 2 {
64            return None; // RTP version must be 2
65        }
66        let has_extension = b0 & 0x10 != 0;
67        let csrc_count = (b0 & 0x0F) as usize;
68        let b1 = r.u8()?;
69        let marker = b1 & 0x80 != 0;
70        let payload_type = b1 & 0x7F;
71        let sequence = r.u16_be()?;
72        let timestamp = r.u32_be()?;
73        let ssrc = r.u32_be()?;
74        r.skip(csrc_count * 4)?; // CSRC list
75
76        if has_extension {
77            // Extension header: 2-byte profile id, 2-byte length (in 32-bit words).
78            r.skip(2)?;
79            let ext_words = r.u16_be()? as usize;
80            r.skip(ext_words * 4)?;
81        }
82        Some(RtpHeader {
83            payload_type,
84            marker,
85            sequence,
86            timestamp,
87            ssrc,
88            payload_offset: r.position(),
89        })
90    }
91}
92
93/// Extract the value of RTP header extension `ext_id` from `buf`, per the
94/// RFC 8285 one-byte (`0xBEDE`) and two-byte (`0x100x`) forms. Returns `None` if
95/// the packet has no extension block or no element with that local id.
96///
97/// Used to read the **RID** (`urn:ietf:params:rtp-hdrext:sdes:rid`) that labels a
98/// simulcast layer — the value is the ASCII rid string.
99pub fn rtp_extension_value(buf: &[u8], ext_id: u8) -> Option<&[u8]> {
100    if buf.len() < 12 || buf[0] >> 6 != 2 || buf[0] & 0x10 == 0 {
101        return None; // not RTPv2 or no extension flag
102    }
103    let csrc = (buf[0] & 0x0F) as usize;
104    let mut p = 12 + csrc * 4;
105    if buf.len() < p + 4 {
106        return None;
107    }
108    let profile = u16::from_be_bytes([buf[p], buf[p + 1]]);
109    let words = u16::from_be_bytes([buf[p + 2], buf[p + 3]]) as usize;
110    p += 4;
111    let end = p.checked_add(words * 4)?;
112    let ext = buf.get(p..end)?;
113
114    if profile == 0xBEDE {
115        // One-byte form: each element is `(id<<4 | len-1)` then `len` bytes.
116        let mut i = 0;
117        while i < ext.len() {
118            let b = ext[i];
119            if b == 0 {
120                i += 1; // padding
121                continue;
122            }
123            let id = b >> 4;
124            let len = (b & 0x0F) as usize + 1;
125            i += 1;
126            let value = ext.get(i..i + len)?;
127            if id == ext_id {
128                return Some(value);
129            }
130            i += len;
131        }
132    } else if profile & 0xFFF0 == 0x1000 {
133        // Two-byte form: each element is `id`, `len`, then `len` bytes.
134        let mut i = 0;
135        while i + 1 < ext.len() {
136            let id = ext[i];
137            let len = ext[i + 1] as usize;
138            i += 2;
139            if id == 0 {
140                continue; // padding
141            }
142            let value = ext.get(i..i + len)?;
143            if id == ext_id {
144                return Some(value);
145            }
146            i += len;
147        }
148    }
149    None
150}
151
152/// Decode the **client-to-mixer audio level** RTP header extension (RFC 6464,
153/// `urn:ietf:params:rtp-hdrext:ssrc-audio-level`) carried under `ext_id`.
154///
155/// Returns `(level_dbov, voice_active)` where `level_dbov` is the audio level in
156/// -dBov (0 = loudest, 127 = silence) and `voice_active` is the voice-activity
157/// flag (the extension's most-significant bit). `None` when the packet carries no
158/// such extension. This is the per-packet input a conference uses to pick the
159/// active (dominant) speaker.
160pub fn audio_level(buf: &[u8], ext_id: u8) -> Option<(u8, bool)> {
161    let b = *rtp_extension_value(buf, ext_id)?.first()?;
162    Some((b & 0x7F, b & 0x80 != 0))
163}
164
165/// Errors surfaced while depacketizing an RTP stream.
166#[derive(Debug, Clone, Copy, PartialEq, Eq)]
167#[non_exhaustive]
168pub enum DepacketizeError {
169    /// The packet was shorter than the format requires.
170    Truncated,
171    /// A sequence-number discontinuity was detected mid-fragment; the partial
172    /// access unit was dropped. The handler should request a keyframe.
173    OutOfOrder,
174    /// An unsupported NAL/aggregation type was encountered.
175    Unsupported(u8),
176}
177
178/// Depacketizes RFC 3640 AAC-hbr RTP payloads into raw AAC access units.
179///
180/// The common RTSP/SDP profile for AAC (`mode=AAC-hbr`, `sizelength=13`,
181/// `indexlength=3`) frames each payload as a 2-byte **AU-headers-length** (in
182/// bits), followed by one 2-byte AU-header per access unit (13-bit size +
183/// 3-bit index), followed by the access units concatenated. One RTP packet may
184/// carry several AAC frames; [`push`](Self::push) returns each as a separate
185/// raw (ADTS-less) [`bytes::Bytes`].
186#[derive(Debug, Clone, Copy, Default)]
187pub struct AacDepacketizer {
188    /// Bits per AU-header `size` field (13 for AAC-hbr).
189    size_length: u8,
190    /// Bits per AU-header `index`/`index-delta` field (3 for AAC-hbr).
191    index_length: u8,
192}
193
194impl AacDepacketizer {
195    /// A depacketizer for the standard AAC-hbr profile (`sizelength=13`,
196    /// `indexlength=3`).
197    pub fn new() -> Self {
198        Self {
199            size_length: 13,
200            index_length: 3,
201        }
202    }
203
204    /// A depacketizer with explicit AU-header field widths from the SDP `fmtp`.
205    pub fn with_lengths(size_length: u8, index_length: u8) -> Self {
206        Self {
207            size_length,
208            index_length,
209        }
210    }
211
212    /// Split one RTP AAC-hbr payload into its constituent access units.
213    pub fn push(&self, payload: &[u8]) -> Result<Vec<Bytes>, DepacketizeError> {
214        if payload.len() < 2 {
215            return Err(DepacketizeError::Truncated);
216        }
217        // Sizes wider than a 16-bit AU-header field are unsupported (and would
218        // otherwise over-shift below). `with_lengths` can supply arbitrary widths.
219        if self.size_length == 0 || self.size_length > 16 {
220            return Err(DepacketizeError::Unsupported(self.size_length));
221        }
222        let header_bits = u16::from_be_bytes([payload[0], payload[1]]) as usize;
223        let au_header_bits = self.size_length as usize + self.index_length as usize;
224        if au_header_bits == 0 {
225            return Err(DepacketizeError::Unsupported(0));
226        }
227        let header_bytes = header_bits.div_ceil(8);
228        let au_count = header_bits / au_header_bits;
229        let headers = payload
230            .get(2..2 + header_bytes)
231            .ok_or(DepacketizeError::Truncated)?;
232        let mut data_off = 2 + header_bytes;
233        let mut out = Vec::with_capacity(au_count);
234        for i in 0..au_count {
235            // Each AU-header is `au_header_bits` wide; for AAC-hbr that is 16
236            // bits, so the size is the top `size_length` bits of a 2-byte field.
237            let bit = i * au_header_bits;
238            let byte = bit / 8;
239            let hdr = headers
240                .get(byte..byte + 2)
241                .ok_or(DepacketizeError::Truncated)?;
242            let size = (u16::from_be_bytes([hdr[0], hdr[1]]) >> (16 - self.size_length)) as usize;
243            let end = data_off + size;
244            let au = payload
245                .get(data_off..end)
246                .ok_or(DepacketizeError::Truncated)?;
247            out.push(Bytes::copy_from_slice(au));
248            data_off = end;
249        }
250        Ok(out)
251    }
252}
253
254/// Packetizes H.264 Annex-B access units into RFC 6184 RTP packets — the inverse
255/// of [`H264Depacketizer`], used for WebRTC/WHEP egress.
256///
257/// Each NAL unit that fits the MTU is sent as a single-NAL packet; larger NALs
258/// are split into FU-A fragments. The RTP marker bit is set on the last packet
259/// of each access unit so the receiver knows the frame is complete.
260#[derive(Debug, Clone)]
261pub struct RtpPacketizer {
262    payload_type: u8,
263    ssrc: u32,
264    sequence: u16,
265    /// Maximum RTP payload size (excluding the 12-byte header).
266    max_payload: usize,
267    /// Which NAL-based RTP payload format to emit (H.264 RFC 6184 vs H.265
268    /// RFC 7798 — they differ in NAL-header width and FU framing).
269    codec: NalCodec,
270}
271
272/// The NAL-based codecs [`RtpPacketizer`] / the depacketizers understand. Both
273/// are Annex-B start-code framed; they differ in NAL-header width (1 vs 2 bytes)
274/// and fragmentation-unit layout.
275#[derive(Debug, Clone, Copy, PartialEq, Eq)]
276enum NalCodec {
277    H264,
278    H265,
279}
280
281impl RtpPacketizer {
282    /// An H.264 packetizer for `payload_type`/`ssrc`. `mtu` is the maximum UDP
283    /// payload (1200 is the WebRTC-safe default); the 12-byte RTP header is
284    /// subtracted.
285    pub fn new(payload_type: u8, ssrc: u32, mtu: usize) -> Self {
286        Self::with_codec(payload_type, ssrc, mtu, NalCodec::H264)
287    }
288
289    /// An H.265 (HEVC) packetizer, emitting the RFC 7798 payload format
290    /// (2-byte NAL header, FU type 49). Counterpart to [`new`](Self::new).
291    pub fn new_h265(payload_type: u8, ssrc: u32, mtu: usize) -> Self {
292        Self::with_codec(payload_type, ssrc, mtu, NalCodec::H265)
293    }
294
295    fn with_codec(payload_type: u8, ssrc: u32, mtu: usize, codec: NalCodec) -> Self {
296        Self {
297            payload_type,
298            ssrc,
299            sequence: 0,
300            max_payload: mtu.saturating_sub(12).max(1),
301            codec,
302        }
303    }
304
305    /// Emit one RTP packet: recycle a buffer from `recycle` (reusing its heap
306    /// allocation), write the 12-byte header, let `fill` append the payload, and
307    /// push it onto `out`. Advances the sequence number.
308    fn emit(
309        &mut self,
310        recycle: &mut Vec<Vec<u8>>,
311        out: &mut Vec<Vec<u8>>,
312        marker: bool,
313        timestamp: u32,
314        fill: impl FnOnce(&mut Vec<u8>),
315    ) {
316        let mut buf = recycle.pop().unwrap_or_default();
317        buf.clear();
318        write_rtp_header(
319            &mut buf,
320            self.payload_type,
321            marker,
322            self.sequence,
323            timestamp,
324            self.ssrc,
325        );
326        self.sequence = self.sequence.wrapping_add(1);
327        fill(&mut buf);
328        out.push(buf);
329    }
330
331    /// Packetize one Annex-B access unit at `timestamp` (90 kHz) into RTP packets.
332    ///
333    /// Each NAL that fits the MTU is sent as a single-NAL packet; larger NALs are
334    /// fragmented (FU-A for H.264, FU for H.265). The marker bit is set on the
335    /// last packet of the access unit.
336    ///
337    /// Allocating variant; prefer [`packetize_into`](Self::packetize_into) on the
338    /// egress hot path to recycle packet buffers across frames.
339    pub fn packetize(&mut self, access_unit: &[u8], timestamp: u32) -> Vec<Vec<u8>> {
340        let mut out = Vec::new();
341        self.packetize_into(access_unit, timestamp, &mut out);
342        out
343    }
344
345    /// Packetize into a caller-owned buffer, recycling the `Vec<u8>` allocations
346    /// it already holds from a previous frame.
347    ///
348    /// The hot-path contract: pass the *same* `out` every frame. On entry the
349    /// previously-produced (already-sent) packet buffers are reclaimed as a free
350    /// pool and refilled in place, so steady-state egress performs no per-packet
351    /// heap allocation.
352    pub fn packetize_into(&mut self, access_unit: &[u8], timestamp: u32, out: &mut Vec<Vec<u8>>) {
353        // Reclaim last frame's buffers (now consumed by the caller) as a pool.
354        let mut recycle = std::mem::take(out);
355        // Both codecs are Annex-B start-code framed; the shared scanner yields the
356        // NAL units (without start codes) in order.
357        let nals: Vec<&[u8]> = crate::codec::nal::iter_nals(access_unit)
358            .filter(|n| !n.is_empty())
359            .collect();
360        for (i, nal) in nals.iter().enumerate() {
361            let last_nal = i + 1 == nals.len();
362            if nal.len() <= self.max_payload {
363                // Single NAL unit packet; marker on the final NAL of the AU.
364                self.emit(&mut recycle, out, last_nal, timestamp, |b| {
365                    b.extend_from_slice(nal)
366                });
367            } else {
368                match self.codec {
369                    NalCodec::H264 => {
370                        self.fragment_fua(nal, timestamp, last_nal, &mut recycle, out)
371                    }
372                    NalCodec::H265 => {
373                        self.fragment_fu_h265(nal, timestamp, last_nal, &mut recycle, out)
374                    }
375                }
376            }
377        }
378    }
379
380    /// Split one oversized NAL into FU-A fragments (RFC 6184 §5.8).
381    fn fragment_fua(
382        &mut self,
383        nal: &[u8],
384        timestamp: u32,
385        last_nal: bool,
386        recycle: &mut Vec<Vec<u8>>,
387        out: &mut Vec<Vec<u8>>,
388    ) {
389        let nal_header = nal[0];
390        let fu_indicator = (nal_header & 0xE0) | 28; // F|NRI from NAL, type 28
391        let nal_type = nal_header & 0x1F;
392        let body = &nal[1..];
393        // Each fragment carries a 2-byte FU header (indicator + FU header).
394        let chunk = self.max_payload.saturating_sub(2).max(1);
395        let n_chunks = body.len().div_ceil(chunk);
396        for (idx, part) in body.chunks(chunk).enumerate() {
397            let start = idx == 0;
398            let end = idx + 1 == n_chunks;
399            let mut fu_header = nal_type;
400            if start {
401                fu_header |= 0x80;
402            }
403            if end {
404                fu_header |= 0x40;
405            }
406            // Marker only on the very last fragment of the final NAL of the AU.
407            self.emit(recycle, out, last_nal && end, timestamp, |pkt| {
408                pkt.push(fu_indicator);
409                pkt.push(fu_header);
410                pkt.extend_from_slice(part);
411            });
412        }
413    }
414
415    /// Split one oversized H.265 NAL into FU fragments (RFC 7798 §4.4.3).
416    ///
417    /// The 2-byte NAL header becomes a PayloadHdr with its type field set to 49
418    /// (the F/LayerId/TID bits are preserved); each fragment then carries a
419    /// 1-byte FU header (`S | E | FuType`) where `FuType` is the original type.
420    fn fragment_fu_h265(
421        &mut self,
422        nal: &[u8],
423        timestamp: u32,
424        last_nal: bool,
425        recycle: &mut Vec<Vec<u8>>,
426        out: &mut Vec<Vec<u8>>,
427    ) {
428        // A well-formed H.265 NAL has a 2-byte header; anything shorter can't be
429        // fragmented meaningfully, so emit it as a single packet.
430        if nal.len() < 2 {
431            self.emit(recycle, out, last_nal, timestamp, |pkt| {
432                pkt.extend_from_slice(nal)
433            });
434            return;
435        }
436        let nal_type = (nal[0] >> 1) & 0x3F;
437        // PayloadHdr: keep F (bit 15) + LayerId + TID, overwrite the 6-bit type
438        // field with 49 (FU). Type occupies bits 9..14, i.e. (byte0 >> 1) & 0x3F.
439        let payload_hdr0 = (nal[0] & 0x81) | (49 << 1);
440        let payload_hdr1 = nal[1];
441        let body = &nal[2..];
442        // Each fragment carries the 2-byte PayloadHdr + 1-byte FU header.
443        let chunk = self.max_payload.saturating_sub(3).max(1);
444        let n_chunks = body.len().div_ceil(chunk);
445        for (idx, part) in body.chunks(chunk).enumerate() {
446            let start = idx == 0;
447            let end = idx + 1 == n_chunks;
448            let mut fu_header = nal_type;
449            if start {
450                fu_header |= 0x80;
451            }
452            if end {
453                fu_header |= 0x40;
454            }
455            self.emit(recycle, out, last_nal && end, timestamp, |pkt| {
456                pkt.push(payload_hdr0);
457                pkt.push(payload_hdr1);
458                pkt.push(fu_header);
459                pkt.extend_from_slice(part);
460            });
461        }
462    }
463}
464
465/// Reassembles RFC 6184 H.264 RTP payloads into Annex-B access units.
466///
467/// Feed each packet's payload (the bytes after [`RtpHeader::payload_offset`])
468/// with its marker bit and timestamp to [`push`](Self::push). When a complete
469/// access unit is ready the method returns `Ok(Some(au))`, where `au` is the
470/// concatenated NAL units each prefixed with a 4-byte Annex-B start code —
471/// exactly the shape the codec parsers and `annexb_to_avcc` expect.
472#[derive(Debug, Default)]
473pub struct H264Depacketizer {
474    /// Bytes accumulated for the current access unit (Annex-B framed).
475    au: Vec<u8>,
476    /// FU-A reassembly buffer for the NAL currently being defragmented.
477    fua: Vec<u8>,
478    /// `true` while an FU-A fragment is in progress (between Start and End bits).
479    in_fragment: bool,
480    /// Reconstructed NAL header byte for the in-progress FU-A NAL.
481    fua_header: u8,
482    /// Timestamp of the access unit currently being assembled.
483    current_ts: Option<u32>,
484    /// Last sequence number seen (for gap detection during fragmentation).
485    last_seq: Option<u16>,
486}
487
488impl H264Depacketizer {
489    /// A fresh depacketizer with no in-progress access unit.
490    pub fn new() -> Self {
491        Self::default()
492    }
493
494    /// Append one NAL unit (Annex-B framed) to the current access unit.
495    fn append_nal(&mut self, nal: &[u8]) {
496        self.au.extend_from_slice(&ANNEXB_START);
497        self.au.extend_from_slice(nal);
498    }
499
500    /// Whether the pending access unit holds an IDR (type 5) NAL — a keyframe.
501    fn pending_is_keyframe(&self) -> bool {
502        // Scan the assembled Annex-B for a NAL header with type 5.
503        let mut i = 0;
504        while i + 4 < self.au.len() {
505            if self.au[i..i + 4] == ANNEXB_START {
506                let nal_type = self.au[i + 4] & 0x1F;
507                if nal_type == 5 {
508                    return true;
509                }
510            }
511            i += 1;
512        }
513        false
514    }
515
516    /// Emit and reset the pending access unit, if any.
517    fn take_au(&mut self) -> Option<AccessUnit> {
518        if self.au.is_empty() {
519            return None;
520        }
521        let keyframe = self.pending_is_keyframe();
522        let timestamp = self.current_ts.unwrap_or(0);
523        let data = Bytes::from(std::mem::take(&mut self.au));
524        self.current_ts = None;
525        Some(AccessUnit {
526            data,
527            timestamp,
528            keyframe,
529        })
530    }
531
532    /// Push one RTP H.264 payload. Returns a completed [`AccessUnit`] when the
533    /// marker bit closes the frame (or the timestamp advances to a new one).
534    pub fn push(
535        &mut self,
536        payload: &[u8],
537        marker: bool,
538        timestamp: u32,
539        sequence: u16,
540    ) -> Result<Option<AccessUnit>, DepacketizeError> {
541        if payload.is_empty() {
542            return Err(DepacketizeError::Truncated);
543        }
544
545        // A timestamp change flushes the previous access unit before starting the
546        // new one (some encoders omit the marker bit).
547        let mut completed = None;
548        if let Some(ts) = self.current_ts {
549            if ts != timestamp && !self.in_fragment {
550                completed = self.take_au();
551            }
552        }
553        self.current_ts = Some(timestamp);
554
555        let nal_type = payload[0] & 0x1F;
556        match nal_type {
557            1..=23 => {
558                // Single NAL unit packet — the payload *is* the NAL.
559                self.append_nal(payload);
560            }
561            24 => {
562                // STAP-A: one byte type, then [u16 size][nal]… aggregates.
563                let mut i = 1;
564                while i + 2 <= payload.len() {
565                    let size = u16::from_be_bytes([payload[i], payload[i + 1]]) as usize;
566                    i += 2;
567                    if i + size > payload.len() {
568                        return Err(DepacketizeError::Truncated);
569                    }
570                    self.append_nal(&payload[i..i + size]);
571                    i += size;
572                }
573            }
574            28 => {
575                // FU-A: byte0 = FU indicator, byte1 = FU header (S|E|R|type).
576                if payload.len() < 2 {
577                    return Err(DepacketizeError::Truncated);
578                }
579                let fu_header = payload[1];
580                let start = fu_header & 0x80 != 0;
581                let end = fu_header & 0x40 != 0;
582                let frag_type = fu_header & 0x1F;
583
584                if start {
585                    // Reconstruct the original NAL header: F|NRI from the indicator,
586                    // type from the FU header.
587                    self.fua_header = (payload[0] & 0xE0) | frag_type;
588                    self.fua.clear();
589                    self.fua.push(self.fua_header);
590                    self.in_fragment = true;
591                } else if !self.in_fragment {
592                    // Mid/last fragment with no start — lost the head.
593                    return Err(DepacketizeError::OutOfOrder);
594                } else if self.seq_gap(sequence) {
595                    self.in_fragment = false;
596                    self.fua.clear();
597                    return Err(DepacketizeError::OutOfOrder);
598                }
599                self.fua.extend_from_slice(&payload[2..]);
600
601                if end && self.in_fragment {
602                    let nal = std::mem::take(&mut self.fua);
603                    self.append_nal(&nal);
604                    self.in_fragment = false;
605                }
606            }
607            other => return Err(DepacketizeError::Unsupported(other)),
608        }
609
610        self.last_seq = Some(sequence);
611
612        if completed.is_some() {
613            return Ok(completed);
614        }
615        if marker {
616            return Ok(self.take_au());
617        }
618        Ok(None)
619    }
620
621    /// Detect a one-step sequence-number gap relative to the previous packet.
622    fn seq_gap(&self, sequence: u16) -> bool {
623        match self.last_seq {
624            Some(prev) => sequence.wrapping_sub(prev) != 1,
625            None => false,
626        }
627    }
628}
629
630/// Reassembles RFC 7798 H.265 (HEVC) RTP payloads into Annex-B access units.
631///
632/// The H.265 counterpart to [`H264Depacketizer`], used for ingesting HEVC IP
633/// cameras and encoders over RTSP/WebRTC. It handles the three packetization
634/// modes: single NAL units, aggregation packets (AP, type 48), and fragmentation
635/// units (FU, type 49). The output shape — NAL units each prefixed with a 4-byte
636/// Annex-B start code — matches [`H264Depacketizer`] and the codec parsers.
637///
638/// DONL/DOND fields (only present when `sprop-max-don-diff > 0` is negotiated in
639/// SDP) are not consumed; the common single-stream profile does not use them.
640#[derive(Debug, Default)]
641pub struct H265Depacketizer {
642    /// Bytes accumulated for the current access unit (Annex-B framed).
643    au: Vec<u8>,
644    /// FU reassembly buffer for the NAL currently being defragmented.
645    fu: Vec<u8>,
646    /// `true` while an FU is in progress (between Start and End bits).
647    in_fragment: bool,
648    /// Timestamp of the access unit currently being assembled.
649    current_ts: Option<u32>,
650    /// Last sequence number seen (for gap detection during fragmentation).
651    last_seq: Option<u16>,
652}
653
654impl H265Depacketizer {
655    /// A fresh depacketizer with no in-progress access unit.
656    pub fn new() -> Self {
657        Self::default()
658    }
659
660    /// Append one NAL unit (Annex-B framed) to the current access unit.
661    fn append_nal(&mut self, nal: &[u8]) {
662        self.au.extend_from_slice(&ANNEXB_START);
663        self.au.extend_from_slice(nal);
664    }
665
666    /// Whether the pending access unit holds an IRAP (BLA/IDR/CRA, types 16–23)
667    /// VCL NAL — i.e. a random-access point / keyframe.
668    fn pending_is_keyframe(&self) -> bool {
669        let mut i = 0;
670        while i + 4 < self.au.len() {
671            if self.au[i..i + 4] == ANNEXB_START {
672                let nal_type = (self.au[i + 4] >> 1) & 0x3F;
673                if (16..=23).contains(&nal_type) {
674                    return true;
675                }
676            }
677            i += 1;
678        }
679        false
680    }
681
682    /// Emit and reset the pending access unit, if any.
683    fn take_au(&mut self) -> Option<AccessUnit> {
684        if self.au.is_empty() {
685            return None;
686        }
687        let keyframe = self.pending_is_keyframe();
688        let timestamp = self.current_ts.unwrap_or(0);
689        let data = Bytes::from(std::mem::take(&mut self.au));
690        self.current_ts = None;
691        Some(AccessUnit {
692            data,
693            timestamp,
694            keyframe,
695        })
696    }
697
698    /// Detect a one-step sequence-number gap relative to the previous packet.
699    fn seq_gap(&self, sequence: u16) -> bool {
700        match self.last_seq {
701            Some(prev) => sequence.wrapping_sub(prev) != 1,
702            None => false,
703        }
704    }
705
706    /// Push one RTP H.265 payload. Returns a completed [`AccessUnit`] when the
707    /// marker bit closes the frame (or the timestamp advances to a new one).
708    pub fn push(
709        &mut self,
710        payload: &[u8],
711        marker: bool,
712        timestamp: u32,
713        sequence: u16,
714    ) -> Result<Option<AccessUnit>, DepacketizeError> {
715        // The H.265 NAL header is two bytes; a single byte cannot carry a type.
716        if payload.len() < 2 {
717            return Err(DepacketizeError::Truncated);
718        }
719
720        // A timestamp change flushes the previous access unit (some encoders omit
721        // the marker bit).
722        let mut completed = None;
723        if let Some(ts) = self.current_ts {
724            if ts != timestamp && !self.in_fragment {
725                completed = self.take_au();
726            }
727        }
728        self.current_ts = Some(timestamp);
729
730        let nal_type = (payload[0] >> 1) & 0x3F;
731        match nal_type {
732            // Single NAL unit packet — the payload *is* the NAL (header included).
733            0..=47 => self.append_nal(payload),
734            48 => {
735                // AP: 2-byte header, then [u16 size][nal]… aggregates.
736                let mut i = 2;
737                while i + 2 <= payload.len() {
738                    let size = u16::from_be_bytes([payload[i], payload[i + 1]]) as usize;
739                    i += 2;
740                    if i + size > payload.len() {
741                        return Err(DepacketizeError::Truncated);
742                    }
743                    self.append_nal(&payload[i..i + size]);
744                    i += size;
745                }
746            }
747            49 => {
748                // FU: 2-byte PayloadHdr, 1-byte FU header (S|E|FuType), then body.
749                if payload.len() < 3 {
750                    return Err(DepacketizeError::Truncated);
751                }
752                let fu_header = payload[2];
753                let start = fu_header & 0x80 != 0;
754                let end = fu_header & 0x40 != 0;
755                let fu_type = fu_header & 0x3F;
756
757                if start {
758                    // Reconstruct the original 2-byte NAL header: restore the type
759                    // field (bits 9..14) from FuType, keep F/LayerId/TID.
760                    let hdr0 = (payload[0] & 0x81) | (fu_type << 1);
761                    let hdr1 = payload[1];
762                    self.fu.clear();
763                    self.fu.push(hdr0);
764                    self.fu.push(hdr1);
765                    self.in_fragment = true;
766                } else if !self.in_fragment {
767                    return Err(DepacketizeError::OutOfOrder);
768                } else if self.seq_gap(sequence) {
769                    self.in_fragment = false;
770                    self.fu.clear();
771                    return Err(DepacketizeError::OutOfOrder);
772                }
773                self.fu.extend_from_slice(&payload[3..]);
774
775                if end && self.in_fragment {
776                    let nal = std::mem::take(&mut self.fu);
777                    self.append_nal(&nal);
778                    self.in_fragment = false;
779                }
780            }
781            other => return Err(DepacketizeError::Unsupported(other)),
782        }
783
784        self.last_seq = Some(sequence);
785
786        if completed.is_some() {
787            return Ok(completed);
788        }
789        if marker {
790            return Ok(self.take_au());
791        }
792        Ok(None)
793    }
794}
795
796/// A reassembled coded video frame from the RTP bus.
797///
798/// For the NAL codecs (H.264/H.265) `data` is the access unit in Annex-B form
799/// (each NAL prefixed with a 4-byte start code); for VP9/AV1 it is the raw coded
800/// frame / temporal unit. `keyframe` marks a decodable random-access point.
801#[derive(Debug, Clone, PartialEq, Eq)]
802pub struct AccessUnit {
803    /// The coded frame bytes (Annex-B NALs for H.26x; raw frame for VP9/AV1).
804    pub data: Bytes,
805    /// RTP media timestamp (90 kHz) of the frame.
806    pub timestamp: u32,
807    /// Whether this is a keyframe / random-access point.
808    pub keyframe: bool,
809}
810
811/// Write the 12-byte RTP fixed header (V=2, no padding/extension/CSRC) for one
812/// packet. Shared by every packetizer in this module.
813fn write_rtp_header(out: &mut Vec<u8>, pt: u8, marker: bool, seq: u16, ts: u32, ssrc: u32) {
814    out.push(0x80); // V=2, P=0, X=0, CC=0
815    out.push(if marker { 0x80 } else { 0 } | (pt & 0x7F));
816    out.extend_from_slice(&seq.to_be_bytes());
817    out.extend_from_slice(&ts.to_be_bytes());
818    out.extend_from_slice(&ssrc.to_be_bytes());
819}
820
821// ── Opus (RFC 7587) ──────────────────────────────────────────────────────────
822
823/// Packetizes Opus audio into RTP: each Opus packet is carried verbatim as the
824/// payload of one RTP packet (RFC 7587 — Opus is self-delimiting, so there is no
825/// payload descriptor). The 48 kHz media clock means the caller passes a
826/// timestamp already scaled to 48 kHz.
827#[derive(Debug, Clone)]
828pub struct OpusPacketizer {
829    payload_type: u8,
830    ssrc: u32,
831    sequence: u16,
832}
833
834impl OpusPacketizer {
835    /// An Opus packetizer for `payload_type`/`ssrc`.
836    pub fn new(payload_type: u8, ssrc: u32) -> Self {
837        Self {
838            payload_type,
839            ssrc,
840            sequence: 0,
841        }
842    }
843
844    /// Packetize one Opus frame at `timestamp` (48 kHz) into a single RTP packet,
845    /// appended to `out` (recycling a buffer it already holds). The RTP marker is
846    /// left clear — continuous audio is not a talkspurt boundary.
847    pub fn packetize_into(&mut self, opus: &[u8], timestamp: u32, out: &mut Vec<Vec<u8>>) {
848        let mut recycle = std::mem::take(out);
849        let mut pkt = recycle.pop().unwrap_or_default();
850        pkt.clear();
851        write_rtp_header(
852            &mut pkt,
853            self.payload_type,
854            false,
855            self.sequence,
856            timestamp,
857            self.ssrc,
858        );
859        self.sequence = self.sequence.wrapping_add(1);
860        pkt.extend_from_slice(opus);
861        out.push(pkt);
862    }
863}
864
865// ── VP9 (draft-ietf-payload-vp9) ─────────────────────────────────────────────
866
867/// Packetizes VP9 coded frames into RTP, using a flexible-mode-off payload
868/// descriptor for the common single-layer (non-scalable) case.
869///
870/// Each frame is carried verbatim after a VP9 payload descriptor (the bytes are
871/// not transformed — VP9 RTP carries the frame opaquely), split across the MTU
872/// with the B (begin) bit on the first packet and E (end) + RTP marker on the
873/// last. A 15-bit picture ID increments per frame. Spatial/temporal scalability
874/// and flexible mode are out of scope.
875#[derive(Debug, Clone)]
876pub struct Vp9Packetizer {
877    payload_type: u8,
878    ssrc: u32,
879    sequence: u16,
880    max_payload: usize,
881    picture_id: u16,
882}
883
884impl Vp9Packetizer {
885    /// A VP9 packetizer for `payload_type`/`ssrc`. `mtu` is the maximum UDP
886    /// payload; the 12-byte RTP header and a 3-byte descriptor are subtracted.
887    pub fn new(payload_type: u8, ssrc: u32, mtu: usize) -> Self {
888        Self {
889            payload_type,
890            ssrc,
891            sequence: 0,
892            // 12-byte RTP header + up to 3-byte descriptor (1 flags + 2 picture id).
893            max_payload: mtu.saturating_sub(12 + 3).max(1),
894            picture_id: 0,
895        }
896    }
897
898    /// Packetize one VP9 frame at `timestamp` (90 kHz). `keyframe` clears the P
899    /// (inter-predicted) bit so receivers can identify random-access points.
900    pub fn packetize(&mut self, frame: &[u8], timestamp: u32, keyframe: bool) -> Vec<Vec<u8>> {
901        let mut out = Vec::new();
902        self.packetize_into(frame, timestamp, keyframe, &mut out);
903        out
904    }
905
906    /// Recycling variant of [`packetize`](Self::packetize): pass the same `out`
907    /// every frame to reuse the packet-buffer allocations across frames.
908    pub fn packetize_into(
909        &mut self,
910        frame: &[u8],
911        timestamp: u32,
912        keyframe: bool,
913        out: &mut Vec<Vec<u8>>,
914    ) {
915        let pid = self.picture_id & 0x7FFF;
916        self.picture_id = self.picture_id.wrapping_add(1);
917
918        let mut recycle = std::mem::take(out);
919        let chunks: Vec<&[u8]> = if frame.is_empty() {
920            vec![&[]]
921        } else {
922            frame.chunks(self.max_payload).collect()
923        };
924        let n = chunks.len();
925        for (i, chunk) in chunks.into_iter().enumerate() {
926            let begin = i == 0;
927            let end = i + 1 == n;
928            let mut pkt = recycle.pop().unwrap_or_default();
929            pkt.clear();
930            write_rtp_header(
931                &mut pkt,
932                self.payload_type,
933                end,
934                self.sequence,
935                timestamp,
936                self.ssrc,
937            );
938            self.sequence = self.sequence.wrapping_add(1);
939
940            // Descriptor octet: I=1, P=!keyframe, L=0, F=0, B, E, V=0, Z=0.
941            let mut desc0 = 0x80; // I = 1 (picture ID present)
942            if !keyframe {
943                desc0 |= 0x40; // P (inter-predicted)
944            }
945            if begin {
946                desc0 |= 0x08; // B (start of frame)
947            }
948            if end {
949                desc0 |= 0x04; // E (end of frame)
950            }
951            pkt.push(desc0);
952            // 15-bit picture ID (M=1): 0x80|hi, lo.
953            pkt.push(0x80 | (pid >> 8) as u8);
954            pkt.push((pid & 0xFF) as u8);
955            pkt.extend_from_slice(chunk);
956            out.push(pkt);
957        }
958    }
959}
960
961/// Reassembles VP9 RTP payloads (draft-ietf-payload-vp9, non-flexible single
962/// layer) into coded frames. Counterpart to [`Vp9Packetizer`].
963#[derive(Debug, Default)]
964pub struct Vp9Depacketizer {
965    frame: Vec<u8>,
966    in_frame: bool,
967    keyframe: bool,
968    current_ts: Option<u32>,
969}
970
971impl Vp9Depacketizer {
972    /// A fresh depacketizer with no in-progress frame.
973    pub fn new() -> Self {
974        Self::default()
975    }
976
977    /// Push one VP9 RTP payload. Returns a completed frame when the E (end) bit
978    /// and RTP marker close it.
979    pub fn push(
980        &mut self,
981        payload: &[u8],
982        marker: bool,
983        timestamp: u32,
984    ) -> Result<Option<AccessUnit>, DepacketizeError> {
985        if payload.is_empty() {
986            return Err(DepacketizeError::Truncated);
987        }
988        let desc0 = payload[0];
989        let has_pid = desc0 & 0x80 != 0;
990        let has_layer = desc0 & 0x20 != 0;
991        let flexible = desc0 & 0x10 != 0;
992        let begin = desc0 & 0x08 != 0;
993        let end = desc0 & 0x04 != 0;
994        let predicted = desc0 & 0x40 != 0;
995
996        // Walk past the variable-length descriptor fields we recognize.
997        let mut off = 1;
998        if has_pid {
999            // M bit selects a 1- or 2-byte picture ID.
1000            let m = payload.get(off).ok_or(DepacketizeError::Truncated)? & 0x80 != 0;
1001            off += if m { 2 } else { 1 };
1002        }
1003        if has_layer {
1004            off += 1; // TID/U/SID/D byte
1005            if !flexible {
1006                off += 1; // TL0PICIDX (non-flexible mode)
1007            }
1008        }
1009        if off > payload.len() {
1010            return Err(DepacketizeError::Truncated);
1011        }
1012
1013        if begin {
1014            self.frame.clear();
1015            self.in_frame = true;
1016            self.keyframe = !predicted;
1017            self.current_ts = Some(timestamp);
1018        } else if !self.in_frame {
1019            return Err(DepacketizeError::OutOfOrder);
1020        }
1021        self.frame.extend_from_slice(&payload[off..]);
1022
1023        if end && marker && self.in_frame {
1024            self.in_frame = false;
1025            return Ok(Some(AccessUnit {
1026                data: Bytes::from(std::mem::take(&mut self.frame)),
1027                timestamp: self.current_ts.unwrap_or(timestamp),
1028                keyframe: self.keyframe,
1029            }));
1030        }
1031        Ok(None)
1032    }
1033}
1034
1035// ── AV1 (AOMedia "RTP Payload Format For AV1") ───────────────────────────────
1036
1037/// Encode `v` as unsigned LEB128 into `out`.
1038#[cfg(feature = "codec-av1")]
1039fn leb128_encode(mut v: u64, out: &mut Vec<u8>) {
1040    loop {
1041        let mut byte = (v & 0x7F) as u8;
1042        v >>= 7;
1043        if v != 0 {
1044            byte |= 0x80;
1045        }
1046        out.push(byte);
1047        if v == 0 {
1048            break;
1049        }
1050    }
1051}
1052
1053#[cfg(feature = "codec-av1")]
1054const AV1_OBU_SEQUENCE_HEADER: u8 = 1;
1055#[cfg(feature = "codec-av1")]
1056const AV1_OBU_TEMPORAL_DELIMITER: u8 = 2;
1057
1058/// Packetizes an AV1 temporal unit into RTP using the AOMedia payload format.
1059///
1060/// Each non-temporal-delimiter OBU is re-framed as a length-delimited *OBU
1061/// element* (the W=0 form, with the OBU's `obu_has_size_field` cleared), the
1062/// elements are concatenated, and the resulting stream is split across the MTU
1063/// with a one-byte aggregation header per packet (Z/Y continuation bits, N on a
1064/// new coded video sequence). The temporal delimiter is dropped per the spec;
1065/// frame boundaries are conveyed by the RTP marker. Scalability structures are
1066/// not emitted.
1067#[cfg(feature = "codec-av1")]
1068#[derive(Debug, Clone)]
1069pub struct Av1Packetizer {
1070    payload_type: u8,
1071    ssrc: u32,
1072    sequence: u16,
1073    max_payload: usize,
1074}
1075
1076#[cfg(feature = "codec-av1")]
1077impl Av1Packetizer {
1078    /// An AV1 packetizer for `payload_type`/`ssrc`. `mtu` is the maximum UDP
1079    /// payload; the 12-byte RTP header and 1-byte aggregation header are removed.
1080    pub fn new(payload_type: u8, ssrc: u32, mtu: usize) -> Self {
1081        Self {
1082            payload_type,
1083            ssrc,
1084            sequence: 0,
1085            max_payload: mtu.saturating_sub(12 + 1).max(1),
1086        }
1087    }
1088
1089    /// Packetize one AV1 temporal unit (low-overhead OBUs) at `timestamp`.
1090    pub fn packetize(&mut self, temporal_unit: &[u8], timestamp: u32) -> Vec<Vec<u8>> {
1091        let mut out = Vec::new();
1092        self.packetize_into(temporal_unit, timestamp, &mut out);
1093        out
1094    }
1095
1096    /// Recycling variant of [`packetize`](Self::packetize): pass the same `out`
1097    /// every temporal unit to reuse the packet-buffer allocations.
1098    pub fn packetize_into(&mut self, temporal_unit: &[u8], timestamp: u32, out: &mut Vec<Vec<u8>>) {
1099        // Re-frame each OBU (minus the temporal delimiter) as a length-delimited
1100        // OBU element with obu_has_size_field cleared.
1101        let mut stream = Vec::with_capacity(temporal_unit.len());
1102        let mut new_cvs = false;
1103        for obu in crate::codec::obu::iter_obus(temporal_unit) {
1104            if obu.obu_type == AV1_OBU_TEMPORAL_DELIMITER {
1105                continue;
1106            }
1107            if obu.obu_type == AV1_OBU_SEQUENCE_HEADER {
1108                new_cvs = true;
1109            }
1110            let header_len = 1 + obu.has_extension as usize;
1111            let mut element = Vec::with_capacity(header_len + obu.payload.len());
1112            element.push(obu.raw[0] & !0x02); // clear obu_has_size_field
1113            if obu.has_extension {
1114                element.push(obu.raw[1]);
1115            }
1116            element.extend_from_slice(obu.payload);
1117            leb128_encode(element.len() as u64, &mut stream);
1118            stream.extend_from_slice(&element);
1119        }
1120
1121        let mut recycle = std::mem::take(out);
1122        let chunks: Vec<&[u8]> = if stream.is_empty() {
1123            vec![&[]]
1124        } else {
1125            stream.chunks(self.max_payload).collect()
1126        };
1127        let n = chunks.len();
1128        for (i, chunk) in chunks.into_iter().enumerate() {
1129            let last = i + 1 == n;
1130            let mut pkt = recycle.pop().unwrap_or_default();
1131            pkt.clear();
1132            write_rtp_header(
1133                &mut pkt,
1134                self.payload_type,
1135                last,
1136                self.sequence,
1137                timestamp,
1138                self.ssrc,
1139            );
1140            self.sequence = self.sequence.wrapping_add(1);
1141
1142            // Aggregation header: Z (continues previous packet) | Y (continues in
1143            // next) | W=0 (length-delimited elements) | N (new coded video seq).
1144            let mut agg = 0u8;
1145            if i > 0 {
1146                agg |= 0x80; // Z
1147            }
1148            if !last {
1149                agg |= 0x40; // Y
1150            }
1151            if i == 0 && new_cvs {
1152                agg |= 0x08; // N
1153            }
1154            pkt.push(agg);
1155            pkt.extend_from_slice(chunk);
1156            out.push(pkt);
1157        }
1158    }
1159}
1160
1161/// Reassembles AV1 RTP payloads into temporal units. Counterpart to
1162/// [`Av1Packetizer`]: it concatenates each packet's OBU-element bytes (past the
1163/// aggregation header) and, on the RTP marker, parses the length-delimited
1164/// elements back into low-overhead OBUs (re-adding each `obu_has_size_field`).
1165#[cfg(feature = "codec-av1")]
1166#[derive(Debug, Default)]
1167pub struct Av1Depacketizer {
1168    stream: Vec<u8>,
1169    new_cvs: bool,
1170    current_ts: Option<u32>,
1171}
1172
1173#[cfg(feature = "codec-av1")]
1174impl Av1Depacketizer {
1175    /// A fresh depacketizer with no in-progress temporal unit.
1176    pub fn new() -> Self {
1177        Self::default()
1178    }
1179
1180    /// Push one AV1 RTP payload. Returns a completed temporal unit when the RTP
1181    /// marker closes it.
1182    pub fn push(
1183        &mut self,
1184        payload: &[u8],
1185        marker: bool,
1186        timestamp: u32,
1187    ) -> Result<Option<AccessUnit>, DepacketizeError> {
1188        if payload.is_empty() {
1189            return Err(DepacketizeError::Truncated);
1190        }
1191        let agg = payload[0];
1192        if agg & 0x08 != 0 {
1193            self.new_cvs = true; // N: new coded video sequence
1194        }
1195        if self.current_ts.is_none() {
1196            self.current_ts = Some(timestamp);
1197        }
1198        self.stream.extend_from_slice(&payload[1..]);
1199
1200        if !marker {
1201            return Ok(None);
1202        }
1203
1204        // Marker: rebuild the temporal unit from length-delimited OBU elements.
1205        let stream = std::mem::take(&mut self.stream);
1206        let mut tu = Vec::with_capacity(stream.len() + 8);
1207        let mut pos = 0;
1208        while pos < stream.len() {
1209            let len = leb128_decode(&stream, &mut pos).ok_or(DepacketizeError::Truncated)?;
1210            let end = pos.checked_add(len).ok_or(DepacketizeError::Truncated)?;
1211            let element = stream.get(pos..end).ok_or(DepacketizeError::Truncated)?;
1212            pos = end;
1213            // Element → low-overhead OBU: set obu_has_size_field, insert the size.
1214            let hdr0 = *element.first().ok_or(DepacketizeError::Truncated)?;
1215            let has_ext = (hdr0 >> 2) & 1 == 1;
1216            let header_len = 1 + has_ext as usize;
1217            let obu_payload = element
1218                .get(header_len..)
1219                .ok_or(DepacketizeError::Truncated)?;
1220            tu.push(hdr0 | 0x02);
1221            if has_ext {
1222                tu.push(element[1]);
1223            }
1224            leb128_encode(obu_payload.len() as u64, &mut tu);
1225            tu.extend_from_slice(obu_payload);
1226        }
1227
1228        let keyframe = std::mem::take(&mut self.new_cvs);
1229        let ts = self.current_ts.take().unwrap_or(timestamp);
1230        Ok(Some(AccessUnit {
1231            data: Bytes::from(tu),
1232            timestamp: ts,
1233            keyframe,
1234        }))
1235    }
1236}
1237
1238/// Decode an unsigned LEB128 integer at `*pos` (advancing it), returning `None`
1239/// on truncation or overflow. Mirrors the codec-side decoder for RTP carriage.
1240#[cfg(feature = "codec-av1")]
1241fn leb128_decode(data: &[u8], pos: &mut usize) -> Option<usize> {
1242    let mut value: u64 = 0;
1243    for i in 0..8 {
1244        let byte = *data.get(*pos)?;
1245        *pos += 1;
1246        value |= ((byte & 0x7F) as u64) << (i * 7);
1247        if byte & 0x80 == 0 {
1248            return usize::try_from(value).ok();
1249        }
1250    }
1251    None
1252}
1253
1254#[cfg(test)]
1255mod tests {
1256    use super::*;
1257
1258    /// Build a minimal 12-byte RTP packet with the given fields and payload.
1259    fn rtp(seq: u16, ts: u32, marker: bool, payload: &[u8]) -> Vec<u8> {
1260        let mut p = vec![0x80, if marker { 0x80 | 96 } else { 96 }];
1261        p.extend_from_slice(&seq.to_be_bytes());
1262        p.extend_from_slice(&ts.to_be_bytes());
1263        p.extend_from_slice(&[0, 0, 0, 1]); // ssrc
1264        p.extend_from_slice(payload);
1265        p
1266    }
1267
1268    #[test]
1269    fn parses_fixed_header_and_payload_offset() {
1270        let pkt = rtp(7, 9000, true, &[0x65, 0xAA]);
1271        let h = RtpHeader::parse(&pkt).unwrap();
1272        assert_eq!(h.sequence, 7);
1273        assert_eq!(h.timestamp, 9000);
1274        assert!(h.marker);
1275        assert_eq!(h.payload_type, 96);
1276        assert_eq!(h.payload_offset, 12);
1277        assert_eq!(&pkt[h.payload_offset..], &[0x65, 0xAA]);
1278    }
1279
1280    #[test]
1281    fn rejects_wrong_version_and_short_buffers() {
1282        assert!(RtpHeader::parse(&[0x00; 12]).is_none()); // version 0
1283        assert!(RtpHeader::parse(&[0x80; 4]).is_none()); // too short
1284    }
1285
1286    /// Build an RTP packet carrying a one-byte (RFC 8285) header extension with a
1287    /// single element `(ext_id, value)`.
1288    fn rtp_with_ext(ext_id: u8, value: &[u8], payload: &[u8]) -> Vec<u8> {
1289        let mut p = vec![0x90, 96]; // X bit set (0x10), PT 96
1290        p.extend_from_slice(&1u16.to_be_bytes()); // seq
1291        p.extend_from_slice(&0u32.to_be_bytes()); // ts
1292        p.extend_from_slice(&7u32.to_be_bytes()); // ssrc
1293        p.extend_from_slice(&0xBEDEu16.to_be_bytes()); // one-byte profile
1294                                                       // One element + padding to a 4-byte word boundary.
1295        let mut ext = vec![(ext_id << 4) | (value.len() as u8 - 1)];
1296        ext.extend_from_slice(value);
1297        while ext.len() % 4 != 0 {
1298            ext.push(0);
1299        }
1300        p.extend_from_slice(&((ext.len() / 4) as u16).to_be_bytes());
1301        p.extend_from_slice(&ext);
1302        p.extend_from_slice(payload);
1303        p
1304    }
1305
1306    #[test]
1307    fn extracts_rid_header_extension() {
1308        let pkt = rtp_with_ext(4, b"hi", &[0xAA, 0xBB]);
1309        assert_eq!(rtp_extension_value(&pkt, 4), Some(&b"hi"[..]));
1310        assert_eq!(rtp_extension_value(&pkt, 5), None, "unknown id");
1311        // Payload still locates correctly past the extension.
1312        let h = RtpHeader::parse(&pkt).unwrap();
1313        assert_eq!(&pkt[h.payload_offset..], &[0xAA, 0xBB]);
1314    }
1315
1316    #[test]
1317    fn extension_value_none_without_extension_flag() {
1318        let pkt = rtp(1, 0, false, &[1, 2, 3]);
1319        assert_eq!(rtp_extension_value(&pkt, 4), None);
1320    }
1321
1322    #[test]
1323    fn decodes_rfc6464_audio_level() {
1324        // Voice active (MSB set) at level 20 -dBov.
1325        let pkt = rtp_with_ext(3, &[0x80 | 20], &[0xAA]);
1326        assert_eq!(audio_level(&pkt, 3), Some((20, true)));
1327        // Voice inactive (MSB clear) at level 90 -dBov (quiet).
1328        let pkt = rtp_with_ext(3, &[90], &[0xAA]);
1329        assert_eq!(audio_level(&pkt, 3), Some((90, false)));
1330        // No such extension id.
1331        assert_eq!(audio_level(&pkt, 7), None);
1332    }
1333
1334    #[test]
1335    fn honors_csrc_count_in_payload_offset() {
1336        let mut pkt = rtp(1, 0, false, &[0x41]);
1337        pkt[0] = 0x82; // version 2, CSRC count = 2
1338        let mut with_csrc = pkt[..12].to_vec();
1339        with_csrc.extend_from_slice(&[0xDE, 0xAD, 0xBE, 0xEF, 0, 0, 0, 0]); // 2 CSRCs
1340        with_csrc.push(0x41);
1341        let h = RtpHeader::parse(&with_csrc).unwrap();
1342        assert_eq!(h.payload_offset, 20);
1343    }
1344
1345    #[test]
1346    fn aac_hbr_splits_two_access_units() {
1347        // AU-headers-length = 32 bits → two 16-bit AU-headers.
1348        // AU sizes 3 and 2 (top 13 bits of each 2-byte header).
1349        let mut p = Vec::new();
1350        p.extend_from_slice(&32u16.to_be_bytes()); // header bits
1351        p.extend_from_slice(&((3u16) << 3).to_be_bytes()); // AU-header: size 3
1352        p.extend_from_slice(&((2u16) << 3).to_be_bytes()); // AU-header: size 2
1353        p.extend_from_slice(&[0xA1, 0xA2, 0xA3]); // AU 1
1354        p.extend_from_slice(&[0xB1, 0xB2]); // AU 2
1355        let aus = AacDepacketizer::new().push(&p).unwrap();
1356        assert_eq!(aus.len(), 2);
1357        assert_eq!(&aus[0][..], &[0xA1, 0xA2, 0xA3]);
1358        assert_eq!(&aus[1][..], &[0xB1, 0xB2]);
1359    }
1360
1361    #[test]
1362    fn aac_hbr_single_au() {
1363        let mut p = Vec::new();
1364        p.extend_from_slice(&16u16.to_be_bytes()); // one 16-bit AU-header
1365        p.extend_from_slice(&((4u16) << 3).to_be_bytes()); // size 4
1366        p.extend_from_slice(&[1, 2, 3, 4]);
1367        let aus = AacDepacketizer::new().push(&p).unwrap();
1368        assert_eq!(aus.len(), 1);
1369        assert_eq!(&aus[0][..], &[1, 2, 3, 4]);
1370    }
1371
1372    #[test]
1373    fn aac_truncated_payload_errors() {
1374        assert_eq!(
1375            AacDepacketizer::new().push(&[0x00]),
1376            Err(DepacketizeError::Truncated)
1377        );
1378        // Declares one AU of size 8 but supplies only 2 data bytes.
1379        let mut p = 16u16.to_be_bytes().to_vec();
1380        p.extend_from_slice(&((8u16) << 3).to_be_bytes());
1381        p.extend_from_slice(&[1, 2]);
1382        assert_eq!(
1383            AacDepacketizer::new().push(&p),
1384            Err(DepacketizeError::Truncated)
1385        );
1386    }
1387
1388    #[test]
1389    fn single_nal_packet_emits_annexb_on_marker() {
1390        let mut d = H264Depacketizer::new();
1391        // Type 1 (non-IDR slice), marker set → one access unit.
1392        let out = d.push(&[0x41, 0x9A, 0xBC], true, 3000, 1).unwrap().unwrap();
1393        assert_eq!(&out.data[..], &[0, 0, 0, 1, 0x41, 0x9A, 0xBC]);
1394        assert!(!out.keyframe);
1395        assert_eq!(out.timestamp, 3000);
1396    }
1397
1398    #[test]
1399    fn idr_single_nal_is_flagged_keyframe() {
1400        let mut d = H264Depacketizer::new();
1401        let out = d.push(&[0x65, 0x01], true, 0, 1).unwrap().unwrap();
1402        assert!(out.keyframe);
1403    }
1404
1405    #[test]
1406    fn packetizer_single_nal_round_trips_through_depacketizer() {
1407        // A small AU (two NALs) → single-NAL packets → reassembled identically.
1408        let au = [0, 0, 0, 1, 0x67, 0x42, 0x00, 0, 0, 0, 1, 0x65, 0x88, 0x99];
1409        let mut pkt = RtpPacketizer::new(96, 0xABCD, 1200);
1410        let packets = pkt.packetize(&au, 3000);
1411        assert_eq!(packets.len(), 2, "one packet per NAL");
1412
1413        let mut depack = H264Depacketizer::new();
1414        let mut out = None;
1415        for p in &packets {
1416            let h = RtpHeader::parse(p).unwrap();
1417            if let Some(au) = depack
1418                .push(&p[h.payload_offset..], h.marker, h.timestamp, h.sequence)
1419                .unwrap()
1420            {
1421                out = Some(au);
1422            }
1423        }
1424        let out = out.expect("AU completed on the marker packet");
1425        assert_eq!(&out.data[..], &au);
1426        assert!(out.keyframe);
1427        assert_eq!(out.timestamp, 3000);
1428    }
1429
1430    #[test]
1431    fn packetize_into_recycles_buffers_without_changing_output() {
1432        // The recycling hot-path API must produce byte-identical packets to the
1433        // allocating `packetize`, frame after frame, including correct sequence
1434        // numbers carried across the reused buffer.
1435        let au1 = [0, 0, 0, 1, 0x67, 0x42, 0x00, 0, 0, 0, 1, 0x65, 0x88, 0x99];
1436        let au2 = [0, 0, 0, 1, 0x65, 0x11, 0x22, 0x33];
1437
1438        let mut a = RtpPacketizer::new(96, 0xABCD, 1200);
1439        let mut b = RtpPacketizer::new(96, 0xABCD, 1200);
1440        let mut reused: Vec<Vec<u8>> = Vec::new();
1441
1442        for au in [&au1[..], &au2[..], &au1[..]] {
1443            let expected = a.packetize(au, 3000);
1444            // Capture the backing pointers to prove buffers are reused, not freed.
1445            b.packetize_into(au, 3000, &mut reused);
1446            assert_eq!(
1447                reused, expected,
1448                "recycled output matches allocating output"
1449            );
1450        }
1451    }
1452
1453    #[test]
1454    fn packetizer_fragments_oversized_nal_and_round_trips() {
1455        // One NAL larger than the MTU → FU-A fragments → reassembled identically.
1456        let mut nal = vec![0, 0, 0, 1, 0x65]; // start code + IDR NAL header
1457        nal.extend((0..600u16).map(|i| i as u8)); // long payload
1458        let mut pkt = RtpPacketizer::new(96, 1, 100); // tiny MTU forces FU-A
1459        let packets = pkt.packetize(&nal, 90);
1460        assert!(packets.len() > 1, "oversized NAL is fragmented");
1461        // Only the last packet carries the marker bit.
1462        let markers: Vec<bool> = packets
1463            .iter()
1464            .map(|p| RtpHeader::parse(p).unwrap().marker)
1465            .collect();
1466        assert_eq!(markers.iter().filter(|m| **m).count(), 1);
1467        assert!(markers.last().unwrap());
1468
1469        let mut depack = H264Depacketizer::new();
1470        let mut out = None;
1471        for p in &packets {
1472            let h = RtpHeader::parse(p).unwrap();
1473            if let Some(au) = depack
1474                .push(&p[h.payload_offset..], h.marker, h.timestamp, h.sequence)
1475                .unwrap()
1476            {
1477                out = Some(au);
1478            }
1479        }
1480        assert_eq!(&out.unwrap().data[..], &nal[..]);
1481    }
1482
1483    #[test]
1484    fn stap_a_splits_aggregated_nals() {
1485        // STAP-A (24): [24][size=2][AA BB][size=3][CC DD EE]
1486        let payload = [24, 0, 2, 0xAA, 0xBB, 0, 3, 0xCC, 0xDD, 0xEE];
1487        let mut d = H264Depacketizer::new();
1488        let out = d.push(&payload, true, 0, 1).unwrap().unwrap();
1489        assert_eq!(
1490            &out.data[..],
1491            &[0, 0, 0, 1, 0xAA, 0xBB, 0, 0, 0, 1, 0xCC, 0xDD, 0xEE]
1492        );
1493    }
1494
1495    #[test]
1496    fn fu_a_reassembles_fragmented_nal() {
1497        let mut d = H264Depacketizer::new();
1498        // FU indicator 0x7C (F=0,NRI=3,type=28), FU header start 0x85 (S=1,type=5).
1499        assert!(d
1500            .push(&[0x7C, 0x85, 0x11, 0x22], false, 0, 1)
1501            .unwrap()
1502            .is_none());
1503        // Middle fragment (S=0,E=0).
1504        assert!(d.push(&[0x7C, 0x05, 0x33], false, 0, 2).unwrap().is_none());
1505        // End fragment (E=1), marker closes the AU.
1506        let out = d.push(&[0x7C, 0x45, 0x44], true, 0, 3).unwrap().unwrap();
1507        // Reconstructed NAL header: NRI 0x60 | type 5 = 0x65, then payload bytes.
1508        assert_eq!(&out.data[..], &[0, 0, 0, 1, 0x65, 0x11, 0x22, 0x33, 0x44]);
1509        assert!(out.keyframe);
1510    }
1511
1512    #[test]
1513    fn fu_a_sequence_gap_reports_out_of_order() {
1514        let mut d = H264Depacketizer::new();
1515        d.push(&[0x7C, 0x85, 0x11], false, 0, 1).unwrap();
1516        // Jump from seq 1 to seq 5 mid-fragment.
1517        assert_eq!(
1518            d.push(&[0x7C, 0x05, 0x22], false, 0, 5),
1519            Err(DepacketizeError::OutOfOrder)
1520        );
1521    }
1522
1523    #[test]
1524    fn timestamp_change_flushes_previous_au_without_marker() {
1525        let mut d = H264Depacketizer::new();
1526        // First AU, no marker.
1527        assert!(d.push(&[0x41, 0x01], false, 1000, 1).unwrap().is_none());
1528        // New timestamp flushes the first AU.
1529        let out = d.push(&[0x41, 0x02], false, 2000, 2).unwrap().unwrap();
1530        assert_eq!(out.timestamp, 1000);
1531        assert_eq!(&out.data[..], &[0, 0, 0, 1, 0x41, 0x01]);
1532    }
1533
1534    // ── H.265 (RFC 7798) ────────────────────────────────────────────────────
1535    // H.265 NAL headers are two bytes; type = (byte0 >> 1) & 0x3F. Examples used
1536    // below: VPS=32 (0x40,0x01), IDR_W_RADL=19 (0x26,0x01).
1537
1538    #[test]
1539    fn h265_single_nal_round_trips_through_depacketizer() {
1540        // VPS (non-VCL) + IDR (VCL keyframe), each a single-NAL packet.
1541        let au = [
1542            0, 0, 0, 1, 0x40, 0x01, 0xAA, // VPS (type 32)
1543            0, 0, 0, 1, 0x26, 0x01, 0x88, 0x99, // IDR (type 19)
1544        ];
1545        let mut pkt = RtpPacketizer::new_h265(96, 0xABCD, 1200);
1546        let packets = pkt.packetize(&au, 3000);
1547        assert_eq!(packets.len(), 2, "one packet per NAL");
1548
1549        let mut depack = H265Depacketizer::new();
1550        let mut out = None;
1551        for p in &packets {
1552            let h = RtpHeader::parse(p).unwrap();
1553            if let Some(au) = depack
1554                .push(&p[h.payload_offset..], h.marker, h.timestamp, h.sequence)
1555                .unwrap()
1556            {
1557                out = Some(au);
1558            }
1559        }
1560        let out = out.expect("AU completed on the marker packet");
1561        assert_eq!(&out.data[..], &au);
1562        assert!(out.keyframe, "IRAP type 19 is a keyframe");
1563        assert_eq!(out.timestamp, 3000);
1564    }
1565
1566    #[test]
1567    fn h265_fragments_oversized_nal_and_round_trips() {
1568        // One IDR NAL larger than the MTU → FU fragments → reassembled identically.
1569        let mut nal = vec![0, 0, 0, 1, 0x26, 0x01]; // start code + 2-byte IDR header
1570        nal.extend((0..600u16).map(|i| i as u8));
1571        let mut pkt = RtpPacketizer::new_h265(96, 1, 100); // tiny MTU forces FU
1572        let packets = pkt.packetize(&nal, 90);
1573        assert!(packets.len() > 1, "oversized NAL is fragmented");
1574        // Exactly one marker, on the last fragment.
1575        let markers: Vec<bool> = packets
1576            .iter()
1577            .map(|p| RtpHeader::parse(p).unwrap().marker)
1578            .collect();
1579        assert_eq!(markers.iter().filter(|m| **m).count(), 1);
1580        assert!(markers.last().unwrap());
1581        // Each FU packet carries a type-49 PayloadHdr.
1582        for p in &packets {
1583            let h = RtpHeader::parse(p).unwrap();
1584            let pt = (p[h.payload_offset] >> 1) & 0x3F;
1585            assert_eq!(pt, 49, "FU payload type");
1586        }
1587
1588        let mut depack = H265Depacketizer::new();
1589        let mut out = None;
1590        for p in &packets {
1591            let h = RtpHeader::parse(p).unwrap();
1592            if let Some(au) = depack
1593                .push(&p[h.payload_offset..], h.marker, h.timestamp, h.sequence)
1594                .unwrap()
1595            {
1596                out = Some(au);
1597            }
1598        }
1599        assert_eq!(&out.unwrap().data[..], &nal[..]);
1600    }
1601
1602    #[test]
1603    fn h265_ap_splits_aggregated_nals() {
1604        // AP (type 48): [0x60,0x01][size=2][AA BB][size=3][CC DD EE]
1605        let payload = [0x60, 0x01, 0, 2, 0xAA, 0xBB, 0, 3, 0xCC, 0xDD, 0xEE];
1606        let mut d = H265Depacketizer::new();
1607        let out = d.push(&payload, true, 0, 1).unwrap().unwrap();
1608        assert_eq!(
1609            &out.data[..],
1610            &[0, 0, 0, 1, 0xAA, 0xBB, 0, 0, 0, 1, 0xCC, 0xDD, 0xEE]
1611        );
1612    }
1613
1614    #[test]
1615    fn h265_rejects_truncated_and_unsupported() {
1616        let mut d = H265Depacketizer::new();
1617        // One byte cannot hold a 2-byte NAL header.
1618        assert_eq!(
1619            d.push(&[0x26], true, 0, 1),
1620            Err(DepacketizeError::Truncated)
1621        );
1622        // PACI (type 50) is not supported.
1623        assert_eq!(
1624            d.push(&[50 << 1, 0x01, 0x00], true, 0, 2),
1625            Err(DepacketizeError::Unsupported(50))
1626        );
1627    }
1628
1629    // ── VP9 ───────────────────────────────────────────────────────────────────
1630
1631    fn vp9_depacketize(packets: &[Vec<u8>]) -> Option<AccessUnit> {
1632        let mut d = Vp9Depacketizer::new();
1633        let mut out = None;
1634        for p in packets {
1635            let h = RtpHeader::parse(p).unwrap();
1636            if let Some(f) = d
1637                .push(&p[h.payload_offset..], h.marker, h.timestamp)
1638                .unwrap()
1639            {
1640                out = Some(f);
1641            }
1642        }
1643        out
1644    }
1645
1646    #[test]
1647    fn vp9_fragmented_frame_round_trips() {
1648        let frame: Vec<u8> = (0..500u16).map(|i| i as u8).collect();
1649        let mut pkt = Vp9Packetizer::new(98, 0x1234, 100); // small MTU → fragments
1650        let packets = pkt.packetize(&frame, 9000, true);
1651        assert!(packets.len() > 1, "frame fragmented");
1652
1653        // Exactly one marker, on the last packet.
1654        let markers: Vec<bool> = packets
1655            .iter()
1656            .map(|p| RtpHeader::parse(p).unwrap().marker)
1657            .collect();
1658        assert_eq!(markers.iter().filter(|m| **m).count(), 1);
1659        assert!(markers.last().unwrap());
1660
1661        let out = vp9_depacketize(&packets).expect("frame completed");
1662        assert_eq!(&out.data[..], &frame[..]);
1663        assert!(out.keyframe, "keyframe → P bit clear");
1664        assert_eq!(out.timestamp, 9000);
1665    }
1666
1667    #[test]
1668    fn vp9_inter_frame_is_not_a_keyframe() {
1669        let mut pkt = Vp9Packetizer::new(98, 1, 1200);
1670        let packets = pkt.packetize(&[1, 2, 3], 0, false);
1671        assert_eq!(packets.len(), 1);
1672        let out = vp9_depacketize(&packets).expect("frame");
1673        assert_eq!(&out.data[..], &[1, 2, 3]);
1674        assert!(!out.keyframe, "P bit set → inter frame");
1675    }
1676
1677    // ── AV1 ───────────────────────────────────────────────────────────────────
1678
1679    #[cfg(feature = "codec-av1")]
1680    fn av1_depacketize(packets: &[Vec<u8>]) -> Option<AccessUnit> {
1681        let mut d = Av1Depacketizer::new();
1682        let mut out = None;
1683        for p in packets {
1684            let h = RtpHeader::parse(p).unwrap();
1685            if let Some(f) = d
1686                .push(&p[h.payload_offset..], h.marker, h.timestamp)
1687                .unwrap()
1688            {
1689                out = Some(f);
1690            }
1691        }
1692        out
1693    }
1694
1695    #[cfg(feature = "codec-av1")]
1696    #[test]
1697    fn av1_temporal_unit_round_trips_without_delimiter() {
1698        // Low-overhead OBUs: temporal delimiter + sequence header + frame.
1699        let td = [0x12u8, 0x00];
1700        let seq = [0x0Au8, 0x02, 0xAA, 0xBB];
1701        let frame = [0x32u8, 0x03, 0x11, 0x22, 0x33];
1702        let mut tu = Vec::new();
1703        tu.extend_from_slice(&td);
1704        tu.extend_from_slice(&seq);
1705        tu.extend_from_slice(&frame);
1706
1707        let mut pkt = Av1Packetizer::new(99, 7, 1200);
1708        let packets = pkt.packetize(&tu, 1000);
1709        let out = av1_depacketize(&packets).expect("TU completed");
1710
1711        // The temporal delimiter is dropped; seq + frame survive, low-overhead.
1712        let mut expected = Vec::new();
1713        expected.extend_from_slice(&seq);
1714        expected.extend_from_slice(&frame);
1715        assert_eq!(&out.data[..], &expected[..]);
1716        assert!(out.keyframe, "sequence header → new coded video sequence");
1717        assert_eq!(out.timestamp, 1000);
1718    }
1719
1720    #[cfg(feature = "codec-av1")]
1721    #[test]
1722    fn av1_large_temporal_unit_fragments_and_round_trips() {
1723        // A frame OBU with a 300-byte payload (size field leb128(300) = AC 02).
1724        let mut frame = vec![0x32u8, 0xAC, 0x02];
1725        frame.extend((0..300u16).map(|i| i as u8));
1726        let mut tu = vec![0x12u8, 0x00]; // temporal delimiter
1727        tu.extend_from_slice(&frame);
1728
1729        let mut pkt = Av1Packetizer::new(99, 1, 64); // tiny MTU forces fragmentation
1730        let packets = pkt.packetize(&tu, 0);
1731        assert!(packets.len() > 1, "large TU fragmented");
1732        // Z set on every packet but the first; Y on every packet but the last.
1733        for (i, p) in packets.iter().enumerate() {
1734            let agg = p[RtpHeader::parse(p).unwrap().payload_offset];
1735            assert_eq!((agg & 0x80 != 0), i > 0, "Z continuation bit");
1736            assert_eq!(
1737                (agg & 0x40 != 0),
1738                i + 1 < packets.len(),
1739                "Y continuation bit"
1740            );
1741        }
1742
1743        let out = av1_depacketize(&packets).expect("TU completed");
1744        assert_eq!(&out.data[..], &frame[..], "frame OBU reconstructed");
1745    }
1746}