enable TWCC sender/receiver as default

Author: rainliu

Cargo.toml

@@ -25,7 +25,7 @@ srtp = { package = "webrtc-srtp", version = "0.8.3" }
 sctp = { package = "webrtc-sctp", version = "0.4.1" }
 data = { package = "webrtc-data", version = "0.3.1" }
 media = { package = "webrtc-media", version = "0.4.1" }
-interceptor = "0.6.4"
+interceptor = "0.7.0"
 tokio = { version = "1.12.0", features = ["full"] }
 log = "0.4.14"
 async-trait = "0.1.42"

crates/ice

@@ -9,8 +9,7 @@ use crate::rtp_transceiver::{rtp_codec::RTPCodecType, RTCPFeedback, TYPE_RTCP_FB
 use interceptor::nack::{generator::Generator, responder::Responder};
 use interceptor::registry::Registry;
 use interceptor::report::{receiver::ReceiverReport, sender::SenderReport};
-use interceptor::twcc::header_extension::HeaderExtension;
-use interceptor::twcc::sender::Sender;
+use interceptor::twcc::{receiver::Receiver, sender::Sender};
 
 /// register_default_interceptors will register some useful interceptors.
 /// If you want to customize which interceptors are loaded, you should copy the
@@ -23,10 +22,7 @@ pub async fn register_default_interceptors(
 
     registry = configure_rtcp_reports(registry);
 
-    //TODO: temporarily disable twcc until all reference cycle memory leak fixed
-    // current when configure_twcc_sender, audio + video cause corrupted audio
-    // https://github.com/webrtc-rs/webrtc/issues/129
-    // registry = configure_twcc_sender(registry, media_engine).await?;
+    registry = configure_twcc(registry, media_engine).await?;
 
     Ok(registry)
 }
@@ -42,9 +38,6 @@ pub fn configure_rtcp_reports(mut registry: Registry) -> Registry {
 
 /// configure_nack will setup everything necessary for handling generating/responding to nack messages.
 pub fn configure_nack(mut registry: Registry, media_engine: &mut MediaEngine) -> Registry {
-    let generator = Box::new(Generator::builder());
-    let responder = Box::new(Responder::builder());
-
     media_engine.register_feedback(
         RTCPFeedback {
             typ: "nack".to_owned(),
@@ -60,17 +53,26 @@ pub fn configure_nack(mut registry: Registry, media_engine: &mut MediaEngine) ->
         RTPCodecType::Video,
     );
 
+    let generator = Box::new(Generator::builder());
+    let responder = Box::new(Responder::builder());
     registry.add(responder);
     registry.add(generator);
     registry
 }
 
-/// configure_twcc_header_extension_sender will setup everything necessary for adding
-/// a TWCC header extension to outgoing RTP packets. This will allow the remote peer to generate TWCC reports.
-pub async fn configure_twcc_header_extension_sender(
+/// configure_twcc will setup everything necessary for adding
+/// a TWCC header extension to outgoing RTP packets and generating TWCC reports.
+pub async fn configure_twcc(
     mut registry: Registry,
     media_engine: &mut MediaEngine,
 ) -> Result<Registry> {
+    media_engine.register_feedback(
+        RTCPFeedback {
+            typ: TYPE_RTCP_FB_TRANSPORT_CC.to_owned(),
+            ..Default::default()
+        },
+        RTPCodecType::Video,
+    );
     media_engine
         .register_header_extension(
             RTCRtpHeaderExtensionCapability {
@@ -81,6 +83,13 @@ pub async fn configure_twcc_header_extension_sender(
         )
         .await?;
 
+    media_engine.register_feedback(
+        RTCPFeedback {
+            typ: TYPE_RTCP_FB_TRANSPORT_CC.to_owned(),
+            ..Default::default()
+        },
+        RTPCodecType::Audio,
+    );
     media_engine
         .register_header_extension(
             RTCRtpHeaderExtensionCapability {
@@ -91,14 +100,46 @@ pub async fn configure_twcc_header_extension_sender(
         )
         .await?;
 
-    let header_extension = Box::new(HeaderExtension::builder());
+    let sender = Box::new(Sender::builder());
+    let receiver = Box::new(Receiver::builder());
+    registry.add(sender);
+    registry.add(receiver);
+    Ok(registry)
+}
+
+/// configure_twcc_sender will setup everything necessary for adding
+/// a TWCC header extension to outgoing RTP packets. This will allow the remote peer to generate TWCC reports.
+pub async fn configure_twcc_sender_only(
+    mut registry: Registry,
+    media_engine: &mut MediaEngine,
+) -> Result<Registry> {
+    media_engine
+        .register_header_extension(
+            RTCRtpHeaderExtensionCapability {
+                uri: sdp::extmap::TRANSPORT_CC_URI.to_owned(),
+            },
+            RTPCodecType::Video,
+            vec![],
+        )
+        .await?;
+
+    media_engine
+        .register_header_extension(
+            RTCRtpHeaderExtensionCapability {
+                uri: sdp::extmap::TRANSPORT_CC_URI.to_owned(),
+            },
+            RTPCodecType::Audio,
+            vec![],
+        )
+        .await?;
 
-    registry.add(header_extension);
+    let sender = Box::new(Sender::builder());
+    registry.add(sender);
     Ok(registry)
 }
 
-/// configure_twcc_sender will setup everything necessary for generating TWCC reports.
-pub async fn configure_twcc_sender(
+/// configure_twcc_receiver will setup everything necessary for generating TWCC reports.
+pub async fn configure_twcc_receiver_only(
     mut registry: Registry,
     media_engine: &mut MediaEngine,
 ) -> Result<Registry> {
@@ -136,7 +177,7 @@ pub async fn configure_twcc_sender(
         )
         .await?;
 
-    let sender = Box::new(Sender::builder());
-    registry.add(sender);
+    let receiver = Box::new(Receiver::builder());
+    registry.add(receiver);
     Ok(registry)
 }

crates/interceptor

@@ -136,11 +136,13 @@ impl RTPReceiverInternal {
 
         //log::debug!("read_rtp enter tracks tid {}", tid);
         let mut rtp_interceptor = None;
+        //let mut ssrc = 0;
         {
             let tracks = self.tracks.lock().await;
             for t in &*tracks {
                 if t.track.tid() == tid {
                     rtp_interceptor = t.stream.rtp_interceptor.clone();
+                    //ssrc = t.track.ssrc();
                     break;
                 }
             }
@@ -153,6 +155,10 @@ impl RTPReceiverInternal {
 
         if let Some(rtp_interceptor) = rtp_interceptor {
             let a = Attributes::new();
+            //println!(
+            //    "read_rtp rtp_interceptor.read enter with tid {} ssrc {}",
+            //    tid, ssrc
+            //);
             tokio::select! {
                 _ = self.closed_rx.notified() => {
                     Err(Error::ErrClosedPipe)
@@ -609,6 +615,7 @@ impl RTCRtpReceiver {
                     let mut b = vec![0u8; receive_mtu];
                     while let Some(repair_rtp_interceptor) = &track.repair_stream.rtp_interceptor {
                         //TODO: cancel repair_rtp_interceptor.read gracefully
+                        //println!("repair_rtp_interceptor read begin with ssrc={}", ssrc);
                         if repair_rtp_interceptor.read(&mut b, &a).await.is_err() {
                             break;
                         }