arcly-stream 0.1.5

An open-extensible live-media streaming kernel: lock-free zero-copy frame fan-out, instant-start GOP cache, a pluggable multi-protocol ingestion layer (RTMP, RTSP, SRT, WHIP/WHEP shipped), and a feature-gated pure-Rust media plane (MPEG-TS/HLS/fMP4) — runtime, config, and metrics free.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
//! Shared RTP/RTCP parsing and H.264 depacketization (RFC 3550 + RFC 6184).
//!
//! Gated behind the internal `_rtp` marker, pulled in by both [`rtsp`] and
//! [`webrtc`]. The two transports differ only in how RTP packets reach the
//! process (TCP-interleaved / UDP for RTSP, DTLS-SRTP for WebRTC); once a packet
//! is in hand, reassembling NAL units into an Annex-B access unit is identical,
//! so it lives here once.
//!
//! [`rtsp`]: crate::protocol::rtsp
//! [`webrtc`]: crate::protocol::webrtc
//!
//! # What it does
//!
//! - [`RtpHeader::parse`] decodes the fixed RTP header (RFC 3550 §5.1), honoring
//!   the CSRC count and the extension-header flag to locate the payload.
//! - [`H264Depacketizer`] turns a sequence of RTP payloads into complete H.264
//!   access units in Annex-B form, handling the three NALU packetization modes
//!   defined by RFC 6184: single NAL units, STAP-A aggregation (type 24), and
//!   FU-A fragmentation (type 28). An access unit is emitted when the RTP marker
//!   bit is set or the RTP timestamp advances.
//!
//! # What it does not do
//!
//! Jitter-buffer reordering and loss concealment are the caller's concern — the
//! depacketizer assumes in-order delivery (true for TCP-interleaved RTSP; for
//! UDP/SRTP a small reorder buffer should sit in front of it). It reports a
//! [`DepacketizeError::OutOfOrder`] gap so a handler can request a keyframe
//! (PLI/FIR) rather than emit a corrupt access unit.

use bytes::Bytes;

/// Annex-B start code prefixed to every reassembled NAL unit.
const ANNEXB_START: [u8; 4] = [0, 0, 0, 1];

/// A parsed RTP fixed header (RFC 3550 §5.1).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RtpHeader {
    /// Payload type (7 bits) — identifies the codec/format binding from SDP.
    pub payload_type: u8,
    /// Marker bit. For H.264 it flags the last packet of an access unit.
    pub marker: bool,
    /// 16-bit sequence number, increments by one per packet (wraps).
    pub sequence: u16,
    /// 32-bit media timestamp in the payload's clock (90 kHz for H.264 video).
    pub timestamp: u32,
    /// Synchronization source identifier.
    pub ssrc: u32,
    /// Byte offset at which the payload begins (past CSRCs and any extension).
    pub payload_offset: usize,
}

impl RtpHeader {
    /// Parse the fixed header from the front of `buf`, returning the header and
    /// the payload offset. Returns `None` if `buf` is too short or the version
    /// field is not 2.
    pub fn parse(buf: &[u8]) -> Option<RtpHeader> {
        use super::byteops::ByteReader;
        let mut r = ByteReader::new(buf);
        let b0 = r.u8()?;
        if b0 >> 6 != 2 {
            return None; // RTP version must be 2
        }
        let has_extension = b0 & 0x10 != 0;
        let csrc_count = (b0 & 0x0F) as usize;
        let b1 = r.u8()?;
        let marker = b1 & 0x80 != 0;
        let payload_type = b1 & 0x7F;
        let sequence = r.u16_be()?;
        let timestamp = r.u32_be()?;
        let ssrc = r.u32_be()?;
        r.skip(csrc_count * 4)?; // CSRC list

        if has_extension {
            // Extension header: 2-byte profile id, 2-byte length (in 32-bit words).
            r.skip(2)?;
            let ext_words = r.u16_be()? as usize;
            r.skip(ext_words * 4)?;
        }
        Some(RtpHeader {
            payload_type,
            marker,
            sequence,
            timestamp,
            ssrc,
            payload_offset: r.position(),
        })
    }
}

