Configuration of outgoing RTX streams (#607)

* interceptor: Add AssociatedStreamInfo and populate for rtx streams

* rtp_sender: Create outgoing rtx streams

* sdp: Assign repair stream regardless of attribute order

Author: anders-avos

interceptor/src/stream_info.rs

@@ -20,6 +20,15 @@ pub struct StreamInfo {
     pub channels: u16,
     pub sdp_fmtp_line: String,
     pub rtcp_feedback: Vec<RTCPFeedback>,
+    pub associated_stream: Option<AssociatedStreamInfo>,
+}
+
+/// AssociatedStreamInfo provides a mapping from an auxiliary stream (RTX, FEC,
+/// etc.) back to the original stream.
+#[derive(Default, Debug, Clone)]
+pub struct AssociatedStreamInfo {
+    pub ssrc: u32,
+    pub payload_type: u8,
 }
 
 /// RTCPFeedback signals the connection to use additional RTCP packet types.

webrtc/src/api/mod.rs

@@ -159,11 +159,11 @@ impl API {
     ) -> RTCRtpSender {
         let kind = track.as_ref().map(|t| t.kind()).unwrap_or_default();
         RTCRtpSender::new(
-            self.setting_engine.get_receive_mtu(),
             track,
             kind,
             transport,
             Arc::clone(&self.media_engine),
+            Arc::clone(&self.setting_engine),
             interceptor,
             false,
         )

webrtc/src/api/setting_engine/mod.rs

@@ -78,6 +78,7 @@ pub struct SettingEngine {
     pub(crate) srtp_protection_profiles: Vec<SrtpProtectionProfile>,
     pub(crate) receive_mtu: usize,
     pub(crate) mid_generator: Option<Arc<dyn Fn(isize) -> String + Send + Sync>>,
+    pub(crate) enable_sender_rtx: bool,
 }
 
 impl SettingEngine {
@@ -334,4 +335,11 @@ impl SettingEngine {
     pub fn set_mid_generator(&mut self, f: impl Fn(isize) -> String + Send + Sync + 'static) {
         self.mid_generator = Some(Arc::new(f));
     }
+
+    /// enable_sender_rtx allows outgoing rtx streams to be created where applicable.
+    /// RTPSender will create an RTP retransmission stream for each source stream where a retransmission
+    /// codec is configured.
+    pub fn enable_sender_rtx(&mut self, is_enabled: bool) {
+        self.enable_sender_rtx = is_enabled;
+    }
 }

webrtc/src/peer_connection/mod.rs

@@ -1415,11 +1415,11 @@ impl RTCPeerConnection {
 
                             let sender = Arc::new(
                                 RTCRtpSender::new(
-                                    receive_mtu,
                                     None,
                                     kind,
                                     Arc::clone(&self.internal.dtls_transport),
                                     Arc::clone(&self.internal.media_engine),
+                                    Arc::clone(&self.internal.setting_engine),
                                     Arc::clone(&self.interceptor),
                                     false,
                                 )

webrtc/src/peer_connection/peer_connection_internal.rs

@@ -528,11 +528,11 @@ impl PeerConnectionInternal {
 
                 let sender = Arc::new(
                     RTCRtpSender::new(
-                        self.setting_engine.get_receive_mtu(),
                         None,
                         kind,
                         Arc::clone(&self.dtls_transport),
                         Arc::clone(&self.media_engine),
+                        Arc::clone(&self.setting_engine),
                         interceptor,
                         false,
                     )
@@ -582,11 +582,11 @@ impl PeerConnectionInternal {
 
         let s = Arc::new(
             RTCRtpSender::new(
-                self.setting_engine.get_receive_mtu(),
                 Some(Arc::clone(&track)),
                 track.kind(),
                 Arc::clone(&self.dtls_transport),
                 Arc::clone(&self.media_engine),
+                Arc::clone(&self.setting_engine),
                 Arc::clone(&interceptor),
                 false,
             )
@@ -1080,6 +1080,7 @@ impl PeerConnectionInternal {
             params.codecs[0].payload_type,
             params.codecs[0].capability.clone(),
             &params.header_extensions,
+            None,
         );
         let (rtp_read_stream, rtp_interceptor, rtcp_read_stream, rtcp_interceptor) = self
             .dtls_transport
@@ -1386,20 +1387,30 @@ impl PeerConnectionInternal {
             let sender = transceiver.sender().await;
             let track_encodings = sender.track_encodings.lock().await;
             for encoding in track_encodings.iter() {
-                let track_id = encoding.track.id().to_string();
+                let track_id = encoding.track.id();
                 let kind = match encoding.track.kind() {
                     RTPCodecType::Unspecified => continue,
                     RTPCodecType::Audio => "audio",
                     RTPCodecType::Video => "video",
                 };
 
                 track_infos.push(TrackInfo {
-                    track_id,
+                    track_id: track_id.to_owned(),
                     ssrc: encoding.ssrc,
                     mid: mid.to_owned(),
                     rid: encoding.track.rid().map(Into::into),
                     kind,
                 });
+
+                if let Some(rtx) = &encoding.rtx {
+                    track_infos.push(TrackInfo {
+                        track_id: track_id.to_owned(),
+                        ssrc: rtx.ssrc,
+                        mid: mid.to_owned(),
+                        rid: encoding.track.rid().map(Into::into),
+                        kind,
+                    });
+                }
             }
         }
 

webrtc/src/peer_connection/sdp/mod.rs

@@ -178,29 +178,19 @@ pub(crate) fn track_details_from_sdp(
                             }
                         }
 
-                        let mut repair_ssrc = 0;
-                        for (repair, base) in &rtx_repair_flows {
-                            if *base == ssrc {
-                                repair_ssrc = *repair;
-                                //TODO: no break?
-                            }
-                        }
-
                         if track_idx < tracks_in_media_section.len() {
                             tracks_in_media_section[track_idx].mid = SmolStr::from(mid_value);
                             tracks_in_media_section[track_idx].kind = codec_type;
                             stream_id.clone_into(&mut tracks_in_media_section[track_idx].stream_id);
                             track_id.clone_into(&mut tracks_in_media_section[track_idx].id);
                             tracks_in_media_section[track_idx].ssrcs = vec![ssrc];
-                            tracks_in_media_section[track_idx].repair_ssrc = repair_ssrc;
                         } else {
                             let track_details = TrackDetails {
                                 mid: SmolStr::from(mid_value),
                                 kind: codec_type,
                                 stream_id: stream_id.to_owned(),
                                 id: track_id.to_owned(),
                                 ssrcs: vec![ssrc],
-                                repair_ssrc,
                                 ..Default::default()
                             };
                             tracks_in_media_section.push(track_details);
@@ -210,6 +200,13 @@ pub(crate) fn track_details_from_sdp(
                 _ => {}
             };
         }
+        for (repair, base) in &rtx_repair_flows {
+            for track in &mut tracks_in_media_section {
+                if track.ssrcs.contains(base) {
+                    track.repair_ssrc = *repair;
+                }
+            }
+        }
 
         // If media line is using RTP Stream Identifier Source Description per RFC8851
         // we will need to override tracks, and remove ssrcs.
@@ -595,6 +592,23 @@ pub(crate) async fn add_transceiver_sdp(
                     track.stream_id().to_owned(), /* streamLabel */
                     track.id().to_owned(),
                 );
+
+                if encoding.rtx.ssrc != 0 {
+                    media = media.with_media_source(
+                        encoding.rtx.ssrc,
+                        track.stream_id().to_owned(),
+                        track.stream_id().to_owned(),
+                        track.id().to_owned(),
+                    );
+
+                    media = media.with_value_attribute(
+                        ATTR_KEY_SSRCGROUP.to_owned(),
+                        format!(
+                            "{} {} {}",
+                            SEMANTIC_TOKEN_FLOW_IDENTIFICATION, encoding.ssrc, encoding.rtx.ssrc
+                        ),
+                    );
+                }
             }
 
             if send_parameters.encodings.len() > 1 {