Merge 90bd28249e into 51f49d3a89

海康解码器播放rtsp流url带了?key=value参数，SETUP时会带着这些参数，导致获取trackid失败

postman/ZLMediaKit.postman_collection.json

@@ -2161,6 +2161,81 @@
 			},
 			"response": []
 		},
+		{
+			"name": "开始双向对讲(startSendRtpTalk)",
+			"request": {
+				"method": "GET",
+				"header": [],
+				"url": {
+					"raw": "{{ZLMediaKit_URL}}/index/api/startSendRtpTalk?secret={{ZLMediaKit_secret}}&vhost={{defaultVhost}}&app=live&stream=obs&ssrc=1&recv_stream_id=",
+					"host": [
+						"{{ZLMediaKit_URL}}"
+					],
+					"path": [
+						"index",
+						"api",
+						"startSendRtpTalk"
+					],
+					"query": [
+						{
+							"key": "secret",
+							"value": "{{ZLMediaKit_secret}}",
+							"description": "api操作密钥(配置文件配置)"
+						},
+						{
+							"key": "vhost",
+							"value": "{{defaultVhost}}",
+							"description": "虚拟主机，例如__defaultVhost__"
+						},
+						{
+							"key": "app",
+							"value": "rtp",
+							"description": "应用名，例如 rtp"
+						},
+						{
+							"key": "stream",
+							"value": "rtc",
+							"description": "流id，例如webrtc推流上来的流id"
+						},
+						{
+							"key": "ssrc",
+							"value": "1",
+							"description": "rtp推流出去的ssrc"
+						},
+						{
+							"key": "recv_stream_id",
+							"value": "",
+							"description": "对方rtp推流上来的流id，我们将通过这个链接回复他rtp流；请注意两个流的app和vhost需一致"
+						},
+						{
+							"key": "from_mp4",
+							"value": "0",
+							"description": "是否推送本地MP4录像，该参数非必选参数",
+							"disabled": true
+						},
+						{
+							"key": "type",
+							"value": "1",
+							"description": "0(ES流)、1(PS流)、2(TS流)，默认1(PS流)；该参数非必选参数",
+							"disabled": true
+						},
+						{
+							"key": "pt",
+							"value": "96",
+							"description": "rtp payload type，默认96；该参数非必选参数",
+							"disabled": true
+						},
+						{
+							"key": "only_audio",
+							"value": "1",
+							"description": "rtp es方式打包时，是否只打包音频；该参数非必选参数",
+							"disabled": true
+						}
+					]
+				}
+			},
+			"response": []
+		},
 		{
 			"name": "停止 发送rtp(stopSendRtp)",
 			"request": {

server/WebApi.cpp

@@ -1606,6 +1606,41 @@ void installWebApi() {
         start_send_rtp(true, API_ARGS_VALUE, invoker);
     });
 
+    api_regist("/index/api/startSendRtpTalk",[](API_ARGS_MAP_ASYNC){
+        CHECK_SECRET();
+        CHECK_ARGS("vhost", "app", "stream", "ssrc", "recv_stream_id");
+        auto src = MediaSource::find(allArgs["vhost"], allArgs["app"], allArgs["stream"], allArgs["from_mp4"].as<int>());
+        if (!src) {
+            throw ApiRetException("can not find the source stream", API::NotFound);
+        }
+        MediaSourceEvent::SendRtpArgs args;
+        args.con_type = mediakit::MediaSourceEvent::SendRtpArgs::kVoiceTalk;
+        args.ssrc = allArgs["ssrc"];
+        args.pt = allArgs["pt"].empty() ? 96 : allArgs["pt"].as<int>();
+        args.data_type = allArgs["type"].empty() ? MediaSourceEvent::SendRtpArgs::kRtpPS : (MediaSourceEvent::SendRtpArgs::DataType)(allArgs["type"].as<int>());
+        args.only_audio = allArgs["only_audio"].as<bool>();
+        args.recv_stream_id = allArgs["recv_stream_id"];
+        args.recv_stream_app = allArgs["app"];
+        args.recv_stream_vhost = allArgs["vhost"];
+
+        src->getOwnerPoller()->async([=]() mutable {
+            try {
+                src->startSendRtp(args, [val, headerOut, invoker](uint16_t local_port, const SockException &ex) mutable {
+                    if (ex) {
+                        val["code"] = API::OtherFailed;
+                        val["msg"] = ex.what();
+                    }
+                    val["local_port"] = local_port;
+                    invoker(200, headerOut, val.toStyledString());
+                });
+            } catch (std::exception &ex) {
+                val["code"] = API::Exception;
+                val["msg"] = ex.what();
+                invoker(200, headerOut, val.toStyledString());
+            }
+        });
+    });
+
     api_regist("/index/api/listRtpSender",[](API_ARGS_MAP_ASYNC){
         CHECK_SECRET();
         CHECK_ARGS("vhost", "app", "stream");

src/Common/MediaSink.cpp

@@ -114,12 +114,14 @@ void MediaSink::checkTrackIfReady() {
         }
     }
 
-    // 等待音频超时时间
+    // 等待音频超时时间  [AUTO-TRANSLATED:5ec16b26]
+    // Wait for audio timeout time
     GET_CONFIG(uint32_t, kWaitAudioTrackDataMS, General::kWaitAudioTrackDataMS);
     if (_max_track_size > 1) {
         for (auto it = _track_map.begin(); it != _track_map.end();) {
             if (it->second.first->getTrackType() == TrackAudio && _ticker.elapsedTime() > kWaitAudioTrackDataMS && !it->second.second) {
-                // 音频超时且完全没收到音频数据，忽略音频
+                // 音频超时且完全没收到音频数据，忽略音频  [AUTO-TRANSLATED:0d0fbb13]
+                // Audio timeout and did not receive any audio data, ignore audio
                 auto index = it->second.first->getIndex();
                 WarnL << "Audio track index " << index << " codec " << it->second.first->getCodecName() << " receive no data for long "
                       << _ticker.elapsedTime() << "ms. Ignore it!";

src/Common/MediaSource.h

@@ -124,7 +124,8 @@ public:
             kTcpActive = 0, // tcp主动模式，tcp客户端主动连接对方并发送rtp
             kUdpActive = 1, // udp主动模式，主动发送数据给对方
             kTcpPassive = 2, // tcp被动模式，tcp服务器，等待对方连接并回复rtp
-            kUdpPassive = 3 // udp被动方式，等待对方发送nat打洞包，然后回复rtp至打洞包源地址
+            kUdpPassive = 3, // udp被动方式，等待对方发送nat打洞包，然后回复rtp至打洞包源地址
+            kVoiceTalk = 4,  // 语音对讲模式，对方必须想推流上来，通过他的推流链路再回复rtp数据
         };
 
         // rtp类型  [AUTO-TRANSLATED:acca40ab]

src/Record/MP4Recorder.cpp

@@ -123,11 +123,15 @@ bool MP4Recorder::inputFrame(const Frame::Ptr &frame) {
     if (!(_have_video && frame->getTrackType() == TrackAudio)) {
         // 如果有视频且输入的是音频，那么应该忽略切片逻辑  [AUTO-TRANSLATED:fbb15d93]
         // If there is video and the input is audio, then the slice logic should be ignored
-        if (_last_dts == 0 || _last_dts > frame->dts()) {
+        if (_last_dts == 0) {
+            // first frame assign dts
+            _last_dts = frame->dts();
+        } else if (_last_dts > frame->dts()) {
             // b帧情况下dts时间戳可能回退  [AUTO-TRANSLATED:1de38f77]
             // In the case of b-frames, the dts timestamp may regress
             _last_dts = MIN(frame->dts(), _last_dts);
         }
+        
         auto duration = 5u; // 默认至少一帧5ms
         if (frame->dts() > 0 && frame->dts() > _last_dts) {
             duration = MAX(duration, frame->dts() - _last_dts);

src/Rtp/RtpProcess.cpp

@@ -346,5 +346,9 @@ float RtpProcess::getLossRate(MediaSource &sender, TrackType type) {
     return getLostInterval() * 100 / expected;
 }
 
+const toolkit::Socket::Ptr& RtpProcess::getSock() const {
+    return _sock;
+}
+
 }//namespace mediakit
 #endif//defined(ENABLE_RTPPROXY)

src/Rtp/RtpProcess.h

@@ -102,6 +102,8 @@ public:
     uint16_t get_peer_port() override;
     std::string getIdentifier() const override;
 
+    const toolkit::Socket::Ptr& getSock() const;
+
 protected:
     bool inputFrame(const Frame::Ptr &frame) override;
     bool addTrack(const Track::Ptr & track) override;

src/Rtp/RtpSender.cpp

@@ -190,6 +190,25 @@ void RtpSender::startSend(const MediaSourceEvent::SendRtpArgs &args, const funct
             }
         }, delay_ms / 1000.0, "::", args.src_port);
         InfoL << "start tcp active send rtp to: " << args.dst_url << ":" << args.dst_port;
+    } else if (args.con_type == MediaSourceEvent::SendRtpArgs::kVoiceTalk) {
+        auto src = MediaSource::find(args.recv_stream_vhost, args.recv_stream_app, args.recv_stream_id);
+        if (!src) {
+            cb(0, SockException(Err_other, "can not find the target stream"));
+            return;
+        }
+        auto processor = src->getRtpProcess();
+        if (!processor) {
+            cb(0, SockException(Err_other, "get rtp processor from target stream failed"));
+            return;
+        }
+        auto sock = processor->getSock();
+        if (!sock) {
+            cb(0, SockException(Err_other, "get sock from rtp processor failed"));
+            return;
+        }
+        _socket_rtp = std::move(sock);
+        onConnect();
+        cb(_socket_rtp->get_local_port(), SockException());
     } else {
         CHECK(0, "invalid con type");
     }
@@ -249,6 +268,8 @@ void RtpSender::onConnect() {
     // 加大发送缓存,防止udp丢包之类的问题  [AUTO-TRANSLATED:6e1cb40a]
     // Increase the send buffer to prevent problems such as UDP packet loss
     SockUtil::setSendBuf(_socket_rtp->rawFD(), 4 * 1024 * 1024);
+
+    if (_args.con_type != MediaSourceEvent::SendRtpArgs::kVoiceTalk) {
         if (_args.con_type == MediaSourceEvent::SendRtpArgs::kTcpActive || _args.con_type == MediaSourceEvent::SendRtpArgs::kTcpPassive) {
             // 关闭tcp no_delay并开启MSG_MORE, 提高发送性能  [AUTO-TRANSLATED:c0f4e378]
             // Close TCP no_delay and enable MSG_MORE to improve sending performance
@@ -291,6 +312,7 @@ void RtpSender::onConnect() {
                 strong_self->onErr(err);
             }
         });
+    }
     InfoL << "startSend rtp success: " << _socket_rtp->get_peer_ip() << ":" << _socket_rtp->get_peer_port() << ", data_type: " << _args.data_type << ", con_type: " << _args.con_type;
 }
 
@@ -378,6 +400,7 @@ void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr>> rtp_list) {
         return;
     }
 
+    auto send_func = [this](const shared_ptr<List<Buffer::Ptr>> &rtp_list) {
         size_t i = 0;
         auto size = rtp_list->size();
         rtp_list->for_each([&](Buffer::Ptr &packet) {
@@ -397,9 +420,31 @@ void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr>> rtp_list) {
                     _socket_rtp->send(std::make_shared<BufferRtp>(std::move(packet), 2), nullptr, 0, ++i == size);
                     break;
                 }
+                case MediaSourceEvent::SendRtpArgs::kVoiceTalk: {
+                    auto type = _socket_rtp->alive() ? _socket_rtp->sockType() : SockNum::Sock_Invalid;
+                    if (type == SockNum::Sock_UDP) {
+                        _socket_rtp->send(std::make_shared<BufferRtp>(std::move(packet), RtpPacket::kRtpTcpHeaderSize), nullptr, 0, ++i == size);
+                    } else if (type == SockNum::Sock_TCP) {
+                        _socket_rtp->send(std::make_shared<BufferRtp>(std::move(packet), 2), nullptr, 0, ++i == size);
+                    } else {
+                        onErr(SockException(Err_other, "dst socket disconnected"));
+                    }
+                    break;
+                }
                 default: CHECK(0);
             }
         });
+    };
+    if (_args.con_type != MediaSourceEvent::SendRtpArgs::kVoiceTalk) {
+        weak_ptr<RtpSender> weak_self = shared_from_this();
+        _socket_rtp->getPoller()->async([weak_self, rtp_list, send_func]() {
+            if (auto strong_self = weak_self.lock()) {
+                send_func(rtp_list);
+            }
+        });
+    } else {
+        send_func(rtp_list);
+    }
 }
 
 void RtpSender::onErr(const SockException &ex) {

src/Rtsp/RtspSession.cpp

@@ -1136,7 +1136,7 @@ int RtspSession::getTrackIndexByTrackType(TrackType type) {
 
 int RtspSession::getTrackIndexByControlUrl(const string &control_url) {
     for (size_t i = 0; i < _sdp_track.size(); ++i) {
-        if (control_url == _sdp_track[i]->getControlUrl(_content_base)) {
+        if (control_url.find(_sdp_track[i]->getControlUrl(_content_base)) == 0) {
             return i;
         }
     }