胡斌

support forward between two origin,to make a full backup of top server

... ... @@ -77,6 +77,57 @@ SrsForwarder::~SrsForwarder()
srs_freep(sh_audio);
}
int SrsForwarder::check_dead_loop(SrsRequest* req, std::string ep_forward)
{
int ret = ERROR_SUCCESS;
std::string server, port, tc_url;
server = ep_forward;
port = SRS_CONSTS_RTMP_DEFAULT_PORT;
// TODO: FIXME: parse complex params
size_t pos = ep_forward.find(":");
if (pos != std::string::npos) {
port = ep_forward.substr(pos + 1);
server = ep_forward.substr(0, pos);
}
// generate tcUrl
tc_url = srs_generate_tc_url(server, req->vhost, req->app, port, req->param);
// dead loop check
std::string source_ep = "rtmp://";
source_ep += req->host;
source_ep += ":";
source_ep += req->port;
source_ep += "?vhost=";
source_ep += req->vhost;
std::string dest_ep = "rtmp://";
if (ep_forward == SRS_CONSTS_LOCALHOST) {
dest_ep += req->host;
}
else {
dest_ep += server;
}
dest_ep += ":";
dest_ep += port;
dest_ep += "?vhost=";
dest_ep += req->vhost;
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;
}
int SrsForwarder::initialize(SrsRequest* req, string ep_forward)
{
int ret = ERROR_SUCCESS;
... ... @@ -99,7 +150,7 @@ void SrsForwarder::set_queue_size(double queue_size)
int SrsForwarder::on_publish()
{
int ret = ERROR_SUCCESS;
#if 0
SrsRequest* req = _req;
// discovery the server port and tcUrl from req and ep_forward.
... ... @@ -134,7 +185,7 @@ int SrsForwarder::on_publish()
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());
#endif
if ((ret = pthread->start()) != ERROR_SUCCESS) {
srs_error("start srs thread failed. ret=%d", ret);
return ret;
... ...
... ... @@ -75,6 +75,7 @@ public:
SrsForwarder(SrsSource* _source);
virtual ~SrsForwarder();
public:
static int check_dead_loop(SrsRequest* req, std::string ep_forward);
virtual int initialize(SrsRequest* req, std::string ep_forward);
virtual void set_queue_size(double queue_size);
public:
... ...
... ... @@ -2427,6 +2427,10 @@ int SrsSource::create_forwarders()
SrsConfDirective* 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)) {
continue;
}
SrsForwarder* forwarder = new SrsForwarder(this);
forwarders.push_back(forwarder);
... ... @@ -2459,31 +2463,39 @@ int SrsSource::create_forwarders()
if(servers_size <= 0){
return ret;
}
for(int i = 0; i < servers_size; i++){
if(server_index >= servers_size){
server_index = 0;
}
if(server_index >= servers_size){
server_index = 0;
}
std::string forward_server = conf->args.at(server_index);
server_index ++;
std::string forward_server = conf->args.at(server_index);
server_index ++;
SrsForwarder* forwarder = new SrsForwarder(this);
forwarders.push_back(forwarder);
if (SrsForwarder::check_dead_loop(_req, forward_server)) {
continue;
}
// initialize the forwarder with request.
if ((ret = forwarder->initialize(_req, forward_server)) != ERROR_SUCCESS) {
return ret;
}
SrsForwarder* forwarder = new SrsForwarder(this);
forwarders.push_back(forwarder);
double queue_size = _srs_config->get_queue_length(_req->vhost);
forwarder->set_queue_size(queue_size);
// initialize the forwarder with request.
if ((ret = forwarder->initialize(_req, forward_server)) != ERROR_SUCCESS) {
return ret;
}
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;
double queue_size = _srs_config->get_queue_length(_req->vhost);
forwarder->set_queue_size(queue_size);
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;
}
break;
}
return ret;
... ...
... ... @@ -1764,6 +1764,11 @@ void SrsRequest::update_auth(SrsRequest* req)
pageUrl = req->pageUrl;
swfUrl = req->swfUrl;
tcUrl = req->tcUrl;
vhost = req->vhost;
app = req->app;
param = req->param;
port = req->port;
if (args) {
srs_freep(args);
... ...