From 03770ff4094005e601c83b59b8e5741423b002d6 Mon Sep 17 00:00:00 2001 From: johzzy Date: Sat, 17 Jun 2023 10:28:01 +0800 Subject: [PATCH] Replace the tuple (vhost/app/stream) with MediaTuple. (#2560) --- api/source/mk_events_objects.cpp | 6 +++--- server/FFmpegSource.cpp | 4 +--- server/WebApi.cpp | 10 +++++++--- server/WebHook.cpp | 22 +++++++--------------- src/Common/MediaSource.cpp | 26 +++++--------------------- src/Common/MediaSource.h | 16 ++++++---------- src/Common/MultiMediaSourceMuxer.cpp | 12 ------------ src/Common/MultiMediaSourceMuxer.h | 3 --- tests/test_pusher.cpp | 5 +++-- tests/test_server.cpp | 5 +++-- webrtc/WebRtcPusher.cpp | 4 ++-- 11 files changed, 37 insertions(+), 76 deletions(-) diff --git a/api/source/mk_events_objects.cpp b/api/source/mk_events_objects.cpp index c8656cec..fd7f696c 100644 --- a/api/source/mk_events_objects.cpp +++ b/api/source/mk_events_objects.cpp @@ -174,17 +174,17 @@ API_EXPORT const char* API_CALL mk_media_source_get_schema(const mk_media_source API_EXPORT const char* API_CALL mk_media_source_get_vhost(const mk_media_source ctx){ assert(ctx); MediaSource *src = (MediaSource *)ctx; - return src->getVhost().c_str(); + return src->getMediaTuple().vhost.c_str(); } API_EXPORT const char* API_CALL mk_media_source_get_app(const mk_media_source ctx){ assert(ctx); MediaSource *src = (MediaSource *)ctx; - return src->getApp().c_str(); + return src->getMediaTuple().app.c_str(); } API_EXPORT const char* API_CALL mk_media_source_get_stream(const mk_media_source ctx){ assert(ctx); MediaSource *src = (MediaSource *)ctx; - return src->getId().c_str(); + return src->getMediaTuple().stream.c_str(); } API_EXPORT int API_CALL mk_media_source_get_reader_count(const mk_media_source ctx){ assert(ctx); diff --git a/server/FFmpegSource.cpp b/server/FFmpegSource.cpp index 5e7fc246..564dca46 100644 --- a/server/FFmpegSource.cpp +++ b/server/FFmpegSource.cpp @@ -184,9 +184,7 @@ void FFmpegSource::findAsync(int maxWaitMS, const function_media_info.schema || - sender.getVhost() != strongSelf->_media_info.vhost || - sender.getApp() != strongSelf->_media_info.app || - sender.getId() != strongSelf->_media_info.stream) { + !equalMediaTuple(sender.getMediaTuple(), strongSelf->_media_info)) { //不是自己感兴趣的事件，忽略之 return; } diff --git a/server/WebApi.cpp b/server/WebApi.cpp index 88e4d4ea..2fe75f73 100755 --- a/server/WebApi.cpp +++ b/server/WebApi.cpp @@ -321,12 +321,16 @@ static void fillSockInfo(Value& val, SockInfo* info) { val["identifier"] = info->getIdentifier(); } +void dumpMediaTuple(const MediaTuple &tuple, Json::Value& item) { + item[VHOST_KEY] = tuple.vhost; + item["app"] = tuple.app; + item["stream"] = tuple.stream; +} + Value makeMediaSourceJson(MediaSource &media){ Value item; item["schema"] = media.getSchema(); - item[VHOST_KEY] = media.getVhost(); - item["app"] = media.getApp(); - item["stream"] = media.getId(); + dumpMediaTuple(media.getMediaTuple(), item); item["createStamp"] = (Json::UInt64) media.getCreateStamp(); item["aliveSecond"] = (Json::UInt64) media.getAliveSecond(); item["bytesSpeed"] = media.getBytesSpeed(); diff --git a/server/WebHook.cpp b/server/WebHook.cpp index d50b1db9..551d9c1d 100755 --- a/server/WebHook.cpp +++ b/server/WebHook.cpp @@ -213,12 +213,12 @@ void do_http_hook(const string &url, const ArgsType &body, const function weakSrc = sender.shared_from_this(); // 执行hook do_http_hook(hook_stream_none_reader, body, [weakSrc](const Value &obj, const string &err) { @@ -584,9 +578,7 @@ void installWebHook() { } ArgsType body; - body[VHOST_KEY] = sender.getVhost(); - body["app"] = sender.getApp(); - body["stream"] = sender.getStreamId(); + dumpMediaTuple(sender.getMediaTuple(), body); body["ssrc"] = ssrc; body["originType"] = (int)sender.getOriginType(MediaSource::NullMediaSource()); body["originTypeStr"] = getOriginTypeString(sender.getOriginType(MediaSource::NullMediaSource())); diff --git a/src/Common/MediaSource.cpp b/src/Common/MediaSource.cpp index 6f3a62e4..7c48c3e1 100644 --- a/src/Common/MediaSource.cpp +++ b/src/Common/MediaSource.cpp @@ -131,23 +131,6 @@ MediaSource::~MediaSource() { } } -const string& MediaSource::getSchema() const { - return _schema; -} - -const string& MediaSource::getVhost() const { - return _tuple.vhost; -} - -const string& MediaSource::getApp() const { - //获取该源的id - return _tuple.app; -} - -const string& MediaSource::getId() const { - return _tuple.stream; -} - std::shared_ptr MediaSource::getOwnership() { if (_owned.test_and_set()) { //已经被所有 @@ -460,9 +443,7 @@ static void findAsync_l(const MediaInfo &info, const std::shared_ptr &s auto on_register = [weak_session, info, cb_once, cancel_all, poller](BroadcastMediaChangedArgs) { if (!bRegist || sender.getSchema() != info.schema || - sender.getVhost() != info.vhost || - sender.getApp() != info.app || - sender.getId() != info.stream) { + !equalMediaTuple(sender.getMediaTuple(), info)) { //不是自己感兴趣的事件，忽略之 return; } @@ -579,6 +560,9 @@ bool MediaSource::unregist() { return ret; } +bool equalMediaTuple(const MediaTuple& a, const MediaTuple& b) { + return a.vhost == b.vhost && a.app == b.app && a.stream == b.stream; +} /////////////////////////////////////MediaInfo////////////////////////////////////// void MediaInfo::parse(const std::string &url_in){ @@ -663,7 +647,7 @@ void MediaSourceEvent::onReaderChanged(MediaSource &sender, int size){ GET_CONFIG(string, record_app, Record::kAppName); GET_CONFIG(int, stream_none_reader_delay, General::kStreamNoneReaderDelayMS); //如果mp4点播, 无人观看时我们强制关闭点播 - bool is_mp4_vod = sender.getApp() == record_app; + bool is_mp4_vod = sender.getMediaTuple().app == record_app; weak_ptr weak_sender = sender.shared_from_this(); _async_close_timer = std::make_shared(stream_none_reader_delay / 1000.0f, [weak_sender, is_mp4_vod]() { diff --git a/src/Common/MediaSource.h b/src/Common/MediaSource.h index bbe8035f..9391e6fb 100644 --- a/src/Common/MediaSource.h +++ b/src/Common/MediaSource.h @@ -274,6 +274,8 @@ public: std::string param_strs; }; +bool equalMediaTuple(const MediaTuple& a, const MediaTuple& b); + /** * 媒体源，任何rtsp/rtmp的直播流都源自该对象 */ @@ -288,21 +290,15 @@ public: ////////////////获取MediaSource相关信息//////////////// // 获取协议类型 - const std::string& getSchema() const; - // 虚拟主机 - const std::string& getVhost() const; - // 应用名 - const std::string& getApp() const; - // 流id - const std::string& getId() const; + const std::string& getSchema() const { + return _schema; + } const MediaTuple& getMediaTuple() const { return _tuple; } - std::string shortUrl() const { return _tuple.shortUrl(); } - - std::string getUrl() const { return _schema + "://" + shortUrl(); } + std::string getUrl() const { return _schema + "://" + _tuple.shortUrl(); } //获取对象所有权 std::shared_ptr getOwnership(); diff --git a/src/Common/MultiMediaSourceMuxer.cpp b/src/Common/MultiMediaSourceMuxer.cpp index a9d904be..f1921a95 100644 --- a/src/Common/MultiMediaSourceMuxer.cpp +++ b/src/Common/MultiMediaSourceMuxer.cpp @@ -70,18 +70,6 @@ static string getTrackInfoStr(const TrackSource *track_src){ return std::move(codec_info); } -const std::string &MultiMediaSourceMuxer::getVhost() const { - return _tuple.vhost; -} - -const std::string &MultiMediaSourceMuxer::getApp() const { - return _tuple.app; -} - -const std::string &MultiMediaSourceMuxer::getStreamId() const { - return _tuple.stream; -} - std::string MultiMediaSourceMuxer::shortUrl() const { auto ret = getOriginUrl(MediaSource::NullMediaSource()); if (!ret.empty()) { diff --git a/src/Common/MultiMediaSourceMuxer.h b/src/Common/MultiMediaSourceMuxer.h index 0e4f6a5b..a9983550 100644 --- a/src/Common/MultiMediaSourceMuxer.h +++ b/src/Common/MultiMediaSourceMuxer.h @@ -126,9 +126,6 @@ public: */ toolkit::EventPoller::Ptr getOwnerPoller(MediaSource &sender) override; - const std::string& getVhost() const; - const std::string& getApp() const; - const std::string& getStreamId() const; const MediaTuple& getMediaTuple() const { return _tuple; } diff --git a/tests/test_pusher.cpp b/tests/test_pusher.cpp index adf85199..8e5ca1f7 100644 --- a/tests/test_pusher.cpp +++ b/tests/test_pusher.cpp @@ -88,8 +88,9 @@ int domain(const string &playUrl, const string &pushUrl) { NoticeCenter::Instance().addListener(nullptr, Broadcast::kBroadcastMediaChanged, [pushUrl,poller](BroadcastMediaChangedArgs) { //媒体源"app/stream"已经注册，这时方可新建一个RtmpPusher对象并绑定该媒体源 - if(bRegist && pushUrl.find(sender.getSchema()) == 0){ - createPusher(poller,sender.getSchema(),sender.getVhost(),sender.getApp(), sender.getId(), pushUrl); + if (bRegist && pushUrl.find(sender.getSchema()) == 0) { + auto tuple = sender.getMediaTuple(); + createPusher(poller, sender.getSchema(), tuple.vhost, tuple.app, tuple.stream, pushUrl); } }); diff --git a/tests/test_server.cpp b/tests/test_server.cpp index 6a15c532..dc123ba2 100644 --- a/tests/test_server.cpp +++ b/tests/test_server.cpp @@ -155,9 +155,10 @@ void initEventListener() { //监听rtsp、rtmp源注册或注销事件；此处用于测试rtmp保存为flv录像，保存在http根目录下 NoticeCenter::Instance().addListener(nullptr, Broadcast::kBroadcastMediaChanged, [](BroadcastMediaChangedArgs) { - if (sender.getSchema() == RTMP_SCHEMA && sender.getApp() == "live") { + auto tuple = sender.getMediaTuple(); + if (sender.getSchema() == RTMP_SCHEMA && tuple.app == "live") { lock_guard lck(s_mtxFlvRecorder); - auto key = sender.shortUrl(); + auto key = tuple.shortUrl(); if (bRegist) { DebugL << "开始录制RTMP：" << sender.getUrl(); GET_CONFIG(string, http_root, Http::kRootPath); diff --git a/webrtc/WebRtcPusher.cpp b/webrtc/WebRtcPusher.cpp index 334236bf..e8e46a40 100644 --- a/webrtc/WebRtcPusher.cpp +++ b/webrtc/WebRtcPusher.cpp @@ -98,8 +98,8 @@ void WebRtcPusher::onRecvRtp(MediaTrack &track, const string &rid, RtpPacket::Pt //视频 auto &src = _push_src_sim[rid]; if (!src) { - auto stream_id = rid.empty() ? _push_src->getId() : _push_src->getId() + "_" + rid; - auto src_imp = _push_src->clone(stream_id); + const auto& stream = _push_src->getMediaTuple().stream; + auto src_imp = _push_src->clone(rid.empty() ? stream : stream + '_' + rid); _push_src_sim_ownership[rid] = src_imp->getOwnership(); src_imp->setListener(static_pointer_cast(shared_from_this())); src = src_imp;