Should be atomic code for checking stream busy.

[winlinvip]

The code is ugly and under risk, which is not atomic:

srs_error_t SrsRtmpConn::acquire_publish(SrsLiveSource* source)
{
    srs_error_t err = srs_success;
    
    SrsRequest* req = info->req;

    // Check whether RTMP stream is busy.
    if (!source->can_publish(info->edge)) {
        return srs_error_new(ERROR_SYSTEM_STREAM_BUSY, "rtmp: stream %s is busy", req->get_stream_url().c_str());
    }

    // Check whether RTC stream is busy.
#ifdef SRS_RTC
    SrsRtcSource* rtc = NULL;
    bool rtc_server_enabled = _srs_config->get_rtc_server_enabled();
    bool rtc_enabled = _srs_config->get_rtc_enabled(req->vhost);
    if (rtc_server_enabled && rtc_enabled && !info->edge) {
        if ((err = _srs_rtc_sources->fetch_or_create(req, &rtc)) != srs_success) {
            return srs_error_wrap(err, "create source");
        }

        if (!rtc->can_publish()) {
            return srs_error_new(ERROR_SYSTEM_STREAM_BUSY, "rtc stream %s busy", req->get_stream_url().c_str());
        }
    }
#endif

    // Check whether SRT stream is busy.
#ifdef SRS_SRT
    SrsSrtSource* srt = NULL;
    bool srt_server_enabled = _srs_config->get_srt_enabled();
    bool srt_enabled = _srs_config->get_srt_enabled(req->vhost);
    if (srt_server_enabled && srt_enabled && !info->edge) {
        if ((err = _srs_srt_sources->fetch_or_create(req, &srt)) != srs_success) {
            return srs_error_wrap(err, "create source");
        }

        if (!srt->can_publish()) {
            return srs_error_new(ERROR_SYSTEM_STREAM_BUSY, "srt stream %s busy", req->get_stream_url().c_str());
        }
    }
#endif

And for SRT stream:

srs_error_t SrsMpegtsSrtConn::acquire_publish()
{
    srs_error_t err = srs_success;

    // Check srt stream is busy.
    if (! srt_source_->can_publish()) {
        return srs_error_new(ERROR_SRT_SOURCE_BUSY, "srt stream %s busy", req_->get_stream_url().c_str());
    }

    // Check rtmp stream is busy.
    SrsLiveSource *live_source = _srs_sources->fetch(req_);
    if (live_source && !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) {
        return srs_error_wrap(err, "create source");
    }

    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);
    live_source->set_cache(enabled_cache);
    live_source->set_gop_cache_max_frames(gcmf);

    // srt->rtmp->rtc
    // TODO: FIXME: the code below is repeat in srs_app_rtmp_conn.cpp, refactor it later, use function instead.

    // Check whether RTC stream is busy.
#ifdef SRS_RTC
    SrsRtcSource* rtc = NULL;
    bool rtc_server_enabled = _srs_config->get_rtc_server_enabled();
    bool rtc_enabled = _srs_config->get_rtc_enabled(req_->vhost);
    bool edge = _srs_config->get_vhost_is_edge(req_->vhost);
    if (rtc_server_enabled && rtc_enabled && ! edge) {
        if ((err = _srs_rtc_sources->fetch_or_create(req_, &rtc)) != srs_success) {
            return srs_error_wrap(err, "create source");
        }

        if (!rtc->can_publish()) {
            return srs_error_new(ERROR_SYSTEM_STREAM_BUSY, "rtc stream %s busy", req_->get_stream_url().c_str());
        }
    }
#endif