startSendRtp接口支持rtcp接收超时主动停止

3rdpart/ZLToolKit

@@ -1 +1 @@
-Subproject commit 4e127f53a93dd1b7665f04a0ec7312d67a34ab6f
+Subproject commit 658271fdf4fb497b0665c06544627789844a003a

postman/ZLMediaKit.postman_collection.json

@@ -1562,6 +1562,12 @@
 							"value": "1",
 							"description": "rtp es方式打包时，是否只打包音频，该参数非必选参数",
 							"disabled": true
+						},
+						{
+							"key": "udp_rtcp_timeout",
+							"value": "0",
+							"description": "udp方式推流时，是否开启rtcp发送和rtcp接收超时判断，开启后(默认关闭)，如果接收rr rtcp超时，将导致主动停止rtp发送",
+							"disabled": true
 						}
 					]
 				}

server/WebApi.cpp

@@ -1147,7 +1147,8 @@ void installWebApi() {
         args.src_port = allArgs["src_port"];
         args.pt = allArgs["pt"].empty() ? 96 : allArgs["pt"].as<int>();
         args.use_ps = allArgs["use_ps"].empty() ? true : allArgs["use_ps"].as<bool>();
-        args.only_audio = allArgs["only_audio"].empty() ? false : allArgs["only_audio"].as<bool>();
+        args.only_audio = allArgs["only_audio"].as<bool>();
+        args.udp_rtcp_timeout = allArgs["udp_rtcp_timeout"];
         TraceL << "startSendRtp, pt " << int(args.pt) << " ps " << args.use_ps << " audio " << args.only_audio;
 
         src->getOwnerPoller()->async([=]() mutable {
@@ -1178,7 +1179,7 @@ void installWebApi() {
         args.src_port = allArgs["src_port"];
         args.pt = allArgs["pt"].empty() ? 96 : allArgs["pt"].as<int>();
         args.use_ps = allArgs["use_ps"].empty() ? true : allArgs["use_ps"].as<bool>();
-        args.only_audio = allArgs["only_audio"].empty() ? false : allArgs["only_audio"].as<bool>();
+        args.only_audio = allArgs["only_audio"].as<bool>();
         TraceL << "startSendRtpPassive, pt " << int(args.pt) << " ps " << args.use_ps << " audio " <<  args.only_audio;
 
         src->getOwnerPoller()->async([=]() mutable {

src/Common/MediaSource.h

@@ -111,6 +111,15 @@ public:
         uint16_t dst_port;
         // 发送目标主机地址，可以是ip或域名
         std::string dst_url;
+
+        //udp发送时，是否开启rr rtcp接收超时判断
+        bool udp_rtcp_timeout = false;
+        //tcp被动发送服务器延时关闭事件，单位毫秒
+        uint32_t tcp_passive_close_delay_ms = 5 * 1000;
+        //udp 发送时，rr rtcp包接收超时时间，单位毫秒
+        uint32_t rtcp_timeout_ms = 30 * 1000;
+        //udp 发送时，发送sr rtcp包间隔，单位毫秒
+        uint32_t rtcp_send_interval_ms = 5 * 1000;
     };
 
     // 开始发送ps-rtp

src/Common/MultiMediaSourceMuxer.cpp

@@ -223,7 +223,7 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &, const MediaSourceEvent::
 #if defined(ENABLE_RTPPROXY)
     auto rtp_sender = std::make_shared<RtpSender>();
     weak_ptr<MultiMediaSourceMuxer> weak_self = shared_from_this();
-    rtp_sender->startSend(args, [args, weak_self, rtp_sender, cb](uint16_t local_port, const SockException &ex) {
+    rtp_sender->startSend(args, [args, weak_self, rtp_sender, cb](uint16_t local_port, const SockException &ex) mutable {
         cb(local_port, ex);
         auto strong_self = weak_self.lock();
         if (!strong_self || ex) {
@@ -233,7 +233,15 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &, const MediaSourceEvent::
             rtp_sender->addTrack(track);
         }
         rtp_sender->addTrackCompleted();
-        strong_self->_rtp_sender[args.ssrc] = rtp_sender;
+
+        auto ssrc = args.ssrc;
+        rtp_sender->setOnClose([weak_self, ssrc]() {
+            if (auto strong_self = weak_self.lock()) {
+                strong_self->_rtp_sender.erase(ssrc);
+                WarnL << "stream:" << strong_self->_get_origin_url() << " stop send rtp:" << ssrc;
+            }
+        });
+        strong_self->_rtp_sender[args.ssrc] = std::move(rtp_sender);
     });
 #else
     cb(0, SockException(Err_other, "该功能未启用，编译时请打开ENABLE_RTPPROXY宏"));

src/Rtp/RtpSender.cpp

@@ -22,7 +22,7 @@ namespace mediakit{
 
 RtpSender::RtpSender() {
     _poller = EventPollerPool::Instance().getPoller();
-    _socket = Socket::createSocket(_poller, false);
+    _socket_rtp = Socket::createSocket(_poller, false);
 }
 
 void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const function<void(uint16_t local_port, const SockException &ex)> &cb){
@@ -55,7 +55,7 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
                 makeSockPair(pr, "::", false, false);
             }
             // tcp服务器默认开启5秒
-            auto delay_task = _poller->doDelayTask(5 * 1000, [tcp_listener, cb]() mutable {
+            auto delay_task = _poller->doDelayTask(_args.tcp_passive_close_delay_ms, [tcp_listener, cb]() mutable {
                 cb(0, SockException(Err_timeout, "wait tcp connection timeout"));
                 tcp_listener = nullptr;
                 return 0;
@@ -67,7 +67,7 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
                 }
                 //立即关闭tcp服务器
                 delay_task->cancel();
-                strong_self->_socket = sock;
+                strong_self->_socket_rtp = sock;
                 strong_self->onConnect();
                 cb(sock->get_local_port(), SockException());
                 InfoL << "accept connection from:" << sock->get_peer_ip() << ":" << sock->get_peer_port();
@@ -103,12 +103,12 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
                 try {
                     if (args.src_port) {
                         //指定端口
-                        if (!strong_self->_socket->bindUdpSock(args.src_port, ifr_ip)) {
+                        if (!strong_self->_socket_rtp->bindUdpSock(args.src_port, ifr_ip)) {
                             throw std::invalid_argument(StrPrinter << "bindUdpSock failed on port:" << args.src_port
                                                                    << ", err:" << get_uv_errmsg(true));
                         }
                     } else {
-                        auto pr = std::make_pair(strong_self->_socket, Socket::createSocket(strong_self->_poller, false));
+                        auto pr = std::make_pair(strong_self->_socket_rtp, Socket::createSocket(strong_self->_poller, false));
                         //从端口池获取随机端口
                         makeSockPair(pr, ifr_ip, true);
                     }
@@ -116,20 +116,20 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
                     cb(0, SockException(Err_other, ex.what()));
                     return;
                 }
-                strong_self->_socket->bindPeerAddr((struct sockaddr *)&addr);
+                strong_self->_socket_rtp->bindPeerAddr((struct sockaddr *)&addr);
                 strong_self->onConnect();
-                cb(strong_self->_socket->get_local_port(), SockException());
+                cb(strong_self->_socket_rtp->get_local_port(), SockException());
             });
         });
     } else {
-        _socket->connect(args.dst_url, args.dst_port, [cb, weak_self](const SockException &err) {
+        _socket_rtp->connect(args.dst_url, args.dst_port, [cb, weak_self](const SockException &err) {
             auto strong_self = weak_self.lock();
             if (strong_self) {
                 if (!err) {
                     //tcp连接成功
                     strong_self->onConnect();
                 }
-                cb(strong_self->_socket->get_local_port(), err);
+                cb(strong_self->_socket_rtp->get_local_port(), err);
             } else {
                 cb(0, err);
             }
@@ -137,26 +137,73 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
     }
 }
 
+void RtpSender::createRtcpSocket() {
+    if (_socket_rtcp) {
+        return;
+    }
+    _socket_rtcp = Socket::createSocket(_socket_rtp->getPoller(), false);
+    //rtcp端口使用户rtp端口+1
+    if(!_socket_rtcp->bindUdpSock(_socket_rtp->get_local_port() + 1, _socket_rtp->get_local_ip(), false)){
+        WarnL << "bind rtcp udp socket failed:" << get_uv_errmsg(true);
+        _socket_rtcp = nullptr;
+        return;
+    }
+
+    struct sockaddr_storage addr;
+    //目标rtp端口
+    SockUtil::get_sock_peer_addr(_socket_rtp->rawFD(), addr);
+    //绑定目标rtcp端口(目标rtp端口 + 1)
+    switch (addr.ss_family) {
+        case AF_INET: ((sockaddr_in *)&addr)->sin_port = htons(ntohs(((sockaddr_in *)&addr)->sin_port) + 1); break;
+        case AF_INET6: ((sockaddr_in6 *)&addr)->sin6_port = htons(ntohs(((sockaddr_in6 *)&addr)->sin6_port) + 1); break;
+        default: assert(0); break;
+    }
+    _socket_rtcp->bindPeerAddr((struct sockaddr *)&addr);
+
+    _rtcp_context = std::make_shared<RtcpContextForSend>();
+    weak_ptr<RtpSender> weak_self = shared_from_this();
+    _socket_rtcp->setOnRead([weak_self](const Buffer::Ptr &buf, struct sockaddr *, int) {
+        //接收receive report rtcp
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            return;
+        }
+        auto rtcp_arr = RtcpHeader::loadFromBytes(buf->data(), buf->size());
+        for (auto &rtcp : rtcp_arr) {
+            strong_self->onRecvRtcp(rtcp);
+        }
+    });
+    InfoL << "open rtcp port success, start check rr rtcp timeout";
+}
+
+void RtpSender::onRecvRtcp(RtcpHeader *rtcp) {
+    _rtcp_context->onRtcp(rtcp);
+    _rtcp_recv_ticker.resetTime();
+}
+
+//连接建立成功事件
 void RtpSender::onConnect(){
     _is_connect = true;
     //加大发送缓存,防止udp丢包之类的问题
-    SockUtil::setSendBuf(_socket->rawFD(), 4 * 1024 * 1024);
+    SockUtil::setSendBuf(_socket_rtp->rawFD(), 4 * 1024 * 1024);
     if (!_args.is_udp) {
         //关闭tcp no_delay并开启MSG_MORE, 提高发送性能
-        SockUtil::setNoDelay(_socket->rawFD(), false);
+        SockUtil::setNoDelay(_socket_rtp->rawFD(), false);
-        _socket->setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
+        _socket_rtp->setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
+    } else if (_args.udp_rtcp_timeout) {
+        createRtcpSocket();
     }
     //连接建立成功事件
     weak_ptr<RtpSender> weak_self = shared_from_this();
-    _socket->setOnErr([weak_self](const SockException &err) {
+    _socket_rtp->setOnErr([weak_self](const SockException &err) {
         auto strong_self = weak_self.lock();
         if (strong_self) {
             strong_self->onErr(err);
         }
     });
     //获取本地端口，断开重连后确保端口不变
-    _args.src_port = _socket->get_local_port();
+    _args.src_port = _socket_rtp->get_local_port();
-    InfoL << "开始发送 rtp:" << _socket->get_peer_ip() << ":" << _socket->get_peer_port() << ", 是否为udp方式:" << _args.is_udp;
+    InfoL << "开始发送 rtp:" << _socket_rtp->get_peer_ip() << ":" << _socket_rtp->get_peer_port() << ", 是否为udp方式:" << _args.is_udp;
 }
 
 bool RtpSender::addTrack(const Track::Ptr &track){
@@ -177,6 +224,41 @@ bool RtpSender::inputFrame(const Frame::Ptr &frame) {
     return _is_connect ? _interface->inputFrame(frame) : false;
 }
 
+void RtpSender::onSendRtpUdp(const toolkit::Buffer::Ptr &buf, bool check) {
+    if (!_socket_rtcp) {
+        return;
+    }
+    auto rtp = static_pointer_cast<RtpPacket>(buf);
+    _rtcp_context->onRtp(rtp->getSeq(), rtp->getStamp(), rtp->getStampMS(), 90000 /*not used*/, rtp->size());
+
+    if (!check) {
+        //减少判断次数
+        return;
+    }
+    //每5秒发送一次rtcp
+    if (_rtcp_send_ticker.elapsedTime() > _args.rtcp_send_interval_ms) {
+        _rtcp_send_ticker.resetTime();
+        //rtcp ssrc为rtp ssrc + 1
+       auto sr = _rtcp_context->createRtcpSR(atoi(_args.ssrc.data()) + 1);
+       //send sender report rtcp
+       _socket_rtcp->send(sr);
+    }
+
+    if (_rtcp_recv_ticker.elapsedTime() > _args.rtcp_timeout_ms) {
+        //接收rr rtcp超时
+        WarnL << "recv rr rtcp timeout";
+        _rtcp_recv_ticker.resetTime();
+        onClose();
+    }
+}
+
+void RtpSender::onClose() {
+    auto cb = _on_close;
+    if (cb) {
+        _poller->async([cb]() { cb(); }, false);
+    }
+}
+
 //此函数在其他线程执行
 void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr> > rtp_list) {
     if(!_is_connect){
@@ -184,18 +266,22 @@ void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr> > rtp_list) {
         return;
     }
 
-    auto is_udp = _args.is_udp;
+    weak_ptr<RtpSender> weak_self = shared_from_this();
-    auto socket = _socket;
+    _poller->async([rtp_list, weak_self]() {
-    _poller->async([rtp_list, is_udp, socket]() {
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            return;
+        }
         size_t i = 0;
         auto size = rtp_list->size();
         rtp_list->for_each([&](Buffer::Ptr &packet) {
-            if (is_udp) {
+            if (strong_self->_args.is_udp) {
+                strong_self->onSendRtpUdp(packet, i == 0);
                 //udp模式，rtp over tcp前4个字节可以忽略
-                socket->send(std::make_shared<BufferRtp>(std::move(packet), 4), nullptr, 0, ++i == size);
+                strong_self->_socket_rtp->send(std::make_shared<BufferRtp>(std::move(packet),  RtpPacket::kRtpTcpHeaderSize), nullptr, 0, ++i == size);
             } else {
                 //tcp模式, rtp over tcp前2个字节可以忽略,只保留后续rtp长度的2个字节
-                socket->send(std::make_shared<BufferRtp>(std::move(packet), 2), nullptr, 0, ++i == size);
+                strong_self->_socket_rtp->send(std::make_shared<BufferRtp>(std::move(packet), 2), nullptr, 0, ++i == size);
             }
         });
     });
@@ -206,6 +292,8 @@ void RtpSender::onErr(const SockException &ex, bool is_connect) {
 
     if (_args.passive) {
         WarnL << "tcp passive connection lost: " << ex.what();
+        //tcp被动模式，如果对方断开连接，应该停止发送rtp
+        onClose();
     } else {
         //监听socket断开事件，方便重连
         if (is_connect) {
@@ -232,5 +320,9 @@ void RtpSender::onErr(const SockException &ex, bool is_connect) {
     }, _poller);
 }
 
+void RtpSender::setOnClose(std::function<void()> on_close){
+    _on_close = std::move(on_close);
+}
+
 }//namespace mediakit
 #endif// defined(ENABLE_RTPPROXY)

src/Rtp/RtpSender.h

@@ -13,6 +13,7 @@
 #if defined(ENABLE_RTPPROXY)
 #include "PSEncoder.h"
 #include "Extension/CommonRtp.h"
+#include "Rtcp/RtcpContext.h"
 
 namespace mediakit{
 
@@ -53,6 +54,8 @@ public:
      */
     virtual void resetTracks() override;
 
+    void setOnClose(std::function<void()> on_close);
+
 private:
     //合并写输出
     void onFlushRtpList(std::shared_ptr<toolkit::List<toolkit::Buffer::Ptr> > rtp_list);
@@ -60,14 +63,23 @@ private:
     void onConnect();
     //异常断开socket事件
     void onErr(const toolkit::SockException &ex, bool is_connect = false);
+    void createRtcpSocket();
+    void onRecvRtcp(RtcpHeader *rtcp);
+    void onSendRtpUdp(const toolkit::Buffer::Ptr &buf, bool check);
+    void onClose();
 
 private:
     bool _is_connect = false;
     MediaSourceEvent::SendRtpArgs _args;
-    toolkit::Socket::Ptr _socket;
+    toolkit::Socket::Ptr _socket_rtp;
+    toolkit::Socket::Ptr _socket_rtcp;
     toolkit::EventPoller::Ptr _poller;
     toolkit::Timer::Ptr _connect_timer;
     MediaSinkInterface::Ptr _interface;
+    std::shared_ptr<RtcpContext> _rtcp_context;
+    toolkit::Ticker _rtcp_send_ticker;
+    toolkit::Ticker _rtcp_recv_ticker;
+    std::function<void()> _on_close;
 };
 
 }//namespace mediakit