完善主动关断流机制

server/WebApi.cpp

@@ -338,7 +338,8 @@ void installWebApi() {
         });
     });
 
-    API_REGIST(api,kick_pusher,{
+    //主动关断流，包括关断拉流、推流
+    API_REGIST(api,close_stream,{
         CHECK_SECRET();
         CHECK_ARGS("schema","vhost","app","stream");
         //踢掉推流器
@@ -405,6 +406,12 @@ void installWebApi() {
             const_cast<PlayerProxy::Ptr &>(player).reset();
             invoker("200 OK", headerOut, val.toStyledString());
         });
+
+        //被主动关闭拉流
+        player->setOnClose([key](){
+            lock_guard<recursive_mutex> lck(s_proxyMapMtx);
+            s_proxyMap.erase(key);
+        });
         player->play(allArgs["url"]);
     });
 

src/Common/MediaSource.h

@@ -148,7 +148,7 @@ public:
         }
         return listener->close();
     }
-    void setListener(const std::weak_ptr<MediaSourceEvent> &listener){
+    virtual void setListener(const std::weak_ptr<MediaSourceEvent> &listener){
         _listener = listener;
     }
 

src/Player/PlayerProxy.cpp

@@ -79,6 +79,11 @@ PlayerProxy::PlayerProxy(const string &strVhost,
 void PlayerProxy::setPlayCallbackOnce(const function<void(const SockException &ex)> &cb){
     _playCB = cb;
 }
+
+void PlayerProxy::setOnClose(const function<void()> &cb){
+    _onClose = cb;
+}
+
 void PlayerProxy::play(const string &strUrlTmp) {
 	weak_ptr<PlayerProxy> weakSelf = shared_from_this();
 	std::shared_ptr<int> piFailedCnt(new int(0)); //连续播放失败次数
@@ -148,6 +153,9 @@ bool PlayerProxy::close() {
 		if (stronSelf) {
 			stronSelf->_mediaMuxer.reset();
 			stronSelf->teardown();
+			if(stronSelf->_onClose){
+                stronSelf->_onClose();
+			}
 		}
 	});
     return true;

src/Player/PlayerProxy.h

@@ -62,8 +62,23 @@ public:
      */
     void setPlayCallbackOnce(const function<void(const SockException &ex)> &cb);
 
-	void play(const string &strUrl) override;
+    /**
+     * 设置主动关闭回调
+     * @param cb
+     */
+    void setOnClose(const function<void()> &cb);
 
+    /**
+     * 开始拉流播放
+     * @param strUrl
+     */
+    void play(const string &strUrl) override;
+
+
+    /**
+     * 被主动关闭
+     * @return
+     */
     bool close() override;
 private:
 	void rePlay(const string &strUrl,int iFailedCnt);
@@ -78,6 +93,7 @@ private:
     string _strSrc;
     Timer::Ptr _timer;
     function<void(const SockException &ex)> _playCB;
+    function<void()> _onClose;
 };
 
 } /* namespace mediakit */

src/Rtmp/RtmpToRtspMediaSource.h

@@ -78,9 +78,17 @@ public:
 				track->addDelegate(_rtspMuxer);
 				track->addDelegate(_recorder);
 			}
+            _rtspMuxer->setListener(_listener);
 		}
 		RtmpMediaSource::onWrite(pkt,key_pos);
 	}
+
+	void setListener(const std::weak_ptr<MediaSourceEvent> &listener) override {
+        RtmpMediaSource::setListener(listener);
+        if(_rtspMuxer){
+            _rtspMuxer->setListener(listener);
+        }
+    }
 private:
 	RtmpDemuxer::Ptr _rtmpDemuxer;
 	RtspMediaSourceMuxer::Ptr _rtspMuxer;

src/Rtsp/RtspToRtmpMediaSource.h

@@ -73,10 +73,18 @@ public:
                     track->addDelegate(_rtmpMuxer);
                     track->addDelegate(_recorder);
                 }
+                _rtmpMuxer->setListener(_listener);
             }
         }
         RtspMediaSource::onWrite(rtp, bKeyPos);
     }
+
+    void setListener(const std::weak_ptr<MediaSourceEvent> &listener) override {
+        RtspMediaSource::setListener(listener);
+        if(_rtmpMuxer){
+            _rtmpMuxer->setListener(listener);
+        }
+    }
 private:
     RtspDemuxer::Ptr _rtspDemuxer;
     RtmpMediaSourceMuxer::Ptr _rtmpMuxer;