From d5e91694d5998354eb200d9a454fc123966f8c13 Mon Sep 17 00:00:00 2001
From: "jinxue.cgh" <jinxue.cgh@alibaba-inc.com>
Date: Tue, 19 May 2020 18:14:48 +0800
Subject: [PATCH] rtc: publisher support twcc

---
 trunk/src/app/srs_app_rtc_conn.cpp | 64 ++++++++++++++++++++++++++++--
 trunk/src/app/srs_app_rtc_conn.hpp | 11 ++++-
 2 files changed, 71 insertions(+), 4 deletions(-)

diff --git a/trunk/src/app/srs_app_rtc_conn.cpp b/trunk/src/app/srs_app_rtc_conn.cpp
index dd92fa1d6..8f6d57130 100644
--- a/trunk/src/app/srs_app_rtc_conn.cpp
+++ b/trunk/src/app/srs_app_rtc_conn.cpp
@@ -1397,6 +1397,9 @@ SrsRtcPublisher::SrsRtcPublisher(SrsRtcSession* session)
     nack_enabled_ = false;
 
     nn_audio_frames = 0;
+    twcc_ext_id_ = 0;
+    last_twcc_feedback_time_ = 0;
+    twcc_fb_count_ = 0;
 }
 
 SrsRtcPublisher::~SrsRtcPublisher()
@@ -1415,14 +1418,19 @@ SrsRtcPublisher::~SrsRtcPublisher()
     srs_freep(audio_queue_);
 }
 
-srs_error_t SrsRtcPublisher::initialize(uint32_t vssrc, uint32_t assrc, SrsRequest* r)
+srs_error_t SrsRtcPublisher::initialize(uint32_t vssrc, uint32_t assrc, uint8_t twcc_ext_id, SrsRequest* r)
 {
     srs_error_t err = srs_success;
 
     video_ssrc = vssrc;
     audio_ssrc = assrc;
+    twcc_ext_id_ = twcc_ext_id;
+    rtcp_twcc_.set_media_ssrc(video_ssrc);
     req = r;
 
+    if (twcc_ext_id_ != 0) {
+        extension_map_.register_by_uri(twcc_ext_id_, kTWCCExt);
+    }
     // TODO: FIXME: Support reload.
     nack_enabled_ = _srs_config->get_rtc_nack_enabled(session_->req->vhost);
 
@@ -1656,6 +1664,34 @@ srs_error_t SrsRtcPublisher::send_rtcp_fb_pli(uint32_t ssrc)
     return err;
 }
 
+srs_error_t SrsRtcPublisher::on_twcc(uint16_t sn) {
+    srs_error_t err = srs_success;
+    srs_utime_t now = srs_get_system_time();
+    srs_trace("get twcc sn:%d, now:%d", sn, now);
+    rtcp_twcc_.recv_packet(sn, now);
+    if(0 == last_twcc_feedback_time_) {
+        last_twcc_feedback_time_ = now;
+        return err;
+    }
+    srs_utime_t diff = now - last_twcc_feedback_time_;
+    if( diff >= 50 * SRS_UTIME_MILLISECONDS) {
+        last_twcc_feedback_time_ = now;
+        char pkt[kRtcpPacketSize];
+        SrsBuffer *buffer = new SrsBuffer(pkt, sizeof(pkt));
+        SrsAutoFree(SrsBuffer, buffer);
+        rtcp_twcc_.set_feedback_count(twcc_fb_count_);
+        twcc_fb_count_++;
+        if((err = rtcp_twcc_.encode(buffer)) != srs_success) {
+            return srs_error_wrap(err, "fail to generate twcc feedback packet");
+        }
+        int nb_protected_buf = buffer->pos();
+        char protected_buf[kRtpPacketSize];
+        if (session_->dtls_->protect_rtcp(protected_buf, pkt, nb_protected_buf) == srs_success) {
+            session_->sendonly_skt->sendto(protected_buf, nb_protected_buf, 0);
+        }
+    }
+    return err;
+}
 srs_error_t SrsRtcPublisher::on_rtp(char* data, int nb_data)
 {
     srs_error_t err = srs_success;
@@ -1699,9 +1735,17 @@ srs_error_t SrsRtcPublisher::on_rtp(char* data, int nb_data)
         pkt->shared_msg->wrap(buf, nb_buf);
 
         SrsBuffer b(buf, nb_buf);
-        if ((err = pkt->decode(&b)) != srs_success) {
+        if ((err = pkt->decode(&b, &extension_map_)) != srs_success) {
             return srs_error_wrap(err, "decode rtp packet");
         }
+        if (0 != twcc_ext_id_) {
+            uint16_t twcc_sn = 0;
+            if ((err = pkt->header.get_twcc_sequence_number(twcc_sn)) == srs_success) {
+                if((err = on_twcc(twcc_sn))) {
+                    return srs_error_wrap(err, "fail to process twcc packet");
+                }
+            }
+        }
     }
 
     // For source to consume packet.
@@ -2502,8 +2546,22 @@ srs_error_t SrsRtcSession::start_publish()
         }
     }
 
