Rtcp sr and rr for rtc play

trunk/src/app/srs_app_rtc_conn.cpp

@@ -438,6 +438,8 @@ SrsRtcPlayStream::SrsRtcPlayStream(SrsRtcConnection* s, const SrsContextId& cid)
 
     cache_ssrc0_ = cache_ssrc1_ = cache_ssrc2_ = 0;
     cache_track0_ = cache_track1_ = cache_track2_ = NULL;
+
+    timer_rtcp_ = new SrsRtcPlayRtcpTimer(this);
 }
 
 SrsRtcPlayStream::~SrsRtcPlayStream()
@@ -448,6 +450,9 @@ SrsRtcPlayStream::~SrsRtcPlayStream()
 
     _srs_config->unsubscribe(this);
 
+    if (timer_rtcp_) {
+        srs_freep(timer_rtcp_);
+    }
     srs_freep(nack_epp);
     srs_freep(pli_worker_);
     srs_freep(trd_);
@@ -681,6 +686,29 @@ srs_error_t SrsRtcPlayStream::cycle()
     }
 }
 
+srs_error_t SrsRtcPlayStream::send_rtcp_sr(int64_t now_ms) {
+    srs_error_t err = srs_success;
+    for(std::map<uint32_t, SrsRtcVideoSendTrack*>::iterator iter = video_tracks_.begin();
+        iter != video_tracks_.end();
+        iter++) {
+        SrsRtcVideoSendTrack* track = iter->second;
+        if ((err = track->send_rtcp_sr(now_ms)) != srs_success) {
+            return srs_error_wrap(err, "video send rtcp sr error track=%s", track->get_track_id().c_str());
+        }
+    }
+
+    for(std::map<uint32_t, SrsRtcAudioSendTrack*>::iterator iter = audio_tracks_.begin();
+        iter != audio_tracks_.end();
+        iter++) {
+        SrsRtcAudioSendTrack* track = iter->second;
+        if ((err = track->send_rtcp_sr(now_ms)) != srs_success) {
+            return srs_error_wrap(err, "audiosend rtcp sr error track=%s", track->get_track_id().c_str());
+        }
+    }
+
+    return err;
+}
+
 srs_error_t SrsRtcPlayStream::send_packet(SrsRtpPacket*& pkt)
 {
     srs_error_t err = srs_success;
@@ -779,8 +807,9 @@ void SrsRtcPlayStream::set_all_tracks_status(bool status)
 srs_error_t SrsRtcPlayStream::on_rtcp(SrsRtcpCommon* rtcp)
 {
     if(SrsRtcpType_rr == rtcp->type()) {
+        int64_t now_ms = srs_update_system_time()/1000;
         SrsRtcpRR* rr = dynamic_cast<SrsRtcpRR*>(rtcp);
-        return on_rtcp_rr(rr);
+        return on_rtcp_rr(rr, now_ms);
     } else if(SrsRtcpType_rtpfb == rtcp->type()) {
         //currently rtpfb of nack will be handle by player. TWCC will be handled by SrsRtcConnection
         SrsRtcpNack* nack = dynamic_cast<SrsRtcpNack*>(rtcp);
@@ -799,12 +828,33 @@ srs_error_t SrsRtcPlayStream::on_rtcp(SrsRtcpCommon* rtcp)
     }
 }
 
-srs_error_t SrsRtcPlayStream::on_rtcp_rr(SrsRtcpRR* rtcp)
+srs_error_t SrsRtcPlayStream::on_rtcp_rr(SrsRtcpRR* rtcp, int64_t now_ms)
 {
     srs_error_t err = srs_success;
+
+    for(std::vector<SrsRtcpRB>::iterator iter = rtcp->rr_blocks_.begin();
+        iter != rtcp->rr_blocks_.end();
+        iter++) {
+        SrsRtcpRB& rb = *iter;
+        uint32_t ssrc = rb.ssrc;
+
+        for(std::map<uint32_t, SrsRtcAudioSendTrack*>::iterator audio_iter = audio_tracks_.begin();
+            audio_iter != audio_tracks_.end();
+            audio_iter++) {
+            if(ssrc == audio_iter->second->track_desc_->ssrc_) {
+                return audio_iter->second->handle_rtcp_rr(rb, now_ms);
+            }
+        }
 
-    // TODO: FIXME: Implements it.
-
+        for(std::map<uint32_t, SrsRtcVideoSendTrack*>::iterator video_iter = video_tracks_.begin();
+            video_iter != video_tracks_.end();
+            video_iter++) {
+            if(ssrc == video_iter->second->track_desc_->ssrc_) {
+                return video_iter->second->handle_rtcp_rr(rb, now_ms);
+            }
+        }
+        srs_warn("rtcp rr find to find track by ssrc:%u", ssrc);
+    }
     return err;
 }
 
@@ -928,6 +978,33 @@ srs_error_t SrsRtcPlayStream::do_request_keyframe(uint32_t ssrc, SrsContextId ci
     return err;
 }
 
+SrsRtcPlayRtcpTimer::SrsRtcPlayRtcpTimer(SrsRtcPlayStream* p) : p_(p)
+{
+    _srs_hybrid->timer1s()->subscribe(this);
+}
+
+SrsRtcPlayRtcpTimer::~SrsRtcPlayRtcpTimer()
+{
+    _srs_hybrid->timer1s()->unsubscribe(this);
+}
+
+srs_error_t SrsRtcPlayRtcpTimer::on_timer(srs_utime_t interval)
+{
+    srs_error_t err = srs_success;
+
+    if (!p_->is_started) {
+        return err;
+    }
+
+    int64_t now_ms = srs_update_system_time();
+    if ((err = p_->send_rtcp_sr(now_ms)) != srs_success) {
+        srs_warn("RR err %s", srs_error_desc(err).c_str());
+        srs_freep(err);
+    }
+
+    return err;
+}
+
 SrsRtcPublishRtcpTimer::SrsRtcPublishRtcpTimer(SrsRtcPublishStream* p) : p_(p)
 {
     _srs_hybrid->timer1s()->subscribe(this);
@@ -1298,17 +1375,21 @@ srs_error_t SrsRtcPublishStream::send_rtcp_rr()
 {
     srs_error_t err = srs_success;
 
-    for (int i = 0; i < (int)video_tracks_.size(); ++i) {
-        SrsRtcVideoRecvTrack* track = video_tracks_.at(i);
+    for (std::vector<SrsRtcVideoRecvTrack*>::iterator iter = video_tracks_.begin();
+         iter != video_tracks_.end();
+         iter++) {
+        SrsRtcVideoRecvTrack* track = *iter;
         if ((err = track->send_rtcp_rr()) != srs_success) {
-            return srs_error_wrap(err, "track=%s", track->get_track_id().c_str());
+            return srs_error_wrap(err, "send rtcp rr error, videotrack=%s", track->get_track_id().c_str());
         }
     }
 
-    for (int i = 0; i < (int)audio_tracks_.size(); ++i) {
-        SrsRtcAudioRecvTrack* track = audio_tracks_.at(i);
+    for (std::vector<SrsRtcAudioRecvTrack*>::iterator iter = audio_tracks_.begin();
+         iter != audio_tracks_.end();
+         iter++) {
+        SrsRtcAudioRecvTrack* track = *iter;
         if ((err = track->send_rtcp_rr()) != srs_success) {
-            return srs_error_wrap(err, "track=%s", track->get_track_id().c_str());
+            return srs_error_wrap(err, "send rtcp rr error: audiotrack=%s", track->get_track_id().c_str());
         }
     }
 
@@ -2079,9 +2160,17 @@ srs_error_t SrsRtcConnection::dispatch_rtcp(SrsRtcpCommon* rtcp)
     // Ignore special packet.
     if (SrsRtcpType_rr == rtcp->type()) {
         SrsRtcpRR* rr = dynamic_cast<SrsRtcpRR*>(rtcp);
-        if (rr->get_rb_ssrc() == 0) { //for native client
+        if (rr->rr_blocks_.empty()) { //for native client
             return err;
         }
+        for (std::vector<SrsRtcpRB>::iterator iter = rr->rr_blocks_.begin();
+             iter != rr->rr_blocks_.end();
+             iter++) {
+            SrsRtcpRB& rb = *iter;
+            if (rb.ssrc == 0) {
+                return err;
+            }
+        }
     }
 
     // The feedback packet for specified SSRC.
@@ -2091,7 +2180,18 @@ srs_error_t SrsRtcConnection::dispatch_rtcp(SrsRtcpCommon* rtcp)
         required_publisher_ssrc = rtcp->get_ssrc();
     } else if (SrsRtcpType_rr == rtcp->type()) {
         SrsRtcpRR* rr = dynamic_cast<SrsRtcpRR*>(rtcp);
-        required_player_ssrc = rr->get_rb_ssrc();
+
+        for (std::vector<SrsRtcpRB>::iterator iter = rr->rr_blocks_.begin();
+             iter != rr->rr_blocks_.end();
+             iter++) {
+            SrsRtcpRB& rb = *iter;
+            uint32_t ssrc = rb.ssrc;
+            std::map<uint32_t, SrsRtcPlayStream*>::iterator it = players_ssrc_map_.find(ssrc);
+            if (it != players_ssrc_map_.end()) {
+                it->second->on_rtcp(rtcp);
+                break;
+            }
+        }
     } else if (SrsRtcpType_rtpfb == rtcp->type()) {
         if(1 == rtcp->get_rc()) {
             SrsRtcpNack* nack = dynamic_cast<SrsRtcpNack*>(rtcp);

trunk/src/app/srs_app_rtc_conn.hpp

@@ -56,6 +56,7 @@ class SrsRtcNetworks;
 class SrsRtcUdpNetwork;
 class ISrsRtcNetwork;
 class SrsRtcTcpNetwork;
+class SrsRtcPlayRtcpTimer;
 
 const uint8_t kSR   = 200;
 const uint8_t kRR   = 201;
@@ -210,6 +211,7 @@ class SrsRtcAsyncCallOnStop : public ISrsAsyncCallTask
 class SrsRtcPlayStream : public ISrsCoroutineHandler, public ISrsReloadHandler
     , public ISrsRtcPLIWorkerHandler, public ISrsRtcSourceChangeCallback
 {
+friend class SrsRtcPlayRtcpTimer;
 private:
     SrsContextId cid_;
     SrsFastCoroutine* trd_;
@@ -223,6 +225,8 @@ class SrsRtcPlayStream : public ISrsCoroutineHandler, public ISrsReloadHandler
     std::map<uint32_t, SrsRtcVideoSendTrack*> video_tracks_;
     // The pithy print for special stage.
     SrsErrorPithyPrint* nack_epp;
+private:
+    SrsRtcPlayRtcpTimer* timer_rtcp_;
 private:
     // Fast cache for tracks.
     uint32_t cache_ssrc0_;
@@ -259,6 +263,8 @@ class SrsRtcPlayStream : public ISrsCoroutineHandler, public ISrsReloadHandler
     virtual void stop();
 public:
     virtual srs_error_t cycle();
+public:
+    srs_error_t send_rtcp_sr(int64_t now_ms);
 private:
     srs_error_t send_packet(SrsRtpPacket*& pkt);
 public:
@@ -270,7 +276,7 @@ class SrsRtcPlayStream : public ISrsCoroutineHandler, public ISrsReloadHandler
     srs_error_t on_rtcp_xr(SrsRtcpXr* rtcp);
     srs_error_t on_rtcp_nack(SrsRtcpNack* rtcp);
     srs_error_t on_rtcp_ps_feedback(SrsRtcpFbCommon* rtcp);
-    srs_error_t on_rtcp_rr(SrsRtcpRR* rtcp);
+    srs_error_t on_rtcp_rr(SrsRtcpRR* rtcp, int64_t now_ms);
     uint32_t get_video_publish_ssrc(uint32_t play_ssrc);
 // Interface ISrsRtcPLIWorkerHandler
 public:
@@ -290,6 +296,19 @@ class SrsRtcPublishRtcpTimer : public ISrsFastTimer
     srs_error_t on_timer(srs_utime_t interval);
 };
 
+// A fast timer for play stream, for RTCP feedback.
+class SrsRtcPlayRtcpTimer : public ISrsFastTimer
+{
+private:
+    SrsRtcPlayStream* p_;
+public:
+    SrsRtcPlayRtcpTimer(SrsRtcPlayStream* p);
+    virtual ~SrsRtcPlayRtcpTimer();
+// interface ISrsFastTimer
+private:
+    srs_error_t on_timer(srs_utime_t interval);
+};
+
 // A fast timer for publish stream, for TWCC feedback.
 class SrsRtcPublishTwccTimer : public ISrsFastTimer
 {