完善addStreamPusherProxy相关功能并修复自动删除相关的bug

752590f8 · ziyue · aa39680c · 752590f8 · 752590f8 · 752590f8
Commit 752590f8 authored Jun 17, 2021 by ziyue
--- a/server/WebApi.cpp
+++ b/server/WebApi.cpp
@@ -616,6 +616,11 @@ void installWebApi() {
                                          float timeout_sec,
                                          const function<void(const SockException &ex, const string &key)> &cb) {
        auto key = getPusherKey(schema, vhost, app, stream, url);
+        auto src = MediaSource::find(schema, vhost, app, stream);
+        if (!src) {
+            cb(SockException(Err_other, "can not find the source stream"), key);
+            return;
+        }
        lock_guard<recursive_mutex> lck(s_proxyPusherMapMtx);
        if (s_proxyPusherMap.find(key) != s_proxyPusherMap.end()) {
            //已经在推流了
@@ -624,7 +629,7 @@ void installWebApi() {
        }
        //添加推流代理
-        PusherProxy::Ptr pusher(new PusherProxy(schema, vhost, app, stream, retry_count ? retry_count : -1));
+        PusherProxy::Ptr pusher(new PusherProxy(src, retry_count ? retry_count : -1));
        s_proxyPusherMap[key] = pusher;
        //指定RTP over TCP(播放rtsp时有效)
@@ -638,18 +643,18 @@ void installWebApi() {
        //开始推流，如果推流失败或者推流中止，将会自动重试若干次，默认一直重试
        pusher->setPushCallbackOnce([cb, key, url](const SockException &ex) {
            if (ex) {
-                InfoL << "key: " << key << ", " << "addStreamPusherProxy pusher callback error: " << ex.what();
+                WarnL << "Push " << url << " failed, key: " << key << ", err: " << ex.what();
-                lock_guard<recursive_mutex> lck(s_proxyMapMtx);
+                lock_guard<recursive_mutex> lck(s_proxyPusherMapMtx);
-                s_proxyMap.erase(key);
+                s_proxyPusherMap.erase(key);
            }
            cb(ex, key);
        });
        //被主动关闭推流
        pusher->setOnClose([key, url](const SockException &ex) {
-            InfoL << "key: " << key << ", " << "addStreamPusherProxy close callback error: " << ex.what();
+            WarnL << "Push " << url << " failed, key: " << key << ", err: " << ex.what();
-            lock_guard<recursive_mutex> lck(s_proxyMapMtx);
+            lock_guard<recursive_mutex> lck(s_proxyPusherMapMtx);
-            s_proxyMap.erase(key);
+            s_proxyPusherMap.erase(key);
        });
        pusher->publish(url);
    };
@@ -659,23 +664,7 @@ void installWebApi() {
    api_regist("/index/api/addStreamPusherProxy", [](API_ARGS_MAP_ASYNC) {
        CHECK_SECRET();
        CHECK_ARGS("schema", "vhost", "app", "stream", "dst_url");
        auto dst_url = allArgs["dst_url"];
-        auto src_url = allArgs["schema"] + "/" + allArgs["vhost"] + "/" + allArgs["app"] + "/" + allArgs["stream"];
-        auto src = MediaSource::find(allArgs["schema"],
-                                     allArgs["vhost"],
-                                     allArgs["app"],
-                                     allArgs["stream"]);
-        if (!src) {
-            WarnL << "addStreamPusherProxy, can not find source stream:" << src_url;
-            val["code"] = API::NotFound;
-            val["msg"] = "can not find the source stream";
-            invoker(200, headerOut, val.toStyledString());
-            return;
-        }
-        InfoL << "addStreamPusherProxy, find stream: " << src_url << ", push dst url: " << dst_url;
        addStreamPusherProxy(allArgs["schema"],
                             allArgs["vhost"],
                             allArgs["app"],
@@ -688,7 +677,6 @@ void installWebApi() {
                                 if (ex) {
                                     val["code"] = API::OtherFailed;
                                     val["msg"] = ex.what();
-                                     WarnL << "Publish stream failed, dst url is: " << dst_url;
                                 } else {
                                     val["data"]["key"] = key;
                                     InfoL << "Publish success, please play with player:" << dst_url;

--- a/src/Pusher/PusherProxy.cpp
+++ b/src/Pusher/PusherProxy.cpp
@@ -14,13 +14,8 @@ using namespace toolkit;
 namespace mediakit {
-PusherProxy::PusherProxy(const string& schema, const string &vhost, const string &app, const string &stream,
+PusherProxy::PusherProxy(const MediaSource::Ptr &src, int retry_count, const EventPoller::Ptr &poller)
-            int retry_count, const EventPoller::Ptr &poller)
+            : MediaPusher(src, poller){
-            : MediaPusher(schema,vhost, app, stream, poller){
-    _schema = schema;
-    _vhost = vhost;
-    _app = app;
-    _stream_id = stream;
    _retry_count = retry_count;
    _on_close = [](const SockException &) {};
 }
@@ -53,7 +48,7 @@ void PusherProxy::publish(const string& dstUrl) {
        if (!err) {
            // 推流成功
            *piFailedCnt = 0;
-            InfoL << "pusher publish " << dstUrl << " success";
+            InfoL << "Publish " << dstUrl << " success";
        } else if (*piFailedCnt < strongSelf->_retry_count || strongSelf->_retry_count < 0) {
            // 推流失败，延时重试推送
            strongSelf->rePublish(dstUrl, (*piFailedCnt)++);
@@ -77,7 +72,6 @@ void PusherProxy::publish(const string& dstUrl) {
    });
    MediaPusher::publish(dstUrl);
-    _publish_url = dstUrl;
 }
 void PusherProxy::rePublish(const string &dstUrl, int iFailedCnt) {

--- a/src/Pusher/PusherProxy.h
+++ b/src/Pusher/PusherProxy.h
@@ -24,10 +24,8 @@ public:
    typedef std::shared_ptr<PusherProxy> Ptr;
    // 如果retry_count<0,则一直重试播放；否则重试retry_count次数
-    // 默认一直重试，创建此对象时候，需要外部保证mediaSource存在
+    // 默认一直重试，创建此对象时候，需要外部保证MediaSource存在
-    PusherProxy(const string& schema, const string &vhost, const string &app, const string &stream,
+    PusherProxy(const MediaSource::Ptr &src, int retry_count = -1, const EventPoller::Ptr &poller = nullptr);
-                int retry_count = -1, const EventPoller::Ptr &poller = nullptr);
    ~PusherProxy() override;
    /**
@@ -54,11 +52,6 @@ private:
 private:
    int _retry_count;
-    std::string _schema;
-    string _vhost;
-    string _app;
-    string _stream_id;
-    std::string _publish_url;
    Timer::Ptr _timer;
    function<void(const SockException &ex)> _on_close;