add the feature of bitrate limitation for rtc publishing through google remb mechanism

specify the bitrate through the following way:
1. according to the configure file
2. http query parameter
3. http json property

(cherry picked from commit c4b4aa6)

Author: hyt-hz

trunk/conf/full.conf

@@ -595,6 +595,9 @@ vhost rtc.vhost.srs.com {
         # [8000, 320000]
         # default: 48000
         aac_bitrate 48000;
+		# The bitrate limitation for publish in bps, 0 - no limiting
+		# default: 0
+		bitrate 0;
     }
     ###############################################################
     # For transmuxing RTMP to RTC, it will impact the default values if RTC is on.

trunk/src/app/srs_app_config.cpp

@@ -2745,7 +2745,7 @@ srs_error_t SrsConfig::check_normal_config()
                         && m != "bframe" && m != "aac" && m != "stun_timeout" && m != "stun_strict_check"
                         && m != "dtls_role" && m != "dtls_version" && m != "drop_for_pt" && m != "rtc_to_rtmp"
                         && m != "pli_for_rtmp" && m != "rtmp_to_rtc" && m != "keep_bframe" && m != "opus_bitrate"
-                        && m != "aac_bitrate" && m != "keep_avc_nalu_sei") {
+                        && m != "aac_bitrate" && m != "keep_avc_nalu_sei" && m != "bitrate") {
                         return srs_error_new(ERROR_SYSTEM_CONFIG_INVALID, "illegal vhost.rtc.%s of %s", m.c_str(), vhost->arg0().c_str());
                     }
                 }
@@ -4651,6 +4651,34 @@ bool SrsConfig::get_rtc_to_rtmp(string vhost)
     return SRS_CONF_PREFER_FALSE(conf->arg0());
 }
 
+uint64_t SrsConfig::get_rtc_bitrate(string vhost)
+{
+    static uint64_t DEFAULT = 0;
+
+    SrsConfDirective* conf = get_rtc(vhost);
+    if (!conf) {
+        return DEFAULT;
+    }
+
+    conf = conf->get("bitrate");
+    if (!conf || conf->arg0().empty()) {
+        return DEFAULT;
+    }
+
+    uint64_t v = (uint64_t)(::atoll(conf->arg0().c_str()));
+    if ( v > 100 * 1024 * 1024) {
+        srs_warn("Invalid bitrate(%ll), No bitrate control", (long long) v);
+        return DEFAULT;
+    }
+    
+    if (v < 64000){
+        
+        srs_warn("Invalid bitrate(%ll), reset it to 64000 bps", (long long) v);
+		v = 64000;
+    }
+
+    return v;
+}
 srs_utime_t SrsConfig::get_rtc_pli_for_rtmp(string vhost)
 {
     static srs_utime_t DEFAULT = 6 * SRS_UTIME_SECONDS;

trunk/src/app/srs_app_config.hpp

@@ -544,6 +544,7 @@ class SrsConfig
     std::string get_rtc_dtls_version(std::string vhost);
     int get_rtc_drop_for_pt(std::string vhost);
     bool get_rtc_to_rtmp(std::string vhost);
+    uint64_t get_rtc_bitrate(std::string vhost);
     srs_utime_t get_rtc_pli_for_rtmp(std::string vhost);
     bool get_rtc_nack_enabled(std::string vhost);
     bool get_rtc_nack_no_copy(std::string vhost);

trunk/src/app/srs_app_rtc_api.cpp

@@ -432,6 +432,12 @@ srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHtt
         tid = prop->to_str();
     }
 
+    uint64_t bitrate = 0;
+    if ((prop = req->ensure_property_integer("bitrate")) != NULL) {
+        bitrate = prop->to_integer();
+    }
+
+
     // The RTC user config object.
     SrsRtcUserConfig ruc;
     ruc.req_->ip = clientip;
@@ -457,16 +463,31 @@ srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHtt
     }
     string codec = r->query_get("codec");
 
-    srs_trace("RTC publish %s, api=%s, tid=%s, clientip=%s, app=%s, stream=%s, offer=%dB, eip=%s, codec=%s",
+    string bitrate_query = r->query_get("bitrate");
+    if(bitrate == 0 && ! bitrate_query.empty()){
+        bitrate = ::atoll(bitrate_query.c_str());
+    }
+	if (bitrate > 100 * 1024 * 1024) {
+        srs_warn("Request Invalid bitrate(%ll), No bitrate control", (long long)bitrate);
+        bitrate = 0;
+    }    
+    if (bitrate < 64000 && bitrate > 0){
+        srs_warn("Request Invalid bitrate(%ll), reset it to 64000 bps", (long long)bitrate);
+		bitrate = 64000;
+    }
+
+    srs_trace("RTC publish %s, api=%s, tid=%s, clientip=%s, app=%s, stream=%s, offer=%dB, eip=%s, codec=%s, bitrate=%llu",
         streamurl.c_str(), api.c_str(), tid.c_str(), clientip.c_str(), ruc.req_->app.c_str(), ruc.req_->stream.c_str(),
-        remote_sdp_str.length(), eip.c_str(), codec.c_str()
+        remote_sdp_str.length(), eip.c_str(), codec.c_str(), bitrate
     );
 
     ruc.eip_ = eip;
     ruc.codec_ = codec;
     ruc.publish_ = true;
     ruc.dtls_ = ruc.srtp_ = true;
 
+    ruc.req_->bitrate = bitrate;
+
     // TODO: FIXME: It seems remote_sdp doesn't represents the full SDP information.
     ruc.remote_sdp_str_ = remote_sdp_str;
     if ((err = ruc.remote_sdp_.parse(remote_sdp_str)) != srs_success) {

trunk/src/app/srs_app_rtc_conn.cpp

@@ -986,6 +986,12 @@ srs_error_t SrsRtcPublishRtcpTimer::on_timer(srs_utime_t interval)
         srs_warn("XR err %s", srs_error_desc(err).c_str());
         srs_freep(err);
     }
+    
+    // For REMB
+    if((err = p_->send_rtcp_remb()) != srs_success){
+        srs_warn("remb err %s", srs_error_desc(err).c_str());
+        srs_freep(err);        
+    }
 
     return err;
 }
