Replace the tuple (vhost/app/stream) with MediaTuple. (#2560)

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);

server/FFmpegSource.cpp

@@ -184,9 +184,7 @@ void FFmpegSource::findAsync(int maxWaitMS, const function<void(const MediaSourc
 
         if (!bRegist ||
             sender.getSchema() != strongSelf->_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;
         }

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();

server/WebHook.cpp

@@ -213,12 +213,12 @@ void do_http_hook(const string &url, const ArgsType &body, const function<void(c
     do_http_hook(url, body, func, hook_retry);
 }
 
+void dumpMediaTuple(const MediaTuple &tuple, Json::Value& item);
+
 static ArgsType make_json(const MediaInfo &args) {
     ArgsType body;
     body["schema"] = args.schema;
-    body[VHOST_KEY] = args.vhost;
-    body["app"] = args.app;
-    body["stream"] = args.stream;
+    dumpMediaTuple(args, body);
     body["params"] = args.param_strs;
     return body;
 }
@@ -437,9 +437,7 @@ void installWebHook() {
             body["regist"] = bRegist;
         } else {
             body["schema"] = sender.getSchema();
-            body[VHOST_KEY] = sender.getVhost();
-            body["app"] = sender.getApp();
-            body["stream"] = sender.getId();
+            dumpMediaTuple(sender.getMediaTuple(), body);
             body["regist"] = bRegist;
         }
         // 执行hook
@@ -503,9 +501,7 @@ void installWebHook() {
         body["file_name"] = info.file_name;
         body["folder"] = info.folder;
         body["url"] = info.url;
-        body["app"] = info.app;
-        body["stream"] = info.stream;
-        body[VHOST_KEY] = info.vhost;
+        dumpMediaTuple(info, body);
         return body;
     };
 
@@ -561,9 +557,7 @@ void installWebHook() {
 
         ArgsType body;
         body["schema"] = sender.getSchema();
-        body[VHOST_KEY] = sender.getVhost();
-        body["app"] = sender.getApp();
-        body["stream"] = sender.getId();
+        dumpMediaTuple(sender.getMediaTuple(), body);
         weak_ptr<MediaSource> 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()));

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<void> MediaSource::getOwnership() {
     if (_owned.test_and_set()) {
         //已经被所有
@@ -460,9 +443,7 @@ static void findAsync_l(const MediaInfo &info, const std::shared_ptr<Session> &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<MediaSource> weak_sender = sender.shared_from_this();
 
     _async_close_timer = std::make_shared<Timer>(stream_none_reader_delay / 1000.0f, [weak_sender, is_mp4_vod]() {

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<void> getOwnership();

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()) {

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;
     }

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);
                                              }
                                          });
 

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<mutex> lck(s_mtxFlvRecorder);
-                auto key = sender.shortUrl();
+                auto key = tuple.shortUrl();
                 if (bRegist) {
                     DebugL << "开始录制RTMP：" << sender.getUrl();
                     GET_CONFIG(string, http_root, Http::kRootPath);

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<WebRtcPusher>(shared_from_this()));
             src = src_imp;