/// Errors surfaced while depacketizing an RTP stream.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum DepacketizeError {
    /// The packet was shorter than the format requires.
    Truncated,
    /// A sequence-number discontinuity was detected mid-fragment; the partial
    /// access unit was dropped. The handler should request a keyframe.
    OutOfOrder,
    /// An unsupported NAL/aggregation type was encountered.
    Unsupported(u8),
}

/// Depacketizes RFC 3640 AAC-hbr RTP payloads into raw AAC access units.
///
/// The common RTSP/SDP profile for AAC (`mode=AAC-hbr`, `sizelength=13`,
/// `indexlength=3`) frames each payload as a 2-byte **AU-headers-length** (in
/// bits), followed by one 2-byte AU-header per access unit (13-bit size +
/// 3-bit index), followed by the access units concatenated. One RTP packet may
/// carry several AAC frames; [`push`](Self::push) returns each as a separate
/// raw (ADTS-less) [`bytes::Bytes`].
#[derive(Debug, Clone, Copy, Default)]
pub struct AacDepacketizer {
    /// Bits per AU-header `size` field (13 for AAC-hbr).
    size_length: u8,
    /// Bits per AU-header `index`/`index-delta` field (3 for AAC-hbr).
    index_length: u8,
}

impl AacDepacketizer {
    /// A depacketizer for the standard AAC-hbr profile (`sizelength=13`,
    /// `indexlength=3`).
    pub fn new() -> Self {
        Self {
            size_length: 13,
            index_length: 3,
        }
    }

    /// A depacketizer with explicit AU-header field widths from the SDP `fmtp`.
    pub fn with_lengths(size_length: u8, index_length: u8) -> Self {
        Self {
            size_length,
            index_length,
        }
    }

    /// Split one RTP AAC-hbr payload into its constituent access units.
    pub fn push(&self, payload: &[u8]) -> Result<Vec<Bytes>, DepacketizeError> {
        if payload.len() < 2 {
            return Err(DepacketizeError::Truncated);
        }
        // Sizes wider than a 16-bit AU-header field are unsupported (and would
        // otherwise over-shift below). `with_lengths` can supply arbitrary widths.
        if self.size_length == 0 || self.size_length > 16 {
            return Err(DepacketizeError::Unsupported(self.size_length));
        }
        let header_bits = u16::from_be_bytes([payload[0], payload[1]]) as usize;
        let au_header_bits = self.size_length as usize + self.index_length as usize;
        if au_header_bits == 0 {
            return Err(DepacketizeError::Unsupported(0));
        }
        let header_bytes = header_bits.div_ceil(8);
        let au_count = header_bits / au_header_bits;
        let headers = payload
            .get(2..2 + header_bytes)
            .ok_or(DepacketizeError::Truncated)?;
        let mut data_off = 2 + header_bytes;
        let mut out = Vec::with_capacity(au_count);
        for i in 0..au_count {
            // Each AU-header is `au_header_bits` wide; for AAC-hbr that is 16
            // bits, so the size is the top `size_length` bits of a 2-byte field.
            let bit = i * au_header_bits;
            let byte = bit / 8;
            let hdr = headers
                .get(byte..byte + 2)
                .ok_or(DepacketizeError::Truncated)?;
            let size = (u16::from_be_bytes([hdr[0], hdr[1]]) >> (16 - self.size_length)) as usize;
            let end = data_off + size;
            let au = payload
                .get(data_off..end)
                .ok_or(DepacketizeError::Truncated)?;
            out.push(Bytes::copy_from_slice(au));
            data_off = end;
        }
        Ok(out)
    }
}