@@ -1117,6 +1123,9 @@ SrsRtcPublishStream::SrsRtcPublishStream(SrsRtcConnection* session, const SrsCon
 
     pli_worker_ = new SrsRtcPLIWorker(this);
     last_time_send_twcc_ = 0;
+    
+    bitrate_ = 0;
+    remb_startup_ = 4;
 
     timer_rtcp_ = new SrsRtcPublishRtcpTimer(this);
     timer_twcc_ = new SrsRtcPublishTwccTimer(this);
@@ -1199,6 +1208,12 @@ srs_error_t SrsRtcPublishStream::initialize(SrsRequest* r, SrsRtcSourceDescripti
     pt_to_drop_ = (uint16_t)_srs_config->get_rtc_drop_for_pt(req_->vhost);
     twcc_enabled_ = _srs_config->get_rtc_twcc_enabled(req_->vhost);
 
+    bitrate_ = _srs_config->get_rtc_bitrate(req_->vhost);
+    if(r->bitrate != 0){
+        bitrate_ = r->bitrate;		
+    }	
+    remb_startup_ = 4;
+    
     // No TWCC when negotiate, disable it.
     if (twcc_id <= 0) {
         twcc_enabled_ = false;
@@ -1355,6 +1370,43 @@ srs_error_t SrsRtcPublishStream::send_rtcp_rr()
     return err;
 }
 
+
+srs_error_t SrsRtcPublishStream::send_rtcp_remb()
+{
+    srs_error_t err = srs_success;
+    uint32_t bitrate = bitrate_;
+    
+    // no bitrate control
+    if(bitrate_ == 0){
+        return err;
+    }
+    
+    // no rtc session
+    if(session_ == NULL){
+        return err;
+    }
+    
+    // no video tracks
+    if(video_tracks_.size() == 0){
+        return err;
+    }
+    
+    if(remb_startup_ > 0){
+        bitrate = bitrate / remb_startup_;
+        remb_startup_ --;
+    }
+	
+    //only send for video 0 track
+    SrsRtcVideoRecvTrack* track = video_tracks_.at(0);
+    uint32_t ssrc = track->get_ssrc();
+    if ((err = session_->send_rtcp_remb(ssrc, bitrate) ) != srs_success) {
+        return srs_error_wrap(err, "rtcp remb send error(ssrc=%u, bitrate=%llu)" , ssrc, (unsigned long long)bitrate);
+    }
+
+
+    return err;
+}
+
 srs_error_t SrsRtcPublishStream::send_rtcp_xr_rrtr()
 {
     srs_error_t err = srs_success;
@@ -1481,6 +1533,16 @@ srs_error_t SrsRtcPublishStream::do_on_rtp_plaintext(SrsRtpPacket*& pkt, SrsBuff
         return srs_error_new(ERROR_RTC_RTP, "unknown ssrc=%u", ssrc);
     }
 
+    // For remb startup, according to janus
+    if(video_track != NULL && bitrate_ != 0 && remb_startup_ > 0){
+        
+        if ((err = send_rtcp_remb()) != srs_success) {
+            srs_warn("send rtcp remb failed, %s", srs_error_desc(err).c_str());
+            srs_freep(err);
+        }
+
+    }
+
     // If circuit-breaker is enabled, disable nack.
     if (_srs_circuit_breaker->hybrid_critical_water_level()) {
         ++_srs_pps_snack4->sugar;
@@ -2370,6 +2432,56 @@ void SrsRtcConnection::check_send_nacks(SrsRtpNackForReceiver* nack, uint32_t ss
     send_rtcp(stream.data(), stream.pos());
 }
 
+srs_error_t SrsRtcConnection::send_rtcp_remb(uint32_t ssrc, const uint64_t& bitrate)
+{
+    srs_error_t err = srs_success;
+    int min_len = 20 + 4;
+    
+    // @see https://datatracker.ietf.org/doc/html/draft-alvestrand-rmcat-remb-03#section-2.2
+    char buf[kRtpPacketSize];
+    SrsBuffer stream(buf, sizeof(buf));
+    //header
+    stream.write_1bytes(0x8F);
+    stream.write_1bytes(kPsFb);
+    stream.write_2bytes((min_len/4) - 1);
+    //SSRC of packet sender      
+    stream.write_4bytes(ssrc); // TODO: FIXME: Should be 1?
+    //SSRC of media source, always 0
+    stream.write_4bytes(0);
+    
+    //Unique identifier (32 bits):  Always 'R' 'E' 'M' 'B' (4 ASCII characters).
+    stream.write_1bytes('R');
+    stream.write_1bytes('E');
+    stream.write_1bytes('M');
+    stream.write_1bytes('B');    
+    
+    //Num SSRC (8 bits)
+    stream.write_1bytes(1);
+    
+	/* bitrate --> brexp/brmantissa */
+	uint8_t b = 0;
+	uint8_t newbrexp = 0;
+	uint32_t newbrmantissa = 0;
+	for(b=0; b<32; b++) {
+		if(bitrate <= ((uint32_t) 0x3FFFF << b)) {
+			newbrexp = b;
+			break;
+		}
+	}
+	if(b > 31)
+		b = 31;
+	newbrmantissa = bitrate >> b;
+    // BR Exp (6 bits) + BR Mantissa (18 bits)
+    stream.write_1bytes((uint8_t)((newbrexp << 2) + ((newbrmantissa >> 16) & 0x03)));
+    stream.write_1bytes((uint8_t)(newbrmantissa >> 8));
+    stream.write_1bytes((uint8_t)(newbrmantissa));
+    
+    //SSRC feedback
+    stream.write_4bytes(ssrc);
+
+    return send_rtcp(stream.data(), stream.pos());
+}
+
 srs_error_t SrsRtcConnection::send_rtcp_rr(uint32_t ssrc, SrsRtpRingBuffer* rtp_queue, const uint64_t& last_send_systime, const SrsNtp& last_send_ntp)
 {
     ++_srs_pps_srtcps->sugar;
@@ -2651,6 +2763,10 @@ srs_error_t SrsRtcConnection::negotiate_publish_capability(SrsRtcUserConfig* ruc
 
     bool nack_enabled = _srs_config->get_rtc_nack_enabled(req->vhost);
     bool twcc_enabled = _srs_config->get_rtc_twcc_enabled(req->vhost);
+	
+    bool remb_enabled = _srs_config->get_rtc_bitrate(req->vhost) > 0 || 
+                        req->bitrate > 0;
+
     // TODO: FIME: Should check packetization-mode=1 also.
     bool has_42e01f = srs_sdp_has_h264_profile(remote_sdp, "42e01f");
 
@@ -2751,6 +2867,11 @@ srs_error_t SrsRtcConnection::negotiate_publish_capability(SrsRtcUserConfig* ruc
                             video_payload->rtcp_fbs_.push_back(rtcp_fb);
                         }
                     }
+                    if(remb_enabled) {
+                        if (rtcp_fb == "goog-remb") {
+                            video_payload->rtcp_fbs_.push_back(rtcp_fb);
+                        }                        
+                    }
                 }
 
                 track_desc->type_ = "video";
@@ -2785,6 +2906,11 @@ srs_error_t SrsRtcConnection::negotiate_publish_capability(SrsRtcUserConfig* ruc
                             video_payload->rtcp_fbs_.push_back(rtcp_fb);
                         }
                     }
+                    if(remb_enabled) {
+                        if (rtcp_fb == "goog-remb") {
+                            video_payload->rtcp_fbs_.push_back(rtcp_fb);
+                        }                        
+                    }
                 }
 
                 track_desc->type_ = "video";
@@ -2836,6 +2962,11 @@ srs_error_t SrsRtcConnection::negotiate_publish_capability(SrsRtcUserConfig* ruc
                                 video_payload->rtcp_fbs_.push_back(rtcp_fb);
                             }
                         }
+                        if(remb_enabled) {
+                            if (rtcp_fb == "goog-remb") {
+                                video_payload->rtcp_fbs_.push_back(rtcp_fb);
+                            }                        
+                        }
                     }
 
                     track_desc->type_ = "video";
