For #307, refine core structure, RTMP base on frame, RTC base on RTP. 4.0.26

README.md

@@ -159,6 +159,7 @@ For previous versions, please read:
 
 ## V4 changes
 
+* v4.0, 2020-05-14, For [#307][bug #307], refine core structure, RTMP base on frame, RTC base on RTP. 4.0.26
 * v4.0, 2020-05-11, For [#307][bug #307], refine RTC publisher structure. 4.0.25
 * v4.0, 2020-04-30, For [#307][bug #307], support publish RTC with passing opus. 4.0.24
 * v4.0, 2020-04-14, For [#307][bug #307], support sendmmsg, GSO and reuseport. 4.0.23

trunk/src/app/srs_app_rtc_conn.cpp

@@ -1759,75 +1759,18 @@ void SrsRtcPublisher::on_before_decode_payload(SrsRtpPacket2* pkt, SrsBuffer* bu
     }
 }
 
-srs_error_t SrsRtcPublisher::on_rtcp(char* data, int nb_data)
+srs_error_t SrsRtcPublisher::on_audio(SrsRtpPacket2* pkt)
 {
     srs_error_t err = srs_success;
 
-    char* ph = data;
+    pkt->frame_type = SrsFrameTypeAudio;
-    int nb_left = nb_data;
-    while (nb_left) {
-        uint8_t payload_type = ph[1];
-        uint16_t length_4bytes = (((uint16_t)ph[2]) << 8) | ph[3];
-
-        int length = (length_4bytes + 1) * 4;
-
-        if (length > nb_data) {
-            return srs_error_new(ERROR_RTC_RTCP, "invalid rtcp packet, length=%u", length);
-        }
-
-        srs_verbose("on rtcp, payload_type=%u", payload_type);
-
-        switch (payload_type) {
-            case kSR: {
-                err = on_rtcp_sr(ph, length);
-                break;
-            }
-            case kRR: {
-                err = on_rtcp_rr(ph, length);
-                break;
-            }
-            case kSDES: {
-                break;
-            }
-            case kBye: {
-                break;
-            }
-            case kApp: {
-                break;
-            }
-            case kRtpFb: {
-                err = on_rtcp_feedback(ph, length);
-                break;
-            }
-            case kPsFb: {
-                err = on_rtcp_ps_feedback(ph, length);
-                break;
-            }
-            case kXR: {
-                err = on_rtcp_xr(ph, length);
-                break;
-            }
-            default:{
-                return srs_error_new(ERROR_RTC_RTCP_CHECK, "unknown rtcp type=%u", payload_type);
-                break;
-            }
-        }
-
-        if (err != srs_success) {
-            return srs_error_wrap(err, "rtcp");
-        }
 
-        ph += length;
+    // TODO: FIXME: Error check.
-        nb_left -= length;
+    source->on_rtp(pkt);
-    }
 
     return err;
-}
-
-srs_error_t SrsRtcPublisher::on_audio(SrsRtpPacket2* pkt)
-{
-    srs_error_t err = srs_success;
 
+    // TODO: FIXME: Directly dispatch to consumer for performance?
     std::vector<SrsRtpPacket2*> frames;
 
     if (nack_enabled_) {
@@ -1850,57 +1793,34 @@ srs_error_t SrsRtcPublisher::on_audio(SrsRtpPacket2* pkt)
         SrsRtpPacket2* frame = frames[i];
 
         // TODO: FIXME: Check error.
-        on_audio_frame(frame);
+        source->on_rtp(frame);
 
-        srs_freep(frame);
+        if (nn_audio_frames++ == 0) {
+            SrsRtpHeader* h = &frame->rtp_header;
+            SrsRtpRawPayload* payload = dynamic_cast<SrsRtpRawPayload*>(frame->payload);
+            srs_trace("RTC got Opus seq=%u, ssrc=%u, ts=%u, %d bytes", h->get_sequence(), h->get_ssrc(), h->get_timestamp(), payload->nn_payload);
+        }
     }
 
     return err;
 }
 
-srs_error_t SrsRtcPublisher::on_audio_frame(SrsRtpPacket2* frame)
+srs_error_t SrsRtcPublisher::on_video(SrsRtpPacket2* pkt)
 {
     srs_error_t err = srs_success;
 
-    SrsRtpRawPayload* payload = dynamic_cast<SrsRtpRawPayload*>(frame->payload);
+    pkt->frame_type = SrsFrameTypeVideo;
-
-    if (!payload) {
-        return srs_error_new(ERROR_RTC_RTP_MUXER, "OPUS payload");
-    }
-
-    // TODO: FIXME: Transcode OPUS to AAC.
-    if (!payload->nn_payload) {
-        return err;
-    }
 
-    SrsMessageHeader header;
+    // TODO: FIXME: Error check.
-    header.message_type = RTMP_MSG_AudioMessage;
+    source->on_rtp(pkt);
-    // TODO: FIXME: Maybe the tbn is not 90k.
-    header.timestamp = frame->rtp_header.get_timestamp() / 90;
-
-    SrsSharedPtrMessage msg;
-    // TODO: FIXME: Check error.
-    msg.create(&header, NULL, 0);
-
-    SrsSample sample;
-    sample.size = payload->nn_payload;
-    sample.bytes = new char[sample.size];
-    memcpy((void*)sample.bytes, payload->payload, sample.size);
-    msg.set_extra_payloads(&sample, 1);
 
+    if (video_queue_->should_request_key_frame()) {
         // TODO: FIXME: Check error.
-    source->on_audio_imp(&msg);
+        send_rtcp_fb_pli(video_ssrc);
-
-    if (nn_audio_frames++ == 0) {
-        SrsRtpHeader* h = &frame->rtp_header;
-        srs_trace("RTC got Opus seq=%u, ssrc=%u, ts=%u, %d bytes", h->get_sequence(), h->get_ssrc(), h->get_timestamp(), payload->nn_payload);
     }
 
     return err;
-}
 
-srs_error_t SrsRtcPublisher::on_video(SrsRtpPacket2* pkt)
-{
     std::vector<SrsRtpPacket2*> frames;
 
     if (nack_enabled_) {
@@ -2042,6 +1962,71 @@ srs_error_t SrsRtcPublisher::on_video_frame(SrsRtpPacket2* frame)
     return source->on_video(shared_video);
 }
 
+srs_error_t SrsRtcPublisher::on_rtcp(char* data, int nb_data)
+{
+    srs_error_t err = srs_success;
+
+    char* ph = data;
+    int nb_left = nb_data;
+    while (nb_left) {
+        uint8_t payload_type = ph[1];
+        uint16_t length_4bytes = (((uint16_t)ph[2]) << 8) | ph[3];
+
+        int length = (length_4bytes + 1) * 4;
+
+        if (length > nb_data) {
+            return srs_error_new(ERROR_RTC_RTCP, "invalid rtcp packet, length=%u", length);
+        }
+
+        srs_verbose("on rtcp, payload_type=%u", payload_type);
+
+        switch (payload_type) {
+            case kSR: {
+                err = on_rtcp_sr(ph, length);
+                break;
+            }
+            case kRR: {
+                err = on_rtcp_rr(ph, length);
+                break;
+            }
+            case kSDES: {
+                break;
+            }
+            case kBye: {
+                break;
+            }
+            case kApp: {
+                break;
+            }
+            case kRtpFb: {
+                err = on_rtcp_feedback(ph, length);
+                break;
+            }
+            case kPsFb: {
+                err = on_rtcp_ps_feedback(ph, length);
+                break;
+            }
+            case kXR: {
+                err = on_rtcp_xr(ph, length);
+                break;
+            }
+            default:{
+                return srs_error_new(ERROR_RTC_RTCP_CHECK, "unknown rtcp type=%u", payload_type);
+                break;
+            }
+        }
+
+        if (err != srs_success) {
+            return srs_error_wrap(err, "rtcp");
+        }
+
+        ph += length;
+        nb_left -= length;
+    }
+
+    return err;
+}
+
 srs_error_t SrsRtcPublisher::on_rtcp_sr(char* buf, int nb_buf)
 {
     srs_error_t err = srs_success;

trunk/src/app/srs_app_rtc_conn.hpp

@@ -314,12 +314,13 @@ private:
 public:
     srs_error_t on_rtp(char* buf, int nb_buf);
     virtual void on_before_decode_payload(SrsRtpPacket2* pkt, SrsBuffer* buf, ISrsCodec** ppayload);
-    srs_error_t on_rtcp(char* data, int nb_data);
 private:
     srs_error_t on_audio(SrsRtpPacket2* pkt);
-    srs_error_t on_audio_frame(SrsRtpPacket2* frame);
     srs_error_t on_video(SrsRtpPacket2* pkt);
     srs_error_t on_video_frame(SrsRtpPacket2* frame);
+public:
+    srs_error_t on_rtcp(char* data, int nb_data);
+private:
     srs_error_t on_rtcp_sr(char* buf, int nb_buf);
     srs_error_t on_rtcp_xr(char* buf, int nb_buf);
     srs_error_t on_rtcp_feedback(char* data, int nb_data);

trunk/src/app/srs_app_rtc_source.cpp

@@ -695,8 +695,8 @@ srs_error_t SrsRtcFromRtmpBridger::package_opus(char* data, int size, SrsRtpPack
     srs_error_t err = srs_success;
 
     SrsRtpPacket2* pkt = new SrsRtpPacket2();
-    pkt->rtp_header.set_marker(true);
     pkt->frame_type = SrsFrameTypeAudio;
+    pkt->rtp_header.set_marker(true);
 
     SrsRtpRawPayload* raw = pkt->reuse_raw();
     raw->payload = new char[size];
@@ -836,6 +836,7 @@ srs_error_t SrsRtcFromRtmpBridger::package_stap_a(SrsRtcSource* source, SrsShare
     }
 
     SrsRtpPacket2* pkt = new SrsRtpPacket2();
+    pkt->frame_type = SrsFrameTypeVideo;
     pkt->rtp_header.set_marker(false);
     pkt->rtp_header.set_timestamp(msg->timestamp * 90);
 
@@ -903,6 +904,7 @@ srs_error_t SrsRtcFromRtmpBridger::package_nalus(SrsSharedPtrMessage* msg, vecto
     if (nn_bytes < kRtpMaxPayloadSize) {
         // Package NALUs in a single RTP packet.
         SrsRtpPacket2* pkt = new SrsRtpPacket2();
+        pkt->frame_type = SrsFrameTypeVideo;
         pkt->rtp_header.set_timestamp(msg->timestamp * 90);
         pkt->payload = raw;
         pkt->original_msg = msg->copy();
@@ -931,6 +933,7 @@ srs_error_t SrsRtcFromRtmpBridger::package_nalus(SrsSharedPtrMessage* msg, vecto
             }
 
             SrsRtpPacket2* pkt = new SrsRtpPacket2();
+            pkt->frame_type = SrsFrameTypeVideo;
             pkt->rtp_header.set_timestamp(msg->timestamp * 90);
 
             fua->nri = (SrsAvcNaluType)header;
@@ -955,6 +958,7 @@ srs_error_t SrsRtcFromRtmpBridger::package_single_nalu(SrsSharedPtrMessage* msg,
     srs_error_t err = srs_success;
 
     SrsRtpPacket2* pkt = new SrsRtpPacket2();
+    pkt->frame_type = SrsFrameTypeVideo;
     pkt->rtp_header.set_timestamp(msg->timestamp * 90);
 
     SrsRtpRawPayload* raw = pkt->reuse_raw();
@@ -981,6 +985,7 @@ srs_error_t SrsRtcFromRtmpBridger::package_fu_a(SrsSharedPtrMessage* msg, SrsSam
         int packet_size = srs_min(nb_left, fu_payload_size);
 
         SrsRtpPacket2* pkt = new SrsRtpPacket2();
+        pkt->frame_type = SrsFrameTypeVideo;
         pkt->rtp_header.set_timestamp(msg->timestamp * 90);
 
         SrsRtpFUAPayload2* fua = pkt->reuse_fua();

trunk/src/core/srs_core_version4.hpp

@@ -24,6 +24,6 @@
 #ifndef SRS_CORE_VERSION4_HPP
 #define SRS_CORE_VERSION4_HPP
 
-#define SRS_VERSION4_REVISION 25
+#define SRS_VERSION4_REVISION 26
 
 #endif