webrtc支持断连续推

webrtc/WebRtcPusher.cpp

@@ -12,8 +12,9 @@
 
 WebRtcPusher::Ptr WebRtcPusher::create(const EventPoller::Ptr &poller,
                                        const RtspMediaSource::Ptr &src,
+                                       const std::shared_ptr<void> &ownership,
                                        const MediaInfo &info) {
-    WebRtcPusher::Ptr ret(new WebRtcPusher(poller, src, info), [](WebRtcPusher *ptr) {
+    WebRtcPusher::Ptr ret(new WebRtcPusher(poller, src, ownership, info), [](WebRtcPusher *ptr) {
         ptr->onDestory();
         delete ptr;
     });
@@ -23,9 +24,11 @@ WebRtcPusher::Ptr WebRtcPusher::create(const EventPoller::Ptr &poller,
 
 WebRtcPusher::WebRtcPusher(const EventPoller::Ptr &poller,
                            const RtspMediaSource::Ptr &src,
+                           const std::shared_ptr<void> &ownership,
                            const MediaInfo &info) : WebRtcTransportImp(poller) {
     _media_info = info;
     _push_src = src;
+    _push_src_ownership = ownership;
     CHECK(_push_src);
 }
 
@@ -48,7 +51,7 @@ bool WebRtcPusher::close(MediaSource &sender, bool force) {
 
 int WebRtcPusher::totalReaderCount(MediaSource &sender) {
     auto total_count = 0;
-    for (auto &src : _push_src_simulcast) {
+    for (auto &src : _push_src_sim) {
         total_count += src.second->totalReaderCount();
     }
     return total_count + _push_src->totalReaderCount();
@@ -75,15 +78,16 @@ void WebRtcPusher::onRecvRtp(MediaTrack &track, const string &rid, RtpPacket::Pt
 
     if (rtp->type == TrackAudio) {
         //音频
-        for (auto &pr : _push_src_simulcast) {
+        for (auto &pr : _push_src_sim) {
             pr.second->onWrite(rtp, false);
         }
     } else {
         //视频
-        auto &src = _push_src_simulcast[rid];
+        auto &src = _push_src_sim[rid];
         if (!src) {
             auto stream_id = rid.empty() ? _push_src->getId() : _push_src->getId() + "_" + rid;
             auto src_imp = std::make_shared<RtspMediaSourceImp>(_push_src->getVhost(), _push_src->getApp(), stream_id);
+            _push_src_sim_ownership[rid] = src_imp->getOwnership();
             src_imp->setSdp(_push_src->getSdp());
             src_imp->setProtocolTranslation(_push_src->isRecording(Recorder::type_hls),
                                             _push_src->isRecording(Recorder::type_mp4));
@@ -121,6 +125,15 @@ void WebRtcPusher::onDestory() {
                                                false, static_cast<SockInfo &>(*getSession()));
         }
     }
+
+    GET_CONFIG(uint32_t, continue_push_ms, General::kContinuePushMS);
+    if (_push_src && continue_push_ms) {
+        //取消所有权
+        _push_src_ownership = nullptr;
+        //延时10秒注销流
+        auto push_src = std::move(_push_src);
+        getPoller()->doDelayTask(continue_push_ms, [push_src]() { return 0; });
+    }
 }
 
 void WebRtcPusher::onRtcConfigure(RtcConfigure &configure) const {

webrtc/WebRtcPusher.h

@@ -17,7 +17,8 @@ class WebRtcPusher : public WebRtcTransportImp, public MediaSourceEvent {
 public:
     using Ptr = std::shared_ptr<WebRtcPusher>;
     ~WebRtcPusher() override = default;
-    static Ptr create(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src, const MediaInfo &info);
+    static Ptr create(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src,
+                      const std::shared_ptr<void> &ownership, const MediaInfo &info);
 
 protected:
     ///////WebRtcTransportImp override///////
@@ -40,7 +41,8 @@ protected:
     std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
 
 private:
-    WebRtcPusher(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src, const MediaInfo &info);
+    WebRtcPusher(const EventPoller::Ptr &poller, const RtspMediaSource::Ptr &src,
+                 const std::shared_ptr<void> &ownership, const MediaInfo &info);
 
 private:
     bool _simulcast = false;
@@ -48,8 +50,11 @@ private:
     MediaInfo _media_info;
     //推流的rtsp源
     RtspMediaSource::Ptr _push_src;
+    //推流所有权
+    std::shared_ptr<void> _push_src_ownership;
     //推流的rtsp源,支持simulcast
-    unordered_map<string/*rid*/, RtspMediaSource::Ptr> _push_src_simulcast;
+    unordered_map<string/*rid*/, RtspMediaSource::Ptr> _push_src_sim;
+    unordered_map<string/*rid*/, std::shared_ptr<void> > _push_src_sim_ownership;
 };
 
 #endif //ZLMEDIAKIT_WEBRTCPUSHER_H

webrtc/WebRtcTransport.cpp

@@ -969,15 +969,41 @@ void push_plugin(Session &sender, const string &offer_sdp, const WebRtcArgs &arg
             cb(WebRtcException(SockException(Err_other, err)));
             return;
         }
-        auto src = dynamic_pointer_cast<RtspMediaSource>(MediaSource::find(RTSP_SCHEMA, info._vhost, info._app, info._streamid));
+
-        if (src) {
+        RtspMediaSourceImp::Ptr push_src;
+        std::shared_ptr<void> push_src_ownership;
+        auto src = MediaSource::find(RTSP_SCHEMA, info._vhost, info._app, info._streamid);
+        auto push_failed = (bool)src;
+
+        while (src) {
+            //尝试断连后继续推流
+            auto rtsp_src = dynamic_pointer_cast<RtspMediaSourceImp>(src);
+            if (!rtsp_src) {
+                //源不是rtsp推流产生的
+                break;
+            }
+            auto ownership = rtsp_src->getOwnership();
+            if (!ownership) {
+                //获取推流源所有权失败
+                break;
+            }
+            push_src = std::move(rtsp_src);
+            push_src_ownership = std::move(ownership);
+            push_failed = false;
+            break;
+        }
+
+        if (push_failed) {
             cb(WebRtcException(SockException(Err_other, "already publishing")));
             return;
         }
 
-        auto push_src = std::make_shared<RtspMediaSourceImp>(info._vhost, info._app, info._streamid);
+        if (!push_src) {
+            push_src = std::make_shared<RtspMediaSourceImp>(info._vhost, info._app, info._streamid);
+            push_src_ownership = push_src->getOwnership();
             push_src->setProtocolTranslation(enable_hls, enable_mp4);
-        auto rtc = WebRtcPusher::create(EventPollerPool::Instance().getPoller(), push_src, info);
+        }
+        auto rtc = WebRtcPusher::create(EventPollerPool::Instance().getPoller(), push_src, push_src_ownership, info);
         push_src->setListener(rtc);
         cb(*rtc);
     };