@@ -2873,6 +3004,11 @@ srs_error_t SrsRtcConnection::negotiate_publish_capability(SrsRtcUserConfig* ruc
                             video_payload->rtcp_fbs_.push_back(rtcp_fb);
                         }
                     }
+                    if(remb_enabled) {
+                        if (rtcp_fb == "goog-remb") {
+                            video_payload->rtcp_fbs_.push_back(rtcp_fb);
+                        }                        
+                    }
                 }
 
                 track_desc->type_ = "video";

trunk/src/app/srs_app_rtc_conn.hpp

@@ -341,6 +341,8 @@ class SrsRtcPublishStream : public ISrsRtspPacketDecodeHandler
     bool nack_enabled_;
     bool nack_no_copy_;
     bool twcc_enabled_;
+    uint64_t bitrate_;
+    uint64_t remb_startup_;
 private:
     bool request_keyframe_;
     SrsErrorPithyPrint* pli_epp;
@@ -372,6 +374,7 @@ class SrsRtcPublishStream : public ISrsRtspPacketDecodeHandler
 private:
     srs_error_t send_rtcp_rr();
     srs_error_t send_rtcp_xr_rrtr();
+    srs_error_t send_rtcp_remb();    
 public:
     srs_error_t on_rtp_cipher(char* buf, int nb_buf);
     srs_error_t on_rtp_plaintext(char* buf, int nb_buf);
@@ -535,6 +538,7 @@ class SrsRtcConnection : public ISrsResource, public ISrsDisposingHandler, publi
     srs_error_t send_rtcp(char *data, int nb_data);
     void check_send_nacks(SrsRtpNackForReceiver* nack, uint32_t ssrc, uint32_t& sent_nacks, uint32_t& timeout_nacks);
     srs_error_t send_rtcp_rr(uint32_t ssrc, SrsRtpRingBuffer* rtp_queue, const uint64_t& last_send_systime, const SrsNtp& last_send_ntp);
+    srs_error_t send_rtcp_remb(uint32_t ssrc, const uint64_t& bitrate);
     srs_error_t send_rtcp_xr_rrtr(uint32_t ssrc);
     srs_error_t send_rtcp_fb_pli(uint32_t ssrc, const SrsContextId& cid_of_subscriber);
 public:

trunk/src/protocol/srs_protocol_rtmp_stack.cpp

@@ -1463,6 +1463,8 @@ SrsRequest::SrsRequest()
     args = NULL;
 
     protocol = "rtmp";
+	
+	bitrate = 0;
 }
 
 SrsRequest::~SrsRequest()
@@ -1492,6 +1494,7 @@ SrsRequest* SrsRequest::copy()
     }
 
     cp->protocol = protocol;
+	cp->bitrate = bitrate;
 
     return cp;
 }
@@ -1521,6 +1524,8 @@ void SrsRequest::update_auth(SrsRequest* req)
     }
 
     protocol = req->protocol;
+
+	bitrate = req->bitrate;
 
     srs_info("update req of soruce for auth ok");
 }

trunk/src/protocol/srs_protocol_rtmp_stack.hpp

@@ -441,6 +441,9 @@ class SrsRequest
     // used for edge traverse to origin authentication,
     // @see https://github.com/ossrs/srs/issues/104
     SrsAmf0Object* args;
+
+    //the max bitrate for client
+    uint64_t bitrate;
 public:
     SrsRequest();
     virtual ~SrsRequest();