修复MediaSource::close线程安全问题

主动或无人观看关闭流可能会由于线程安全问题导致崩溃

server/WebApi.cpp

@@ -1106,9 +1106,8 @@ void installWebApi() {
 
         bool force = allArgs["force"].as<bool>();
         for (auto &media : media_list) {
-            if (media->close(force)) {
+            media->getOwnerPoller()->async([media, force]() { media->close(force); });
-                ++count_closed;
+            ++count_closed;
-            }
         }
         val["count_hit"] = count_hit;
         val["count_closed"] = count_closed;

server/WebHook.cpp

@@ -634,7 +634,10 @@ void installWebHook() {
             // 边沿站无人观看时如果是拉流的则立即停止溯源  [AUTO-TRANSLATED:a1429c77]
             // If no one is watching at the edge station, stop tracing immediately if it is pulling
             if (!auto_close) {
-                sender.close(false);
+                auto ptr = sender.shared_from_this();
+                sender.getOwnerPoller()->async([ptr]() {
+                    ptr->close(false);
+                });
                 WarnL << "Auto close stream when none reader: " << sender.getOriginUrl();
             }
             return;
@@ -661,7 +664,7 @@ void installWebHook() {
             if (!flag || !err.empty() || !strongSrc) {
                 return;
             }
-            strongSrc->close(false);
+            strongSrc->getOwnerPoller()->async([strongSrc]() { strongSrc->close(false); });
             WarnL << "无人观看主动关闭流:" << strongSrc->getOriginUrl();
         });
     });

src/Common/MediaSource.cpp

@@ -698,13 +698,13 @@ void MediaSourceEvent::onReaderChanged(MediaSource &sender, int size){
                 // 此流被标记为无人观看自动关闭流  [AUTO-TRANSLATED:64a0dac3]
                 // This stream is marked as an automatically closed stream with no viewers.
                 WarnL << "Auto close stream when none reader: " << strong_sender->getUrl();
-                strong_sender->close(false);
+                strong_sender->getOwnerPoller()->async([strong_sender]() { strong_sender->close(false); });
             }
         } else {
             // 这个是mp4点播，我们自动关闭  [AUTO-TRANSLATED:8a7b9a90]
             // This is an mp4 on-demand, we automatically close it.
             WarnL << "MP4点播无人观看,自动关闭:" << strong_sender->getUrl();
-            strong_sender->close(false);
+            strong_sender->getOwnerPoller()->async([strong_sender]() { strong_sender->close(false); });
         }
         return false;
     }, specified_poller);

src/Player/PlayerProxy.cpp

@@ -254,16 +254,9 @@ void PlayerProxy::rePlay(const string &strUrl, int iFailedCnt) {
 bool PlayerProxy::close(MediaSource &sender) {
     // 通知其停止推流  [AUTO-TRANSLATED:d69d10d8]
     // Notify it to stop pushing the stream
-    weak_ptr<PlayerProxy> weakSelf = dynamic_pointer_cast<PlayerProxy>(shared_from_this());
+    _muxer = nullptr;
-    getPoller()->async_first([weakSelf]() {
+    setMediaSource(nullptr);
-        auto strongSelf = weakSelf.lock();
+    teardown();
-        if (!strongSelf) {
-            return;
-        }
-        strongSelf->_muxer.reset();
-        strongSelf->setMediaSource(nullptr);
-        strongSelf->teardown();
-    });
     _on_close(SockException(Err_shutdown, "closed by user"));
     WarnL << "close media: " << sender.getUrl();
     return true;

src/Rtmp/RtmpSession.cpp

@@ -591,9 +591,7 @@ void RtmpSession::onSendMedia(const RtmpPacket::Ptr &pkt) {
 }
 
 bool RtmpSession::close(MediaSource &sender) {
-    //此回调在其他线程触发
+    shutdown(SockException(Err_shutdown, "close media: " + sender.getUrl()));
-    string err = StrPrinter << "close media: " << sender.getUrl();
-    safeShutdown(SockException(Err_shutdown, err));
     return true;
 }
 

src/Rtsp/RtspSession.cpp

@@ -1175,9 +1175,7 @@ int RtspSession::getTrackIndexByInterleaved(int interleaved) {
 }
 
 bool RtspSession::close(MediaSource &sender) {
-    //此回调在其他线程触发
+    shutdown(SockException(Err_shutdown,"close media: " + sender.getUrl()));
-    string err = StrPrinter << "close media: " << sender.getUrl();
-    safeShutdown(SockException(Err_shutdown,err));
     return true;
 }
 

src/Shell/ShellCMD.h

@@ -36,9 +36,9 @@ public:
                             if (!media) {
                                 break;
                             }
-                            if (!media->close(true)) {
+                            media->getOwnerPoller()->async([media]() {
-                                break;
+                                media->close(true);
-                            }
+                            });
                             (*stream) << "\t踢出成功:" << media->getUrl() << "\r\n";
                             return;
                         } while (0);

srt/SrtTransportImp.cpp

@@ -145,16 +145,8 @@ void SrtTransportImp::onShutdown(const SockException &ex) {
 }
 
 bool SrtTransportImp::close(mediakit::MediaSource &sender) {
-    std::string err = StrPrinter << "close media: " << sender.getUrl();
+    onShutdown(SockException(Err_shutdown, "close media: " + sender.getUrl()));
-    weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this());
+    _muxer = nullptr;
-    getPoller()->async([weak_self, err]() {
-        auto strong_self = weak_self.lock();
-        if (strong_self) {
-            strong_self->onShutdown(SockException(Err_shutdown, err));
-            // 主动关闭推流，那么不延时注销
-            strong_self->_muxer = nullptr;
-        }
-    });
     return true;
 }
 

webrtc/WebRtcPusher.cpp

@@ -42,19 +42,10 @@ WebRtcPusher::WebRtcPusher(const EventPoller::Ptr &poller,
 }
 
 bool WebRtcPusher::close(MediaSource &sender) {
-    // 此回调在其他线程触发  [AUTO-TRANSLATED:c98e7686]
+    onShutdown(SockException(Err_shutdown, "close media: " + sender.getUrl()));
-    // This callback is triggered in another thread
+    // 主动关闭推流，那么不延时注销  [AUTO-TRANSLATED:ee7cc580]
-    string err = StrPrinter << "close media: " << sender.getUrl();
+    // Actively close the stream, then do not delay the logout
-    weak_ptr<WebRtcPusher> weak_self = static_pointer_cast<WebRtcPusher>(shared_from_this());
+    _push_src = nullptr;
-    getPoller()->async([weak_self, err]() {
-        auto strong_self = weak_self.lock();
-        if (strong_self) {
-            strong_self->onShutdown(SockException(Err_shutdown, err));
-            // 主动关闭推流，那么不延时注销  [AUTO-TRANSLATED:ee7cc580]
-            // Actively close the stream, then do not delay the logout
-            strong_self->_push_src = nullptr;
-        }
-    });
     return true;
 }