startSendRtp接口支持同时接收流：#2109,#2149

postman/ZLMediaKit.postman_collection.json

@@ -1661,6 +1661,12 @@
 							"value": "0",
 							"value": "0",
 							"description": "udp方式推流时，是否开启rtcp发送和rtcp接收超时判断，开启后(默认关闭)，如果接收rr rtcp超时，将导致主动停止rtp发送",
 							"description": "udp方式推流时，是否开启rtcp发送和rtcp接收超时判断，开启后(默认关闭)，如果接收rr rtcp超时，将导致主动停止rtp发送",
 							"disabled": true
 							"disabled": true
+						},
+						{
+							"key": "recv_stream_id",
+							"value": "",
+							"description": "发送rtp同时接收，一般用于双向语言对讲, 如果不为空，说明开启接收，值为接收流的id",
+							"disabled": true
 						}
 						}
 					]
 					]
 				}
 				}

server/WebApi.cpp

@@ -1203,6 +1203,7 @@ void installWebApi() {
         args.use_ps = allArgs["use_ps"].empty() ? true : allArgs["use_ps"].as<bool>();
         args.use_ps = allArgs["use_ps"].empty() ? true : allArgs["use_ps"].as<bool>();
         args.only_audio = allArgs["only_audio"].as<bool>();
         args.only_audio = allArgs["only_audio"].as<bool>();
         args.udp_rtcp_timeout = allArgs["udp_rtcp_timeout"];
         args.udp_rtcp_timeout = allArgs["udp_rtcp_timeout"];
+        args.recv_stream_id = allArgs["recv_stream_id"];
         TraceL << "startSendRtp, pt " << int(args.pt) << " ps " << args.use_ps << " audio " << args.only_audio;
         TraceL << "startSendRtp, pt " << int(args.pt) << " ps " << args.use_ps << " audio " << args.only_audio;
 
 
         src->getOwnerPoller()->async([=]() mutable {
         src->getOwnerPoller()->async([=]() mutable {

src/Common/MediaSource.h

@@ -118,6 +118,9 @@ public:
         uint32_t rtcp_timeout_ms = 30 * 1000;
         uint32_t rtcp_timeout_ms = 30 * 1000;
         //udp 发送时，发送sr rtcp包间隔，单位毫秒
         //udp 发送时，发送sr rtcp包间隔，单位毫秒
         uint32_t rtcp_send_interval_ms = 5 * 1000;
         uint32_t rtcp_send_interval_ms = 5 * 1000;
+
+        //发送rtp同时接收，一般用于双向语言对讲, 如果不为空，说明开启接收
+        std::string recv_stream_id;
     };
     };
 
 
     // 开始发送ps-rtp
     // 开始发送ps-rtp

src/Rtp/RtpSender.cpp

@@ -10,6 +10,7 @@
 
 
 #if defined(ENABLE_RTPPROXY)
 #if defined(ENABLE_RTPPROXY)
 #include "RtpSender.h"
 #include "RtpSender.h"
+#include "RtpSession.h"
 #include "Rtsp/RtspSession.h"
 #include "Rtsp/RtspSession.h"
 #include "Thread/WorkThreadPool.h"
 #include "Thread/WorkThreadPool.h"
 #include "Util/uv_errno.h"
 #include "Util/uv_errno.h"
@@ -213,6 +214,25 @@ void RtpSender::onConnect(){
     }
     }
     //连接建立成功事件
     //连接建立成功事件
     weak_ptr<RtpSender> weak_self = shared_from_this();
     weak_ptr<RtpSender> weak_self = shared_from_this();
+    if (!_args.recv_stream_id.empty()) {
+        mINI ini;
+        ini[RtpSession::kStreamID] = _args.recv_stream_id;
+        _rtp_session = std::make_shared<RtpSession>(_socket_rtp);
+        _rtp_session->setParams(ini);
+
+        _socket_rtp->setOnRead([weak_self](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) {
+            auto strong_self = weak_self.lock();
+            if (!strong_self) {
+                return;
+            }
+            try {
+                strong_self->_rtp_session->onRecv(buf);
+            } catch (std::exception &ex){
+                SockException err(toolkit::Err_shutdown, ex.what());
+                strong_self->_rtp_session->shutdown(err);
+            }
+        });
+    }
     _socket_rtp->setOnErr([weak_self](const SockException &err) {
     _socket_rtp->setOnErr([weak_self](const SockException &err) {
         auto strong_self = weak_self.lock();
         auto strong_self = weak_self.lock();
         if (strong_self) {
         if (strong_self) {

src/Rtp/RtpSender.h

@@ -19,6 +19,8 @@
 
 
 namespace mediakit{
 namespace mediakit{
 
 
+class RtpSession;
+
 //rtp发送客户端，支持发送GB28181协议
 //rtp发送客户端，支持发送GB28181协议
 class RtpSender final : public MediaSinkInterface, public std::enable_shared_from_this<RtpSender>{
 class RtpSender final : public MediaSinkInterface, public std::enable_shared_from_this<RtpSender>{
 public:
 public:
@@ -85,6 +87,7 @@ private:
     std::shared_ptr<RtcpContext> _rtcp_context;
     std::shared_ptr<RtcpContext> _rtcp_context;
     toolkit::Ticker _rtcp_send_ticker;
     toolkit::Ticker _rtcp_send_ticker;
     toolkit::Ticker _rtcp_recv_ticker;
     toolkit::Ticker _rtcp_recv_ticker;
+    std::shared_ptr<RtpSession> _rtp_session;
     std::function<void(const toolkit::SockException &ex)> _on_close;
     std::function<void(const toolkit::SockException &ex)> _on_close;
 };
 };
 
 

src/Rtp/RtpSession.cpp

@@ -24,8 +24,12 @@ const string RtpSession::kStreamID = "stream_id";
 const string RtpSession::kSSRC = "ssrc";
 const string RtpSession::kSSRC = "ssrc";
 
 
 void RtpSession::attachServer(const Server &server) {
 void RtpSession::attachServer(const Server &server) {
-    _stream_id = const_cast<Server &>(server)[kStreamID];
-    _ssrc = const_cast<Server &>(server)[kSSRC];
+    setParams(const_cast<Server &>(server));
+}
+
+void RtpSession::setParams(mINI &ini) {
+    _stream_id = ini[kStreamID];
+    _ssrc = ini[kSSRC];
 }
 }
 
 
 RtpSession::RtpSession(const Socket::Ptr &sock) : Session(sock) {
 RtpSession::RtpSession(const Socket::Ptr &sock) : Session(sock) {

src/Rtp/RtpSession.h

@@ -30,6 +30,7 @@ public:
     void onRecv(const toolkit::Buffer::Ptr &) override;
     void onRecv(const toolkit::Buffer::Ptr &) override;
     void onError(const toolkit::SockException &err) override;
     void onError(const toolkit::SockException &err) override;
     void onManager() override;
     void onManager() override;
+    void setParams(toolkit::mINI &ini);
     void attachServer(const toolkit::Server &server) override;
     void attachServer(const toolkit::Server &server) override;
 
 
 protected:
 protected: