胡斌

add forward_peer,to support forward between origins.the streams pushed from peer

origins will not be forward any more.
remove forward_server_srs,which used to indicate the forward target server is or is not srs,
add forward_server_other, to list the target rtmp server other than srs in.
... ... @@ -11,6 +11,9 @@ vhost __defaultVhost__ {
forward 127.0.0.1:19350;
#select on of the listed servers in turn to forward,not all of them
forward_in_turn 127.0.0.1:19351 127.0.0.1:19352;
#the forward destination server type,default true. if the server is not srs server,connect app maybe fail,try config as false
forward_server_srs true;
#if the forward destination server is not srs,such as some rtmp server in cdn,write the address of them below
forward_server_other 127.0.0.1:19350;
#if the forward server is same as this server , a origin,try use forward_peer,
#the stream pushed from other forward peer will not forward any more in this server
forward_peer 127.0.0.1:1936;
}
... ...
... ... @@ -1823,7 +1823,7 @@ int SrsConfig::check_config()
&& n != "dvr" && n != "ingest" && n != "hls" && n != "http_hooks"
&& n != "gop_cache" && n != "queue_length"
&& n != "refer" && n != "refer_publish" && n != "refer_play"
&& n != "forward" && n != "forward_server_srs" && n != "forward_in_turn" && n != "transcode" && n != "bandcheck"
&& n != "forward" && n != "forward_server_other" && n != "forward_in_turn" && n != "forward_peer" && n != "transcode" && n != "bandcheck"
&& n != "time_jitter" && n != "mix_correct"
&& n != "atc" && n != "atc_auto"
&& n != "debug_srs_upnode"
... ... @@ -1939,15 +1939,6 @@ int SrsConfig::check_config()
return ret;
}
}*/
} else if (n == "forward_server_srs") {
for (int j = 0; j < (int)conf->directives.size(); j++) {
string m = conf->at(j)->name.c_str();
if (m != "true" && m != "false") {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("unsupported vhost forward_server_srs directive %s(only support true or flase), ret=%d", m.c_str(), ret);
return ret;
}
}
}
else if (n == "security") {
for (int j = 0; j < (int)conf->directives.size(); j++) {
... ... @@ -2756,20 +2747,15 @@ SrsConfDirective* SrsConfig::get_forward(string vhost)
return conf->get("forward");
}
bool SrsConfig::get_forward_server_srs(string vhost)
SrsConfDirective* SrsConfig::get_forward_server_other(string vhost)
{
SrsConfDirective* conf = get_vhost(vhost);
if (!conf) {
return true;
}
conf = conf->get("forward_server_srs");
if (!conf || conf->arg0().empty()) {
return true;
return NULL;
}
return conf->arg0()=="true";
return conf->get("forward_server_other");
}
SrsConfDirective* SrsConfig::get_forward_in_turn(string vhost)
... ... @@ -2783,6 +2769,17 @@ SrsConfDirective* SrsConfig::get_forward_in_turn(string vhost)
return conf->get("forward_in_turn");
}
SrsConfDirective* SrsConfig::get_forward_peer(string vhost)
{
SrsConfDirective* conf = get_vhost(vhost);
if (!conf) {
return NULL;
}
return conf->get("forward_peer");
}
SrsConfDirective* SrsConfig::get_vhost_http_hooks(string vhost)
{
SrsConfDirective* conf = get_vhost(vhost);
... ...
... ... @@ -570,13 +570,17 @@ public:
*/
virtual SrsConfDirective* get_forward(std::string vhost);
/**
* get the forward server type of vhost,if srs return true,else false.
* get the forward_server_other directive of vhost
*/
virtual bool get_forward_server_srs(std::string vhost);
virtual SrsConfDirective* get_forward_server_other(std::string vhost);
/**
* get the forward_in_turn directive of vhost.
*/
virtual SrsConfDirective* get_forward_in_turn(std::string vhost);
/**
* get the forward_peer directive of vhost.
*/
virtual SrsConfDirective* get_forward_peer(std::string vhost);
// http_hooks section
private:
/**
... ...
... ... @@ -52,6 +52,8 @@ SrsForwarder::SrsForwarder(SrsSource* _source)
{
source = _source;
server_type = 0;
_req = NULL;
io = NULL;
client = NULL;
... ... @@ -117,13 +119,7 @@ int SrsForwarder::check_dead_loop(SrsRequest* req, std::string ep_forward)
if (source_ep == dest_ep) {
ret = ERROR_SYSTEM_FORWARD_LOOP;
srs_warn("forward loop detected. src=%s, dest=%s, ret=%d",
source_ep.c_str(), dest_ep.c_str(), ret);
return ret;
}
srs_trace("start forward %s to %s, tcUrl=%s, stream=%s",
source_ep.c_str(), dest_ep.c_str(), tc_url.c_str(),
req->stream.c_str());
return ret;
}
... ... @@ -139,6 +135,15 @@ int SrsForwarder::initialize(SrsRequest* req, string ep_forward)
// the ep(endpoint) to forward to
_ep_forward = ep_forward;
SrsConfDirective* conf = _srs_config->get_forward_server_other(_req->vhost);
for (int i = 0; conf && i < (int)conf->args.size(); i++) {
std::string forward_server = conf->args.at(i);
if(ep_forward == forward_server){
server_type = 1;
break;
}
}
return ret;
}
... ... @@ -428,13 +433,12 @@ int SrsForwarder::connect_app(string ep_server, string ep_port)
bool debug_srs_upnode = _srs_config->get_debug_srs_upnode(req->vhost);
//if forward_server is not srs,or srs compatible,don't send req when connect app,in case of be rejected
bool forward_server_srs = _srs_config->get_forward_server_srs(req->vhost);
if (!forward_server_srs) {
if (server_type) {
srs_trace("forward to srs incompatible server, tcUrl=%s,vhost:%s",
tc_url.c_str(), req->vhost.c_str());
}
if ((ret = client->connect_app(req->app, tc_url, forward_server_srs ? req : NULL, debug_srs_upnode)) != ERROR_SUCCESS) {
if ((ret = client->connect_app(req->app, tc_url, server_type ? NULL : req, debug_srs_upnode)) != ERROR_SUCCESS) {
srs_error("connect with server failed, tcUrl=%s, dsu=%d. ret=%d",
tc_url.c_str(), debug_srs_upnode, ret);
return ret;
... ...
... ... @@ -71,6 +71,8 @@ private:
*/
SrsSharedPtrMessage* sh_audio;
SrsSharedPtrMessage* sh_video;
int server_type; //server_type 0, srs, 1,other
public:
SrsForwarder(SrsSource* _source);
virtual ~SrsForwarder();
... ...
... ... @@ -2424,11 +2424,57 @@ int SrsSource::create_forwarders()
{
int ret = ERROR_SUCCESS;
SrsConfDirective* conf = _srs_config->get_forward(_req->vhost);
//forward_peer
SrsConfDirective* conf = _srs_config->get_forward_peer(_req->vhost);
for (int i = 0; conf && i < (int)conf->args.size(); i++) {
std::string forward_server = conf->args.at(i);
if (SrsForwarder::check_dead_loop(_req, forward_server)) {//if the source is from any of the peer list,don't forward it
srs_trace("the source is come from forward_peer,don't forward it any more: "
"vhost=%s, app=%s, stream=%s, forward-to=%s",
_req->vhost.c_str(), _req->app.c_str(), _req->stream.c_str(),
forward_server.c_str());
return ret;
}
}
for (int i = 0; conf && i < (int)conf->args.size(); i++) {
std::string forward_server = conf->args.at(i);
SrsForwarder* forwarder = new SrsForwarder(this);
forwarders.push_back(forwarder);
// initialize the forwarder with request.
if ((ret = forwarder->initialize(_req, forward_server)) != ERROR_SUCCESS) {
return ret;
}
double queue_size = _srs_config->get_queue_length(_req->vhost);
forwarder->set_queue_size(queue_size);
srs_trace("start forward_peer: "
"vhost=%s, app=%s, stream=%s, forward-to=%s",
_req->vhost.c_str(), _req->app.c_str(), _req->stream.c_str(),
forward_server.c_str());
if ((ret = forwarder->on_publish()) != ERROR_SUCCESS) {
srs_error("start forwarder failed. "
"vhost=%s, app=%s, stream=%s, forward-to=%s",
_req->vhost.c_str(), _req->app.c_str(), _req->stream.c_str(),
forward_server.c_str());
return ret;
}
}
//forward
conf = _srs_config->get_forward(_req->vhost);
for (int i = 0; conf && i < (int)conf->args.size(); i++) {
std::string forward_server = conf->args.at(i);
if (SrsForwarder::check_dead_loop(_req, forward_server)) {
srs_trace("start forward,dead loop checked,don't forward it: "
"vhost=%s, app=%s, stream=%s, forward-to=%s",
_req->vhost.c_str(), _req->app.c_str(), _req->stream.c_str(),
forward_server.c_str());
continue;
}
... ... @@ -2474,6 +2520,10 @@ int SrsSource::create_forwarders()
server_index ++;
if (SrsForwarder::check_dead_loop(_req, forward_server)) {
srs_trace("start forward_in_turn,dead loop checked,don't forward it: "
"vhost=%s, app=%s, stream=%s, forward-to=%s",
_req->vhost.c_str(), _req->app.c_str(), _req->stream.c_str(),
forward_server.c_str());
continue;
}
... ... @@ -2488,6 +2538,11 @@ int SrsSource::create_forwarders()
double queue_size = _srs_config->get_queue_length(_req->vhost);
forwarder->set_queue_size(queue_size);
srs_trace("start forward_in_turn: "
"vhost=%s, app=%s, stream=%s, forward-to=%s",
_req->vhost.c_str(), _req->app.c_str(), _req->stream.c_str(),
forward_server.c_str());
if ((ret = forwarder->on_publish()) != ERROR_SUCCESS) {
srs_error("start forwarder failed. "
"vhost=%s, app=%s, stream=%s, forward-to=%s",
... ...