From 8f0ba6988b76cb4aeab1cfe6a0d62239f99f45ca Mon Sep 17 00:00:00 2001 From: xiongziliang <771730766@qq.com> Date: Fri, 17 Feb 2023 22:43:45 +0800 Subject: [PATCH] =?UTF-8?q?openRtpServer=E6=8E=A5=E5=8F=A3=E6=96=B0?= =?UTF-8?q?=E5=A2=9Eonly=5Faudio=E5=8F=82=E6=95=B0=EF=BC=8C=E4=BC=98?= =?UTF-8?q?=E5=8C=96=E8=AF=AD=E9=9F=B3=E5=AF=B9=E8=AE=B2=E5=9C=BA=E6=99=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- postman/ZLMediaKit.postman_collection.json | 6 ++++ server/WebApi.cpp | 6 ++-- src/Common/MediaSink.cpp | 40 ++++++++++++++++------ src/Common/MediaSink.h | 12 +++++++ src/Rtp/RtpProcess.cpp | 7 ++++ src/Rtp/RtpProcess.h | 7 ++++ src/Rtp/RtpServer.cpp | 11 ++++-- src/Rtp/RtpServer.h | 3 +- src/Rtp/RtpSession.cpp | 3 ++ src/Rtp/RtpSession.h | 2 ++ 10 files changed, 80 insertions(+), 17 deletions(-) diff --git a/postman/ZLMediaKit.postman_collection.json b/postman/ZLMediaKit.postman_collection.json index 7287f1ed..6bf86d54 100644 --- a/postman/ZLMediaKit.postman_collection.json +++ b/postman/ZLMediaKit.postman_collection.json @@ -1404,6 +1404,12 @@ "value": "0", "description": "是否指定收流的rtp ssrc, 十进制数字，不指定或指定0时则不过滤rtp，非必选参数", "disabled": true + }, + { + "key": "only_audio", + "value": "1", + "description": "是否为单音频track，用于语音对讲", + "disabled": true } ] } diff --git a/server/WebApi.cpp b/server/WebApi.cpp index 4c9063ce..3f5b36dd 100755 --- a/server/WebApi.cpp +++ b/server/WebApi.cpp @@ -391,7 +391,7 @@ Value makeMediaSourceJson(MediaSource &media){ } #if defined(ENABLE_RTPPROXY) -uint16_t openRtpServer(uint16_t local_port, const string &stream_id, int tcp_mode, const string &local_ip, bool re_use_port, uint32_t ssrc) { +uint16_t openRtpServer(uint16_t local_port, const string &stream_id, int tcp_mode, const string &local_ip, bool re_use_port, uint32_t ssrc, bool only_audio) { lock_guard lck(s_rtpServerMapMtx); if (s_rtpServerMap.find(stream_id) != s_rtpServerMap.end()) { //为了防止RtpProcess所有权限混乱的问题，不允许重复添加相同的stream_id @@ -399,7 +399,7 @@ uint16_t openRtpServer(uint16_t local_port, const string &stream_id, int tcp_mod } RtpServer::Ptr server = std::make_shared(); - server->start(local_port, stream_id, (RtpServer::TcpMode)tcp_mode, local_ip.c_str(), re_use_port, ssrc); + server->start(local_port, stream_id, (RtpServer::TcpMode)tcp_mode, local_ip.c_str(), re_use_port, ssrc, only_audio); server->setOnDetach([stream_id]() { //设置rtp超时移除事件 lock_guard lck(s_rtpServerMapMtx); @@ -1140,7 +1140,7 @@ void installWebApi() { tcp_mode = 1; } auto port = openRtpServer(allArgs["port"], stream_id, tcp_mode, "::", allArgs["re_use_port"].as(), - allArgs["ssrc"].as()); + allArgs["ssrc"].as(), allArgs["only_audio"].as()); if (port == 0) { throw InvalidArgsException("该stream_id已存在"); } diff --git a/src/Common/MediaSink.cpp b/src/Common/MediaSink.cpp index 5f3bf2be..e88f56ee 100644 --- a/src/Common/MediaSink.cpp +++ b/src/Common/MediaSink.cpp @@ -17,16 +17,20 @@ using namespace std; namespace mediakit{ bool MediaSink::addTrack(const Track::Ptr &track_in) { + if (_only_audio && track_in->getTrackType() != TrackAudio) { + InfoL << "Only audio enabled, track ignored: " << track_in->getCodecName(); + return false; + } if (!_enable_audio) { - //关闭音频时，加快单视频流注册速度 - _max_track_size = 1; + // 关闭音频时，加快单视频流注册速度 if (track_in->getTrackType() == TrackAudio) { - //音频被全局忽略 + // 音频被全局忽略 + InfoL << "Audio disabled, audio track ignored"; return false; } } if (_all_track_ready) { - WarnL << "all track is ready, add this track too late!"; + WarnL << "All track is ready, add track too late: " << track_in->getCodecName(); return false; } //克隆Track，只拷贝其数据，不拷贝其数据转发关系 @@ -48,7 +52,7 @@ bool MediaSink::addTrack(const Track::Ptr &track_in) { if (frame_unread.size() > kMaxUnreadyFrame) { //未就绪的的track，不能缓存太多的帧，否则可能内存溢出 frame_unread.clear(); - WarnL << "cached frame of unready track(" << frame->getCodecName() << ") is too much, now cleared"; + WarnL << "Cached frame of unready track(" << frame->getCodecName() << ") is too much, now cleared"; } //还有Track未就绪，先缓存之 frame_unread.emplace_back(Frame::getCacheAbleFrame(frame)); @@ -124,8 +128,16 @@ void MediaSink::checkTrackIfReady(){ } } -void MediaSink::addTrackCompleted(){ - _max_track_size = _track_map.size(); +void MediaSink::addTrackCompleted() { + setMaxTrackCount(_track_map.size()); +} + +void MediaSink::setMaxTrackCount(size_t i) { + if (_all_track_ready) { + WarnL << "All track is ready, set max track count ignored"; + return; + } + _max_track_size = MAX(MIN(i, 2), 1); checkTrackIfReady(); } @@ -134,14 +146,14 @@ void MediaSink::emitAllTrackReady() { return; } - DebugL << "all track ready use " << _ticker.elapsedTime() << "ms"; + DebugL << "All track ready use " << _ticker.elapsedTime() << "ms"; if (!_track_ready_callback.empty()) { //这是超时强制忽略未准备好的Track _track_ready_callback.clear(); //移除未准备好的Track for (auto it = _track_map.begin(); it != _track_map.end();) { if (!it->second.second || !it->second.first->ready()) { - WarnL << "track not ready for a long time, ignored: " << it->second.first->getCodecName(); + WarnL << "Track not ready for a long time, ignored: " << it->second.first->getCodecName(); it = _track_map.erase(it); continue; } @@ -256,7 +268,7 @@ bool MediaSink::addMuteAudioTrack() { return audio->inputFrame(frame); }); onTrackReady(audio); - TraceL << "mute aac track added"; + TraceL << "Mute aac track added"; return true; } @@ -266,6 +278,14 @@ bool MediaSink::isAllTrackReady() const { void MediaSink::enableAudio(bool flag) { _enable_audio = flag; + _max_track_size = flag ? 2 : 1; +} + +void MediaSink::setOnlyAudio(){ + _only_audio = true; + _enable_audio = true; + _add_mute_audio = false; + _max_track_size = 1; } void MediaSink::enableMuteAudio(bool flag) { diff --git a/src/Common/MediaSink.h b/src/Common/MediaSink.h index fd8206b9..88420593 100644 --- a/src/Common/MediaSink.h +++ b/src/Common/MediaSink.h @@ -94,6 +94,12 @@ public: */ void addTrackCompleted() override; + /** + * 设置最大track数，取值范围1~2；该方法与addTrackCompleted类型； + * 在设置单track时，可以加快媒体注册速度 + */ + void setMaxTrackCount(size_t i); + /** * 重置track */ @@ -115,6 +121,11 @@ public: */ void enableAudio(bool flag); + /** + * 设置单音频 + */ + void setOnlyAudio(); + /** * 设置是否开启添加静音音频 */ @@ -157,6 +168,7 @@ private: private: bool _enable_audio = true; + bool _only_audio = false; bool _add_mute_audio = true; bool _all_track_ready = false; size_t _max_track_size = 2; diff --git a/src/Rtp/RtpProcess.cpp b/src/Rtp/RtpProcess.cpp index 868e27e9..0673931e 100644 --- a/src/Rtp/RtpProcess.cpp +++ b/src/Rtp/RtpProcess.cpp @@ -187,6 +187,10 @@ void RtpProcess::setStopCheckRtp(bool is_check){ } } +void RtpProcess::setOnlyAudio(bool only_audio){ + _only_audio = only_audio; +} + void RtpProcess::onDetach() { if (_on_detach) { _on_detach(); @@ -247,6 +251,9 @@ void RtpProcess::emitOnPublish() { strong_self->_media_info._app, strong_self->_media_info._streamid,0.0f, option); + if (strong_self->_only_audio) { + strong_self->_muxer->setOnlyAudio(); + } strong_self->_muxer->setMediaListener(strong_self); strong_self->doCachedFunc(); InfoP(strong_self) << "允许RTP推流"; diff --git a/src/Rtp/RtpProcess.h b/src/Rtp/RtpProcess.h index d6d04fca..860be01c 100644 --- a/src/Rtp/RtpProcess.h +++ b/src/Rtp/RtpProcess.h @@ -57,6 +57,12 @@ public: */ void setStopCheckRtp(bool is_check=false); + /** + * 设置为单track，单音频时可以加快媒体注册速度 + * 请在inputRtp前调用此方法，否则可能会是空操作 + */ + void setOnlyAudio(bool only_audio); + /** * flush输出缓存 */ @@ -87,6 +93,7 @@ private: void doCachedFunc(); private: + bool _only_audio = false; uint64_t _dts = 0; uint64_t _total_bytes = 0; std::unique_ptr _addr; diff --git a/src/Rtp/RtpServer.cpp b/src/Rtp/RtpServer.cpp index fc256110..9d276c77 100644 --- a/src/Rtp/RtpServer.cpp +++ b/src/Rtp/RtpServer.cpp @@ -42,11 +42,12 @@ public: } } - void setRtpServerInfo(uint16_t local_port,RtpServer::TcpMode mode,bool re_use_port,uint32_t ssrc){ + void setRtpServerInfo(uint16_t local_port,RtpServer::TcpMode mode,bool re_use_port,uint32_t ssrc, bool only_audio) { _local_port = local_port; _tcp_mode = mode; _re_use_port = re_use_port; _ssrc = ssrc; + _only_audio = only_audio; } void setOnDetach(function cb) { @@ -60,6 +61,7 @@ public: void onRecvRtp(const Socket::Ptr &sock, const Buffer::Ptr &buf, struct sockaddr *addr) { if (!_process) { _process = RtpSelector::Instance().getProcess(_stream_id, true); + _process->setOnlyAudio(_only_audio); _process->setOnDetach(std::move(_on_detach)); cancelDelayTask(); } @@ -137,6 +139,7 @@ private: private: bool _re_use_port = false; + bool _only_audio = false; uint16_t _local_port = 0; uint32_t _ssrc = 0; RtpServer::TcpMode _tcp_mode = RtpServer::NONE; @@ -150,7 +153,7 @@ private: EventPoller::DelayTask::Ptr _delay_task; }; -void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_mode, const char *local_ip, bool re_use_port, uint32_t ssrc) { +void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_mode, const char *local_ip, bool re_use_port, uint32_t ssrc, bool only_audio) { //创建udp服务器 Socket::Ptr rtp_socket = Socket::createSocket(nullptr, true); Socket::Ptr rtcp_socket = Socket::createSocket(nullptr, true); @@ -176,6 +179,7 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_ tcp_server = std::make_shared(rtp_socket->getPoller()); (*tcp_server)[RtpSession::kStreamID] = stream_id; (*tcp_server)[RtpSession::kSSRC] = ssrc; + (*tcp_server)[RtpSession::kOnlyAudio] = only_audio; if (tcp_mode == PASSIVE) { tcp_server->start(rtp_socket->get_local_port(), local_ip); } else if (stream_id.empty()) { @@ -191,7 +195,7 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_ //指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流) helper = std::make_shared(std::move(rtcp_socket), stream_id); helper->startRtcp(); - helper->setRtpServerInfo(local_port,tcp_mode,re_use_port,ssrc); + 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 { RtpHeader *header = (RtpHeader *)buf->data(); @@ -211,6 +215,7 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_ #if 1 //单端口多线程接收多个流，根据ssrc区分流 udp_server = std::make_shared(rtp_socket->getPoller()); + (*udp_server)[RtpSession::kOnlyAudio] = only_audio; udp_server->start(rtp_socket->get_local_port(), local_ip); rtp_socket = nullptr; #else diff --git a/src/Rtp/RtpServer.h b/src/Rtp/RtpServer.h index 409671a1..c8033416 100644 --- a/src/Rtp/RtpServer.h +++ b/src/Rtp/RtpServer.h @@ -44,7 +44,7 @@ public: * @param ssrc 指定的ssrc */ void start(uint16_t local_port, const std::string &stream_id = "", TcpMode tcp_mode = PASSIVE, - const char *local_ip = "::", bool re_use_port = true, uint32_t ssrc = 0); + const char *local_ip = "::", bool re_use_port = true, uint32_t ssrc = 0, bool only_audio = false); /** * 连接到tcp服务(tcp主动模式) @@ -75,6 +75,7 @@ protected: std::shared_ptr _rtcp_helper; std::function _on_cleanup; + bool _only_audio = false; //用于tcp主动模式 TcpMode _tcp_mode = NONE; }; diff --git a/src/Rtp/RtpSession.cpp b/src/Rtp/RtpSession.cpp index 190dff5f..f99734c8 100644 --- a/src/Rtp/RtpSession.cpp +++ b/src/Rtp/RtpSession.cpp @@ -22,6 +22,7 @@ namespace mediakit{ const string RtpSession::kStreamID = "stream_id"; const string RtpSession::kSSRC = "ssrc"; +const string RtpSession::kOnlyAudio = "only_audio"; void RtpSession::attachServer(const Server &server) { setParams(const_cast(server)); @@ -30,6 +31,7 @@ void RtpSession::attachServer(const Server &server) { void RtpSession::setParams(mINI &ini) { _stream_id = ini[kStreamID]; _ssrc = ini[kSSRC]; + _only_audio = ini[kOnlyAudio]; } RtpSession::RtpSession(const Socket::Ptr &sock) : Session(sock) { @@ -101,6 +103,7 @@ void RtpSession::onRtpPacket(const char *data, size_t len) { } //tcp情况下，一个tcp链接只可能是一路流，不需要通过多个ssrc来区分，所以不需要频繁getProcess _process = RtpSelector::Instance().getProcess(_stream_id, true); + _process->setOnlyAudio(_only_audio); _process->setDelegate(dynamic_pointer_cast(shared_from_this())); } try { diff --git a/src/Rtp/RtpSession.h b/src/Rtp/RtpSession.h index 14cd33dd..586e2df2 100644 --- a/src/Rtp/RtpSession.h +++ b/src/Rtp/RtpSession.h @@ -24,6 +24,7 @@ class RtpSession : public toolkit::Session, public RtpSplitter, public MediaSour public: static const std::string kStreamID; static const std::string kSSRC; + static const std::string kOnlyAudio; RtpSession(const toolkit::Socket::Ptr &sock); ~RtpSession() override; @@ -45,6 +46,7 @@ private: bool _is_udp = false; bool _search_rtp = false; bool _search_rtp_finished = false; + bool _only_audio = false; uint32_t _ssrc = 0; toolkit::Ticker _ticker; std::string _stream_id;