From bfe118d94ef8dccaba1bb856e2731f1fbec9531e Mon Sep 17 00:00:00 2001
From: ziyue <1213642868@qq.com>
Date: Tue, 1 Nov 2022 16:05:23 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8DRtpProcess=E6=9E=90=E6=9E=84?=
 =?UTF-8?q?=E5=AF=BC=E8=87=B4=E7=9A=84=E7=BA=BF=E7=A8=8B=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/Rtp/RtpProcess.cpp  | 5 ++++-
 src/Rtp/RtpProcess.h    | 7 ++++++-
 src/Rtp/RtpSelector.cpp | 7 +++++++
 3 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/src/Rtp/RtpProcess.cpp b/src/Rtp/RtpProcess.cpp
index e4ca2dac..b9aa5f1b 100644
--- a/src/Rtp/RtpProcess.cpp
+++ b/src/Rtp/RtpProcess.cpp
@@ -49,10 +49,13 @@ RtpProcess::RtpProcess(const string &stream_id) {
     }
 }
 
-RtpProcess::~RtpProcess() {
+void RtpProcess::flush() {
     if (_process) {
         _process->flush();
     }
+}
+
+RtpProcess::~RtpProcess() {
     uint64_t duration = (_last_frame_time.createdTime() - _last_frame_time.elapsedTime()) / 1000;
     WarnP(this) << "RTP推流器("
                 << _media_info.shortUrl()
diff --git a/src/Rtp/RtpProcess.h b/src/Rtp/RtpProcess.h
index 3c3244a2..8b951216 100644
--- a/src/Rtp/RtpProcess.h
+++ b/src/Rtp/RtpProcess.h
@@ -18,7 +18,7 @@
 
 namespace mediakit {
 
-class RtpProcess : public RtcpContextForRecv, public toolkit::SockInfo, public MediaSinkInterface, public MediaSourceEventInterceptor, public std::enable_shared_from_this<RtpProcess>{
+class RtpProcess final : public RtcpContextForRecv, public toolkit::SockInfo, public MediaSinkInterface, public MediaSourceEventInterceptor, public std::enable_shared_from_this<RtpProcess>{
 public:
     typedef std::shared_ptr<RtpProcess> Ptr;
     friend class RtpProcessHelper;
@@ -57,6 +57,11 @@ public:
      */
     void setStopCheckRtp(bool is_check=false);
 
+    /**
+     * flush输出缓存
+     */
+    void flush() override;
+
     /// SockInfo override
     std::string get_local_ip() override;
     uint16_t get_local_port() override;
diff --git a/src/Rtp/RtpSelector.cpp b/src/Rtp/RtpSelector.cpp
index 66d44828..8ac620ac 100644
--- a/src/Rtp/RtpSelector.cpp
+++ b/src/Rtp/RtpSelector.cpp
@@ -112,6 +112,13 @@ RtpProcessHelper::RtpProcessHelper(const string &stream_id, const weak_ptr<RtpSe
 }
 
 RtpProcessHelper::~RtpProcessHelper() {
+    auto process = std::move(_process);
+    try {
+        // flush时，确保线程安全
+        process->getOwnerPoller(MediaSource::NullMediaSource())->async([process]() { process->flush(); });
+    } catch (...) {
+        // 忽略getOwnerPoller可能抛出的异常
+    }
 }
 
 void RtpProcessHelper::attachEvent() {