+    uint32_t twcc_ext_id = 0;
+    for (size_t i = 0; i < local_sdp.media_descs_.size(); ++i) {
+        const SrsMediaDesc& media_desc = remote_sdp.media_descs_[i];
+        map<int, string> extmaps = media_desc.get_extmaps();
+        for(map<int, string>::iterator it_ext = extmaps.begin(); it_ext != extmaps.end(); ++it_ext) {
+            if(kTWCCExt == it_ext->second) {
+                twcc_ext_id = it_ext->first;
+                break;
+            }
+        }
+        if (twcc_ext_id != 0){
+            break;
+        }
+    }
     // FIXME: err process.
-    if ((err = publisher_->initialize(video_ssrc, audio_ssrc, req)) != srs_success) {
+    if ((err = publisher_->initialize(video_ssrc, audio_ssrc, twcc_ext_id, req)) != srs_success) {
         return srs_error_wrap(err, "rtc publisher init");
     }
 
diff --git a/trunk/src/app/srs_app_rtc_conn.hpp b/trunk/src/app/srs_app_rtc_conn.hpp
index fa9985bff..f92820df3 100644
--- a/trunk/src/app/srs_app_rtc_conn.hpp
+++ b/trunk/src/app/srs_app_rtc_conn.hpp
@@ -34,6 +34,7 @@
 #include <srs_app_rtc_sdp.hpp>
 #include <srs_app_reload.hpp>
 #include <srs_kernel_rtc_rtp.hpp>
+#include <srs_kernel_rtc_rtcp.hpp>
 #include <srs_app_rtc_queue.hpp>
 
 #include <string>
@@ -285,11 +286,17 @@ private:
 private:
     std::map<uint32_t, uint64_t> last_sender_report_sys_time;
     std::map<uint32_t, SrsNtp> last_sender_report_ntp;
+private:
+    srs_utime_t last_twcc_feedback_time_;
+    uint8_t twcc_ext_id_;
+    uint8_t twcc_fb_count_;
+    SrsRtcpTWCC rtcp_twcc_;
+    SrsRtpHeaderExtensionMap extension_map_;
 public:
     SrsRtcPublisher(SrsRtcSession* session);
     virtual ~SrsRtcPublisher();
 public:
-    srs_error_t initialize(uint32_t vssrc, uint32_t assrc, SrsRequest* req);
+    srs_error_t initialize(uint32_t vssrc, uint32_t assrc, uint8_t twcc_ext_id, SrsRequest* req);
 private:
     void check_send_nacks(SrsRtpNackForReceiver* nack, uint32_t ssrc);
     srs_error_t send_rtcp_rr(uint32_t ssrc, SrsRtpRingBuffer* rtp_queue);
@@ -319,6 +326,8 @@ public:
     void simulate_nack_drop(int nn);
 private:
     void simulate_drop_packet(SrsRtpHeader* h, int nn_bytes);
+private:
+    srs_error_t on_twcc(uint16_t sn);
 };
 
 class SrsRtcSession