From e7069788e9b628850df35f4efa6b08488bdcccfd Mon Sep 17 00:00:00 2001 From: Winlin Date: Sat, 15 Jun 2024 07:54:56 +0800 Subject: [PATCH] SmartPtr: Support shared ptr for live source. v6.0.129 (#4089) Detail change log: 1. [Simple,Refactor] Remove member fields of http entry, etc. https://github.com/ossrs/srs/pull/4089/commits/e34b3d3aa44f56b49c5c0b8e58c0285c4d9094d4 2. [Ignore] Rename source to live_source. https://github.com/ossrs/srs/pull/4089/commits/846f95ec96f183909070c4b3f5ef0e9dd7d5448d 3. [Ignore] Use directly ptr in consumer. https://github.com/ossrs/srs/pull/4089/commits/d38af021ad532ed9b4043c529dd5e54b9cbf5c01 4. [Complex, Important] Use shared ptr for live source. https://github.com/ossrs/srs/pull/4089/commits/88f922413a1fb5cb920ab64f8b3805420932602a The object relationship: ![live-source](https://github.com/ossrs/srs/assets/2777660/1adb59af-6e7a-40f3-9a4a-1cc849d7dae1) --- Co-authored-by: Jacob Su --- trunk/doc/CHANGELOG.md | 1 + trunk/src/app/srs_app_coworkers.cpp | 4 +- trunk/src/app/srs_app_coworkers.hpp | 4 +- trunk/src/app/srs_app_edge.cpp | 36 +++--- trunk/src/app/srs_app_edge.hpp | 17 ++- trunk/src/app/srs_app_http_conn.cpp | 8 +- trunk/src/app/srs_app_http_conn.hpp | 4 +- trunk/src/app/srs_app_http_stream.cpp | 68 +++++------ trunk/src/app/srs_app_http_stream.hpp | 16 +-- trunk/src/app/srs_app_recv_thread.cpp | 6 +- trunk/src/app/srs_app_recv_thread.hpp | 5 +- trunk/src/app/srs_app_rtc_api.cpp | 4 +- trunk/src/app/srs_app_rtc_conn.cpp | 12 +- trunk/src/app/srs_app_rtc_source.cpp | 6 +- trunk/src/app/srs_app_rtc_source.hpp | 8 +- trunk/src/app/srs_app_rtmp_conn.cpp | 38 +++---- trunk/src/app/srs_app_rtmp_conn.hpp | 17 +-- trunk/src/app/srs_app_server.cpp | 12 +- trunk/src/app/srs_app_server.hpp | 4 +- trunk/src/app/srs_app_source.cpp | 143 +++++++++++------------- trunk/src/app/srs_app_source.hpp | 23 ++-- trunk/src/app/srs_app_srt_conn.cpp | 10 +- trunk/src/app/srs_app_srt_source.cpp | 6 +- trunk/src/app/srs_app_srt_source.hpp | 8 +- trunk/src/app/srs_app_stream_bridge.cpp | 4 +- trunk/src/app/srs_app_stream_bridge.hpp | 4 +- trunk/src/core/srs_core_autofree.hpp | 7 +- trunk/src/core/srs_core_version6.hpp | 2 +- trunk/src/kernel/srs_kernel_error.hpp | 3 +- trunk/src/utest/srs_utest_core.cpp | 11 ++ 30 files changed, 256 insertions(+), 235 deletions(-) diff --git a/trunk/doc/CHANGELOG.md b/trunk/doc/CHANGELOG.md index 5885d58d6c..d60e25bcc4 100644 --- a/trunk/doc/CHANGELOG.md +++ b/trunk/doc/CHANGELOG.md @@ -7,6 +7,7 @@ The changelog for SRS. ## 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 index 114ec0a703..1c33640ff3 100644 --- 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 index ce0322e6e8..cd284d410f 100644 --- 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 index f4a52115c3..ba9e12f21d 100644 --- 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 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(pkt)) { SrsOnMetaDataPacket* metadata = dynamic_cast(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 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 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 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 index efb5705204..a1f049c49f 100644 --- a/trunk/src/app/srs_app_edge.hpp +++ b/trunk/src/app/srs_app_edge.hpp @@ -10,6 +10,7 @@ #include #include +#include #include @@ -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 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 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 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 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 index 721f038a67..f7bab8a4c8 100644 --- 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 index 7952b40a37..de78c8b3c0 100644 --- 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 diff --git a/trunk/src/app/srs_app_http_stream.cpp b/trunk/src/app/srs_app_http_stream.cpp index 181ee14c09..914cefd5c6 100755 --- a/trunk/src/app/srs_app_http_stream.cpp +++ b/trunk/src/app/srs_app_http_stream.cpp @@ -40,10 +40,9 @@ using namespace std; #include #include -SrsBufferCache::SrsBufferCache(SrsLiveSource* s, SrsRequest* r) +SrsBufferCache::SrsBufferCache(SrsRequest* r) { req = r->copy()->as_http(); - source = s; queue = new SrsMessageQueue(true); trd = new SrsSTCoroutine("http-stream", this); @@ -59,12 +58,11 @@ SrsBufferCache::~SrsBufferCache() srs_freep(req); } -srs_error_t SrsBufferCache::update_auth(SrsLiveSource* s, SrsRequest* r) +srs_error_t SrsBufferCache::update_auth(SrsRequest* r) { srs_freep(req); req = r->copy(); - source = s; - + return srs_success; } @@ -107,15 +105,20 @@ srs_error_t SrsBufferCache::cycle() srs_usleep(SRS_STREAM_CACHE_CYCLE); return err; } + + SrsSharedPtr live_source = _srs_sources->fetch(req); + if (!live_source.get()) { + return srs_error_new(ERROR_NO_SOURCE, "no source for %s", req->get_stream_url().c_str()); + } // the stream cache will create consumer to cache stream, // which will trigger to fetch stream from origin for edge. SrsLiveConsumer* consumer = NULL; SrsAutoFree(SrsLiveConsumer, consumer); - if ((err = source->create_consumer(consumer)) != srs_success) { + if ((err = live_source->create_consumer(consumer)) != srs_success) { return srs_error_wrap(err, "create consumer"); } - if ((err = source->consumer_dumps(consumer, false, false, true)) != srs_success) { + if ((err = live_source->consumer_dumps(consumer, false, false, true)) != srs_success) { return srs_error_wrap(err, "dumps consumer"); } @@ -553,9 +556,8 @@ srs_error_t SrsBufferWriter::writev(const iovec* iov, int iovcnt, ssize_t* pnwri return writer->writev(iov, iovcnt, pnwrite); } -SrsLiveStream::SrsLiveStream(SrsLiveSource* s, SrsRequest* r, SrsBufferCache* c) +SrsLiveStream::SrsLiveStream(SrsRequest* r, SrsBufferCache* c) { - source = s; cache = c; req = r->copy()->as_http(); security_ = new SrsSecurity(); @@ -567,10 +569,8 @@ SrsLiveStream::~SrsLiveStream() srs_freep(security_); } -srs_error_t SrsLiveStream::update_auth(SrsLiveSource* s, SrsRequest* r) +srs_error_t SrsLiveStream::update_auth(SrsRequest* r) { - source = s; - srs_freep(req); req = r->copy()->as_http(); @@ -660,14 +660,19 @@ srs_error_t SrsLiveStream::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMess // Enter chunked mode, because we didn't set the content-length. w->write_header(SRS_CONSTS_HTTP_OK); + + SrsSharedPtr live_source = _srs_sources->fetch(req); + if (!live_source.get()) { + return srs_error_new(ERROR_NO_SOURCE, "no source for %s", req->get_stream_url().c_str()); + } // create consumer of souce, ignore gop cache, use the audio gop cache. SrsLiveConsumer* consumer = NULL; SrsAutoFree(SrsLiveConsumer, consumer); - if ((err = source->create_consumer(consumer)) != srs_success) { + if ((err = live_source->create_consumer(consumer)) != srs_success) { return srs_error_wrap(err, "create consumer"); } - if ((err = source->consumer_dumps(consumer, true, true, !enc->has_cache())) != srs_success) { + if ((err = live_source->consumer_dumps(consumer, true, true, !enc->has_cache())) != srs_success) { return srs_error_wrap(err, "dumps consumer"); } @@ -689,7 +694,7 @@ srs_error_t SrsLiveStream::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMess // if gop cache enabled for encoder, dump to consumer. if (enc->has_cache()) { - if ((err = enc->dump_cache(consumer, source->jitter())) != srs_success) { + if ((err = enc->dump_cache(consumer, live_source->jitter())) != srs_success) { return srs_error_wrap(err, "encoder dump cache"); } } @@ -876,7 +881,6 @@ SrsLiveEntry::SrsLiveEntry(std::string m) cache = NULL; req = NULL; - source = NULL; std::string ext = srs_path_filext(m); _is_flv = (ext == ".flv"); @@ -954,7 +958,7 @@ srs_error_t SrsHttpStreamServer::initialize() } // TODO: FIXME: rename for HTTP FLV mount. -srs_error_t SrsHttpStreamServer::http_mount(SrsLiveSource* s, SrsRequest* r) +srs_error_t SrsHttpStreamServer::http_mount(SrsRequest* r) { srs_error_t err = srs_success; @@ -982,10 +986,9 @@ srs_error_t SrsHttpStreamServer::http_mount(SrsLiveSource* s, SrsRequest* r) entry = new SrsLiveEntry(mount); - entry->source = s; entry->req = r->copy()->as_http(); - entry->cache = new SrsBufferCache(s, r); - entry->stream = new SrsLiveStream(s, r, entry->cache); + entry->cache = new SrsBufferCache(r); + entry->stream = new SrsLiveStream(r, entry->cache); // TODO: FIXME: maybe refine the logic of http remux service. // if user push streams followed: @@ -994,8 +997,7 @@ srs_error_t SrsHttpStreamServer::http_mount(SrsLiveSource* s, SrsRequest* r) // and they will using the same template, such as: [vhost]/[app]/[stream].flv // so, need to free last request object, otherwise, it will cause memory leak. srs_freep(tmpl->req); - - tmpl->source = s; + tmpl->req = r->copy()->as_http(); sflvs[sid] = entry; @@ -1015,8 +1017,8 @@ srs_error_t SrsHttpStreamServer::http_mount(SrsLiveSource* s, SrsRequest* r) } else { // The entry exists, we reuse it and update the request of stream and cache. entry = sflvs[sid]; - entry->stream->update_auth(s, r); - entry->cache->update_auth(s, r); + entry->stream->update_auth(r); + entry->cache->update_auth(r); } if (entry->stream) { @@ -1027,7 +1029,7 @@ srs_error_t SrsHttpStreamServer::http_mount(SrsLiveSource* s, SrsRequest* r) return err; } -void SrsHttpStreamServer::http_unmount(SrsLiveSource* s, SrsRequest* r) +void SrsHttpStreamServer::http_unmount(SrsRequest* r) { std::string sid = r->get_stream_url(); @@ -1133,20 +1135,20 @@ srs_error_t SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandle } } } - - SrsLiveSource* s = NULL; - if ((err = _srs_sources->fetch_or_create(r, server, &s)) != srs_success) { + + SrsSharedPtr live_source; + if ((err = _srs_sources->fetch_or_create(r, server, live_source)) != srs_success) { return srs_error_wrap(err, "source create"); } - srs_assert(s != NULL); + srs_assert(live_source.get() != NULL); bool enabled_cache = _srs_config->get_gop_cache(r->vhost); int gcmf = _srs_config->get_gop_cache_max_frames(r->vhost); - s->set_cache(enabled_cache); - s->set_gop_cache_max_frames(gcmf); + live_source->set_cache(enabled_cache); + live_source->set_gop_cache_max_frames(gcmf); // create http streaming handler. - if ((err = http_mount(s, r)) != srs_success) { + if ((err = http_mount(r)) != srs_success) { return srs_error_wrap(err, "http mount"); } @@ -1161,7 +1163,7 @@ srs_error_t SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandle // trigger edge to fetch from origin. bool vhost_is_edge = _srs_config->get_vhost_is_edge(r->vhost); srs_trace("flv: source url=%s, is_edge=%d, source_id=%s/%s", - r->get_stream_url().c_str(), vhost_is_edge, s->source_id().c_str(), s->pre_source_id().c_str()); + r->get_stream_url().c_str(), vhost_is_edge, live_source->source_id().c_str(), live_source->pre_source_id().c_str()); return err; } diff --git a/trunk/src/app/srs_app_http_stream.hpp b/trunk/src/app/srs_app_http_stream.hpp index 138eb003c4..b3962d072c 100755 --- a/trunk/src/app/srs_app_http_stream.hpp +++ b/trunk/src/app/srs_app_http_stream.hpp @@ -23,13 +23,12 @@ class SrsBufferCache : public ISrsCoroutineHandler srs_utime_t fast_cache; private: SrsMessageQueue* queue; - SrsLiveSource* source; SrsRequest* req; SrsCoroutine* trd; public: - SrsBufferCache(SrsLiveSource* s, SrsRequest* r); + SrsBufferCache(SrsRequest* r); virtual ~SrsBufferCache(); - virtual srs_error_t update_auth(SrsLiveSource* s, SrsRequest* r); + virtual srs_error_t update_auth(SrsRequest* r); public: virtual srs_error_t start(); virtual srs_error_t dump_cache(SrsLiveConsumer* consumer, SrsRtmpJitterAlgorithm jitter); @@ -178,13 +177,12 @@ class SrsLiveStream : public ISrsHttpHandler { private: SrsRequest* req; - SrsLiveSource* source; SrsBufferCache* cache; SrsSecurity* security_; public: - SrsLiveStream(SrsLiveSource* s, SrsRequest* r, SrsBufferCache* c); + SrsLiveStream(SrsRequest* r, SrsBufferCache* c); virtual ~SrsLiveStream(); - virtual srs_error_t update_auth(SrsLiveSource* s, SrsRequest* r); + virtual srs_error_t update_auth(SrsRequest* r); public: virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); private: @@ -205,8 +203,6 @@ struct SrsLiveEntry public: // We will free the request. SrsRequest* req; - // Shared source. - SrsLiveSource* source; public: // For template, the mount contains variables. // For concrete stream, the mount is url to access. @@ -244,8 +240,8 @@ class SrsHttpStreamServer : public ISrsReloadHandler virtual srs_error_t initialize(); public: // HTTP flv/ts/mp3/aac stream - 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); // Interface ISrsHttpMatchHijacker public: virtual srs_error_t hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph); diff --git a/trunk/src/app/srs_app_recv_thread.cpp b/trunk/src/app/srs_app_recv_thread.cpp index 5e77245fac..b25fdc9f1b 100644 --- a/trunk/src/app/srs_app_recv_thread.cpp +++ b/trunk/src/app/srs_app_recv_thread.cpp @@ -259,13 +259,13 @@ void SrsQueueRecvThread::on_stop() } SrsPublishRecvThread::SrsPublishRecvThread(SrsRtmpServer* rtmp_sdk, SrsRequest* _req, - int mr_sock_fd, srs_utime_t tm, SrsRtmpConn* conn, SrsLiveSource* source, SrsContextId parent_cid) + int mr_sock_fd, srs_utime_t tm, SrsRtmpConn* conn, SrsSharedPtr source, SrsContextId parent_cid) : trd(this, rtmp_sdk, tm, parent_cid) { rtmp = rtmp_sdk; _conn = conn; - _source = source; + source_ = source; nn_msgs_for_yield_ = 0; recv_error = srs_success; @@ -370,7 +370,7 @@ srs_error_t SrsPublishRecvThread::consume(SrsCommonMessage* msg) srs_update_system_time(), msg->header.timestamp, msg->size); // the rtmp connection will handle this message - err = _conn->handle_publish_message(_source, msg); + err = _conn->handle_publish_message(source_, msg); // must always free it, // the source will copy it if need to use. diff --git a/trunk/src/app/srs_app_recv_thread.hpp b/trunk/src/app/srs_app_recv_thread.hpp index fdef672702..5fa011db57 100644 --- a/trunk/src/app/srs_app_recv_thread.hpp +++ b/trunk/src/app/srs_app_recv_thread.hpp @@ -16,6 +16,7 @@ #include #include #include +#include class SrsRtmpServer; class SrsCommonMessage; @@ -146,7 +147,7 @@ class SrsPublishRecvThread : public ISrsMessagePumper, public ISrsReloadHandler srs_error_t recv_error; SrsRtmpConn* _conn; // The params for conn callback. - SrsLiveSource* _source; + SrsSharedPtr source_; // The error timeout cond srs_cond_t error; // The merged context id. @@ -154,7 +155,7 @@ class SrsPublishRecvThread : public ISrsMessagePumper, public ISrsReloadHandler SrsContextId ncid; public: SrsPublishRecvThread(SrsRtmpServer* rtmp_sdk, SrsRequest* _req, - int mr_sock_fd, srs_utime_t tm, SrsRtmpConn* conn, SrsLiveSource* source, SrsContextId parent_cid); + int mr_sock_fd, srs_utime_t tm, SrsRtmpConn* conn, SrsSharedPtr source, SrsContextId parent_cid); virtual ~SrsPublishRecvThread(); public: // Wait for error for some timeout. diff --git a/trunk/src/app/srs_app_rtc_api.cpp b/trunk/src/app/srs_app_rtc_api.cpp index 393f3869f3..1f2880633f 100644 --- a/trunk/src/app/srs_app_rtc_api.cpp +++ b/trunk/src/app/srs_app_rtc_api.cpp @@ -224,8 +224,8 @@ srs_error_t SrsGoApiRtcPlay::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessa // For RTMP to RTC, fail if disabled and RTMP is active, see https://github.com/ossrs/srs/issues/2728 if (!is_rtc_stream_active && !_srs_config->get_rtc_from_rtmp(ruc->req_->vhost)) { - SrsLiveSource* rtmp = _srs_sources->fetch(ruc->req_); - if (rtmp && !rtmp->inactive()) { + SrsSharedPtr live_source = _srs_sources->fetch(ruc->req_); + if (live_source.get() && !live_source->inactive()) { return srs_error_new(ERROR_RTC_DISABLED, "Disabled rtmp_to_rtc of %s, see #2728", ruc->req_->vhost.c_str()); } } diff --git a/trunk/src/app/srs_app_rtc_conn.cpp b/trunk/src/app/srs_app_rtc_conn.cpp index 1ef65a8caf..52d142a4ca 100644 --- a/trunk/src/app/srs_app_rtc_conn.cpp +++ b/trunk/src/app/srs_app_rtc_conn.cpp @@ -646,7 +646,7 @@ srs_error_t SrsRtcPlayStream::cycle() SrsRtcConsumer* consumer = NULL; SrsAutoFree(SrsRtcConsumer, consumer); - if ((err = source->create_consumer(source_, consumer)) != srs_success) { + if ((err = source->create_consumer(consumer)) != srs_success) { return srs_error_wrap(err, "create consumer, source=%s", req_->get_stream_url().c_str()); } @@ -1202,8 +1202,8 @@ srs_error_t SrsRtcPublishStream::initialize(SrsRequest* r, SrsRtcSourceDescripti source_->set_publish_stream(this); // TODO: FIMXE: Check it in SrsRtcConnection::add_publisher? - SrsLiveSource *rtmp = _srs_sources->fetch(r); - if (rtmp && !rtmp->can_publish(false)) { + SrsSharedPtr live_source = _srs_sources->fetch(r); + if (live_source.get() && !live_source->can_publish(false)) { return srs_error_new(ERROR_SYSTEM_STREAM_BUSY, "rtmp stream %s busy", r->get_stream_url().c_str()); } @@ -1227,16 +1227,16 @@ srs_error_t SrsRtcPublishStream::initialize(SrsRequest* r, SrsRtcSourceDescripti #if defined(SRS_RTC) && defined(SRS_FFMPEG_FIT) bool rtc_to_rtmp = _srs_config->get_rtc_to_rtmp(req_->vhost); if (rtc_to_rtmp) { - if ((err = _srs_sources->fetch_or_create(r, _srs_hybrid->srs()->instance(), &rtmp)) != srs_success) { + if ((err = _srs_sources->fetch_or_create(r, _srs_hybrid->srs()->instance(), live_source)) != srs_success) { return srs_error_wrap(err, "create source"); } // Disable GOP cache for RTC2RTMP bridge, to keep the streams in sync, // especially for stream merging. - rtmp->set_cache(false); + live_source->set_cache(false); SrsCompositeBridge* bridge = new SrsCompositeBridge(); - bridge->append(new SrsFrameToRtmpBridge(rtmp)); + bridge->append(new SrsFrameToRtmpBridge(live_source)); if ((err = bridge->initialize(r)) != srs_success) { srs_freep(bridge); diff --git a/trunk/src/app/srs_app_rtc_source.cpp b/trunk/src/app/srs_app_rtc_source.cpp index c8abd37981..7957903d99 100644 --- a/trunk/src/app/srs_app_rtc_source.cpp +++ b/trunk/src/app/srs_app_rtc_source.cpp @@ -154,7 +154,7 @@ ISrsRtcSourceChangeCallback::~ISrsRtcSourceChangeCallback() { } -SrsRtcConsumer::SrsRtcConsumer(SrsSharedPtr s) +SrsRtcConsumer::SrsRtcConsumer(SrsRtcSource* s) { source_ = s; should_update_source_id = false; @@ -486,11 +486,11 @@ void SrsRtcSource::set_bridge(ISrsStreamBridge* bridge) #endif } -srs_error_t SrsRtcSource::create_consumer(SrsSharedPtr source, SrsRtcConsumer*& consumer) +srs_error_t SrsRtcSource::create_consumer(SrsRtcConsumer*& consumer) { srs_error_t err = srs_success; - consumer = new SrsRtcConsumer(source); + consumer = new SrsRtcConsumer(this); consumers.push_back(consumer); // TODO: FIXME: Implements edge cluster. diff --git a/trunk/src/app/srs_app_rtc_source.hpp b/trunk/src/app/srs_app_rtc_source.hpp index f3b904556b..00328527b0 100644 --- a/trunk/src/app/srs_app_rtc_source.hpp +++ b/trunk/src/app/srs_app_rtc_source.hpp @@ -80,7 +80,9 @@ class ISrsRtcSourceChangeCallback class SrsRtcConsumer { private: - SrsSharedPtr source_; + // Because source references to this object, so we should directly use the source ptr. + SrsRtcSource* source_; +private: std::vector queue; // when source id changed, notice all consumers bool should_update_source_id; @@ -92,7 +94,7 @@ class SrsRtcConsumer // The callback for stream change event. ISrsRtcSourceChangeCallback* handler_; public: - SrsRtcConsumer(SrsSharedPtr s); + SrsRtcConsumer(SrsRtcSource* s); virtual ~SrsRtcConsumer(); public: // When source id changed, notice client to print. @@ -215,7 +217,7 @@ class SrsRtcSource : public ISrsFastTimer public: // Create consumer // @param consumer, output the create consumer. - virtual srs_error_t create_consumer(SrsSharedPtr source, SrsRtcConsumer*& consumer); + virtual srs_error_t create_consumer(SrsRtcConsumer*& consumer); // Dumps packets in cache to consumer. // @param ds, whether dumps the sequence header. // @param dm, whether dumps the metadata. diff --git a/trunk/src/app/srs_app_rtmp_conn.cpp b/trunk/src/app/srs_app_rtmp_conn.cpp index 58ed3e3caf..fd01a831ef 100644 --- a/trunk/src/app/srs_app_rtmp_conn.cpp +++ b/trunk/src/app/srs_app_rtmp_conn.cpp @@ -571,19 +571,19 @@ srs_error_t SrsRtmpConn::stream_service_cycle() rtmp->set_send_timeout(SRS_CONSTS_RTMP_TIMEOUT); // find a source to serve. - SrsLiveSource* source = NULL; - if ((err = _srs_sources->fetch_or_create(req, server, &source)) != srs_success) { + SrsSharedPtr live_source; + if ((err = _srs_sources->fetch_or_create(req, server, live_source)) != srs_success) { return srs_error_wrap(err, "rtmp: fetch source"); } - srs_assert(source != NULL); + srs_assert(live_source.get() != NULL); bool enabled_cache = _srs_config->get_gop_cache(req->vhost); int gcmf = _srs_config->get_gop_cache_max_frames(req->vhost); srs_trace("source url=%s, ip=%s, cache=%d/%d, is_edge=%d, source_id=%s/%s", - req->get_stream_url().c_str(), ip.c_str(), enabled_cache, gcmf, info->edge, source->source_id().c_str(), - source->pre_source_id().c_str()); - source->set_cache(enabled_cache); - source->set_gop_cache_max_frames(gcmf); + req->get_stream_url().c_str(), ip.c_str(), enabled_cache, gcmf, info->edge, live_source->source_id().c_str(), + live_source->pre_source_id().c_str()); + live_source->set_cache(enabled_cache); + live_source->set_gop_cache_max_frames(gcmf); switch (info->type) { case SrsRtmpConnPlay: { @@ -610,7 +610,7 @@ srs_error_t SrsRtmpConn::stream_service_cycle() span_main_->end(); #endif - err = playing(source); + err = playing(live_source); http_hooks_on_stop(); return err; @@ -627,7 +627,7 @@ srs_error_t SrsRtmpConn::stream_service_cycle() span_main_->end(); #endif - return publishing(source); + return publishing(live_source); } case SrsRtmpConnHaivisionPublish: { if ((err = rtmp->start_haivision_publish(info->res->stream_id)) != srs_success) { @@ -641,7 +641,7 @@ srs_error_t SrsRtmpConn::stream_service_cycle() span_main_->end(); #endif - return publishing(source); + return publishing(live_source); } case SrsRtmpConnFlashPublish: { if ((err = rtmp->start_flash_publish(info->res->stream_id)) != srs_success) { @@ -655,7 +655,7 @@ srs_error_t SrsRtmpConn::stream_service_cycle() span_main_->end(); #endif - return publishing(source); + return publishing(live_source); } default: { return srs_error_new(ERROR_SYSTEM_CLIENT_INVALID, "rtmp: unknown client type=%d", info->type); @@ -699,7 +699,7 @@ srs_error_t SrsRtmpConn::check_vhost(bool try_default_vhost) return err; } -srs_error_t SrsRtmpConn::playing(SrsLiveSource* source) +srs_error_t SrsRtmpConn::playing(SrsSharedPtr source) { srs_error_t err = srs_success; @@ -786,7 +786,7 @@ srs_error_t SrsRtmpConn::playing(SrsLiveSource* source) return err; } -srs_error_t SrsRtmpConn::do_playing(SrsLiveSource* source, SrsLiveConsumer* consumer, SrsQueueRecvThread* rtrd) +srs_error_t SrsRtmpConn::do_playing(SrsSharedPtr source, SrsLiveConsumer* consumer, SrsQueueRecvThread* rtrd) { srs_error_t err = srs_success; @@ -923,7 +923,7 @@ srs_error_t SrsRtmpConn::do_playing(SrsLiveSource* source, SrsLiveConsumer* cons return err; } -srs_error_t SrsRtmpConn::publishing(SrsLiveSource* source) +srs_error_t SrsRtmpConn::publishing(SrsSharedPtr source) { srs_error_t err = srs_success; @@ -969,7 +969,7 @@ srs_error_t SrsRtmpConn::publishing(SrsLiveSource* source) return err; } -srs_error_t SrsRtmpConn::do_publishing(SrsLiveSource* source, SrsPublishRecvThread* rtrd) +srs_error_t SrsRtmpConn::do_publishing(SrsSharedPtr source, SrsPublishRecvThread* rtrd) { srs_error_t err = srs_success; @@ -1073,7 +1073,7 @@ srs_error_t SrsRtmpConn::do_publishing(SrsLiveSource* source, SrsPublishRecvThre return err; } -srs_error_t SrsRtmpConn::acquire_publish(SrsLiveSource* source) +srs_error_t SrsRtmpConn::acquire_publish(SrsSharedPtr source) { srs_error_t err = srs_success; @@ -1141,7 +1141,7 @@ srs_error_t SrsRtmpConn::acquire_publish(SrsLiveSource* source) return err; } -void SrsRtmpConn::release_publish(SrsLiveSource* source) +void SrsRtmpConn::release_publish(SrsSharedPtr source) { // when edge, notice edge to change state. // when origin, notice all service to unpublish. @@ -1152,7 +1152,7 @@ void SrsRtmpConn::release_publish(SrsLiveSource* source) } } -srs_error_t SrsRtmpConn::handle_publish_message(SrsLiveSource* source, SrsCommonMessage* msg) +srs_error_t SrsRtmpConn::handle_publish_message(SrsSharedPtr& source, SrsCommonMessage* msg) { srs_error_t err = srs_success; @@ -1193,7 +1193,7 @@ srs_error_t SrsRtmpConn::handle_publish_message(SrsLiveSource* source, SrsCommon return err; } -srs_error_t SrsRtmpConn::process_publish_message(SrsLiveSource* source, SrsCommonMessage* msg) +srs_error_t SrsRtmpConn::process_publish_message(SrsSharedPtr& source, SrsCommonMessage* msg) { srs_error_t err = srs_success; diff --git a/trunk/src/app/srs_app_rtmp_conn.hpp b/trunk/src/app/srs_app_rtmp_conn.hpp index 9b86f4fab2..8c86627fec 100644 --- a/trunk/src/app/srs_app_rtmp_conn.hpp +++ b/trunk/src/app/srs_app_rtmp_conn.hpp @@ -16,6 +16,7 @@ #include #include #include +#include class SrsServer; class SrsRtmpServer; @@ -145,14 +146,14 @@ class SrsRtmpConn : public ISrsConnection, public ISrsStartable, public ISrsRelo // The stream(play/publish) service cycle, identify client first. virtual srs_error_t stream_service_cycle(); virtual srs_error_t check_vhost(bool try_default_vhost); - virtual srs_error_t playing(SrsLiveSource* source); - virtual srs_error_t do_playing(SrsLiveSource* source, SrsLiveConsumer* consumer, SrsQueueRecvThread* trd); - virtual srs_error_t publishing(SrsLiveSource* source); - virtual srs_error_t do_publishing(SrsLiveSource* source, SrsPublishRecvThread* trd); - virtual srs_error_t acquire_publish(SrsLiveSource* source); - virtual void release_publish(SrsLiveSource* source); - virtual srs_error_t handle_publish_message(SrsLiveSource* source, SrsCommonMessage* msg); - virtual srs_error_t process_publish_message(SrsLiveSource* source, SrsCommonMessage* msg); + virtual srs_error_t playing(SrsSharedPtr source); + virtual srs_error_t do_playing(SrsSharedPtr source, SrsLiveConsumer* consumer, SrsQueueRecvThread* trd); + virtual srs_error_t publishing(SrsSharedPtr source); + virtual srs_error_t do_publishing(SrsSharedPtr source, SrsPublishRecvThread* trd); + virtual srs_error_t acquire_publish(SrsSharedPtr source); + virtual void release_publish(SrsSharedPtr source); + virtual srs_error_t handle_publish_message(SrsSharedPtr& source, SrsCommonMessage* msg); + virtual srs_error_t process_publish_message(SrsSharedPtr& source, SrsCommonMessage* msg); virtual srs_error_t process_play_control_msg(SrsLiveConsumer* consumer, SrsCommonMessage* msg); virtual void set_sock_options(); private: diff --git a/trunk/src/app/srs_app_server.cpp b/trunk/src/app/srs_app_server.cpp index af962f38b1..2db8738f88 100644 --- a/trunk/src/app/srs_app_server.cpp +++ b/trunk/src/app/srs_app_server.cpp @@ -1302,28 +1302,28 @@ srs_error_t SrsServer::on_reload_listen() return err; } -srs_error_t SrsServer::on_publish(SrsLiveSource* s, SrsRequest* r) +srs_error_t SrsServer::on_publish(SrsRequest* r) { srs_error_t err = srs_success; - if ((err = http_server->http_mount(s, r)) != srs_success) { + if ((err = http_server->http_mount(r)) != srs_success) { return srs_error_wrap(err, "http mount"); } SrsCoWorkers* coworkers = SrsCoWorkers::instance(); - if ((err = coworkers->on_publish(s, r)) != srs_success) { + if ((err = coworkers->on_publish(r)) != srs_success) { return srs_error_wrap(err, "coworkers"); } return err; } -void SrsServer::on_unpublish(SrsLiveSource* s, SrsRequest* r) +void SrsServer::on_unpublish(SrsRequest* r) { - http_server->http_unmount(s, r); + http_server->http_unmount(r); SrsCoWorkers* coworkers = SrsCoWorkers::instance(); - coworkers->on_unpublish(s, r); + coworkers->on_unpublish(r); } SrsServerAdapter::SrsServerAdapter() diff --git a/trunk/src/app/srs_app_server.hpp b/trunk/src/app/srs_app_server.hpp index 795e84f148..07fb52cc44 100644 --- a/trunk/src/app/srs_app_server.hpp +++ b/trunk/src/app/srs_app_server.hpp @@ -234,8 +234,8 @@ class SrsServer : public ISrsReloadHandler, public ISrsLiveSourceHandler, public virtual srs_error_t on_reload_listen(); // Interface ISrsLiveSourceHandler 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); }; // The SRS server adapter, the master server. diff --git a/trunk/src/app/srs_app_source.cpp b/trunk/src/app/srs_app_source.cpp index 7e6368d727..8e84d24aab 100755 --- a/trunk/src/app/srs_app_source.cpp +++ b/trunk/src/app/srs_app_source.cpp @@ -407,7 +407,7 @@ ISrsWakable::~ISrsWakable() SrsLiveConsumer::SrsLiveConsumer(SrsLiveSource* s) { - source = s; + source_ = s; paused = false; jitter = new SrsRtmpJitter(); queue = new SrsMessageQueue(); @@ -423,7 +423,7 @@ SrsLiveConsumer::SrsLiveConsumer(SrsLiveSource* s) SrsLiveConsumer::~SrsLiveConsumer() { - source->on_consumer_destroy(this); + source_->on_consumer_destroy(this); srs_freep(jitter); srs_freep(queue); @@ -506,7 +506,7 @@ srs_error_t SrsLiveConsumer::dump_packets(SrsMessageArray* msgs, int& count) count = 0; if (should_update_source_id) { - srs_trace("update source_id=%s/%s", source->source_id().c_str(), source->pre_source_id().c_str()); + srs_trace("update source_id=%s/%s", source_->source_id().c_str(), source_->pre_source_id().c_str()); should_update_source_id = false; } @@ -822,7 +822,7 @@ SrsSharedPtrMessage* SrsMixQueue::pop() SrsOriginHub::SrsOriginHub() { - source = NULL; + source_ = NULL; req_ = NULL; is_active = false; @@ -861,12 +861,13 @@ SrsOriginHub::~SrsOriginHub() #endif } -srs_error_t SrsOriginHub::initialize(SrsLiveSource* s, SrsRequest* r) +srs_error_t SrsOriginHub::initialize(SrsSharedPtr s, SrsRequest* r) { srs_error_t err = srs_success; req_ = r; - source = s; + // Because source references to this object, so we should directly use the source ptr. + source_ = s.get(); if ((err = hls->initialize(this, req_)) != srs_success) { return srs_error_wrap(err, "hls initialize"); @@ -936,7 +937,7 @@ srs_error_t SrsOriginHub::on_audio(SrsSharedPtrMessage* shared_audio) srs_error_t err = srs_success; SrsSharedPtrMessage* msg = shared_audio; - SrsRtmpFormat* format = source->format_; + SrsRtmpFormat* format = source_->format_; // Handle the metadata when got sequence header. if (format->is_aac_sequence_header() || format->is_mp3_sequence_header()) { @@ -973,7 +974,7 @@ srs_error_t SrsOriginHub::on_audio(SrsSharedPtrMessage* shared_audio) hls->on_unpublish(); srs_error_reset(err); } else if (srs_config_hls_is_on_error_continue(hls_error_strategy)) { - if (srs_hls_can_continue(srs_error_code(err), source->meta->ash(), msg)) { + if (srs_hls_can_continue(srs_error_code(err), source_->meta->ash(), msg)) { srs_error_reset(err); } else { return srs_error_wrap(err, "hls: audio"); @@ -1022,7 +1023,7 @@ srs_error_t SrsOriginHub::on_video(SrsSharedPtrMessage* shared_video, bool is_se srs_error_t err = srs_success; SrsSharedPtrMessage* msg = shared_video; - SrsRtmpFormat* format = source->format_; + SrsRtmpFormat* format = source_->format_; // cache the sequence header if h264 // donot cache the sequence header to gop_cache, return here. @@ -1066,7 +1067,7 @@ srs_error_t SrsOriginHub::on_video(SrsSharedPtrMessage* shared_video, bool is_se hls->on_unpublish(); srs_error_reset(err); } else if (srs_config_hls_is_on_error_continue(hls_error_strategy)) { - if (srs_hls_can_continue(srs_error_code(err), source->meta->vsh(), msg)) { + if (srs_hls_can_continue(srs_error_code(err), source_->meta->vsh(), msg)) { srs_error_reset(err); } else { return srs_error_wrap(err, "hls: video"); @@ -1177,9 +1178,9 @@ srs_error_t SrsOriginHub::on_forwarder_start(SrsForwarder* forwarder) { srs_error_t err = srs_success; - SrsSharedPtrMessage* cache_metadata = source->meta->data(); - SrsSharedPtrMessage* cache_sh_video = source->meta->vsh(); - SrsSharedPtrMessage* cache_sh_audio = source->meta->ash(); + SrsSharedPtrMessage* cache_metadata = source_->meta->data(); + SrsSharedPtrMessage* cache_sh_video = source_->meta->vsh(); + SrsSharedPtrMessage* cache_sh_audio = source_->meta->ash(); // feed the forwarder the metadata/sequence header, // when reload to enable the forwarder. @@ -1200,9 +1201,9 @@ srs_error_t SrsOriginHub::on_dvr_request_sh() { srs_error_t err = srs_success; - SrsSharedPtrMessage* cache_metadata = source->meta->data(); - SrsSharedPtrMessage* cache_sh_video = source->meta->vsh(); - SrsSharedPtrMessage* cache_sh_audio = source->meta->ash(); + SrsSharedPtrMessage* cache_metadata = source_->meta->data(); + SrsSharedPtrMessage* cache_sh_video = source_->meta->vsh(); + SrsSharedPtrMessage* cache_sh_audio = source_->meta->ash(); // feed the dvr the metadata/sequence header, // when reload to start dvr, dvr will never get the sequence header in stream, @@ -1212,13 +1213,13 @@ srs_error_t SrsOriginHub::on_dvr_request_sh() } if (cache_sh_video) { - if ((err = dvr->on_video(cache_sh_video, source->meta->vsh_format())) != srs_success) { + if ((err = dvr->on_video(cache_sh_video, source_->meta->vsh_format())) != srs_success) { return srs_error_wrap(err, "dvr video"); } } if (cache_sh_audio) { - if ((err = dvr->on_audio(cache_sh_audio, source->meta->ash_format())) != srs_success) { + if ((err = dvr->on_audio(cache_sh_audio, source_->meta->ash_format())) != srs_success) { return srs_error_wrap(err, "dvr audio"); } } @@ -1230,16 +1231,16 @@ srs_error_t SrsOriginHub::on_hls_request_sh() { srs_error_t err = srs_success; - SrsSharedPtrMessage* cache_sh_video = source->meta->vsh(); + SrsSharedPtrMessage* cache_sh_video = source_->meta->vsh(); if (cache_sh_video) { - if ((err = hls->on_video(cache_sh_video, source->meta->vsh_format())) != srs_success) { + if ((err = hls->on_video(cache_sh_video, source_->meta->vsh_format())) != srs_success) { return srs_error_wrap(err, "hls video"); } } - SrsSharedPtrMessage* cache_sh_audio = source->meta->ash(); + SrsSharedPtrMessage* cache_sh_audio = source_->meta->ash(); if (cache_sh_audio) { - if ((err = hls->on_audio(cache_sh_audio, source->meta->ash_format())) != srs_success) { + if ((err = hls->on_audio(cache_sh_audio, source_->meta->ash_format())) != srs_success) { return srs_error_wrap(err, "hls audio"); } } @@ -1295,9 +1296,9 @@ srs_error_t SrsOriginHub::on_reload_vhost_dash(string vhost) return srs_error_wrap(err, "dash start publish"); } - SrsRtmpFormat* format = source->format_; + SrsRtmpFormat* format = source_->format_; - SrsSharedPtrMessage* cache_sh_video = source->meta->vsh(); + SrsSharedPtrMessage* cache_sh_video = source_->meta->vsh(); if (cache_sh_video) { if ((err = format->on_video(cache_sh_video)) != srs_success) { return srs_error_wrap(err, "format on_video"); @@ -1307,7 +1308,7 @@ srs_error_t SrsOriginHub::on_reload_vhost_dash(string vhost) } } - SrsSharedPtrMessage* cache_sh_audio = source->meta->ash(); + SrsSharedPtrMessage* cache_sh_audio = source_->meta->ash(); if (cache_sh_audio) { if ((err = format->on_audio(cache_sh_audio)) != srs_success) { return srs_error_wrap(err, "format on_audio"); @@ -1759,67 +1760,64 @@ srs_error_t SrsLiveSourceManager::initialize() return setup_ticks(); } -srs_error_t SrsLiveSourceManager::fetch_or_create(SrsRequest* r, ISrsLiveSourceHandler* h, SrsLiveSource** pps) +srs_error_t SrsLiveSourceManager::fetch_or_create(SrsRequest* r, ISrsLiveSourceHandler* h, SrsSharedPtr& pps) { srs_error_t err = srs_success; // Use lock to protect coroutine switch. // @bug https://github.com/ossrs/srs/issues/1230 - // TODO: FIXME: Use smaller lock. + // TODO: FIXME: Use smaller scope lock. SrsLocker(lock); - - SrsLiveSource* source = NULL; - if ((source = fetch(r)) != NULL) { + + string stream_url = r->get_stream_url(); + std::map< std::string, SrsSharedPtr >::iterator it = pool.find(stream_url); + + if (it != pool.end()) { + SrsSharedPtr& source = it->second; + // we always update the request of resource, // for origin auth is on, the token in request maybe invalid, // and we only need to update the token of request, it's simple. source->update_auth(r); - *pps = source; + pps = source; return err; } - - string stream_url = r->get_stream_url(); - string vhost = r->vhost; - - // should always not exists for create a source. - srs_assert (pool.find(stream_url) == pool.end()); + SrsSharedPtr source = new SrsLiveSource(); srs_trace("new live source, stream_url=%s", stream_url.c_str()); - source = new SrsLiveSource(); - if ((err = source->initialize(r, h)) != srs_success) { - err = srs_error_wrap(err, "init source %s", r->get_stream_url().c_str()); - goto failed; + if ((err = source->initialize(source, r, h)) != srs_success) { + return srs_error_wrap(err, "init source %s", r->get_stream_url().c_str()); } pool[stream_url] = source; - *pps = source; - return err; - -failed: - srs_freep(source); + pps = source; return err; } -SrsLiveSource* SrsLiveSourceManager::fetch(SrsRequest* r) +SrsSharedPtr SrsLiveSourceManager::fetch(SrsRequest* r) { - SrsLiveSource* source = NULL; + // Use lock to protect coroutine switch. + // @bug https://github.com/ossrs/srs/issues/1230 + // TODO: FIXME: Use smaller scope lock. + SrsLocker(lock); string stream_url = r->get_stream_url(); - if (pool.find(stream_url) == pool.end()) { - return NULL; + std::map< std::string, SrsSharedPtr >::iterator it = pool.find(stream_url); + + if (it == pool.end()) { + return SrsSharedPtr(NULL); } - - source = pool[stream_url]; - + + SrsSharedPtr& source = it->second; return source; } void SrsLiveSourceManager::dispose() { - std::map::iterator it; + std::map< std::string, SrsSharedPtr >::iterator it; for (it = pool.begin(); it != pool.end(); ++it) { - SrsLiveSource* source = it->second; + SrsSharedPtr& source = it->second; source->dispose(); } return; @@ -1844,9 +1842,9 @@ srs_error_t SrsLiveSourceManager::notify(int event, srs_utime_t interval, srs_ut { srs_error_t err = srs_success; - std::map::iterator it; + std::map< std::string, SrsSharedPtr >::iterator it; for (it = pool.begin(); it != pool.end();) { - SrsLiveSource* source = it->second; + SrsSharedPtr& source = it->second; // Do cycle source to cleanup components, such as hls dispose. if ((err = source->cycle()) != srs_success) { @@ -1856,19 +1854,11 @@ srs_error_t SrsLiveSourceManager::notify(int event, srs_utime_t interval, srs_ut // See SrsSrtSource::on_consumer_destroy // TODO: FIXME: support source cleanup. // @see https://github.com/ossrs/srs/issues/713 -#if 0 +#if 1 // When source expired, remove it. if (source->stream_is_dead()) { - int cid = source->source_id(); - if (cid == -1 && source->pre_source_id() > 0) { - cid = source->pre_source_id(); - } - if (cid > 0) { - _srs_context->set_id(cid); - } - srs_trace("cleanup die source, total=%d", (int)pool.size()); - - srs_freep(source); + const SrsContextId& cid = source->source_id(); + srs_trace("cleanup die source, id=[%s], total=%d", cid.c_str(), (int)pool.size()); pool.erase(it++); } else { ++it; @@ -1883,11 +1873,6 @@ srs_error_t SrsLiveSourceManager::notify(int event, srs_utime_t interval, srs_ut void SrsLiveSourceManager::destroy() { - std::map::iterator it; - for (it = pool.begin(); it != pool.end(); ++it) { - SrsLiveSource* source = it->second; - srs_freep(source); - } pool.clear(); } @@ -1995,7 +1980,7 @@ bool SrsLiveSource::publisher_is_idle_for(srs_utime_t timeout) return false; } -srs_error_t SrsLiveSource::initialize(SrsRequest* r, ISrsLiveSourceHandler* h) +srs_error_t SrsLiveSource::initialize(SrsSharedPtr wrapper, SrsRequest* r, ISrsLiveSourceHandler* h) { srs_error_t err = srs_success; @@ -2013,14 +1998,14 @@ srs_error_t SrsLiveSource::initialize(SrsRequest* r, ISrsLiveSourceHandler* h) // Setup the SPS/PPS parsing strategy. format_->try_annexb_first = _srs_config->try_annexb_first(r->vhost); - if ((err = hub->initialize(this, req)) != srs_success) { + if ((err = hub->initialize(wrapper, req)) != srs_success) { return srs_error_wrap(err, "hub"); } - if ((err = play_edge->initialize(this, req)) != srs_success) { + if ((err = play_edge->initialize(wrapper, req)) != srs_success) { return srs_error_wrap(err, "edge(play)"); } - if ((err = publish_edge->initialize(this, req)) != srs_success) { + if ((err = publish_edge->initialize(wrapper, req)) != srs_success) { return srs_error_wrap(err, "edge(publish)"); } @@ -2600,7 +2585,7 @@ srs_error_t SrsLiveSource::on_publish() // notify the handler. srs_assert(handler); - if ((err = handler->on_publish(this, req)) != srs_success) { + if ((err = handler->on_publish(req)) != srs_success) { return srs_error_wrap(err, "handle publish"); } @@ -2652,7 +2637,7 @@ void SrsLiveSource::on_unpublish() SrsStatistic* stat = SrsStatistic::instance(); stat->on_stream_close(req); - handler->on_unpublish(this, req); + handler->on_unpublish(req); if (bridge_) { bridge_->on_unpublish(); diff --git a/trunk/src/app/srs_app_source.hpp b/trunk/src/app/srs_app_source.hpp index 75a0b6a765..802c2cb202 100644 --- a/trunk/src/app/srs_app_source.hpp +++ b/trunk/src/app/srs_app_source.hpp @@ -19,6 +19,7 @@ #include #include #include +#include class SrsFormat; class SrsRtmpFormat; @@ -168,9 +169,11 @@ class ISrsWakable // The consumer for SrsLiveSource, that is a play client. class SrsLiveConsumer : public ISrsWakable { +private: + // Because source references to this object, so we should directly use the source ptr. + SrsLiveSource* source_; private: SrsRtmpJitter* jitter; - SrsLiveSource* source; SrsMessageQueue* queue; bool paused; // when source id changed, notice all consumers @@ -288,9 +291,9 @@ class ISrsLiveSourceHandler virtual ~ISrsLiveSourceHandler(); public: // when stream start publish, mount stream. - virtual srs_error_t on_publish(SrsLiveSource* s, SrsRequest* r) = 0; + virtual srs_error_t on_publish(SrsRequest* r) = 0; // when stream stop publish, unmount stream. - virtual void on_unpublish(SrsLiveSource* s, SrsRequest* r) = 0; + virtual void on_unpublish(SrsRequest* r) = 0; }; // The mix queue to correct the timestamp for mix_correct algorithm. @@ -315,7 +318,9 @@ class SrsMixQueue class SrsOriginHub : public ISrsReloadHandler { private: - SrsLiveSource* source; + // Because source references to this object, so we should directly use the source ptr. + SrsLiveSource* source_; +private: SrsRequest* req_; bool is_active; private: @@ -341,7 +346,7 @@ class SrsOriginHub : public ISrsReloadHandler public: // Initialize the hub with source and request. // @param r The request object, managed by source. - virtual srs_error_t initialize(SrsLiveSource* s, SrsRequest* r); + virtual srs_error_t initialize(SrsSharedPtr s, SrsRequest* r); // Dispose the hub, release utilities resource, // For example, delete all HLS pieces. virtual void dispose(); @@ -443,7 +448,7 @@ class SrsLiveSourceManager : public ISrsHourGlass { private: srs_mutex_t lock; - std::map pool; + std::map< std::string, SrsSharedPtr > pool; SrsHourGlass* timer_; public: SrsLiveSourceManager(); @@ -454,10 +459,10 @@ class SrsLiveSourceManager : public ISrsHourGlass // @param r the client request. // @param h the event handler for source. // @param pps the matched source, if success never be NULL. - virtual srs_error_t fetch_or_create(SrsRequest* r, ISrsLiveSourceHandler* h, SrsLiveSource** pps); + virtual srs_error_t fetch_or_create(SrsRequest* r, ISrsLiveSourceHandler* h, SrsSharedPtr& pps); public: // Get the exists source, NULL when not exists. - virtual SrsLiveSource* fetch(SrsRequest* r); + virtual SrsSharedPtr fetch(SrsRequest* r); public: // dispose and cycle all sources. virtual void dispose(); @@ -539,7 +544,7 @@ class SrsLiveSource : public ISrsReloadHandler bool publisher_is_idle_for(srs_utime_t timeout); public: // Initialize the hls with handlers. - virtual srs_error_t initialize(SrsRequest* r, ISrsLiveSourceHandler* h); + virtual srs_error_t initialize(SrsSharedPtr wrapper, SrsRequest* r, ISrsLiveSourceHandler* h); // Bridge to other source, forward packets to it. void set_bridge(ISrsStreamBridge* v); // Interface ISrsReloadHandler diff --git a/trunk/src/app/srs_app_srt_conn.cpp b/trunk/src/app/srs_app_srt_conn.cpp index a1a4673381..47d3428953 100644 --- a/trunk/src/app/srs_app_srt_conn.cpp +++ b/trunk/src/app/srs_app_srt_conn.cpp @@ -368,16 +368,16 @@ srs_error_t SrsMpegtsSrtConn::acquire_publish() } // Check rtmp stream is busy. - SrsLiveSource *live_source = _srs_sources->fetch(req_); - if (live_source && !live_source->can_publish(false)) { + SrsSharedPtr live_source = _srs_sources->fetch(req_); + if (live_source.get() && !live_source->can_publish(false)) { return srs_error_new(ERROR_SYSTEM_STREAM_BUSY, "live_source stream %s busy", req_->get_stream_url().c_str()); } - if ((err = _srs_sources->fetch_or_create(req_, _srs_hybrid->srs()->instance(), &live_source)) != srs_success) { + if ((err = _srs_sources->fetch_or_create(req_, _srs_hybrid->srs()->instance(), live_source)) != srs_success) { return srs_error_wrap(err, "create source"); } - srs_assert(live_source != NULL); + srs_assert(live_source.get() != NULL); bool enabled_cache = _srs_config->get_gop_cache(req_->vhost); int gcmf = _srs_config->get_gop_cache_max_frames(req_->vhost); @@ -489,7 +489,7 @@ srs_error_t SrsMpegtsSrtConn::do_playing() SrsSrtConsumer* consumer = NULL; SrsAutoFree(SrsSrtConsumer, consumer); - if ((err = srt_source_->create_consumer(srt_source_, consumer)) != srs_success) { + if ((err = srt_source_->create_consumer(consumer)) != srs_success) { return srs_error_wrap(err, "create consumer, ts source=%s", req_->get_stream_url().c_str()); } srs_assert(consumer); diff --git a/trunk/src/app/srs_app_srt_source.cpp b/trunk/src/app/srs_app_srt_source.cpp index a0dfa60204..6b65c34773 100644 --- a/trunk/src/app/srs_app_srt_source.cpp +++ b/trunk/src/app/srs_app_srt_source.cpp @@ -152,7 +152,7 @@ void SrsSrtSourceManager::eliminate(SrsRequest* r) SrsSrtSourceManager* _srs_srt_sources = NULL; -SrsSrtConsumer::SrsSrtConsumer(SrsSharedPtr s) +SrsSrtConsumer::SrsSrtConsumer(SrsSrtSource* s) { source_ = s; should_update_source_id = false; @@ -942,11 +942,11 @@ void SrsSrtSource::set_bridge(ISrsStreamBridge* bridge) frame_builder_ = new SrsSrtFrameBuilder(bridge); } -srs_error_t SrsSrtSource::create_consumer(SrsSharedPtr source, SrsSrtConsumer*& consumer) +srs_error_t SrsSrtSource::create_consumer(SrsSrtConsumer*& consumer) { srs_error_t err = srs_success; - consumer = new SrsSrtConsumer(source); + consumer = new SrsSrtConsumer(this); consumers.push_back(consumer); return err; diff --git a/trunk/src/app/srs_app_srt_source.hpp b/trunk/src/app/srs_app_srt_source.hpp index 46d5b8d7e2..4bb3a014bb 100644 --- a/trunk/src/app/srs_app_srt_source.hpp +++ b/trunk/src/app/srs_app_srt_source.hpp @@ -70,10 +70,12 @@ extern SrsSrtSourceManager* _srs_srt_sources; class SrsSrtConsumer { public: - SrsSrtConsumer(SrsSharedPtr source); + SrsSrtConsumer(SrsSrtSource* source); virtual ~SrsSrtConsumer(); private: - SrsSharedPtr source_; + // Because source references to this object, so we should directly use the source ptr. + SrsSrtSource* source_; +private: std::vector queue; // when source id changed, notice all consumers bool should_update_source_id; @@ -167,7 +169,7 @@ class SrsSrtSource public: // Create consumer // @param consumer, output the create consumer. - virtual srs_error_t create_consumer(SrsSharedPtr source, SrsSrtConsumer*& consumer); + virtual srs_error_t create_consumer(SrsSrtConsumer*& consumer); // Dumps packets in cache to consumer. virtual srs_error_t consumer_dumps(SrsSrtConsumer* consumer); virtual void on_consumer_destroy(SrsSrtConsumer* consumer); diff --git a/trunk/src/app/srs_app_stream_bridge.cpp b/trunk/src/app/srs_app_stream_bridge.cpp index c024974ad8..d043384159 100644 --- a/trunk/src/app/srs_app_stream_bridge.cpp +++ b/trunk/src/app/srs_app_stream_bridge.cpp @@ -25,9 +25,9 @@ ISrsStreamBridge::~ISrsStreamBridge() { } -SrsFrameToRtmpBridge::SrsFrameToRtmpBridge(SrsLiveSource *src) +SrsFrameToRtmpBridge::SrsFrameToRtmpBridge(SrsSharedPtr source) { - source_ = src; + source_ = source; } SrsFrameToRtmpBridge::~SrsFrameToRtmpBridge() diff --git a/trunk/src/app/srs_app_stream_bridge.hpp b/trunk/src/app/srs_app_stream_bridge.hpp index de9378669d..e78aad96b8 100644 --- a/trunk/src/app/srs_app_stream_bridge.hpp +++ b/trunk/src/app/srs_app_stream_bridge.hpp @@ -42,9 +42,9 @@ class ISrsStreamBridge class SrsFrameToRtmpBridge : public ISrsStreamBridge { private: - SrsLiveSource *source_; + SrsSharedPtr source_; public: - SrsFrameToRtmpBridge(SrsLiveSource *src); + SrsFrameToRtmpBridge(SrsSharedPtr source); virtual ~SrsFrameToRtmpBridge(); public: srs_error_t initialize(SrsRequest* r); diff --git a/trunk/src/core/srs_core_autofree.hpp b/trunk/src/core/srs_core_autofree.hpp index a84aaeca1f..7bc3935c83 100644 --- a/trunk/src/core/srs_core_autofree.hpp +++ b/trunk/src/core/srs_core_autofree.hpp @@ -11,6 +11,9 @@ #include +// The auto free helper, which is actually the unique ptr, without the move feature, +// see https://github.com/ossrs/srs/discussions/3667#discussioncomment-8969107 +// // To free the instance in the current scope, for instance, MyClass* ptr, // which is a ptr and this class will: // 1. free the ptr. @@ -81,7 +84,9 @@ class impl_SrsAutoFree } }; -// Shared ptr smart pointer, see https://github.com/ossrs/srs/discussions/3667#discussioncomment-8969107 +// Shared ptr smart pointer, only support shared ptr, no weak ptr, no shared from this, no inheritance, +// no comparing, see https://github.com/ossrs/srs/discussions/3667#discussioncomment-8969107 +// // Usage: // SrsSharedPtr ptr(new MyClass()); // ptr->do_something(); diff --git a/trunk/src/core/srs_core_version6.hpp b/trunk/src/core/srs_core_version6.hpp index 30399a8b48..50ebfc6477 100644 --- a/trunk/src/core/srs_core_version6.hpp +++ b/trunk/src/core/srs_core_version6.hpp @@ -9,6 +9,6 @@ #define VERSION_MAJOR 6 #define VERSION_MINOR 0 -#define VERSION_REVISION 128 +#define VERSION_REVISION 129 #endif diff --git a/trunk/src/kernel/srs_kernel_error.hpp b/trunk/src/kernel/srs_kernel_error.hpp index 46a94f436a..2b1fa01c1d 100644 --- a/trunk/src/kernel/srs_kernel_error.hpp +++ b/trunk/src/kernel/srs_kernel_error.hpp @@ -107,6 +107,7 @@ XX(ERROR_BACKTRACE_ADDR2LINE , 1094, "BacktraceAddr2Line", "Backtrace addr2line failed") \ XX(ERROR_SYSTEM_FILE_NOT_OPEN , 1095, "FileNotOpen", "File is not opened") \ XX(ERROR_SYSTEM_FILE_SETVBUF , 1096, "FileSetVBuf", "Failed to set file vbuf") \ + XX(ERROR_NO_SOURCE , 1097, "NoSource", "No source found") /**************************************************/ /* RTMP protocol error. */ @@ -334,7 +335,7 @@ XX(ERROR_STREAM_CASTER_HEVC_FORMAT , 4057, "CasterTsHevcFormat", "Invalid ts HEVC Format for stream caster") \ XX(ERROR_HTTP_JSONP , 4058, "HttpJsonp", "Invalid callback for JSONP") \ XX(ERROR_HEVC_NALU_UEV , 4059, "HevcNaluUev", "Failed to read UEV for HEVC NALU") \ - XX(ERROR_HEVC_NALU_SEV , 4060, "HevcNaluSev", "Failed to read SEV for HEVC NALU") \ + XX(ERROR_HEVC_NALU_SEV , 4060, "HevcNaluSev", "Failed to read SEV for HEVC NALU") /**************************************************/ diff --git a/trunk/src/utest/srs_utest_core.cpp b/trunk/src/utest/srs_utest_core.cpp index 2344e911ef..155621be0b 100644 --- a/trunk/src/utest/srs_utest_core.cpp +++ b/trunk/src/utest/srs_utest_core.cpp @@ -139,6 +139,17 @@ VOID TEST(CoreLogger, SharedPtrReset) } } +SrsSharedPtr mock_create_from_ptr(SrsSharedPtr p) { + return p; +} + +VOID TEST(CoreLogger, SharedPtrContructor) +{ + int* p = new int(100); + SrsSharedPtr q = mock_create_from_ptr(p); + EXPECT_EQ(100, *q); +} + VOID TEST(CoreLogger, SharedPtrObject) { SrsSharedPtr p(new MyNormalObject(100));