新增动态更新rtp server过滤ssrc接口updateRtpServerSSRC(#2390)

postman/ZLMediaKit.postman_collection.json

@@ -1488,6 +1488,42 @@
 			},
 			"response": []
 		},
+		{
+			"name": "更新RTP服务器过滤SSRC(updateRtpServerSSRC)",
+			"request": {
+				"method": "GET",
+				"header": [],
+				"url": {
+					"raw": "{{ZLMediaKit_URL}}/index/api/updateRtpServerSSRC?secret={{ZLMediaKit_secret}}&stream_id=test&ssrc=123456",
+					"host": [
+						"{{ZLMediaKit_URL}}"
+					],
+					"path": [
+						"index",
+						"api",
+						"updateRtpServerSSRC"
+					],
+					"query": [
+						{
+							"key": "secret",
+							"value": "{{ZLMediaKit_secret}}",
+							"description": "api操作密钥(配置文件配置)，如果操作ip是127.0.0.1，则不需要此参数"
+						},
+						{
+							"key": "stream_id",
+							"value": "test",
+							"description": "该端口绑定的流id"
+						},
+						{
+							"key": "ssrc",
+							"value": "123456",
+							"description": "十进制ssrc"
+						}
+					]
+				}
+			},
+			"response": []
+		},
 		{
 			"name": "暂停RTP超时检查(pauseRtpCheck)",
 			"request": {

server/WebApi.cpp

@@ -1182,6 +1182,18 @@ void installWebApi() {
         val["hit"] = 1;
     });
 
+    api_regist("/index/api/updateRtpServerSSRC",[](API_ARGS_MAP){
+        CHECK_SECRET();
+        CHECK_ARGS("stream_id", "ssrc");
+
+        lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
+        auto it = s_rtpServerMap.find(allArgs["stream_id"]);
+        if (it == s_rtpServerMap.end()) {
+            throw ApiRetException("RtpServer not found by stream_id", API::NotFound);
+        }
+        it->second->updateSSRC(allArgs["ssrc"]);
+    });
+
     api_regist("/index/api/listRtpServer",[](API_ARGS_MAP){
         CHECK_SECRET();
 

src/Rtp/RtpServer.cpp

@@ -102,7 +102,8 @@ public:
                     process->setOnDetach(std::move(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);
+                    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;
@@ -198,11 +199,14 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
         helper->startRtcp();
         helper->setRtpServerInfo(local_port, tcp_mode, re_use_port, ssrc, only_audio);
         bool bind_peer_addr = false;
-        rtp_socket->setOnRead([rtp_socket, helper, ssrc, bind_peer_addr](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) mutable {
+        auto ssrc_ptr = std::make_shared<uint32_t>(ssrc);
+        _ssrc = ssrc_ptr;
+        rtp_socket->setOnRead([rtp_socket, helper, ssrc_ptr, bind_peer_addr](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) mutable {
             RtpHeader *header = (RtpHeader *)buf->data();
             auto rtp_ssrc = ntohl(header->ssrc);
+            auto ssrc = *ssrc_ptr;
             if (ssrc && rtp_ssrc != ssrc) {
-                WarnL << "ssrc不匹配,rtp已丢弃:" << rtp_ssrc << " != " << ssrc;
+                WarnL << "ssrc mismatched, rtp dropped: " << rtp_ssrc << " != " << ssrc;
             } else {
                 if (!bind_peer_addr) {
                     //绑定对方ip+端口，防止多个设备或一个设备多次推流从而日志报ssrc不匹配问题
@@ -213,19 +217,11 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
             }
         });
     } else {
-#if 1
         //单端口多线程接收多个流，根据ssrc区分流
         udp_server = std::make_shared<UdpServer>(rtp_socket->getPoller());
         (*udp_server)[RtpSession::kOnlyAudio] = only_audio;
         udp_server->start<RtpSession>(local_port, local_ip);
         rtp_socket = nullptr;
-#else
-        //单端口单线程接收多个流
-        auto &ref = RtpSelector::Instance();
-        rtp_socket->setOnRead([&ref, rtp_socket](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
-            ref.inputRtp(rtp_socket, buf->data(), buf->size(), addr);
-        });
-#endif
     }
 
     _on_cleanup = [rtp_socket, stream_id]() {
@@ -288,5 +284,15 @@ void RtpServer::onConnect() {
     });
 }
 
+void RtpServer::updateSSRC(uint32_t ssrc) {
+    if (_ssrc) {
+        *_ssrc = ssrc;
+    }
+
+    if (_tcp_server) {
+        (*_tcp_server)[RtpSession::kSSRC] = ssrc;
+    }
+}
+
 }//namespace mediakit
 #endif//defined(ENABLE_RTPPROXY)

src/Rtp/RtpServer.h

@@ -64,6 +64,11 @@ public:
      */
     void setOnDetach(std::function<void()> cb);
 
+    /**
+     * 更新ssrc
+     */
+    void updateSSRC(uint32_t ssrc);
+
 private:
     // tcp主动模式连接服务器成功回调
     void onConnect();
@@ -72,6 +77,7 @@ protected:
     toolkit::Socket::Ptr _rtp_socket;
     toolkit::UdpServer::Ptr _udp_server;
     toolkit::TcpServer::Ptr _tcp_server;
+    std::shared_ptr<uint32_t> _ssrc;
     std::shared_ptr<RtcpHelper> _rtcp_helper;
     std::function<void()> _on_cleanup;
 

src/Rtp/RtpSession.cpp

@@ -130,7 +130,7 @@ void RtpSession::onRtpPacket(const char *data, size_t len) {
         uint32_t rtp_ssrc = 0;
         RtpSelector::getSSRC(data, len, rtp_ssrc);
         if (rtp_ssrc != _ssrc) {
-            WarnP(this) << "ssrc不匹配,rtp已丢弃:" << rtp_ssrc << " != " << _ssrc;
+            WarnP(this) << "ssrc mismatched, rtp dropped: " << rtp_ssrc << " != " << _ssrc;
             return;
         }
         _process->inputRtp(false, getSock(), data, len, (struct sockaddr *)&_addr);