openipc_core/
rtp.rs

1use std::collections::BTreeMap;
2
3const DEFAULT_RTP_REORDER_WINDOW: usize = 15;
4
5/// Error returned while parsing or depacketizing RTP video.
6#[derive(Debug, Clone, Copy, PartialEq, Eq)]
7pub enum RtpError {
8    /// Packet is shorter than the RTP fixed header or declared extension.
9    TooShort,
10    /// RTP version was not 2.
11    InvalidVersion(u8),
12    /// RTP extension header length is malformed.
13    InvalidExtension,
14    /// RTP padding length is malformed.
15    InvalidPadding,
16    /// Packet has no payload after header/extension/padding.
17    EmptyPayload,
18    /// Payload could not be interpreted as H.264 or H.265.
19    UnsupportedPayload,
20    /// Fragmented access unit exceeded the configured size guard.
21    FragmentOverflow,
22}
23
24/// Encoded video codec carried by a depacketized frame.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum Codec {
27    /// H.264/AVC video.
28    H264,
29    /// H.265/HEVC video.
30    H265,
31}
32
33/// Dynamic RTP payload type used by OpenIPC for H.264.
34pub const RTP_PAYLOAD_TYPE_H264: u8 = 96;
35/// Dynamic RTP payload type used by OpenIPC for H.265.
36pub const RTP_PAYLOAD_TYPE_H265: u8 = 97;
37/// Dynamic RTP payload type used by OpenIPC/Majestic for Opus audio.
38pub const RTP_PAYLOAD_TYPE_OPUS: u8 = 98;
39
40/// Decoder configuration NAL units observed by the RTP depacketizer.
41///
42/// H.264 needs SPS and PPS before a decoder can be configured. H.265 needs
43/// VPS, SPS and PPS. PixelPilot starts its decoder as soon as these parameter
44/// sets have been observed, then feeds subsequent NAL units without requiring a
45/// fresh IDR for every startup path.
46#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
47pub struct CodecConfigState {
48    /// H.264 sequence parameter set has been seen.
49    pub h264_sps: bool,
50    /// H.264 picture parameter set has been seen.
51    pub h264_pps: bool,
52    /// H.265 video parameter set has been seen.
53    pub h265_vps: bool,
54    /// H.265 sequence parameter set has been seen.
55    pub h265_sps: bool,
56    /// H.265 picture parameter set has been seen.
57    pub h265_pps: bool,
58}
59
60impl CodecConfigState {
61    /// Return true when all parameter sets required for `codec` are cached.
62    pub const fn is_complete_for(self, codec: Codec) -> bool {
63        match codec {
64            Codec::H264 => self.h264_sps && self.h264_pps,
65            Codec::H265 => self.h265_vps && self.h265_sps && self.h265_pps,
66        }
67    }
68}
69
70/// Cumulative RTP depacketizer diagnostics.
71#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
72pub struct RtpDepacketizerStatus {
73    /// RTP packets submitted to the depacketizer.
74    pub packets: u64,
75    /// Annex-B video frames emitted.
76    pub frames_emitted: u64,
77    /// Video NAL units dropped because decoder config was not complete yet.
78    pub config_wait_drops: u64,
79    /// Keyframes emitted with cached decoder config prepended.
80    pub keyframes_with_prepended_config: u64,
81    /// Cached SPS/PPS/VPS parameter-set NAL units prepended to keyframes.
82    pub parameter_sets_prepended: u64,
83    /// Fragment chains dropped because an RTP sequence gap was observed.
84    pub fragment_sequence_gaps: u64,
85    /// Fragment chains that exceeded the configured size guard.
86    pub fragment_overflows: u64,
87    /// Packets rejected because they were not H.264/H.265 video.
88    pub unsupported_payloads: u64,
89    /// Packets rejected because the RTP header or payload was malformed.
90    pub malformed_packets: u64,
91    /// Most recent RTP payload type.
92    pub last_payload_type: Option<u8>,
93    /// Most recent RTP sequence number.
94    pub last_sequence_number: Option<u16>,
95    /// Most recent RTP timestamp.
96    pub last_timestamp: Option<u32>,
97    /// Most recent detected video codec.
98    pub last_codec: Option<Codec>,
99    /// Most recent H.264/H.265 NAL unit type.
100    pub last_nal_type: Option<u8>,
101    /// Current decoder configuration state.
102    pub codec_config: CodecConfigState,
103}
104
105/// Parsed RTP header metadata.
106#[derive(Debug, Clone, Copy, PartialEq, Eq)]
107pub struct RtpHeader {
108    /// RTP marker bit, usually set at an access-unit boundary.
109    pub marker: bool,
110    /// RTP payload type.
111    pub payload_type: u8,
112    /// RTP sequence number.
113    pub sequence_number: u16,
114    /// RTP timestamp.
115    pub timestamp: u32,
116    /// RTP synchronization source.
117    pub ssrc: u32,
118    /// Number of CSRC entries.
119    pub csrc_count: u8,
120    /// True if the packet has an RTP header extension.
121    pub has_extension: bool,
122    /// Header length including CSRC and extension bytes.
123    pub header_len: usize,
124    /// Payload length after header and padding removal.
125    pub payload_len: usize,
126}
127
128impl RtpHeader {
129    /// Parse an RTP header and validate extension/padding bounds.
130    pub fn parse(packet: &[u8]) -> Result<Self, RtpError> {
131        if packet.len() < 12 {
132            return Err(RtpError::TooShort);
133        }
134        let version = packet[0] >> 6;
135        if version != 2 {
136            return Err(RtpError::InvalidVersion(version));
137        }
138
139        let padding = packet[0] & 0x20 != 0;
140        let extension = packet[0] & 0x10 != 0;
141        let csrc_count = packet[0] & 0x0f;
142        let mut header_len = 12 + csrc_count as usize * 4;
143        if packet.len() < header_len {
144            return Err(RtpError::TooShort);
145        }
146
147        if extension {
148            if packet.len() < header_len + 4 {
149                return Err(RtpError::InvalidExtension);
150            }
151            let ext_words =
152                u16::from_be_bytes([packet[header_len + 2], packet[header_len + 3]]) as usize;
153            header_len += 4 + ext_words * 4;
154            if packet.len() < header_len {
155                return Err(RtpError::InvalidExtension);
156            }
157        }
158
159        let padding_len = if padding {
160            let len = *packet.last().ok_or(RtpError::InvalidPadding)? as usize;
161            if len == 0 || len > packet.len() - header_len {
162                return Err(RtpError::InvalidPadding);
163            }
164            len
165        } else {
166            0
167        };
168
169        let payload_len = packet.len() - header_len - padding_len;
170        if payload_len == 0 {
171            return Err(RtpError::EmptyPayload);
172        }
173
174        Ok(Self {
175            marker: packet[1] & 0x80 != 0,
176            payload_type: packet[1] & 0x7f,
177            sequence_number: u16::from_be_bytes([packet[2], packet[3]]),
178            timestamp: u32::from_be_bytes([packet[4], packet[5], packet[6], packet[7]]),
179            ssrc: u32::from_be_bytes([packet[8], packet[9], packet[10], packet[11]]),
180            csrc_count,
181            has_extension: extension,
182            header_len,
183            payload_len,
184        })
185    }
186
187    /// Borrow this packet's payload using the parsed header offsets.
188    pub fn payload<'a>(&self, packet: &'a [u8]) -> &'a [u8] {
189        &packet[self.header_len..self.header_len + self.payload_len]
190    }
191}
192
193/// Cumulative status for [`RtpReorderBuffer`].
194#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
195pub struct RtpReorderStatus {
196    /// Packets currently held while waiting for missing sequence numbers.
197    pub buffered_packets: usize,
198    /// Out-of-order packets accepted into the reorder window.
199    pub reordered_packets: u64,
200    /// Packets dropped because their sequence number was older than the window.
201    pub late_packets: u64,
202    /// Times the window flushed ahead after the missing packet did not arrive.
203    pub forced_flushes: u64,
204}
205
206/// Small RTP sequence reorder buffer.
207///
208/// PixelPilot keeps a short queue before its RTP parser so FU-A/FU fragments
209/// survive small USB/radio delivery inversions. This buffer does the same for
210/// the shared Rust receiver runtime while keeping the in-order path immediate.
211#[derive(Debug, Clone)]
212pub struct RtpReorderBuffer {
213    next_sequence: Option<u16>,
214    pending: BTreeMap<u16, Vec<u8>>,
215    max_depth: usize,
216    status: RtpReorderStatus,
217}
218
219impl Default for RtpReorderBuffer {
220    fn default() -> Self {
221        Self::new(DEFAULT_RTP_REORDER_WINDOW)
222    }
223}
224
225impl RtpReorderBuffer {
226    /// Create a reorder buffer with a maximum pending packet depth.
227    pub fn new(max_depth: usize) -> Self {
228        Self {
229            next_sequence: None,
230            pending: BTreeMap::new(),
231            max_depth: max_depth.max(1),
232            status: RtpReorderStatus::default(),
233        }
234    }
235
236    /// Push one RTP packet and return packets that are ready in sequence order.
237    pub fn push(&mut self, packet: &[u8]) -> Result<Vec<Vec<u8>>, RtpError> {
238        let header = RtpHeader::parse(packet)?;
239        let sequence = header.sequence_number;
240        let mut ready = Vec::new();
241
242        let Some(expected) = self.next_sequence else {
243            self.next_sequence = Some(sequence.wrapping_add(1));
244            ready.push(packet.to_vec());
245            return Ok(ready);
246        };
247
248        if sequence == expected {
249            self.next_sequence = Some(expected.wrapping_add(1));
250            ready.push(packet.to_vec());
251            self.drain_ready(&mut ready);
252            return Ok(ready);
253        }
254
255        if sequence_is_before(sequence, expected) {
256            self.status.late_packets = self.status.late_packets.saturating_add(1);
257            return Ok(ready);
258        }
259
260        if self.pending.insert(sequence, packet.to_vec()).is_none() {
261            self.status.reordered_packets = self.status.reordered_packets.saturating_add(1);
262        }
263        if self.pending.len() >= self.max_depth {
264            self.force_flush(expected, &mut ready);
265        }
266        self.status.buffered_packets = self.pending.len();
267        Ok(ready)
268    }
269
270    /// Return current reorder-buffer status.
271    pub fn status(&self) -> RtpReorderStatus {
272        RtpReorderStatus {
273            buffered_packets: self.pending.len(),
274            ..self.status
275        }
276    }
277
278    fn drain_ready(&mut self, ready: &mut Vec<Vec<u8>>) {
279        while let Some(expected) = self.next_sequence {
280            let Some(packet) = self.pending.remove(&expected) else {
281                break;
282            };
283            self.next_sequence = Some(expected.wrapping_add(1));
284            ready.push(packet);
285        }
286        self.status.buffered_packets = self.pending.len();
287    }
288
289    fn force_flush(&mut self, expected: u16, ready: &mut Vec<Vec<u8>>) {
290        let Some(sequence) = self
291            .pending
292            .keys()
293            .copied()
294            .min_by_key(|sequence| sequence.wrapping_sub(expected))
295        else {
296            return;
297        };
298        if let Some(packet) = self.pending.remove(&sequence) {
299            self.status.forced_flushes = self.status.forced_flushes.saturating_add(1);
300            self.next_sequence = Some(sequence.wrapping_add(1));
301            ready.push(packet);
302            self.drain_ready(ready);
303        }
304    }
305}
306
307fn sequence_is_before(sequence: u16, expected: u16) -> bool {
308    let backward = expected.wrapping_sub(sequence);
309    backward != 0 && backward < 0x8000
310}
311
312/// Encoded Annex-B H.264/H.265 access unit emitted by [`RtpDepacketizer`].
313#[derive(Debug, Clone, PartialEq, Eq)]
314pub struct DepacketizedFrame {
315    /// Annex-B byte stream data including start codes.
316    pub data: Vec<u8>,
317    /// RTP timestamp associated with the access unit.
318    pub timestamp: u32,
319    /// True when the frame contains an IDR/keyframe entry point.
320    pub is_keyframe: bool,
321    /// Video codec for this frame.
322    pub codec: Codec,
323    /// RTP payload type that produced this frame.
324    pub payload_type: u8,
325    /// RTP sequence number of the packet that completed this frame.
326    pub sequence_number: u16,
327    /// H.264/H.265 NAL unit type for the frame payload.
328    pub nal_type: u8,
329    /// Decoder parameter-set state at the time this frame was emitted.
330    pub codec_config: CodecConfigState,
331}
332
333#[derive(Debug, Default, Clone)]
334struct FragmentState {
335    data: Vec<u8>,
336    timestamp: u32,
337    next_sequence: Option<u16>,
338    corrupted: bool,
339}
340
341#[derive(Debug, Clone, Copy)]
342struct FrameMeta {
343    timestamp: u32,
344    is_keyframe: bool,
345    codec: Codec,
346    payload_type: u8,
347    sequence_number: u16,
348    nal_type: u8,
349}
350
351/// Stateful RTP depacketizer for OpenIPC H.264/H.265 video.
352///
353/// The depacketizer buffers fragmented NAL units, drops incomplete fragments
354/// across sequence gaps, and emits complete Annex-B access units.
355#[derive(Debug, Clone)]
356pub struct RtpDepacketizer {
357    h264: FragmentState,
358    h265: FragmentState,
359    h264_sps: Option<Vec<u8>>,
360    h264_pps: Option<Vec<u8>>,
361    h265_vps: Option<Vec<u8>>,
362    h265_sps: Option<Vec<u8>>,
363    h265_pps: Option<Vec<u8>>,
364    max_fragment: usize,
365    status: RtpDepacketizerStatus,
366}
367
368impl Default for RtpDepacketizer {
369    fn default() -> Self {
370        Self::new()
371    }
372}
373
374impl RtpDepacketizer {
375    /// Create a depacketizer with the default fragment-size guard.
376    pub fn new() -> Self {
377        Self {
378            h264: FragmentState::default(),
379            h265: FragmentState::default(),
380            h264_sps: None,
381            h264_pps: None,
382            h265_vps: None,
383            h265_sps: None,
384            h265_pps: None,
385            max_fragment: 1024 * 1024,
386            status: RtpDepacketizerStatus::default(),
387        }
388    }
389
390    /// Return cumulative depacketizer status and codec configuration state.
391    pub fn status(&self) -> RtpDepacketizerStatus {
392        RtpDepacketizerStatus {
393            codec_config: self.codec_config(),
394            ..self.status
395        }
396    }
397
398    /// Return the current decoder parameter-set state.
399    pub fn codec_config(&self) -> CodecConfigState {
400        CodecConfigState {
401            h264_sps: self.h264_sps.is_some(),
402            h264_pps: self.h264_pps.is_some(),
403            h265_vps: self.h265_vps.is_some(),
404            h265_sps: self.h265_sps.is_some(),
405            h265_pps: self.h265_pps.is_some(),
406        }
407    }
408
409    /// Push one RTP packet and return a complete frame when one is ready.
410    pub fn push(&mut self, packet: &[u8]) -> Result<Option<DepacketizedFrame>, RtpError> {
411        self.status.packets = self.status.packets.saturating_add(1);
412        let header = match RtpHeader::parse(packet) {
413            Ok(header) => header,
414            Err(err) => {
415                self.record_error(err);
416                return Err(err);
417            }
418        };
419        self.status.last_payload_type = Some(header.payload_type);
420        self.status.last_sequence_number = Some(header.sequence_number);
421        self.status.last_timestamp = Some(header.timestamp);
422        let payload = header.payload(packet);
423        if header.payload_type == RTP_PAYLOAD_TYPE_OPUS {
424            self.record_error(RtpError::UnsupportedPayload);
425            return Err(RtpError::UnsupportedPayload);
426        }
427        let Some(codec) =
428            codec_from_payload_type(header.payload_type).or_else(|| detect_codec(payload))
429        else {
430            self.record_error(RtpError::UnsupportedPayload);
431            return Err(RtpError::UnsupportedPayload);
432        };
433        self.status.last_codec = Some(codec);
434        let result = match codec {
435            Codec::H264 => self.push_h264(payload, header),
436            Codec::H265 => self.push_h265(payload, header),
437        };
438        match &result {
439            Ok(Some(_)) => {
440                self.status.frames_emitted = self.status.frames_emitted.saturating_add(1)
441            }
442            Err(err) => self.record_error(*err),
443            _ => {}
444        }
445        result
446    }
447
448    fn push_h264(
449        &mut self,
450        payload: &[u8],
451        header: RtpHeader,
452    ) -> Result<Option<DepacketizedFrame>, RtpError> {
453        let nal_type = payload[0] & 0x1f;
454        self.status.last_nal_type = Some(nal_type);
455        match nal_type {
456            7 => {
457                self.h264_sps = Some(payload.to_vec());
458                Ok(None)
459            }
460            8 => {
461                self.h264_pps = Some(payload.to_vec());
462                Ok(None)
463            }
464            24 => self.h264_stap_a(payload, header),
465            28 => self.h264_fu_a(payload, header),
466            _ if self.has_decoder_config(Codec::H264) && is_h264_vcl_nal(nal_type) => {
467                Ok(Some(self.frame_with_prefix(
468                    payload,
469                    FrameMeta {
470                        timestamp: header.timestamp,
471                        is_keyframe: nal_type == 5,
472                        codec: Codec::H264,
473                        payload_type: header.payload_type,
474                        sequence_number: header.sequence_number,
475                        nal_type,
476                    },
477                )))
478            }
479            _ if !is_h264_vcl_nal(nal_type) => Ok(None),
480            _ => {
481                self.status.config_wait_drops = self.status.config_wait_drops.saturating_add(1);
482                Ok(None)
483            }
484        }
485    }
486
487    fn push_h265(
488        &mut self,
489        payload: &[u8],
490        header: RtpHeader,
491    ) -> Result<Option<DepacketizedFrame>, RtpError> {
492        if payload.len() < 2 {
493            return Err(RtpError::UnsupportedPayload);
494        }
495        let nal_type = (payload[0] >> 1) & 0x3f;
496        self.status.last_nal_type = Some(nal_type);
497        match nal_type {
498            32 => {
499                self.h265_vps = Some(payload.to_vec());
500                Ok(None)
501            }
502            33 => {
503                self.h265_sps = Some(payload.to_vec());
504                Ok(None)
505            }
506            34 => {
507                self.h265_pps = Some(payload.to_vec());
508                Ok(None)
509            }
510            48 => self.h265_ap(payload, header),
511            49 => self.h265_fu(payload, header),
512            _ if self.has_decoder_config(Codec::H265) && is_h265_vcl_nal(nal_type) => {
513                Ok(Some(self.frame_with_prefix(
514                    payload,
515                    FrameMeta {
516                        timestamp: header.timestamp,
517                        is_keyframe: (16..=23).contains(&nal_type),
518                        codec: Codec::H265,
519                        payload_type: header.payload_type,
520                        sequence_number: header.sequence_number,
521                        nal_type,
522                    },
523                )))
524            }
525            _ if !is_h265_vcl_nal(nal_type) => Ok(None),
526            _ => {
527                self.status.config_wait_drops = self.status.config_wait_drops.saturating_add(1);
528                Ok(None)
529            }
530        }
531    }
532
533    fn h264_fu_a(
534        &mut self,
535        payload: &[u8],
536        header: RtpHeader,
537    ) -> Result<Option<DepacketizedFrame>, RtpError> {
538        if payload.len() < 2 {
539            return Err(RtpError::UnsupportedPayload);
540        }
541        let fu_indicator = payload[0];
542        let fu_header = payload[1];
543        let start = fu_header & 0x80 != 0;
544        let end = fu_header & 0x40 != 0;
545        let nal_type = fu_header & 0x1f;
546        if start {
547            self.h264.data.clear();
548            self.h264.timestamp = header.timestamp;
549            self.h264.next_sequence = Some(header.sequence_number.wrapping_add(1));
550            self.h264.corrupted = false;
551            self.h264.data.push((fu_indicator & 0xe0) | nal_type);
552        } else if !self.accept_fragment_sequence(Codec::H264, header.sequence_number) {
553            return Ok(None);
554        }
555        if !self.h264.corrupted {
556            self.append_fragment(Codec::H264, &payload[2..])?;
557        }
558        if end {
559            if !is_h264_vcl_nal(nal_type) {
560                self.reset_fragment(Codec::H264);
561                return Ok(None);
562            }
563            if self.h264.corrupted || !self.has_decoder_config(Codec::H264) {
564                if !self.has_decoder_config(Codec::H264) {
565                    self.status.config_wait_drops = self.status.config_wait_drops.saturating_add(1);
566                }
567                self.reset_fragment(Codec::H264);
568                return Ok(None);
569            }
570            let data = self.h264.data.clone();
571            let frame = self.frame_with_prefix(
572                &data,
573                FrameMeta {
574                    timestamp: self.h264.timestamp,
575                    is_keyframe: nal_type == 5,
576                    codec: Codec::H264,
577                    payload_type: header.payload_type,
578                    sequence_number: header.sequence_number,
579                    nal_type,
580                },
581            );
582            self.reset_fragment(Codec::H264);
583            Ok(Some(frame))
584        } else {
585            Ok(None)
586        }
587    }
588
589    fn h265_fu(
590        &mut self,
591        payload: &[u8],
592        header: RtpHeader,
593    ) -> Result<Option<DepacketizedFrame>, RtpError> {
594        if payload.len() < 3 {
595            return Err(RtpError::UnsupportedPayload);
596        }
597        let fu_header = payload[2];
598        let start = fu_header & 0x80 != 0;
599        let end = fu_header & 0x40 != 0;
600        let nal_type = fu_header & 0x3f;
601        if start {
602            self.h265.data.clear();
603            self.h265.timestamp = header.timestamp;
604            self.h265.next_sequence = Some(header.sequence_number.wrapping_add(1));
605            self.h265.corrupted = false;
606            self.h265.data.push((nal_type << 1) | (payload[0] & 0x01));
607            self.h265.data.push(payload[1]);
608        } else if !self.accept_fragment_sequence(Codec::H265, header.sequence_number) {
609            return Ok(None);
610        }
611        if !self.h265.corrupted {
612            self.append_fragment(Codec::H265, &payload[3..])?;
613        }
614        if end {
615            if !is_h265_vcl_nal(nal_type) {
616                self.reset_fragment(Codec::H265);
617                return Ok(None);
618            }
619            if self.h265.corrupted || !self.has_decoder_config(Codec::H265) {
620                if !self.has_decoder_config(Codec::H265) {
621                    self.status.config_wait_drops = self.status.config_wait_drops.saturating_add(1);
622                }
623                self.reset_fragment(Codec::H265);
624                return Ok(None);
625            }
626            let data = self.h265.data.clone();
627            let frame = self.frame_with_prefix(
628                &data,
629                FrameMeta {
630                    timestamp: self.h265.timestamp,
631                    is_keyframe: (16..=23).contains(&nal_type),
632                    codec: Codec::H265,
633                    payload_type: header.payload_type,
634                    sequence_number: header.sequence_number,
635                    nal_type,
636                },
637            );
638            self.reset_fragment(Codec::H265);
639            Ok(Some(frame))
640        } else {
641            Ok(None)
642        }
643    }
644
645    fn accept_fragment_sequence(&mut self, codec: Codec, sequence_number: u16) -> bool {
646        let state = match codec {
647            Codec::H264 => &mut self.h264,
648            Codec::H265 => &mut self.h265,
649        };
650        let Some(expected) = state.next_sequence else {
651            return false;
652        };
653        state.next_sequence = Some(sequence_number.wrapping_add(1));
654        if sequence_number != expected {
655            state.data.clear();
656            state.corrupted = true;
657            self.status.fragment_sequence_gaps =
658                self.status.fragment_sequence_gaps.saturating_add(1);
659            return false;
660        }
661        true
662    }
663
664    fn reset_fragment(&mut self, codec: Codec) {
665        let state = match codec {
666            Codec::H264 => &mut self.h264,
667            Codec::H265 => &mut self.h265,
668        };
669        state.data.clear();
670        state.next_sequence = None;
671        state.corrupted = false;
672    }
673
674    fn h264_stap_a(
675        &mut self,
676        payload: &[u8],
677        header: RtpHeader,
678    ) -> Result<Option<DepacketizedFrame>, RtpError> {
679        let mut out = Vec::new();
680        let mut offset = 1;
681        let mut keyframe = false;
682        let mut has_slice = false;
683        let mut has_sps = false;
684        let mut has_pps = false;
685        while offset + 2 <= payload.len() {
686            let len = u16::from_be_bytes([payload[offset], payload[offset + 1]]) as usize;
687            offset += 2;
688            if offset + len > payload.len() {
689                break;
690            }
691            let nalu = &payload[offset..offset + len];
692            let nal_type = nalu.first().map(|b| b & 0x1f).unwrap_or(0);
693            self.status.last_nal_type = Some(nal_type);
694            match nal_type {
695                7 => {
696                    has_sps = true;
697                    self.h264_sps = Some(nalu.to_vec());
698                }
699                8 => {
700                    has_pps = true;
701                    self.h264_pps = Some(nalu.to_vec());
702                }
703                _ => {}
704            }
705            has_slice |= (1..=5).contains(&nal_type);
706            keyframe |= nal_type == 5;
707            append_annex_b(&mut out, nalu);
708            offset += len;
709        }
710        if !has_slice || !self.has_decoder_config(Codec::H264) {
711            if has_slice {
712                self.status.config_wait_drops = self.status.config_wait_drops.saturating_add(1);
713            }
714            return Ok(None);
715        }
716        let data = if keyframe && !(has_sps && has_pps) {
717            let mut data = Vec::new();
718            self.prepend_cached_config(&mut data, Codec::H264);
719            data.extend_from_slice(&out);
720            data
721        } else {
722            out
723        };
724        Ok(Some(DepacketizedFrame {
725            data,
726            timestamp: header.timestamp,
727            is_keyframe: keyframe,
728            codec: Codec::H264,
729            payload_type: header.payload_type,
730            sequence_number: header.sequence_number,
731            nal_type: self.status.last_nal_type.unwrap_or(0),
732            codec_config: self.codec_config(),
733        }))
734    }
735
736    fn h265_ap(
737        &mut self,
738        payload: &[u8],
739        header: RtpHeader,
740    ) -> Result<Option<DepacketizedFrame>, RtpError> {
741        let mut out = Vec::new();
742        let mut offset = 2;
743        let mut keyframe = false;
744        let mut has_slice = false;
745        let mut has_vps = false;
746        let mut has_sps = false;
747        let mut has_pps = false;
748        while offset + 2 <= payload.len() {
749            let len = u16::from_be_bytes([payload[offset], payload[offset + 1]]) as usize;
750            offset += 2;
751            if offset + len > payload.len() {
752                break;
753            }
754            let nalu = &payload[offset..offset + len];
755            let nal_type = nalu.first().map(|b| (b >> 1) & 0x3f).unwrap_or(0);
756            self.status.last_nal_type = Some(nal_type);
757            match nal_type {
758                32 => {
759                    has_vps = true;
760                    self.h265_vps = Some(nalu.to_vec());
761                }
762                33 => {
763                    has_sps = true;
764                    self.h265_sps = Some(nalu.to_vec());
765                }
766                34 => {
767                    has_pps = true;
768                    self.h265_pps = Some(nalu.to_vec());
769                }
770                _ => {}
771            }
772            has_slice |= !nalu.is_empty() && nal_type <= 31;
773            keyframe |= (16..=23).contains(&nal_type);
774            append_annex_b(&mut out, nalu);
775            offset += len;
776        }
777        if !has_slice || !self.has_decoder_config(Codec::H265) {
778            if has_slice {
779                self.status.config_wait_drops = self.status.config_wait_drops.saturating_add(1);
780            }
781            return Ok(None);
782        }
783        let data = if keyframe && !(has_vps && has_sps && has_pps) {
784            let mut data = Vec::new();
785            self.prepend_cached_config(&mut data, Codec::H265);
786            data.extend_from_slice(&out);
787            data
788        } else {
789            out
790        };
791        Ok(Some(DepacketizedFrame {
792            data,
793            timestamp: header.timestamp,
794            is_keyframe: keyframe,
795            codec: Codec::H265,
796            payload_type: header.payload_type,
797            sequence_number: header.sequence_number,
798            nal_type: self.status.last_nal_type.unwrap_or(0),
799            codec_config: self.codec_config(),
800        }))
801    }
802
803    fn append_fragment(&mut self, codec: Codec, bytes: &[u8]) -> Result<(), RtpError> {
804        let state = match codec {
805            Codec::H264 => &mut self.h264,
806            Codec::H265 => &mut self.h265,
807        };
808        if state.data.len() + bytes.len() > self.max_fragment {
809            self.status.fragment_overflows = self.status.fragment_overflows.saturating_add(1);
810            return Err(RtpError::FragmentOverflow);
811        }
812        state.data.extend_from_slice(bytes);
813        Ok(())
814    }
815
816    fn frame_with_prefix(&mut self, nalu: &[u8], meta: FrameMeta) -> DepacketizedFrame {
817        let mut data = Vec::new();
818        if meta.is_keyframe {
819            self.prepend_cached_config(&mut data, meta.codec);
820        }
821        append_annex_b(&mut data, nalu);
822        DepacketizedFrame {
823            data,
824            timestamp: meta.timestamp,
825            is_keyframe: meta.is_keyframe,
826            codec: meta.codec,
827            payload_type: meta.payload_type,
828            sequence_number: meta.sequence_number,
829            nal_type: meta.nal_type,
830            codec_config: self.codec_config(),
831        }
832    }
833
834    fn prepend_cached_config(&mut self, data: &mut Vec<u8>, codec: Codec) {
835        let mut prepended = 0u64;
836        match codec {
837            Codec::H264 => {
838                if let Some(sps) = &self.h264_sps {
839                    append_annex_b(data, sps);
840                    prepended += 1;
841                }
842                if let Some(pps) = &self.h264_pps {
843                    append_annex_b(data, pps);
844                    prepended += 1;
845                }
846            }
847            Codec::H265 => {
848                if let Some(vps) = &self.h265_vps {
849                    append_annex_b(data, vps);
850                    prepended += 1;
851                }
852                if let Some(sps) = &self.h265_sps {
853                    append_annex_b(data, sps);
854                    prepended += 1;
855                }
856                if let Some(pps) = &self.h265_pps {
857                    append_annex_b(data, pps);
858                    prepended += 1;
859                }
860            }
861        }
862        if prepended > 0 {
863            self.status.keyframes_with_prepended_config = self
864                .status
865                .keyframes_with_prepended_config
866                .saturating_add(1);
867            self.status.parameter_sets_prepended = self
868                .status
869                .parameter_sets_prepended
870                .saturating_add(prepended);
871        }
872    }
873
874    fn has_decoder_config(&self, codec: Codec) -> bool {
875        match codec {
876            Codec::H264 => self.h264_sps.is_some() && self.h264_pps.is_some(),
877            Codec::H265 => {
878                self.h265_vps.is_some() && self.h265_sps.is_some() && self.h265_pps.is_some()
879            }
880        }
881    }
882
883    fn record_error(&mut self, err: RtpError) {
884        match err {
885            RtpError::UnsupportedPayload => {
886                self.status.unsupported_payloads =
887                    self.status.unsupported_payloads.saturating_add(1);
888            }
889            RtpError::FragmentOverflow => {}
890            _ => {
891                self.status.malformed_packets = self.status.malformed_packets.saturating_add(1);
892            }
893        }
894    }
895}
896
897fn codec_from_payload_type(payload_type: u8) -> Option<Codec> {
898    match payload_type {
899        RTP_PAYLOAD_TYPE_H264 => Some(Codec::H264),
900        RTP_PAYLOAD_TYPE_H265 => Some(Codec::H265),
901        _ => None,
902    }
903}
904
905fn detect_codec(payload: &[u8]) -> Option<Codec> {
906    if payload.is_empty() {
907        return None;
908    }
909    if payload.len() >= 2 {
910        let h265_nal_type = (payload[0] >> 1) & 0x3f;
911        if h265_nal_type == 48 || h265_nal_type == 49 || (32..=40).contains(&h265_nal_type) {
912            return Some(Codec::H265);
913        }
914    }
915    let h264_nal_type = payload[0] & 0x1f;
916    if h264_nal_type == 24 || h264_nal_type == 28 || (1..=12).contains(&h264_nal_type) {
917        return Some(Codec::H264);
918    }
919    None
920}
921
922fn is_h264_vcl_nal(nal_type: u8) -> bool {
923    (1..=5).contains(&nal_type)
924}
925
926fn is_h265_vcl_nal(nal_type: u8) -> bool {
927    nal_type <= 31
928}
929
930fn append_annex_b(out: &mut Vec<u8>, nalu: &[u8]) {
931    out.extend_from_slice(&[0, 0, 0, 1]);
932    out.extend_from_slice(nalu);
933}
934
935#[cfg(test)]
936mod tests {
937    use super::*;
938
939    fn rtp(payload: &[u8], marker: bool, seq: u16, timestamp: u32) -> Vec<u8> {
940        rtp_with_payload_type(payload, RTP_PAYLOAD_TYPE_H264, marker, seq, timestamp)
941    }
942
943    fn rtp_with_payload_type(
944        payload: &[u8],
945        payload_type: u8,
946        marker: bool,
947        seq: u16,
948        timestamp: u32,
949    ) -> Vec<u8> {
950        let mut packet = vec![
951            0x80,
952            (if marker { 0x80 } else { 0x00 }) | (payload_type & 0x7f),
953        ];
954        packet.extend_from_slice(&seq.to_be_bytes());
955        packet.extend_from_slice(&timestamp.to_be_bytes());
956        packet.extend_from_slice(&0x1122_3344u32.to_be_bytes());
957        packet.extend_from_slice(payload);
958        packet
959    }
960
961    fn stap_a(units: &[&[u8]]) -> Vec<u8> {
962        let mut payload = vec![24];
963        for unit in units {
964            payload.extend_from_slice(&(unit.len() as u16).to_be_bytes());
965            payload.extend_from_slice(unit);
966        }
967        payload
968    }
969
970    fn h265_ap(units: &[&[u8]]) -> Vec<u8> {
971        let mut payload = vec![0x60, 0x01];
972        for unit in units {
973            payload.extend_from_slice(&(unit.len() as u16).to_be_bytes());
974            payload.extend_from_slice(unit);
975        }
976        payload
977    }
978
979    fn prime_h264(depay: &mut RtpDepacketizer) {
980        assert!(depay
981            .push(&rtp(&[0x67, 0x64, 0x00, 0x1f], true, 1, 10))
982            .unwrap()
983            .is_none());
984        assert!(depay
985            .push(&rtp(&[0x68, 0xee], true, 2, 10))
986            .unwrap()
987            .is_none());
988    }
989
990    fn prime_h265(depay: &mut RtpDepacketizer) {
991        for (seq, payload) in [
992            (1, &[0x40, 0x01, 0xaa][..]),
993            (2, &[0x42, 0x01, 0xbb][..]),
994            (3, &[0x44, 0x01, 0xcc][..]),
995        ] {
996            assert!(depay
997                .push(&rtp_with_payload_type(
998                    payload,
999                    RTP_PAYLOAD_TYPE_H265,
1000                    true,
1001                    seq,
1002                    10,
1003                ))
1004                .unwrap()
1005                .is_none());
1006        }
1007    }
1008
1009    #[test]
1010    fn parses_rtp_header() {
1011        let packet = rtp(&[0x65, 1, 2], true, 7, 1234);
1012        let header = RtpHeader::parse(&packet).unwrap();
1013        assert!(header.marker);
1014        assert_eq!(header.payload_type, 96);
1015        assert_eq!(header.sequence_number, 7);
1016        assert_eq!(header.timestamp, 1234);
1017        assert_eq!(header.payload(&packet), &[0x65, 1, 2]);
1018    }
1019
1020    #[test]
1021    fn depacketizes_h264_single_nalu_as_annex_b() {
1022        let mut depay = RtpDepacketizer::new();
1023        prime_h264(&mut depay);
1024        let frame = depay
1025            .push(&rtp(&[0x65, 0xaa], true, 1, 42))
1026            .unwrap()
1027            .unwrap();
1028        assert_eq!(frame.codec, Codec::H264);
1029        assert!(frame.is_keyframe);
1030        assert_eq!(
1031            frame.data,
1032            [
1033                &[0, 0, 0, 1, 0x67, 0x64, 0x00, 0x1f][..],
1034                &[0, 0, 0, 1, 0x68, 0xee][..],
1035                &[0, 0, 0, 1, 0x65, 0xaa][..],
1036            ]
1037            .concat()
1038        );
1039    }
1040
1041    #[test]
1042    fn drops_h264_video_until_sps_and_pps_are_seen() {
1043        let mut depay = RtpDepacketizer::new();
1044        assert!(depay
1045            .push(&rtp(&[0x65, 0xaa], true, 1, 42))
1046            .unwrap()
1047            .is_none());
1048        let status = depay.status();
1049        assert_eq!(status.config_wait_drops, 1);
1050        assert!(!status.codec_config.is_complete_for(Codec::H264));
1051        assert_eq!(status.last_nal_type, Some(5));
1052    }
1053
1054    #[test]
1055    fn h264_payload_type_prevents_h265_false_positive() {
1056        let mut depay = RtpDepacketizer::new();
1057        prime_h264(&mut depay);
1058        let frame = depay
1059            .push(&rtp(&[0x41, 0xaa], true, 1, 42))
1060            .unwrap()
1061            .unwrap();
1062        assert_eq!(frame.codec, Codec::H264);
1063        assert!(!frame.is_keyframe);
1064        assert_eq!(frame.data, &[0, 0, 0, 1, 0x41, 0xaa]);
1065    }
1066
1067    #[test]
1068    fn h264_non_vcl_nal_is_not_emitted_as_video_frame() {
1069        let mut depay = RtpDepacketizer::new();
1070        prime_h264(&mut depay);
1071        assert!(depay
1072            .push(&rtp(&[0x06, 0x05, 0xff], true, 3, 42))
1073            .unwrap()
1074            .is_none());
1075    }
1076
1077    #[test]
1078    fn opus_payload_type_is_not_sniffed_as_video() {
1079        let mut depay = RtpDepacketizer::new();
1080        prime_h264(&mut depay);
1081        let err = depay
1082            .push(&rtp_with_payload_type(
1083                &[0x65, 0xaa],
1084                RTP_PAYLOAD_TYPE_OPUS,
1085                true,
1086                1,
1087                42,
1088            ))
1089            .unwrap_err();
1090        assert_eq!(err, RtpError::UnsupportedPayload);
1091    }
1092
1093    #[test]
1094    fn depacketizes_h265_single_nalu_by_payload_type() {
1095        let mut depay = RtpDepacketizer::new();
1096        prime_h265(&mut depay);
1097        let frame = depay
1098            .push(&rtp_with_payload_type(
1099                &[0x02, 0x01, 0xaa],
1100                RTP_PAYLOAD_TYPE_H265,
1101                true,
1102                1,
1103                42,
1104            ))
1105            .unwrap()
1106            .unwrap();
1107        assert_eq!(frame.codec, Codec::H265);
1108        assert!(!frame.is_keyframe);
1109        assert_eq!(frame.data, &[0, 0, 0, 1, 0x02, 0x01, 0xaa]);
1110    }
1111
1112    #[test]
1113    fn h265_non_vcl_nal_is_not_emitted_as_video_frame() {
1114        let mut depay = RtpDepacketizer::new();
1115        prime_h265(&mut depay);
1116        assert!(depay
1117            .push(&rtp_with_payload_type(
1118                &[0x4e, 0x01, 0xff],
1119                RTP_PAYLOAD_TYPE_H265,
1120                true,
1121                4,
1122                42,
1123            ))
1124            .unwrap()
1125            .is_none());
1126    }
1127
1128    #[test]
1129    fn h264_stap_a_caches_parameter_sets_for_later_keyframe() {
1130        let mut depay = RtpDepacketizer::new();
1131        let sps = &[0x67, 0x64, 0x00, 0x1f][..];
1132        let pps = &[0x68, 0xee][..];
1133        let aggregate = depay.push(&rtp(&stap_a(&[sps, pps]), true, 1, 10)).unwrap();
1134        assert!(aggregate.is_none());
1135
1136        let frame = depay
1137            .push(&rtp(&[0x65, 0xaa], true, 2, 20))
1138            .unwrap()
1139            .unwrap();
1140        assert!(frame.is_keyframe);
1141        assert_eq!(
1142            frame.data,
1143            [
1144                &[0, 0, 0, 1][..],
1145                sps,
1146                &[0, 0, 0, 1][..],
1147                pps,
1148                &[0, 0, 0, 1, 0x65, 0xaa][..],
1149            ]
1150            .concat()
1151        );
1152    }
1153
1154    #[test]
1155    fn h264_stap_a_prepends_cached_parameter_sets_for_idr_without_inband_config() {
1156        let mut depay = RtpDepacketizer::new();
1157        let sps = &[0x67, 0x64, 0x00, 0x1f][..];
1158        let pps = &[0x68, 0xee][..];
1159        depay.push(&rtp(&stap_a(&[sps, pps]), true, 1, 10)).unwrap();
1160
1161        let frame = depay
1162            .push(&rtp(&stap_a(&[&[0x65, 0xaa, 0xbb]]), true, 2, 20))
1163            .unwrap()
1164            .unwrap();
1165
1166        assert!(frame.is_keyframe);
1167        assert_eq!(
1168            frame.data,
1169            [
1170                &[0, 0, 0, 1][..],
1171                sps,
1172                &[0, 0, 0, 1][..],
1173                pps,
1174                &[0, 0, 0, 1, 0x65, 0xaa, 0xbb][..],
1175            ]
1176            .concat()
1177        );
1178        let status = depay.status();
1179        assert_eq!(status.keyframes_with_prepended_config, 1);
1180        assert_eq!(status.parameter_sets_prepended, 2);
1181    }
1182
1183    #[test]
1184    fn h264_stap_a_does_not_duplicate_inband_parameter_sets() {
1185        let mut depay = RtpDepacketizer::new();
1186        let sps = &[0x67, 0x64, 0x00, 0x1f][..];
1187        let pps = &[0x68, 0xee][..];
1188        let frame = depay
1189            .push(&rtp(&stap_a(&[sps, pps, &[0x65, 0xaa]]), true, 1, 20))
1190            .unwrap()
1191            .unwrap();
1192
1193        assert_eq!(
1194            frame.data,
1195            [
1196                &[0, 0, 0, 1][..],
1197                sps,
1198                &[0, 0, 0, 1][..],
1199                pps,
1200                &[0, 0, 0, 1, 0x65, 0xaa][..],
1201            ]
1202            .concat()
1203        );
1204        let status = depay.status();
1205        assert_eq!(status.keyframes_with_prepended_config, 0);
1206        assert_eq!(status.parameter_sets_prepended, 0);
1207    }
1208
1209    #[test]
1210    fn h265_ap_prepends_cached_parameter_sets_for_keyframe_without_inband_config() {
1211        let mut depay = RtpDepacketizer::new();
1212        prime_h265(&mut depay);
1213        let frame = depay
1214            .push(&rtp_with_payload_type(
1215                &h265_ap(&[&[0x26, 0x01, 0xaa]]),
1216                RTP_PAYLOAD_TYPE_H265,
1217                true,
1218                4,
1219                20,
1220            ))
1221            .unwrap()
1222            .unwrap();
1223
1224        assert!(frame.is_keyframe);
1225        assert_eq!(
1226            frame.data,
1227            [
1228                &[0, 0, 0, 1, 0x40, 0x01, 0xaa][..],
1229                &[0, 0, 0, 1, 0x42, 0x01, 0xbb][..],
1230                &[0, 0, 0, 1, 0x44, 0x01, 0xcc][..],
1231                &[0, 0, 0, 1, 0x26, 0x01, 0xaa][..],
1232            ]
1233            .concat()
1234        );
1235        let status = depay.status();
1236        assert_eq!(status.keyframes_with_prepended_config, 1);
1237        assert_eq!(status.parameter_sets_prepended, 3);
1238    }
1239
1240    #[test]
1241    fn depacketizes_h264_fu_a() {
1242        let mut depay = RtpDepacketizer::new();
1243        prime_h264(&mut depay);
1244        assert!(depay
1245            .push(&rtp(&[0x7c, 0x85, 1, 2], false, 1, 99))
1246            .unwrap()
1247            .is_none());
1248        let frame = depay
1249            .push(&rtp(&[0x7c, 0x45, 3, 4], true, 2, 99))
1250            .unwrap()
1251            .unwrap();
1252        assert_eq!(
1253            frame.data,
1254            [
1255                &[0, 0, 0, 1, 0x67, 0x64, 0x00, 0x1f][..],
1256                &[0, 0, 0, 1, 0x68, 0xee][..],
1257                &[0, 0, 0, 1, 0x65, 1, 2, 3, 4][..],
1258            ]
1259            .concat()
1260        );
1261    }
1262
1263    #[test]
1264    fn drops_h264_fu_a_after_sequence_gap() {
1265        let mut depay = RtpDepacketizer::new();
1266        prime_h264(&mut depay);
1267        assert!(depay
1268            .push(&rtp(&[0x7c, 0x85, 1, 2], false, 10, 99))
1269            .unwrap()
1270            .is_none());
1271        assert!(depay
1272            .push(&rtp(&[0x7c, 0x45, 3, 4], true, 12, 99))
1273            .unwrap()
1274            .is_none());
1275
1276        assert!(depay
1277            .push(&rtp(&[0x7c, 0x85, 5, 6], false, 13, 100))
1278            .unwrap()
1279            .is_none());
1280        let frame = depay
1281            .push(&rtp(&[0x7c, 0x45, 7, 8], true, 14, 100))
1282            .unwrap()
1283            .unwrap();
1284        assert!(frame.data.ends_with(&[0, 0, 0, 1, 0x65, 5, 6, 7, 8]));
1285    }
1286
1287    #[test]
1288    fn drops_fragment_end_without_start() {
1289        let mut depay = RtpDepacketizer::new();
1290        prime_h264(&mut depay);
1291        assert!(depay
1292            .push(&rtp(&[0x7c, 0x45, 1, 2], true, 10, 99))
1293            .unwrap()
1294            .is_none());
1295    }
1296
1297    #[test]
1298    fn status_tracks_h264_decoder_config() {
1299        let mut depay = RtpDepacketizer::new();
1300        depay
1301            .push(&rtp(&[0x67, 0x64, 0x00, 0x1f], true, 1, 10))
1302            .unwrap();
1303        let status = depay.status();
1304        assert!(status.codec_config.h264_sps);
1305        assert!(!status.codec_config.h264_pps);
1306        assert!(!status.codec_config.is_complete_for(Codec::H264));
1307
1308        depay.push(&rtp(&[0x68, 0xee], true, 2, 10)).unwrap();
1309        let status = depay.status();
1310        assert!(status.codec_config.is_complete_for(Codec::H264));
1311    }
1312
1313    #[test]
1314    fn reorder_buffer_restores_short_out_of_order_burst() {
1315        let mut reorder = RtpReorderBuffer::default();
1316        let first = rtp(&[0x61, 1], true, 10, 90);
1317        let second = rtp(&[0x61, 2], true, 11, 90);
1318        let third = rtp(&[0x61, 3], true, 12, 90);
1319
1320        assert_eq!(reorder.push(&first).unwrap(), vec![first.clone()]);
1321        assert!(reorder.push(&third).unwrap().is_empty());
1322        assert_eq!(reorder.status().buffered_packets, 1);
1323        assert_eq!(reorder.status().reordered_packets, 1);
1324
1325        let ready = reorder.push(&second).unwrap();
1326        assert_eq!(ready, vec![second, third]);
1327        assert_eq!(reorder.status().buffered_packets, 0);
1328    }
1329}
openipc_core/rtp.rs

openipc_core/
rtp.rs
