Skip to content

Commit

Permalink
SmartPtr: Rename source to live_source.
Browse files Browse the repository at this point in the history
  • Loading branch information
winlinvip committed Jun 14, 2024
1 parent e34b3d3 commit 846f95e
Show file tree
Hide file tree
Showing 13 changed files with 81 additions and 81 deletions.
24 changes: 12 additions & 12 deletions trunk/src/app/srs_app_edge.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -417,7 +417,7 @@ SrsEdgeIngester::~SrsEdgeIngester()

srs_error_t SrsEdgeIngester::initialize(SrsLiveSource* s, SrsPlayEdge* e, SrsRequest* r)
{
source = s;
source_ = s;
edge = e;
req = r;

Expand All @@ -435,7 +435,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");
}

Expand All @@ -455,8 +455,8 @@ void SrsEdgeIngester::stop()
upstream->close();

// Notify source to un-publish if exists.
if (source) {
source->on_unpublish();
if (source_) {
source_->on_unpublish();
}
}

Expand Down Expand Up @@ -549,7 +549,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");
}

Expand Down Expand Up @@ -635,21 +635,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;
Expand All @@ -665,7 +665,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;
Expand Down Expand Up @@ -725,7 +725,7 @@ SrsEdgeForwarder::SrsEdgeForwarder()
edge = NULL;
req = NULL;
send_error_code = ERROR_SUCCESS;
source = NULL;
source_ = NULL;

sdk = NULL;
lb = new SrsLbRoundRobin();
Expand All @@ -749,7 +749,7 @@ void SrsEdgeForwarder::set_queue_size(srs_utime_t queue_size)

srs_error_t SrsEdgeForwarder::initialize(SrsLiveSource* s, SrsPublishEdge* e, SrsRequest* r)
{
source = s;
source_ = s;
edge = e;
req = r;

Expand Down
4 changes: 2 additions & 2 deletions trunk/src/app/srs_app_edge.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ class SrsEdgeFlvUpstream : public SrsEdgeUpstream
class SrsEdgeIngester : public ISrsCoroutineHandler
{
private:
SrsLiveSource* source;
SrsLiveSource* source_;
SrsPlayEdge* edge;
SrsRequest* req;
SrsCoroutine* trd;
Expand Down Expand Up @@ -172,7 +172,7 @@ class SrsEdgeIngester : public ISrsCoroutineHandler
class SrsEdgeForwarder : public ISrsCoroutineHandler
{
private:
SrsLiveSource* source;
SrsLiveSource* source_;
SrsPublishEdge* edge;
SrsRequest* req;
SrsCoroutine* trd;
Expand Down
30 changes: 15 additions & 15 deletions trunk/src/app/srs_app_http_stream.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -106,19 +106,19 @@ srs_error_t SrsBufferCache::cycle()
return err;
}

SrsLiveSource* source = _srs_sources->fetch(req);
if (!source) {
SrsLiveSource* live_source = _srs_sources->fetch(req);
if (!live_source) {
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");
}

Expand Down Expand Up @@ -661,18 +661,18 @@ 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);

SrsLiveSource* source = _srs_sources->fetch(req);
if (!source) {
SrsLiveSource* live_source = _srs_sources->fetch(req);
if (!live_source) {
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");
}

Expand All @@ -694,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");
}
}
Expand Down Expand Up @@ -1136,16 +1136,16 @@ srs_error_t SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandle
}
}

SrsLiveSource* s = NULL;
if ((err = _srs_sources->fetch_or_create(r, server, &s)) != srs_success) {
SrsLiveSource* live_source = NULL;
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 != 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(r)) != srs_success) {
Expand All @@ -1163,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;
}
Expand Down
4 changes: 2 additions & 2 deletions trunk/src/app/srs_app_recv_thread.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -265,7 +265,7 @@ SrsPublishRecvThread::SrsPublishRecvThread(SrsRtmpServer* rtmp_sdk, SrsRequest*
rtmp = rtmp_sdk;

_conn = conn;
_source = source;
source_ = source;

nn_msgs_for_yield_ = 0;
recv_error = srs_success;
Expand Down Expand Up @@ -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.
Expand Down
2 changes: 1 addition & 1 deletion trunk/src/app/srs_app_recv_thread.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ class SrsPublishRecvThread : public ISrsMessagePumper, public ISrsReloadHandler
srs_error_t recv_error;
SrsRtmpConn* _conn;
// The params for conn callback.
SrsLiveSource* _source;
SrsLiveSource* source_;
// The error timeout cond
srs_cond_t error;
// The merged context id.
Expand Down
4 changes: 2 additions & 2 deletions trunk/src/app/srs_app_rtc_api.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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()) {
SrsLiveSource* live_source = _srs_sources->fetch(ruc->req_);
if (live_source && !live_source->inactive()) {
return srs_error_new(ERROR_RTC_DISABLED, "Disabled rtmp_to_rtc of %s, see #2728", ruc->req_->vhost.c_str());
}
}
Expand Down
10 changes: 5 additions & 5 deletions trunk/src/app/srs_app_rtc_conn.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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)) {
SrsLiveSource* live_source = _srs_sources->fetch(r);
if (live_source && !live_source->can_publish(false)) {
return srs_error_new(ERROR_SYSTEM_STREAM_BUSY, "rtmp stream %s busy", r->get_stream_url().c_str());
}

Expand All @@ -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);
Expand Down
22 changes: 11 additions & 11 deletions trunk/src/app/srs_app_rtmp_conn.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
SrsLiveSource* live_source = NULL;
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 != 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: {
Expand All @@ -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;
Expand All @@ -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) {
Expand All @@ -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) {
Expand All @@ -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);
Expand Down
Loading

0 comments on commit 846f95e

Please sign in to comment.