feat: add PLI/FIR RTCP feedback packets

Author: kbalt

media/rtc/src/rtp_session/inbound/mod.rs

@@ -13,6 +13,9 @@ mod stats;
 
 pub use stats::{RtpInboundRemoteStats, RtpInboundStats};
 
+/// Minimum interval in which FIR/PLI requests can be sent
+const RTCP_FEEDBACK_COOLDOWN: Duration = Duration::from_millis(250);
+
 /// RTP receive stream
 pub struct RtpInboundStream {
     ssrc: Ssrc,
@@ -22,6 +25,15 @@ pub struct RtpInboundStream {
     last_received_sender_report: Option<NtpTimestamp>,
 
     remote_stats: Option<RtpInboundRemoteStats>,
+
+    // RTCP feedback NACK PLI
+    want_nack_pli: bool,
+    last_nack_pli: Option<Instant>,
+
+    // RTCP feedback CCM FIR
+    want_ccm_fir: bool,
+    next_fir_seq: u8,
+    last_ccm_fir: Option<Instant>,
 }
 
 impl RtpInboundStream {
@@ -33,32 +45,87 @@ impl RtpInboundStream {
             last_report_sent: None,
             last_received_sender_report: None,
             remote_stats: None,
+
+            want_nack_pli: false,
+            last_nack_pli: None,
+
+            want_ccm_fir: false,
+            next_fir_seq: rand::random(),
+            last_ccm_fir: None,
         }
     }
 
+    pub fn request_nack_pli(&mut self) {
+        self.want_nack_pli = true
+    }
+
+    pub fn request_ccm_fir(&mut self) {
+        self.want_ccm_fir = true
+    }
+
     pub(crate) fn timeout(&self, now: Instant) -> Option<Duration> {
         let queue = self.queue.timeout(now);
 
         let report = if self.queue.highest_sequence_number_received().is_some() {
-            Some(
-                self.last_report_sent
-                    .and_then(|(last_report_sent, _)| {
-                        (last_report_sent + self.report_interval).checked_duration_since(now)
-                    })
-                    .unwrap_or_default(),
-            )
+            let report_interval = self
+                .last_report_sent
+                .and_then(|(last_report_sent, _)| {
+                    (last_report_sent + self.report_interval).checked_duration_since(now)
+                })
+                .unwrap_or_default();
+
+            let nack_pli = self
+                .last_nack_pli
+                .map(|ts| (ts + RTCP_FEEDBACK_COOLDOWN).saturating_duration_since(now));
+
+            let ccm_fir = self
+                .last_ccm_fir
+                .map(|ts| (ts + RTCP_FEEDBACK_COOLDOWN).saturating_duration_since(now));
+
+            opt_min(Some(report_interval), opt_min(nack_pli, ccm_fir))
         } else {
             None
         };
 
         opt_min(queue, report)
     }
 
-    pub(crate) fn collect_reports(&mut self, now: Instant, reports: &mut ReportsQueue) {
-        let make_report = self
+    pub(super) fn collect_reports(&mut self, now: Instant, reports: &mut ReportsQueue) {
+        if self.want_nack_pli {
+            let cooldown_elapsed = self
+                .last_nack_pli
+                .is_none_or(|i| i + RTCP_FEEDBACK_COOLDOWN <= now);
+
+            if cooldown_elapsed {
+                self.want_nack_pli = false;
+                self.last_nack_pli = Some(now);
+                reports.add_nack_pli(self.ssrc);
+            }
+        }
+
+        if self.want_ccm_fir {
+            let cooldown_elapsed = self
+                .last_ccm_fir
+                .is_none_or(|i| i + RTCP_FEEDBACK_COOLDOWN <= now);
+
+            if cooldown_elapsed {
+                self.want_ccm_fir = false;
+                self.last_ccm_fir = Some(now);
+                reports.add_ccm_fir(self.ssrc, self.next_fir_seq);
+                self.next_fir_seq = self.next_fir_seq.wrapping_add(1);
+            }
+        }
+
+        // When emitting feedback packets & reduced size RTCP is not supported:
+        // Emit a receiver report so the ReportsQueue can generate a valid RTCP packet
+        let receiver_report_for_feedback = reports.has_feedback() && !reports.rtcp_rsize();
+
+        let report_interval_elapsed = self
             .last_report_sent
             .is_none_or(|(instant, _)| now > instant + self.report_interval);
 
+        let make_report = receiver_report_for_feedback || report_interval_elapsed;
+
         if !make_report {
             return;
         }

media/rtc/src/rtp_session/mod.rs

@@ -9,7 +9,7 @@ use crate::Mtu;
 use report::ReportsQueue;
 use rtp::{
     RtpPacket, Ssrc,
-    rtcp_types::{Compound, Packet as RtcpPacket},
+    rtcp_types::{Compound, Fir, Packet as RtcpPacket, Pli, RtcpPacketParser},
 };
 use ssrc_hasher::SsrcHasher;
 use std::{
@@ -44,15 +44,20 @@ pub struct RtpSession {
 
 impl RtpSession {
     /// Create a new RTP session
-    pub fn new() -> Self {
+    pub fn new(rtcp_rsize: bool) -> Self {
         Self {
             next_tx_ssrc: Ssrc(rand::random()),
             tx: HashMap::default(),
             rx: HashMap::default(),
-            reports: ReportsQueue::default(),
+            reports: ReportsQueue::new(rtcp_rsize),
         }
     }
 
+    /// Is reduced size RTCP allowed
+    pub fn rtcp_rsize(&self) -> bool {
+        self.reports.rtcp_rsize()
+    }
+
     /// Create a new outbound RTP stream with the given parameters
     pub fn new_tx_stream(&mut self, clock_rate: u32) -> &mut RtpOutboundStream {
         let ssrc = self.next_tx_ssrc;
@@ -100,13 +105,20 @@ impl RtpSession {
     }
 
     /// Hand of the RTCP packet to the RTP session
-    pub fn receive_rtcp(&mut self, now: Instant, rtcp_packet: Compound<'_>) {
+    #[must_use]
+    pub fn receive_rtcp(
+        &mut self,
+        now: Instant,
+        rtcp_packet: Compound<'_>,
+    ) -> Vec<RtpSessionReceiveRtcpEvent> {
+        let mut events = vec![];
+
         for rtcp_packet in rtcp_packet {
             let rtcp_packet = match rtcp_packet {
                 Ok(rtcp_packet) => rtcp_packet,
                 Err(e) => {
                     log::warn!("Failed to parse RTCP packet in compound packet, {e}");
-                    return;
+                    return events;
                 }
             };
 
@@ -139,14 +151,36 @@ impl RtpSession {
                 RtcpPacket::TransportFeedback(_transport_feedback) => {
                     // TODO: handle feedback
                 }
-                RtcpPacket::PayloadFeedback(_payload_feedback) => {
-                    // TODO: handle feedback
+                RtcpPacket::PayloadFeedback(payload_feedback) => {
+                    if payload_feedback.parse_fci::<Pli>().is_ok() {
+                        events.push(RtpSessionReceiveRtcpEvent::NackPliReceived(Ssrc(
+                            payload_feedback.media_ssrc(),
+                        )));
+                    } else if let Ok(fir) = payload_feedback.parse_fci::<Fir>() {
+                        for entry in fir.entries() {
+                            events.push(RtpSessionReceiveRtcpEvent::CcmFirReceived(Ssrc(
+                                entry.ssrc(),
+                            )));
+                        }
+                    } else {
+                        log::warn!(
+                            "Received unknown RTCP payload feedback packet header={:02X?} sender_ssrc={} media_ssrc={}",
+                            payload_feedback.header_data(),
+                            payload_feedback.sender_ssrc(),
+                            payload_feedback.media_ssrc(),
+                        )
+                    }
+                }
+                RtcpPacket::Xr(_xr) => {
+                    // ignore
                 }
                 RtcpPacket::Unknown(..) => {
                     // ignore
                 }
             }
         }
+
+        events
     }
 
     /// Returns the duration to wait from the given Instant before polling again
@@ -165,7 +199,7 @@ impl RtpSession {
     }
 
     /// Poll the session for any new events
-    pub fn poll(&mut self, now: Instant, mtu: Mtu) -> Option<RtpSessionEvent> {
+    pub fn poll(&mut self, now: Instant, mtu: Mtu) -> Option<RtpSessionPollEvent> {
         let fallback_sender_ssrc = *self.tx.keys().next().unwrap_or(&self.next_tx_ssrc);
 
         for tx in self.tx.values_mut() {
@@ -177,34 +211,34 @@ impl RtpSession {
         }
 
         if let Some(report) = self.reports.make_report(fallback_sender_ssrc, mtu) {
-            return Some(RtpSessionEvent::SendRtcp(report));
+            return Some(RtpSessionPollEvent::SendRtcp(report));
         }
 
         for tx in self.tx.values_mut() {
             if let Some(rtp_packet) = tx.pop(now) {
-                return Some(RtpSessionEvent::SendRtp(rtp_packet));
+                return Some(RtpSessionPollEvent::SendRtp(rtp_packet));
             }
         }
 
         for rx in self.rx.values_mut() {
             if let Some(rtp_packet) = rx.pop(now) {
-                return Some(RtpSessionEvent::ReceiveRtp(rtp_packet));
+                return Some(RtpSessionPollEvent::ReceiveRtp(rtp_packet));
             }
         }
 
         None
     }
 }
 
-impl Default for RtpSession {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 /// Event returned by [`RtpSession::poll`]
-pub enum RtpSessionEvent {
+pub enum RtpSessionPollEvent {
     ReceiveRtp(RtpPacket),
     SendRtp(RtpPacket),
     SendRtcp(Vec<u8>),
 }
+
+/// Event returned by [`RtpSession::receive_rtcp`]
+pub enum RtpSessionReceiveRtcpEvent {
+    NackPliReceived(Ssrc),
+    CcmFirReceived(Ssrc),
+}

media/rtc/src/rtp_session/outbound/mod.rs

@@ -61,7 +61,7 @@ impl RtpOutboundStream {
         opt_min(queue, report)
     }
 
-    pub(crate) fn collect_reports(&mut self, now: Instant, reports: &mut ReportsQueue) {
+    pub(super) fn collect_reports(&mut self, now: Instant, reports: &mut ReportsQueue) {
         let make_report = self
             .last_report_sent
             .is_none_or(|instant| now > instant + self.report_interval);