api/source/mk_rtp_server.cpp

@@ -102,7 +102,7 @@ API_EXPORT void API_CALL mk_rtp_pause_check(const char *app, const char *stream)
     auto src = MediaSource::find(DEFAULT_VHOST, app, stream);
     auto process = src ? src->getRtpProcess() : nullptr;
     if (process) {
-        process->pauseRtpTimeout(true);
+        process->setStopCheckRtp(true);
     }
 }
 
@@ -110,7 +110,7 @@ API_EXPORT void API_CALL mk_rtp_resume_check(const char *app, const char *stream
     auto src = MediaSource::find(DEFAULT_VHOST, app, stream);
     auto process = src ? src->getRtpProcess() : nullptr;
     if (process) {
-        process->pauseRtpTimeout(false);
+        process->setStopCheckRtp(false);
     }
 }
 

postman/ZLMediaKit.postman_collection.json

@@ -2019,12 +2019,6 @@
 							"key": "stream_id",
 							"value": "test",
 							"description": "该端口绑定的流id"
-						},
-						{
-							"key": "pause_seconds",
-							"value": "300",
-							"description": "暂停超时监测后，将在pause_seconds时间后恢复",
-							"disabled": true
 						}
 					]
 				}

server/WebApi.cpp

@@ -1106,8 +1106,9 @@ void installWebApi() {
 
         bool force = allArgs["force"].as<bool>();
         for (auto &media : media_list) {
-            media->getOwnerPoller()->async([media, force]() { media->close(force); });
-            ++count_closed;
+            if (media->close(force)) {
+                ++count_closed;
+            }
         }
         val["count_hit"] = count_hit;
         val["count_closed"] = count_closed;
@@ -1734,7 +1735,7 @@ void installWebApi() {
         auto src = MediaSource::find(vhost, app, allArgs["stream_id"]);
         auto process = src ? src->getRtpProcess() : nullptr;
         if (process) {
-            process->pauseRtpTimeout(true, allArgs["pause_seconds"]);
+            process->setStopCheckRtp(true);
         } else {
             val["code"] = API::NotFound;
         }
@@ -1754,7 +1755,7 @@ void installWebApi() {
         auto src = MediaSource::find(vhost, app, allArgs["stream_id"]);
         auto process = src ? src->getRtpProcess() : nullptr;
         if (process) {
-            process->pauseRtpTimeout(false);
+            process->setStopCheckRtp(false);
         } else {
             val["code"] = API::NotFound;
         }

server/WebHook.cpp

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

src/Player/PlayerProxy.cpp

@@ -254,9 +254,16 @@ void PlayerProxy::rePlay(const string &strUrl, int iFailedCnt) {
 bool PlayerProxy::close(MediaSource &sender) {
     // 通知其停止推流  [AUTO-TRANSLATED:d69d10d8]
     // Notify it to stop pushing the stream
-    _muxer = nullptr;
-    setMediaSource(nullptr);
-    teardown();
+    weak_ptr<PlayerProxy> weakSelf = dynamic_pointer_cast<PlayerProxy>(shared_from_this());
+    getPoller()->async_first([weakSelf]() {
+        auto strongSelf = weakSelf.lock();
+        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,7 +591,9 @@ 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/Rtp/RtpProcess.cpp

@@ -203,24 +203,27 @@ void RtpProcess::doCachedFunc() {
 }
 
 bool RtpProcess::alive() {
-    if (_pause_timeout) {
-        if (_last_check_alive.elapsedTime() < _pause_seconds * 1000) {
+    if (_stop_rtp_check.load()) {
+        if(_last_check_alive.elapsedTime() > 5 * 60 * 1000){
+            // 最多暂停5分钟的rtp超时检测，因为NAT映射有效期一般不会太长  [AUTO-TRANSLATED:2df59aad]
+            // Pause the RTP timeout detection for a maximum of 5 minutes, because the NAT mapping validity period is generally not very long.
+            _stop_rtp_check = false;
+        } else {
             return true;
         }
-        // 最多暂停_pause_seconds秒的rtp超时检测，因为NAT映射有效期一般不会太长
-        _pause_timeout = false;
     }
 
     _last_check_alive.resetTime();
     GET_CONFIG(uint64_t, timeoutSec, RtpProxy::kTimeoutSec)
-    return _last_frame_time.elapsedTime() < timeoutSec * 1000;
+    if (_last_frame_time.elapsedTime() / 1000 < timeoutSec) {
+        return true;
+    }
+    return false;
 }
 
-void RtpProcess::pauseRtpTimeout(bool pause, uint32_t pause_seconds) {
-    _pause_timeout = pause;
-    // 默认5分钟恢复超时监测
-    _pause_seconds = pause_seconds ? pause_seconds : 300;
-    if (!pause) {
+void RtpProcess::setStopCheckRtp(bool is_check){
+    _stop_rtp_check = is_check;
+    if (!is_check) {
         _last_frame_time.resetTime();
     }
 }

src/Rtp/RtpProcess.h

@@ -69,11 +69,12 @@ public:
     void setOnDetach(onDetachCB cb);
 
     /**
-     * 暂停或恢复rtp超时监测
-     * @param pause 是否暂停超时检测
-     * @param pause_seconds 暂停超时检测最大时间(单位秒)，超过这个时间后将恢复超时检测; 设置为0时默认为300
+     * 设置onDetach事件回调,false检查RTP超时，true停止
+     * Set onDetach event callback, false checks RTP timeout, true stops
+     
+     * [AUTO-TRANSLATED:2780397f]
      */
-    void pauseRtpTimeout(bool pause, uint32_t pause_seconds = 0);
+    void setStopCheckRtp(bool is_check=false);
 
     /**
      * 设置为单track，单音频/单视频时可以加快媒体注册速度
@@ -128,12 +129,10 @@ private:
     void createTimer();
 
 private:
-    bool _pause_timeout = false;
-    uint32_t _pause_seconds = 5 * 60;
-    uint64_t _dts = 0;
-    uint64_t _total_bytes = 0;
     OnlyTrack _only_track = kAll;
     std::string _auth_err;
+    uint64_t _dts = 0;
+    uint64_t _total_bytes = 0;
     std::unique_ptr<sockaddr_storage> _addr;
     toolkit::Socket::Ptr _sock;
     MediaInfo _media_info;
@@ -143,6 +142,7 @@ private:
     std::shared_ptr<FILE> _save_file_video;
     ProcessInterface::Ptr _process;
     MultiMediaSourceMuxer::Ptr _muxer;
+    std::atomic_bool _stop_rtp_check{false};
     toolkit::Timer::Ptr _timer;
     toolkit::Ticker _last_check_alive;
     std::recursive_mutex _func_mtx;

src/Rtsp/RtspSession.cpp

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

srt/SrtTransportImp.cpp

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