SmartPtr: Support shared ptr for live source. v6.0.129 (#4089)

Detail change log: 1. [Simple,Refactor] Remove member fields of http entry, etc. e34b3d3 2. [Ignore] Rename source to live_source. 846f95e 3. [Ignore] Use directly ptr in consumer. d38af02 4. [Complex, Important] Use shared ptr for live source. 88f9224 The object relationship: ![live-source](https://github.com/ossrs/srs/assets/2777660/1adb59af-6e7a-40f3-9a4a-1cc849d7dae1) --- Co-authored-by: Jacob Su <[email protected]>
ossrs · Jun 14, 2024 · e706978 · e706978 · winlinvip · Jun 14, 2024
1 parent 908c2f2
commit e706978
Show file tree

Hide file tree

Showing 30 changed files with 256 additions and 235 deletions.
diff --git a/trunk/doc/CHANGELOG.md b/trunk/doc/CHANGELOG.md
@@ -7,6 +7,7 @@ The changelog for SRS.
 <a name="v6-changes"></a>
 
 ## SRS 6.0 Changelog
+* v6.0, 2024-06-15, Merge [#4089](https://github.com/ossrs/srs/pull/4089): SmartPtr: Support shared ptr for live source. v6.0.129 (#4089)
 * v6.0, 2024-06-14, Merge [#4085](https://github.com/ossrs/srs/pull/4085): SmartPtr: Support shared ptr for RTC source. v6.0.128 (#4085)
 * v6.0, 2024-06-13, Merge [#4083](https://github.com/ossrs/srs/pull/4083): SmartPtr: Use shared ptr in RTC TCP connection. v6.0.127 (#4083)
 * v6.0, 2024-06-12, Merge [#4080](https://github.com/ossrs/srs/pull/4080): SmartPtr: Use shared ptr to manage GB objects. v6.0.126 (#4080)

diff --git a/trunk/src/app/srs_app_coworkers.cpp b/trunk/src/app/srs_app_coworkers.cpp
@@ -122,7 +122,7 @@ SrsRequest* SrsCoWorkers::find_stream_info(string vhost, string app, string stre
     return it->second;
 }
 
-srs_error_t SrsCoWorkers::on_publish(SrsLiveSource* s, SrsRequest* r)
+srs_error_t SrsCoWorkers::on_publish(SrsRequest* r)
 {
     srs_error_t err = srs_success;
 
@@ -140,7 +140,7 @@ srs_error_t SrsCoWorkers::on_publish(SrsLiveSource* s, SrsRequest* r)
     return err;
 }
 
-void SrsCoWorkers::on_unpublish(SrsLiveSource* s, SrsRequest* r)
+void SrsCoWorkers::on_unpublish(SrsRequest* r)
 {
     string url = r->get_stream_url();
 

diff --git a/trunk/src/app/srs_app_coworkers.hpp b/trunk/src/app/srs_app_coworkers.hpp
@@ -33,8 +33,8 @@ class SrsCoWorkers
 private:
     virtual SrsRequest* find_stream_info(std::string vhost, std::string app, std::string stream);
 public:
-    virtual srs_error_t on_publish(SrsLiveSource* s, SrsRequest* r);
-    virtual void on_unpublish(SrsLiveSource* s, SrsRequest* r);
+    virtual srs_error_t on_publish(SrsRequest* r);
+    virtual void on_unpublish(SrsRequest* r);
 };
 
 #endif
diff --git a/trunk/src/app/srs_app_edge.cpp b/trunk/src/app/srs_app_edge.cpp
@@ -391,7 +391,7 @@ void SrsEdgeFlvUpstream::kbps_sample(const char* label, srs_utime_t age)
 
 SrsEdgeIngester::SrsEdgeIngester()
 {
-    source = NULL;
+    source_ = NULL;
     edge = NULL;
     req = NULL;
 #ifdef SRS_APM
@@ -415,9 +415,11 @@ SrsEdgeIngester::~SrsEdgeIngester()
     srs_freep(trd);
 }
 
-srs_error_t SrsEdgeIngester::initialize(SrsLiveSource* s, SrsPlayEdge* e, SrsRequest* r)
+srs_error_t SrsEdgeIngester::initialize(SrsSharedPtr<SrsLiveSource> s, SrsPlayEdge* e, SrsRequest* r)
 {
-    source = s;
+    // Because source references to this object, so we should directly use the source ptr.
+    source_ = s.get();
+
     edge = e;
     req = r;
 
@@ -435,7 +437,7 @@ srs_error_t SrsEdgeIngester::start()
 {
     srs_error_t err = srs_success;
 
-    if ((err = source->on_publish()) != srs_success) {
+    if ((err = source_->on_publish()) != srs_success) {
         return srs_error_wrap(err, "notify source");
     }
 
@@ -455,8 +457,8 @@ void SrsEdgeIngester::stop()
     upstream->close();
 
     // Notify source to un-publish if exists.
-    if (source) {
-        source->on_unpublish();
+    if (source_) {
+        source_->on_unpublish();
     }
 }
 
@@ -549,7 +551,7 @@ srs_error_t SrsEdgeIngester::do_cycle()
             upstream = new SrsEdgeRtmpUpstream(redirect);
         }
 
-        if ((err = source->on_source_id_changed(_srs_context->get_id())) != srs_success) {
+        if ((err = source_->on_source_id_changed(_srs_context->get_id())) != srs_success) {
             return srs_error_wrap(err, "on source id changed");
         }
 
@@ -635,21 +637,21 @@ srs_error_t SrsEdgeIngester::process_publish_message(SrsCommonMessage* msg, stri
 
     // process audio packet
     if (msg->header.is_audio()) {
-        if ((err = source->on_audio(msg)) != srs_success) {
+        if ((err = source_->on_audio(msg)) != srs_success) {
             return srs_error_wrap(err, "source consume audio");
         }
     }
 
     // process video packet
     if (msg->header.is_video()) {
-        if ((err = source->on_video(msg)) != srs_success) {
+        if ((err = source_->on_video(msg)) != srs_success) {
             return srs_error_wrap(err, "source consume video");
         }
     }
 
     // process aggregate packet
     if (msg->header.is_aggregate()) {
-        if ((err = source->on_aggregate(msg)) != srs_success) {
+        if ((err = source_->on_aggregate(msg)) != srs_success) {
             return srs_error_wrap(err, "source consume aggregate");
         }
         return err;
@@ -665,7 +667,7 @@ srs_error_t SrsEdgeIngester::process_publish_message(SrsCommonMessage* msg, stri
 
         if (dynamic_cast<SrsOnMetaDataPacket*>(pkt)) {
             SrsOnMetaDataPacket* metadata = dynamic_cast<SrsOnMetaDataPacket*>(pkt);
-            if ((err = source->on_meta_data(msg, metadata)) != srs_success) {
+            if ((err = source_->on_meta_data(msg, metadata)) != srs_success) {
                 return srs_error_wrap(err, "source consume metadata");
             }
             return err;
@@ -725,7 +727,7 @@ SrsEdgeForwarder::SrsEdgeForwarder()
     edge = NULL;
     req = NULL;
     send_error_code = ERROR_SUCCESS;
-    source = NULL;
+    source_ = NULL;
 
     sdk = NULL;
     lb = new SrsLbRoundRobin();
@@ -747,9 +749,11 @@ void SrsEdgeForwarder::set_queue_size(srs_utime_t queue_size)
     return queue->set_queue_size(queue_size);
 }
 
-srs_error_t SrsEdgeForwarder::initialize(SrsLiveSource* s, SrsPublishEdge* e, SrsRequest* r)
+srs_error_t SrsEdgeForwarder::initialize(SrsSharedPtr<SrsLiveSource> s, SrsPublishEdge* e, SrsRequest* r)
 {
-    source = s;
+    // Because source references to this object, so we should directly use the source ptr.
+    source_ = s.get();
+
     edge = e;
     req = r;
 
@@ -956,7 +960,7 @@ SrsPlayEdge::~SrsPlayEdge()
     srs_freep(ingester);
 }
 
-srs_error_t SrsPlayEdge::initialize(SrsLiveSource* source, SrsRequest* req)
+srs_error_t SrsPlayEdge::initialize(SrsSharedPtr<SrsLiveSource> source, SrsRequest* req)
 {
     srs_error_t err = srs_success;
 
@@ -1048,7 +1052,7 @@ void SrsPublishEdge::set_queue_size(srs_utime_t queue_size)
     return forwarder->set_queue_size(queue_size);
 }
 
-srs_error_t SrsPublishEdge::initialize(SrsLiveSource* source, SrsRequest* req)
+srs_error_t SrsPublishEdge::initialize(SrsSharedPtr<SrsLiveSource> source, SrsRequest* req)
 {
     srs_error_t err = srs_success;
 

diff --git a/trunk/src/app/srs_app_edge.hpp b/trunk/src/app/srs_app_edge.hpp
@@ -10,6 +10,7 @@
 #include <srs_core.hpp>
 
 #include <srs_app_st.hpp>
+#include <srs_core_autofree.hpp>
 
 #include <string>
 
@@ -137,7 +138,9 @@ class SrsEdgeFlvUpstream : public SrsEdgeUpstream
 class SrsEdgeIngester : public ISrsCoroutineHandler
 {
 private:
-    SrsLiveSource* source;
+    // Because source references to this object, so we should directly use the source ptr.
+    SrsLiveSource* source_;
+private:
     SrsPlayEdge* edge;
     SrsRequest* req;
     SrsCoroutine* trd;
@@ -150,7 +153,7 @@ class SrsEdgeIngester : public ISrsCoroutineHandler
     SrsEdgeIngester();
     virtual ~SrsEdgeIngester();
 public:
-    virtual srs_error_t initialize(SrsLiveSource* s, SrsPlayEdge* e, SrsRequest* r);
+    virtual srs_error_t initialize(SrsSharedPtr<SrsLiveSource> s, SrsPlayEdge* e, SrsRequest* r);
     virtual srs_error_t start();
     virtual void stop();
     virtual std::string get_curr_origin();
@@ -172,7 +175,9 @@ class SrsEdgeIngester : public ISrsCoroutineHandler
 class SrsEdgeForwarder : public ISrsCoroutineHandler
 {
 private:
-    SrsLiveSource* source;
+    // Because source references to this object, so we should directly use the source ptr.
+    SrsLiveSource* source_;
+private:
     SrsPublishEdge* edge;
     SrsRequest* req;
     SrsCoroutine* trd;
@@ -191,7 +196,7 @@ class SrsEdgeForwarder : public ISrsCoroutineHandler
 public:
     virtual void set_queue_size(srs_utime_t queue_size);
 public:
-    virtual srs_error_t initialize(SrsLiveSource* s, SrsPublishEdge* e, SrsRequest* r);
+    virtual srs_error_t initialize(SrsSharedPtr<SrsLiveSource> s, SrsPublishEdge* e, SrsRequest* r);
     virtual srs_error_t start();
     virtual void stop();
 // Interface ISrsReusableThread2Handler
@@ -216,7 +221,7 @@ class SrsPlayEdge
     // Always use the req of source,
     // For we assume all client to edge is invalid,
     // if auth open, edge must valid it from origin, then service it.
-    virtual srs_error_t initialize(SrsLiveSource* source, SrsRequest* req);
+    virtual srs_error_t initialize(SrsSharedPtr<SrsLiveSource> source, SrsRequest* req);
     // When client play stream on edge.
     virtual srs_error_t on_client_play();
     // When all client stopped play, disconnect to origin.
@@ -239,7 +244,7 @@ class SrsPublishEdge
 public:
     virtual void set_queue_size(srs_utime_t queue_size);
 public:
-    virtual srs_error_t initialize(SrsLiveSource* source, SrsRequest* req);
+    virtual srs_error_t initialize(SrsSharedPtr<SrsLiveSource> source, SrsRequest* req);
     virtual bool can_publish();
     // When client publish stream on edge.
     virtual srs_error_t on_client_publish();

diff --git a/trunk/src/app/srs_app_http_conn.cpp b/trunk/src/app/srs_app_http_conn.cpp
@@ -547,13 +547,13 @@ srs_error_t SrsHttpServer::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage
     return http_static->mux.serve_http(w, r);
 }
 
-srs_error_t SrsHttpServer::http_mount(SrsLiveSource* s, SrsRequest* r)
+srs_error_t SrsHttpServer::http_mount(SrsRequest* r)
 {
-    return http_stream->http_mount(s, r);
+    return http_stream->http_mount(r);
 }
 
-void SrsHttpServer::http_unmount(SrsLiveSource* s, SrsRequest* r)
+void SrsHttpServer::http_unmount(SrsRequest* r)
 {
-    http_stream->http_unmount(s, r);
+    http_stream->http_unmount(r);
 }
 
diff --git a/trunk/src/app/srs_app_http_conn.hpp b/trunk/src/app/srs_app_http_conn.hpp
@@ -187,8 +187,8 @@ class SrsHttpServer : public ISrsHttpServeMux
 public:
     virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
 public:
-    virtual srs_error_t http_mount(SrsLiveSource* s, SrsRequest* r);
-    virtual void http_unmount(SrsLiveSource* s, SrsRequest* r);
+    virtual srs_error_t http_mount(SrsRequest* r);
+    virtual void http_unmount(SrsRequest* r);
 };
 
 #endif