/// Packetizes H.264 Annex-B access units into RFC 6184 RTP packets — the inverse
/// of [`H264Depacketizer`], used for WebRTC/WHEP egress.
///
/// Each NAL unit that fits the MTU is sent as a single-NAL packet; larger NALs
/// are split into FU-A fragments. The RTP marker bit is set on the last packet
/// of each access unit so the receiver knows the frame is complete.
#[derive(Debug, Clone)]
pub struct RtpPacketizer {
    payload_type: u8,
    ssrc: u32,
    sequence: u16,
    /// Maximum RTP payload size (excluding the 12-byte header).
    max_payload: usize,
}

impl RtpPacketizer {
    /// A packetizer for `payload_type`/`ssrc`. `mtu` is the maximum UDP payload
    /// (1200 is the WebRTC-safe default); the 12-byte RTP header is subtracted.
    pub fn new(payload_type: u8, ssrc: u32, mtu: usize) -> Self {
        Self {
            payload_type,
            ssrc,
            sequence: 0,
            max_payload: mtu.saturating_sub(12).max(1),
        }
    }

    /// Build the 12-byte RTP header for the next packet and advance the sequence.
    fn header(&mut self, marker: bool, timestamp: u32, out: &mut Vec<u8>) {
        out.push(0x80); // V=2, no padding/extension/CSRC
        out.push(if marker { 0x80 } else { 0 } | (self.payload_type & 0x7F));
        out.extend_from_slice(&self.sequence.to_be_bytes());
        out.extend_from_slice(&timestamp.to_be_bytes());
        out.extend_from_slice(&self.ssrc.to_be_bytes());
        self.sequence = self.sequence.wrapping_add(1);
    }

    /// Packetize one Annex-B access unit at `timestamp` (90 kHz) into RTP packets.
    pub fn packetize(&mut self, access_unit: &[u8], timestamp: u32) -> Vec<Vec<u8>> {
        // Collect NAL units (without start codes) in order.
        let nals: Vec<&[u8]> = crate::codec::h264::iter_nals_annexb(access_unit)
            .filter(|n| !n.is_empty())
            .collect();
        let mut packets = Vec::new();
        for (i, nal) in nals.iter().enumerate() {
            let last_nal = i + 1 == nals.len();
            if nal.len() <= self.max_payload {
                // Single NAL unit packet; marker on the final NAL of the AU.
                let mut pkt = Vec::with_capacity(12 + nal.len());
                self.header(last_nal, timestamp, &mut pkt);
                pkt.extend_from_slice(nal);
                packets.push(pkt);
            } else {
                self.fragment_fua(nal, timestamp, last_nal, &mut packets);
            }
        }
        packets
    }

    /// Split one oversized NAL into FU-A fragments (RFC 6184 §5.8).
    fn fragment_fua(&mut self, nal: &[u8], timestamp: u32, last_nal: bool, out: &mut Vec<Vec<u8>>) {
        let nal_header = nal[0];
        let fu_indicator = (nal_header & 0xE0) | 28; // F|NRI from NAL, type 28
        let nal_type = nal_header & 0x1F;
        let body = &nal[1..];
        // Each fragment carries a 2-byte FU header (indicator + FU header).
        let chunk = self.max_payload.saturating_sub(2).max(1);
        let n_chunks = body.len().div_ceil(chunk);
        for (idx, part) in body.chunks(chunk).enumerate() {
            let start = idx == 0;
            let end = idx + 1 == n_chunks;
            let mut fu_header = nal_type;
            if start {
                fu_header |= 0x80;
            }
            if end {
                fu_header |= 0x40;
            }
            let mut pkt = Vec::with_capacity(12 + 2 + part.len());
            // Marker only on the very last fragment of the final NAL of the AU.
            self.header(last_nal && end, timestamp, &mut pkt);
            pkt.push(fu_indicator);
            pkt.push(fu_header);
            pkt.extend_from_slice(part);
            out.push(pkt);
        }
    }
}

