zhengfl

fix #438

问题:edge模式,推流时异常断开。
    解决方法:增加edge模式推流检测。
... ... @@ -806,6 +806,11 @@ int SrsPublishEdge::initialize(SrsSource* source, SrsRequest* req)
return ret;
}
bool SrsPublishEdge::can_publish()
{
return state != SrsEdgeStatePublish;
}
int SrsPublishEdge::on_client_publish()
{
int ret = ERROR_SUCCESS;
... ...
... ... @@ -207,6 +207,7 @@ public:
virtual void set_queue_size(double queue_size);
public:
virtual int initialize(SrsSource* source, SrsRequest* req);
virtual bool can_publish();
/**
* when client publish stream on edge.
*/
... ...
... ... @@ -423,27 +423,20 @@ int SrsRtmpConn::stream_service_cycle()
}
}
srs_assert(source != NULL);
// check ASAP, to fail it faster if invalid.
if (type != SrsRtmpConnPlay) {
if ((ret = prepare_publish(source, vhost_is_edge)) != ERROR_SUCCESS) {
return ret;
}
}
// update the statistic when source disconveried.
SrsStatistic* stat = SrsStatistic::instance();
if ((ret = stat->on_client(_srs_context->get_id(), req)) != ERROR_SUCCESS) {
srs_error("stat client failed. ret=%d", ret);
return ret;
}
// check ASAP, to fail it faster if invalid.
if (type != SrsRtmpConnPlay && !vhost_is_edge) {
// check publish available
// for edge, never check it, for edge use proxy mode.
if (!source->can_publish()) {
ret = ERROR_SYSTEM_STREAM_BUSY;
srs_warn("stream %s is already publishing. ret=%d",
req->get_stream_url().c_str(), ret);
// to delay request
st_usleep(SRS_STREAM_BUSY_SLEEP_US);
return ret;
}
}
bool enabled_cache = _srs_config->get_gop_cache(req->vhost);
srs_trace("source url=%s, ip=%s, cache=%d, is_edge=%d, source_id=%d[%d]",
... ... @@ -1260,6 +1253,31 @@ int SrsRtmpConn::do_token_traverse_auth(SrsRtmpClient* client)
return ret;
}
int SrsRtmpConn::prepare_publish(SrsSource* source, bool vhost_is_edge)
{
int ret = ERROR_SUCCESS;
srs_assert(source);
// check publish available
bool can_publish = false;
if (vhost_is_edge) {
can_publish = source->proxy_can_publish();
} else {
can_publish = source->can_publish();
}
if (!can_publish) {
ret = ERROR_SYSTEM_STREAM_BUSY;
srs_warn("stream %s is already publishing. ret=%d",
req->get_stream_url().c_str(), ret);
// to delay request
st_usleep(SRS_STREAM_BUSY_SLEEP_US);
return ret;
}
return ret;
}
int SrsRtmpConn::http_hooks_on_connect()
{
int ret = ERROR_SUCCESS;
... ...
... ... @@ -123,6 +123,7 @@ private:
virtual int check_edge_token_traverse_auth();
virtual int connect_server(int origin_index, st_netfd_t* pstsock);
virtual int do_token_traverse_auth(SrsRtmpClient* client);
virtual int prepare_publish(SrsSource* source, bool vhost_is_edge);
private:
virtual int http_hooks_on_connect();
virtual void http_hooks_on_close();
... ...
... ... @@ -1355,6 +1355,11 @@ bool SrsSource::can_publish()
return _can_publish;
}
bool SrsSource::proxy_can_publish()
{
return publish_edge->can_publish();
}
int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata)
{
int ret = ERROR_SUCCESS;
... ...
... ... @@ -538,6 +538,7 @@ public:
// logic data methods
public:
virtual bool can_publish();
virtual bool proxy_can_publish();
virtual int on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata);
public:
virtual int on_audio(SrsCommonMessage* audio);
... ...