add rtp server timeout hook

conf/config.ini

@@ -155,6 +155,10 @@ on_server_started=https://127.0.0.1/index/hook/on_server_started
 on_server_keepalive=https://127.0.0.1/index/hook/on_server_keepalive
 #发送rtp(startSendRtp)被动关闭时回调
 on_send_rtp_stopped=https://127.0.0.1/index/hook/on_send_rtp_stopped
+
+#rtp server 超时未收到数据
+on_rtp_server_timeout=https://127.0.0.1/index/hook/on_rtp_server_timeout
+
 #hook api最大等待回复时间，单位秒
 timeoutSec=10
 #keepalive hook触发间隔,单位秒，float类型

server/WebApi.cpp

@@ -1761,6 +1761,11 @@ void installWebApi() {
     api_regist("/index/hook/on_server_keepalive",[](API_ARGS_JSON){
         //心跳hook
     });
+
+    api_regist("/index/hook/on_rtp_server_timeout",[](API_ARGS_JSON){
+        //rtp server 超时
+        TraceL <<allArgs.getArgs().toStyledString();
+    });
 }
 
 void unInstallWebApi(){

server/WebHook.cpp

@@ -46,6 +46,7 @@ const string kOnHttpAccess = HOOK_FIELD"on_http_access";
 const string kOnServerStarted = HOOK_FIELD"on_server_started";
 const string kOnServerKeepalive = HOOK_FIELD"on_server_keepalive";
 const string kOnSendRtpStopped = HOOK_FIELD"on_send_rtp_stopped";
+const string kOnRtpServerTimeout = HOOK_FIELD"on_rtp_server_timeout";
 const string kAdminParams = HOOK_FIELD"admin_params";
 const string kAliveInterval = HOOK_FIELD"alive_interval";
 const string kRetry = HOOK_FIELD"retry";
@@ -70,6 +71,7 @@ onceToken token([](){
     mINI::Instance()[kOnServerStarted] = "";
     mINI::Instance()[kOnServerKeepalive] = "";
     mINI::Instance()[kOnSendRtpStopped] = "";
+    mINI::Instance()[kOnRtpServerTimeout] = "";
     mINI::Instance()[kAdminParams] = "secret=035c73f7-bb6b-4889-a715-d9eb2d1925cc";
     mINI::Instance()[kAliveInterval] = 30.0;
     mINI::Instance()[kRetry] = 1;
@@ -648,6 +650,22 @@ void installWebHook(){
         });
     });
 
+    NoticeCenter::Instance().addListener(&web_hook_tag,Broadcast::KBroadcastRtpServerTimeout,[](BroadcastRtpServerTimeout){
+        GET_CONFIG(string,rtp_server_timeout,Hook::kOnRtpServerTimeout);
+        if(!hook_enable || rtp_server_timeout.empty()){
+            return;
+        }
+
+        ArgsType body;
+        body["local_port"] = local_port;
+        body["stream_id"] = stream_id;
+        body["tcp_mode"] = tcp_mode;
+        body["re_use_port"] = re_use_port;
+        body["ssrc"] = ssrc;
+        do_http_hook(rtp_server_timeout,body);
+
+    });
+
     //汇报服务器重新启动
     reportServerStarted();
 

src/Common/config.cpp

@@ -56,6 +56,7 @@ const string kBroadcastNotFoundStream = "kBroadcastNotFoundStream";
 const string kBroadcastStreamNoneReader = "kBroadcastStreamNoneReader";
 const string kBroadcastHttpBeforeAccess = "kBroadcastHttpBeforeAccess";
 const string kBroadcastSendRtpStopped = "kBroadcastSendRtpStopped";
+const string KBroadcastRtpServerTimeout = "KBroadcastRtpServerTimeout";
 
 } // namespace Broadcast
 

src/Common/config.h

@@ -114,6 +114,10 @@ extern const std::string kBroadcastSendRtpStopped;
 extern const std::string kBroadcastReloadConfig;
 #define BroadcastReloadConfigArgs void
 
+// rtp server 超时
+extern const std::string KBroadcastRtpServerTimeout;
+#define BroadcastRtpServerTimeout uint16_t &local_port, const string &stream_id,int &tcp_mode, bool &re_use_port, uint32_t &ssrc
+
 #define ReloadConfigTag ((void *)(0xFF))
 #define RELOAD_KEY(arg, key)                                                                                           \
     do {                                                                                                               \

src/Rtp/RtpServer.cpp

@@ -36,12 +36,20 @@ public:
         _stream_id = std::move(stream_id);
     }
 
+    
+
     ~RtcpHelper() {
         if (_process) {
             // 删除rtp处理器
             RtpSelector::Instance().delProcess(_stream_id, _process.get());
         }
     }
+    void setRtpServerInfo(uint16_t local_port,RtpServer::TcpMode mode,bool re_use_port,uint32_t ssrc){
+        _local_port = local_port;
+        _tcp_mode = mode;
+        _re_use_port = re_use_port;
+        _ssrc = ssrc;
+    }
 
     void setOnDetach(function<void()> cb) {
         if (_process) {
@@ -55,8 +63,7 @@ public:
         if (!_process) {
             _process = RtpSelector::Instance().getProcess(_stream_id, true);
             _process->setOnDetach(std::move(_on_detach));
-            _delay_task->cancel();
+            cancelDelayTask();
-            _delay_task = nullptr;
         }
         _process->inputRtp(true, sock, buf->data(), buf->size(), addr);
 
@@ -91,11 +98,21 @@ public:
                 if (!process && strong_self->_on_detach) {
                     strong_self->_on_detach();
                 }
+             if(!process){ // process 未创建，触发rtp server 超时事件
+                NoticeCenter::Instance().emitEvent(Broadcast::KBroadcastRtpServerTimeout,strong_self->_local_port,strong_self->_stream_id,(int)strong_self->_tcp_mode,strong_self->_re_use_port,strong_self->_ssrc);
+             }
             }
             return 0;
         });
     }
 
+    void cancelDelayTask(){
+        if(_delay_task){
+            _delay_task->cancel();
+            _delay_task = nullptr;
+        }
+    }
+
 private:
     void sendRtcp(uint32_t rtp_ssrc, struct sockaddr *addr) {
         // 每5秒发送一次rtcp
@@ -118,6 +135,12 @@ private:
     }
 
 private:
+    uint16_t _local_port = 0;
+    RtpServer::TcpMode _tcp_mode = RtpServer::NONE;
+    bool _re_use_port = false;
+    uint32_t _ssrc = 0;
+
+
     Ticker _ticker;
     Socket::Ptr _rtcp_sock;
     RtpProcess::Ptr _process;
@@ -169,6 +192,7 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
         //指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流)
         helper = std::make_shared<RtcpHelper>(std::move(rtcp_socket), stream_id);
         helper->startRtcp();
+        helper->setRtpServerInfo(local_port,tcp_mode,re_use_port,ssrc);
         rtp_socket->setOnRead([rtp_socket, helper, ssrc](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) {
             RtpHeader *header = (RtpHeader *)buf->data();
             auto rtp_ssrc = ntohl(header->ssrc);