/// Reassembles RFC 6184 H.264 RTP payloads into Annex-B access units.
///
/// Feed each packet's payload (the bytes after [`RtpHeader::payload_offset`])
/// with its marker bit and timestamp to [`push`](Self::push). When a complete
/// access unit is ready the method returns `Ok(Some(au))`, where `au` is the
/// concatenated NAL units each prefixed with a 4-byte Annex-B start code —
/// exactly the shape the codec parsers and `annexb_to_avcc` expect.
#[derive(Debug, Default)]
pub struct H264Depacketizer {
    /// Bytes accumulated for the current access unit (Annex-B framed).
    au: Vec<u8>,
    /// FU-A reassembly buffer for the NAL currently being defragmented.
    fua: Vec<u8>,
    /// `true` while an FU-A fragment is in progress (between Start and End bits).
    in_fragment: bool,
    /// Reconstructed NAL header byte for the in-progress FU-A NAL.
    fua_header: u8,
    /// Timestamp of the access unit currently being assembled.
    current_ts: Option<u32>,
    /// Last sequence number seen (for gap detection during fragmentation).
    last_seq: Option<u16>,
}

impl H264Depacketizer {
    /// A fresh depacketizer with no in-progress access unit.
    pub fn new() -> Self {
        Self::default()
    }

    /// Append one NAL unit (Annex-B framed) to the current access unit.
    fn append_nal(&mut self, nal: &[u8]) {
        self.au.extend_from_slice(&ANNEXB_START);
        self.au.extend_from_slice(nal);
    }

    /// Whether the pending access unit holds an IDR (type 5) NAL — a keyframe.
    fn pending_is_keyframe(&self) -> bool {
        // Scan the assembled Annex-B for a NAL header with type 5.
        let mut i = 0;
        while i + 4 < self.au.len() {
            if self.au[i..i + 4] == ANNEXB_START {
                let nal_type = self.au[i + 4] & 0x1F;
                if nal_type == 5 {
                    return true;
                }
            }
            i += 1;
        }
        false
    }

    /// Emit and reset the pending access unit, if any.
    fn take_au(&mut self) -> Option<AccessUnit> {
        if self.au.is_empty() {
            return None;
        }
        let keyframe = self.pending_is_keyframe();
        let timestamp = self.current_ts.unwrap_or(0);
        let data = Bytes::from(std::mem::take(&mut self.au));
        self.current_ts = None;
        Some(AccessUnit {
            data,
            timestamp,
            keyframe,
        })
    }

    /// Push one RTP H.264 payload. Returns a completed [`AccessUnit`] when the
    /// marker bit closes the frame (or the timestamp advances to a new one).
    pub fn push(
        &mut self,
        payload: &[u8],
        marker: bool,
        timestamp: u32,
        sequence: u16,
    ) -> Result<Option<AccessUnit>, DepacketizeError> {
        if payload.is_empty() {
            return Err(DepacketizeError::Truncated);
        }

        // A timestamp change flushes the previous access unit before starting the
        // new one (some encoders omit the marker bit).
        let mut completed = None;
        if let Some(ts) = self.current_ts {
            if ts != timestamp && !self.in_fragment {
                completed = self.take_au();
            }
        }
        self.current_ts = Some(timestamp);

        let nal_type = payload[0] & 0x1F;
        match nal_type {
            1..=23 => {
                // Single NAL unit packet — the payload *is* the NAL.
                self.append_nal(payload);
            }
            24 => {
                // STAP-A: one byte type, then [u16 size][nal]… aggregates.
                let mut i = 1;
                while i + 2 <= payload.len() {
                    let size = u16::from_be_bytes([payload[i], payload[i + 1]]) as usize;
                    i += 2;
                    if i + size > payload.len() {
                        return Err(DepacketizeError::Truncated);
                    }
                    self.append_nal(&payload[i..i + size]);
                    i += size;
                }
            }
            28 => {
                // FU-A: byte0 = FU indicator, byte1 = FU header (S|E|R|type).
                if payload.len() < 2 {
                    return Err(DepacketizeError::Truncated);
                }
                let fu_header = payload[1];
                let start = fu_header & 0x80 != 0;
                let end = fu_header & 0x40 != 0;
                let frag_type = fu_header & 0x1F;

                if start {
                    // Reconstruct the original NAL header: F|NRI from the indicator,
                    // type from the FU header.
                    self.fua_header = (payload[0] & 0xE0) | frag_type;
                    self.fua.clear();
                    self.fua.push(self.fua_header);
                    self.in_fragment = true;
                } else if !self.in_fragment {
                    // Mid/last fragment with no start — lost the head.
                    return Err(DepacketizeError::OutOfOrder);
                } else if self.seq_gap(sequence) {
                    self.in_fragment = false;
                    self.fua.clear();
                    return Err(DepacketizeError::OutOfOrder);
                }
                self.fua.extend_from_slice(&payload[2..]);

                if end && self.in_fragment {
                    let nal = std::mem::take(&mut self.fua);
                    self.append_nal(&nal);
                    self.in_fragment = false;
                }
            }
            other => return Err(DepacketizeError::Unsupported(other)),
        }

        self.last_seq = Some(sequence);

        if completed.is_some() {
            return Ok(completed);
        }
        if marker {
            return Ok(self.take_au());
        }
        Ok(None)
    }

    /// Detect a one-step sequence-number gap relative to the previous packet.
    fn seq_gap(&self, sequence: u16) -> bool {
        match self.last_seq {
            Some(prev) => sequence.wrapping_sub(prev) != 1,
            None => false,
        }
    }
}

/// A reassembled H.264 access unit in Annex-B form.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AccessUnit {
    /// Concatenated NAL units, each prefixed with a 4-byte start code.
    pub data: Bytes,
    /// RTP media timestamp (90 kHz) of the access unit.
    pub timestamp: u32,
    /// Whether the access unit contains an IDR (keyframe) NAL.
    pub keyframe: bool,
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Build a minimal 12-byte RTP packet with the given fields and payload.
    fn rtp(seq: u16, ts: u32, marker: bool, payload: &[u8]) -> Vec<u8> {
        let mut p = vec![0x80, if marker { 0x80 | 96 } else { 96 }];
        p.extend_from_slice(&seq.to_be_bytes());
        p.extend_from_slice(&ts.to_be_bytes());
        p.extend_from_slice(&[0, 0, 0, 1]); // ssrc
        p.extend_from_slice(payload);
        p
    }

    #[test]
    fn parses_fixed_header_and_payload_offset() {
        let pkt = rtp(7, 9000, true, &[0x65, 0xAA]);
        let h = RtpHeader::parse(&pkt).unwrap();
        assert_eq!(h.sequence, 7);
        assert_eq!(h.timestamp, 9000);
        assert!(h.marker);
        assert_eq!(h.payload_type, 96);
        assert_eq!(h.payload_offset, 12);
        assert_eq!(&pkt[h.payload_offset..], &[0x65, 0xAA]);
    }

    #[test]
    fn rejects_wrong_version_and_short_buffers() {
        assert!(RtpHeader::parse(&[0x00; 12]).is_none()); // version 0
        assert!(RtpHeader::parse(&[0x80; 4]).is_none()); // too short
    }

    #[test]
    fn honors_csrc_count_in_payload_offset() {
        let mut pkt = rtp(1, 0, false, &[0x41]);
        pkt[0] = 0x82; // version 2, CSRC count = 2
        let mut with_csrc = pkt[..12].to_vec();
        with_csrc.extend_from_slice(&[0xDE, 0xAD, 0xBE, 0xEF, 0, 0, 0, 0]); // 2 CSRCs
        with_csrc.push(0x41);
        let h = RtpHeader::parse(&with_csrc).unwrap();
        assert_eq!(h.payload_offset, 20);
    }

    #[test]
    fn aac_hbr_splits_two_access_units() {
        // AU-headers-length = 32 bits → two 16-bit AU-headers.
        // AU sizes 3 and 2 (top 13 bits of each 2-byte header).
        let mut p = Vec::new();
        p.extend_from_slice(&32u16.to_be_bytes()); // header bits
        p.extend_from_slice(&((3u16) << 3).to_be_bytes()); // AU-header: size 3
        p.extend_from_slice(&((2u16) << 3).to_be_bytes()); // AU-header: size 2
        p.extend_from_slice(&[0xA1, 0xA2, 0xA3]); // AU 1
        p.extend_from_slice(&[0xB1, 0xB2]); // AU 2
        let aus = AacDepacketizer::new().push(&p).unwrap();
        assert_eq!(aus.len(), 2);
        assert_eq!(&aus[0][..], &[0xA1, 0xA2, 0xA3]);
        assert_eq!(&aus[1][..], &[0xB1, 0xB2]);
    }

    #[test]
    fn aac_hbr_single_au() {
        let mut p = Vec::new();
        p.extend_from_slice(&16u16.to_be_bytes()); // one 16-bit AU-header
        p.extend_from_slice(&((4u16) << 3).to_be_bytes()); // size 4
        p.extend_from_slice(&[1, 2, 3, 4]);
        let aus = AacDepacketizer::new().push(&p).unwrap();
        assert_eq!(aus.len(), 1);
        assert_eq!(&aus[0][..], &[1, 2, 3, 4]);
    }

    #[test]
    fn aac_truncated_payload_errors() {
        assert_eq!(
            AacDepacketizer::new().push(&[0x00]),
            Err(DepacketizeError::Truncated)
        );
        // Declares one AU of size 8 but supplies only 2 data bytes.
        let mut p = 16u16.to_be_bytes().to_vec();
        p.extend_from_slice(&((8u16) << 3).to_be_bytes());
        p.extend_from_slice(&[1, 2]);
        assert_eq!(
            AacDepacketizer::new().push(&p),
            Err(DepacketizeError::Truncated)
        );
    }

    #[test]
    fn single_nal_packet_emits_annexb_on_marker() {
        let mut d = H264Depacketizer::new();
        // Type 1 (non-IDR slice), marker set → one access unit.
        let out = d.push(&[0x41, 0x9A, 0xBC], true, 3000, 1).unwrap().unwrap();
        assert_eq!(&out.data[..], &[0, 0, 0, 1, 0x41, 0x9A, 0xBC]);
        assert!(!out.keyframe);
        assert_eq!(out.timestamp, 3000);
    }

    #[test]
    fn idr_single_nal_is_flagged_keyframe() {
        let mut d = H264Depacketizer::new();
        let out = d.push(&[0x65, 0x01], true, 0, 1).unwrap().unwrap();
        assert!(out.keyframe);
    }

    #[test]
    fn packetizer_single_nal_round_trips_through_depacketizer() {
        // A small AU (two NALs) → single-NAL packets → reassembled identically.
        let au = [0, 0, 0, 1, 0x67, 0x42, 0x00, 0, 0, 0, 1, 0x65, 0x88, 0x99];
        let mut pkt = RtpPacketizer::new(96, 0xABCD, 1200);
        let packets = pkt.packetize(&au, 3000);
        assert_eq!(packets.len(), 2, "one packet per NAL");

        let mut depack = H264Depacketizer::new();
        let mut out = None;
        for p in &packets {
            let h = RtpHeader::parse(p).unwrap();
            if let Some(au) = depack
                .push(&p[h.payload_offset..], h.marker, h.timestamp, h.sequence)
                .unwrap()
            {
                out = Some(au);
            }
        }
        let out = out.expect("AU completed on the marker packet");
        assert_eq!(&out.data[..], &au);
        assert!(out.keyframe);
        assert_eq!(out.timestamp, 3000);
    }

    #[test]
    fn packetizer_fragments_oversized_nal_and_round_trips() {
        // One NAL larger than the MTU → FU-A fragments → reassembled identically.
        let mut nal = vec![0, 0, 0, 1, 0x65]; // start code + IDR NAL header
        nal.extend((0..600u16).map(|i| i as u8)); // long payload
        let mut pkt = RtpPacketizer::new(96, 1, 100); // tiny MTU forces FU-A
        let packets = pkt.packetize(&nal, 90);
        assert!(packets.len() > 1, "oversized NAL is fragmented");
        // Only the last packet carries the marker bit.
        let markers: Vec<bool> = packets
            .iter()
            .map(|p| RtpHeader::parse(p).unwrap().marker)
            .collect();
        assert_eq!(markers.iter().filter(|m| **m).count(), 1);
        assert!(markers.last().unwrap());

        let mut depack = H264Depacketizer::new();
        let mut out = None;
        for p in &packets {
            let h = RtpHeader::parse(p).unwrap();
            if let Some(au) = depack
                .push(&p[h.payload_offset..], h.marker, h.timestamp, h.sequence)
                .unwrap()
            {
                out = Some(au);
            }
        }
        assert_eq!(&out.unwrap().data[..], &nal[..]);
    }

    #[test]
    fn stap_a_splits_aggregated_nals() {
        // STAP-A (24): [24][size=2][AA BB][size=3][CC DD EE]
        let payload = [24, 0, 2, 0xAA, 0xBB, 0, 3, 0xCC, 0xDD, 0xEE];
        let mut d = H264Depacketizer::new();
        let out = d.push(&payload, true, 0, 1).unwrap().unwrap();
        assert_eq!(
            &out.data[..],
            &[0, 0, 0, 1, 0xAA, 0xBB, 0, 0, 0, 1, 0xCC, 0xDD, 0xEE]
        );
    }

    #[test]
    fn fu_a_reassembles_fragmented_nal() {
        let mut d = H264Depacketizer::new();
        // FU indicator 0x7C (F=0,NRI=3,type=28), FU header start 0x85 (S=1,type=5).
        assert!(d
            .push(&[0x7C, 0x85, 0x11, 0x22], false, 0, 1)
            .unwrap()
            .is_none());
        // Middle fragment (S=0,E=0).
        assert!(d.push(&[0x7C, 0x05, 0x33], false, 0, 2).unwrap().is_none());
        // End fragment (E=1), marker closes the AU.
        let out = d.push(&[0x7C, 0x45, 0x44], true, 0, 3).unwrap().unwrap();
        // Reconstructed NAL header: NRI 0x60 | type 5 = 0x65, then payload bytes.
        assert_eq!(&out.data[..], &[0, 0, 0, 1, 0x65, 0x11, 0x22, 0x33, 0x44]);
        assert!(out.keyframe);
    }

    #[test]
    fn fu_a_sequence_gap_reports_out_of_order() {
        let mut d = H264Depacketizer::new();
        d.push(&[0x7C, 0x85, 0x11], false, 0, 1).unwrap();
        // Jump from seq 1 to seq 5 mid-fragment.
        assert_eq!(
            d.push(&[0x7C, 0x05, 0x22], false, 0, 5),
            Err(DepacketizeError::OutOfOrder)
        );
    }

    #[test]
    fn timestamp_change_flushes_previous_au_without_marker() {
        let mut d = H264Depacketizer::new();
        // First AU, no marker.
        assert!(d.push(&[0x41, 0x01], false, 1000, 1).unwrap().is_none());
        // New timestamp flushes the first AU.
        let out = d.push(&[0x41, 0x02], false, 2000, 2).unwrap().unwrap();
        assert_eq!(out.timestamp, 1000);
        assert_eq!(&out.data[..], &[0, 0, 0, 1, 0x41, 0x01]);
    }
}