external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

For #913, http callback use complex error

Browse source
This commit is contained in:
winlin 2017-09-22 19:54:50 +08:00
parent 59b53dab8b
commit 20a42599f3
22 changed files with 264 additions and 268 deletions
Download patch file Download diff file Expand all files Collapse all files

6
trunk/src/app/srs_app_async_call.cpp
View file

@ -111,9 +111,9 @@ srs_error_t SrsAsyncCallWorker::cycle()
for (it = copy.begin(); it != copy.end(); ++it) { for (it = copy.begin(); it != copy.end(); ++it) {
ISrsAsyncCallTask* task = *it; ISrsAsyncCallTask* task = *it;
int ret = ERROR_SUCCESS; if ((err = task->call()) != srs_success) {
if ((ret = task->call()) != ERROR_SUCCESS) { srs_warn("ignore task failed %s", srs_error_desc(err).c_str());
srs_warn("ignore async callback %s, ret=%d", task->to_string().c_str(), ret); srs_freep(err);
} }
srs_freep(task); srs_freep(task);
} }

2
trunk/src/app/srs_app_async_call.hpp
View file

@ -50,7 +50,7 @@ public:
* this method is the actual execute method of task, * this method is the actual execute method of task,
* for example, to notify callback server. * for example, to notify callback server.
*/ */
virtual int call() = 0; virtual srs_error_t call() = 0;
/** /**
* convert task to string to describe it. * convert task to string to describe it.
* used for logger. * used for logger.

5
trunk/src/app/srs_app_conn.cpp
View file

@ -82,11 +82,10 @@ void SrsConnection::dispose()
srs_error_t SrsConnection::start() srs_error_t SrsConnection::start()
{ {
int ret = ERROR_SUCCESS;
srs_error_t err = srs_success; srs_error_t err = srs_success;
if ((ret = skt->initialize(stfd)) != ERROR_SUCCESS) { if ((err = skt->initialize(stfd)) != srs_success) {
return srs_error_new(ret, "socket"); return srs_error_wrap(err, "init socket");
} }
if ((err = trd->start()) != srs_success) { if ((err = trd->start()) != srs_success) {

15
trunk/src/app/srs_app_dvr.cpp
View file

@ -553,12 +553,12 @@ SrsDvrAsyncCallOnDvr::~SrsDvrAsyncCallOnDvr()
srs_freep(req); srs_freep(req);
} }
int SrsDvrAsyncCallOnDvr::call() srs_error_t SrsDvrAsyncCallOnDvr::call()
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) { if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
return ret; return err;
} }
// the http hooks will cause context switch, // the http hooks will cause context switch,
@ -571,7 +571,7 @@ int SrsDvrAsyncCallOnDvr::call()
if (!conf) { if (!conf) {
srs_info("ignore the empty http callback: on_dvr"); srs_info("ignore the empty http callback: on_dvr");
return ret; return err;
} }
hooks = conf->args; hooks = conf->args;
@ -579,13 +579,12 @@ int SrsDvrAsyncCallOnDvr::call()
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
if ((ret = SrsHttpHooks::on_dvr(cid, url, req, path)) != ERROR_SUCCESS) { if ((err = SrsHttpHooks::on_dvr(cid, url, req, path)) != srs_success) {
srs_error("hook client on_dvr failed. url=%s, ret=%d", url.c_str(), ret); return srs_error_wrap(err, "callback on_dvr %s", url.c_str());
return ret;
} }
} }
return ret; return err;
} }
string SrsDvrAsyncCallOnDvr::to_string() string SrsDvrAsyncCallOnDvr::to_string()

2
trunk/src/app/srs_app_dvr.hpp
View file

@ -174,7 +174,7 @@ public:
SrsDvrAsyncCallOnDvr(int c, SrsRequest* r, std::string p); SrsDvrAsyncCallOnDvr(int c, SrsRequest* r, std::string p);
virtual ~SrsDvrAsyncCallOnDvr(); virtual ~SrsDvrAsyncCallOnDvr();
public: public:
virtual int call(); virtual srs_error_t call();
virtual std::string to_string(); virtual std::string to_string();
}; };

9
trunk/src/app/srs_app_heartbeat.cpp
View file

@ -47,6 +47,7 @@ SrsHttpHeartbeat::~SrsHttpHeartbeat()
void SrsHttpHeartbeat::heartbeat() void SrsHttpHeartbeat::heartbeat()
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
std::string url = _srs_config->get_heartbeat_url(); std::string url = _srs_config->get_heartbeat_url();
@ -78,13 +79,17 @@ void SrsHttpHeartbeat::heartbeat()
} }
SrsHttpClient http; SrsHttpClient http;
if ((ret = http.initialize(uri.get_host(), uri.get_port())) != ERROR_SUCCESS) { if ((err = http.initialize(uri.get_host(), uri.get_port())) != srs_success) {
srs_freep(err);
return; return;
} }
std::string req = obj->dumps(); std::string req = obj->dumps();
ISrsHttpMessage* msg = NULL; ISrsHttpMessage* msg = NULL;
if ((ret = http.post(uri.get_path(), req, &msg)) != ERROR_SUCCESS) { if ((err = http.post(uri.get_path(), req, &msg)) != srs_success) {
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
srs_info("http post hartbeart uri failed. url=%s, request=%s, ret=%d", srs_info("http post hartbeart uri failed. url=%s, request=%s, ret=%d",
url.c_str(), req.c_str(), ret); url.c_str(), req.c_str(), ret);
return; return;

30
trunk/src/app/srs_app_hls.cpp
View file

@ -90,12 +90,12 @@ SrsDvrAsyncCallOnHls::~SrsDvrAsyncCallOnHls()
srs_freep(req); srs_freep(req);
} }
int SrsDvrAsyncCallOnHls::call() srs_error_t SrsDvrAsyncCallOnHls::call()
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) { if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
return ret; return err;
} }
// the http hooks will cause context switch, // the http hooks will cause context switch,
@ -108,7 +108,7 @@ int SrsDvrAsyncCallOnHls::call()
if (!conf) { if (!conf) {
srs_info("ignore the empty http callback: on_hls"); srs_info("ignore the empty http callback: on_hls");
return ret; return err;
} }
hooks = conf->args; hooks = conf->args;
@ -116,13 +116,12 @@ int SrsDvrAsyncCallOnHls::call()
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
if ((ret = SrsHttpHooks::on_hls(cid, url, req, path, ts_url, m3u8, m3u8_url, seq_no, duration)) != ERROR_SUCCESS) { if ((err = SrsHttpHooks::on_hls(cid, url, req, path, ts_url, m3u8, m3u8_url, seq_no, duration)) != srs_success) {
srs_error("hook client on_hls failed. url=%s, ret=%d", url.c_str(), ret); return srs_error_wrap(err, "callback on_hls %s", url.c_str());
return ret;
} }
} }
return ret; return err;
} }
string SrsDvrAsyncCallOnHls::to_string() string SrsDvrAsyncCallOnHls::to_string()
@ -142,12 +141,12 @@ SrsDvrAsyncCallOnHlsNotify::~SrsDvrAsyncCallOnHlsNotify()
srs_freep(req); srs_freep(req);
} }
int SrsDvrAsyncCallOnHlsNotify::call() srs_error_t SrsDvrAsyncCallOnHlsNotify::call()
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) { if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
return ret; return err;
} }
// the http hooks will cause context switch, // the http hooks will cause context switch,
@ -160,7 +159,7 @@ int SrsDvrAsyncCallOnHlsNotify::call()
if (!conf) { if (!conf) {
srs_info("ignore the empty http callback: on_hls_notify"); srs_info("ignore the empty http callback: on_hls_notify");
return ret; return err;
} }
hooks = conf->args; hooks = conf->args;
@ -169,13 +168,12 @@ int SrsDvrAsyncCallOnHlsNotify::call()
int nb_notify = _srs_config->get_vhost_hls_nb_notify(req->vhost); int nb_notify = _srs_config->get_vhost_hls_nb_notify(req->vhost);
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
if ((ret = SrsHttpHooks::on_hls_notify(cid, url, req, ts_url, nb_notify)) != ERROR_SUCCESS) { if ((err = SrsHttpHooks::on_hls_notify(cid, url, req, ts_url, nb_notify)) != srs_success) {
srs_error("hook client on_hls_notify failed. url=%s, ret=%d", url.c_str(), ret); return srs_error_wrap(err, "callback on_hls_notify %s", url.c_str());
return ret;
} }
} }
return ret; return err;
} }
string SrsDvrAsyncCallOnHlsNotify::to_string() string SrsDvrAsyncCallOnHlsNotify::to_string()

4
trunk/src/app/srs_app_hls.hpp
View file

@ -91,7 +91,7 @@ public:
SrsDvrAsyncCallOnHls(int c, SrsRequest* r, std::string p, std::string t, std::string m, std::string mu, int s, double d); SrsDvrAsyncCallOnHls(int c, SrsRequest* r, std::string p, std::string t, std::string m, std::string mu, int s, double d);
virtual ~SrsDvrAsyncCallOnHls(); virtual ~SrsDvrAsyncCallOnHls();
public: public:
virtual int call(); virtual srs_error_t call();
virtual std::string to_string(); virtual std::string to_string();
}; };
@ -108,7 +108,7 @@ public:
SrsDvrAsyncCallOnHlsNotify(int c, SrsRequest* r, std::string u); SrsDvrAsyncCallOnHlsNotify(int c, SrsRequest* r, std::string u);
virtual ~SrsDvrAsyncCallOnHlsNotify(); virtual ~SrsDvrAsyncCallOnHlsNotify();
public: public:
virtual int call(); virtual srs_error_t call();
virtual std::string to_string(); virtual std::string to_string();
}; };

4
trunk/src/app/srs_app_http_conn.cpp
View file

@ -217,8 +217,8 @@ srs_error_t SrsResponseOnlyHttpConn::pop_message(ISrsHttpMessage** preq)
SrsStSocket skt; SrsStSocket skt;
if ((ret = skt.initialize(stfd)) != ERROR_SUCCESS) { if ((err = skt.initialize(stfd)) != srs_success) {
return srs_error_new(ret, "init socket"); return srs_error_wrap(err, "init socket");
} }
if ((ret = parser->parse_message(&skt, this, preq)) != ERROR_SUCCESS) { if ((ret = parser->parse_message(&skt, this, preq)) != ERROR_SUCCESS) {

208
trunk/src/app/srs_app_http_hooks.cpp
View file

@ -56,9 +56,9 @@ SrsHttpHooks::~SrsHttpHooks()
{ {
} }
int SrsHttpHooks::on_connect(string url, SrsRequest* req) srs_error_t SrsHttpHooks::on_connect(string url, SrsRequest* req)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
int client_id = _srs_context->get_id(); int client_id = _srs_context->get_id();
@ -78,23 +78,20 @@ int SrsHttpHooks::on_connect(string url, SrsRequest* req)
int status_code; int status_code;
SrsHttpClient http; SrsHttpClient http;
if ((ret = do_post(&http, url, data, status_code, res)) != ERROR_SUCCESS) { if ((err = do_post(&http, url, data, status_code, res)) != srs_success) {
srs_error("http post on_connect uri failed. " return srs_error_wrap(err, "http: on_connect failed, client_id=%d, url=%s, request=%s, response=%s, code=%d",
"client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str(), status_code);
client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
return ret;
} }
srs_trace("http hook on_connect success. " srs_trace("http: on_connect ok, client_id=%d, url=%s, request=%s, response=%s",
"client_id=%d, url=%s, request=%s, response=%s, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str());
client_id, url.c_str(), data.c_str(), res.c_str(), ret);
return ret; return err;
} }
void SrsHttpHooks::on_close(string url, SrsRequest* req, int64_t send_bytes, int64_t recv_bytes) void SrsHttpHooks::on_close(string url, SrsRequest* req, int64_t send_bytes, int64_t recv_bytes)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
int client_id = _srs_context->get_id(); int client_id = _srs_context->get_id();
@ -114,23 +111,23 @@ void SrsHttpHooks::on_close(string url, SrsRequest* req, int64_t send_bytes, int
int status_code; int status_code;
SrsHttpClient http; SrsHttpClient http;
if ((ret = do_post(&http, url, data, status_code, res)) != ERROR_SUCCESS) { if ((err = do_post(&http, url, data, status_code, res)) != srs_success) {
srs_warn("http post on_close uri failed, ignored. " int ret = srs_error_code(err);
"client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d", srs_freep(err);
client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret); srs_warn("http: ignore on_close failed, client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d",
client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
return; return;
} }
srs_trace("http hook on_close success. " srs_trace("http: on_close ok, client_id=%d, url=%s, request=%s, response=%s",
"client_id=%d, url=%s, request=%s, response=%s, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str());
client_id, url.c_str(), data.c_str(), res.c_str(), ret);
return; return;
} }
int SrsHttpHooks::on_publish(string url, SrsRequest* req) srs_error_t SrsHttpHooks::on_publish(string url, SrsRequest* req)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
int client_id = _srs_context->get_id(); int client_id = _srs_context->get_id();
@ -150,23 +147,20 @@ int SrsHttpHooks::on_publish(string url, SrsRequest* req)
int status_code; int status_code;
SrsHttpClient http; SrsHttpClient http;
if ((ret = do_post(&http, url, data, status_code, res)) != ERROR_SUCCESS) { if ((err = do_post(&http, url, data, status_code, res)) != srs_success) {
srs_error("http post on_publish uri failed. " return srs_error_wrap(err, "http: on_publish failed, client_id=%d, url=%s, request=%s, response=%s, code=%d",
"client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str(), status_code);
client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
return ret;
} }
srs_trace("http hook on_publish success. " srs_trace("http: on_publish ok, client_id=%d, url=%s, request=%s, response=%s",
"client_id=%d, url=%s, request=%s, response=%s, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str());
client_id, url.c_str(), data.c_str(), res.c_str(), ret);
return ret; return err;
} }
void SrsHttpHooks::on_unpublish(string url, SrsRequest* req) void SrsHttpHooks::on_unpublish(string url, SrsRequest* req)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
int client_id = _srs_context->get_id(); int client_id = _srs_context->get_id();
@ -185,23 +179,23 @@ void SrsHttpHooks::on_unpublish(string url, SrsRequest* req)
int status_code; int status_code;
SrsHttpClient http; SrsHttpClient http;
if ((ret = do_post(&http, url, data, status_code, res)) != ERROR_SUCCESS) { if ((err = do_post(&http, url, data, status_code, res)) != srs_success) {
srs_warn("http post on_unpublish uri failed, ignored. " int ret = srs_error_code(err);
"client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d", srs_freep(err);
client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret); srs_warn("http: ignore on_unpublish failed, client_id=%d, url=%s, request=%s, response=%s, status=%d, ret=%d",
client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
return; return;
} }
srs_trace("http hook on_unpublish success. " srs_trace("http: on_unpublish ok, client_id=%d, url=%s, request=%s, response=%s",
"client_id=%d, url=%s, request=%s, response=%s, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str());
client_id, url.c_str(), data.c_str(), res.c_str(), ret);
return; return;
} }
int SrsHttpHooks::on_play(string url, SrsRequest* req) srs_error_t SrsHttpHooks::on_play(string url, SrsRequest* req)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
int client_id = _srs_context->get_id(); int client_id = _srs_context->get_id();
@ -221,23 +215,20 @@ int SrsHttpHooks::on_play(string url, SrsRequest* req)
int status_code; int status_code;
SrsHttpClient http; SrsHttpClient http;
if ((ret = do_post(&http, url, data, status_code, res)) != ERROR_SUCCESS) { if ((err = do_post(&http, url, data, status_code, res)) != srs_success) {
srs_error("http post on_play uri failed. " return srs_error_wrap(err, "http: on_play failed, client_id=%d, url=%s, request=%s, response=%s, status=%d",
"client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str(), status_code);
client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
return ret;
} }
srs_trace("http hook on_play success. " srs_trace("http: on_play ok, client_id=%d, url=%s, request=%s, response=%s",
"client_id=%d, url=%s, request=%s, response=%s, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str());
client_id, url.c_str(), data.c_str(), res.c_str(), ret);
return ret; return err;
} }
void SrsHttpHooks::on_stop(string url, SrsRequest* req) void SrsHttpHooks::on_stop(string url, SrsRequest* req)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
int client_id = _srs_context->get_id(); int client_id = _srs_context->get_id();
@ -256,23 +247,23 @@ void SrsHttpHooks::on_stop(string url, SrsRequest* req)
int status_code; int status_code;
SrsHttpClient http; SrsHttpClient http;
if ((ret = do_post(&http, url, data, status_code, res)) != ERROR_SUCCESS) { if ((err = do_post(&http, url, data, status_code, res)) != srs_success) {
srs_warn("http post on_stop uri failed, ignored. " int ret = srs_error_code(err);
"client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d", srs_freep(err);
client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret); srs_warn("http: ignore on_stop failed, client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d",
client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
return; return;
} }
srs_trace("http hook on_stop success. " srs_trace("http: on_stop ok, client_id=%d, url=%s, request=%s, response=%s",
"client_id=%d, url=%s, request=%s, response=%s, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str());
client_id, url.c_str(), data.c_str(), res.c_str(), ret);
return; return;
} }
int SrsHttpHooks::on_dvr(int cid, string url, SrsRequest* req, string file) srs_error_t SrsHttpHooks::on_dvr(int cid, string url, SrsRequest* req, string file)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
int client_id = cid; int client_id = cid;
std::string cwd = _srs_config->cwd(); std::string cwd = _srs_config->cwd();
@ -294,23 +285,20 @@ int SrsHttpHooks::on_dvr(int cid, string url, SrsRequest* req, string file)
int status_code; int status_code;
SrsHttpClient http; SrsHttpClient http;
if ((ret = do_post(&http, url, data, status_code, res)) != ERROR_SUCCESS) { if ((err = do_post(&http, url, data, status_code, res)) != srs_success) {
srs_error("http post on_dvr uri failed, ignored. " return srs_error_wrap(err, "http post on_dvr uri failed, client_id=%d, url=%s, request=%s, response=%s, code=%d",
"client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str(), status_code);
client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
return ret;
} }
srs_trace("http hook on_dvr success. " srs_trace("http hook on_dvr success. client_id=%d, url=%s, request=%s, response=%s",
"client_id=%d, url=%s, request=%s, response=%s, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str());
client_id, url.c_str(), data.c_str(), res.c_str(), ret);
return ret; return err;
} }
int SrsHttpHooks::on_hls(int cid, string url, SrsRequest* req, string file, string ts_url, string m3u8, string m3u8_url, int sn, double duration) srs_error_t SrsHttpHooks::on_hls(int cid, string url, SrsRequest* req, string file, string ts_url, string m3u8, string m3u8_url, int sn, double duration)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
int client_id = cid; int client_id = cid;
std::string cwd = _srs_config->cwd(); std::string cwd = _srs_config->cwd();
@ -343,23 +331,20 @@ int SrsHttpHooks::on_hls(int cid, string url, SrsRequest* req, string file, stri
int status_code; int status_code;
SrsHttpClient http; SrsHttpClient http;
if ((ret = do_post(&http, url, data, status_code, res)) != ERROR_SUCCESS) { if ((err = do_post(&http, url, data, status_code, res)) != srs_success) {
srs_error("http post on_hls uri failed, ignored. " return srs_error_wrap(err, "http: post %s with %s, status=%d, res=%s", url.c_str(), data.c_str(), status_code, res.c_str());
"client_id=%d, url=%s, request=%s, response=%s, code=%d, ret=%d",
client_id, url.c_str(), data.c_str(), res.c_str(), status_code, ret);
return ret;
} }
srs_trace("http hook on_hls success. " srs_trace("http: on_hls ok, client_id=%d, url=%s, request=%s, response=%s",
"client_id=%d, url=%s, request=%s, response=%s, ret=%d", client_id, url.c_str(), data.c_str(), res.c_str());
client_id, url.c_str(), data.c_str(), res.c_str(), ret);
return ret; return err;
} }
int SrsHttpHooks::on_hls_notify(int cid, std::string url, SrsRequest* req, std::string ts_url, int nb_notify) srs_error_t SrsHttpHooks::on_hls_notify(int cid, std::string url, SrsRequest* req, std::string ts_url, int nb_notify)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
int client_id = cid; int client_id = cid;
std::string cwd = _srs_config->cwd(); std::string cwd = _srs_config->cwd();
@ -376,13 +361,12 @@ int SrsHttpHooks::on_hls_notify(int cid, std::string url, SrsRequest* req, std::
SrsHttpUri uri; SrsHttpUri uri;
if ((ret = uri.initialize(url)) != ERROR_SUCCESS) { if ((ret = uri.initialize(url)) != ERROR_SUCCESS) {
srs_error("http: post failed. url=%s, ret=%d", url.c_str(), ret); return srs_error_new(ret, "http: init url=%s", url.c_str());
return ret;
} }
SrsHttpClient http; SrsHttpClient http;
if ((ret = http.initialize(uri.get_host(), uri.get_port(), SRS_HLS_NOTIFY_TMMS)) != ERROR_SUCCESS) { if ((err = http.initialize(uri.get_host(), uri.get_port(), SRS_HLS_NOTIFY_TMMS)) != srs_success) {
return ret; return srs_error_wrap(err, "http: init client for %s", url.c_str());
} }
std::string path = uri.get_query(); std::string path = uri.get_query();
@ -393,11 +377,11 @@ int SrsHttpHooks::on_hls_notify(int cid, std::string url, SrsRequest* req, std::
path += "?"; path += "?";
path += uri.get_query(); path += uri.get_query();
} }
srs_warn("GET %s", path.c_str()); srs_info("GET %s", path.c_str());
ISrsHttpMessage* msg = NULL; ISrsHttpMessage* msg = NULL;
if ((ret = http.get(path.c_str(), "", &msg)) != ERROR_SUCCESS) { if ((err = http.get(path.c_str(), "", &msg)) != srs_success) {
return ret; return srs_error_wrap(err, "http: get %s", url.c_str());
} }
SrsAutoFree(ISrsHttpMessage, msg); SrsAutoFree(ISrsHttpMessage, msg);
@ -417,87 +401,75 @@ int SrsHttpHooks::on_hls_notify(int cid, std::string url, SrsRequest* req, std::
int spenttime = (int)(srs_update_system_time_ms() - starttime); int spenttime = (int)(srs_update_system_time_ms() - starttime);
srs_trace("http hook on_hls_notify success. client_id=%d, url=%s, code=%d, spent=%dms, read=%dB, ret=%d", srs_trace("http hook on_hls_notify success. client_id=%d, url=%s, code=%d, spent=%dms, read=%dB, ret=%d",
client_id, url.c_str(), msg->status_code(), spenttime, nb_read, ret); client_id, url.c_str(), msg->status_code(), spenttime, nb_read, ret);
// ignore any error for on_hls_notify. // ignore any error for on_hls_notify.
ret = ERROR_SUCCESS; ret = ERROR_SUCCESS;
return ret; return err;
} }
int SrsHttpHooks::do_post(SrsHttpClient* hc, std::string url, std::string req, int& code, string& res) srs_error_t SrsHttpHooks::do_post(SrsHttpClient* hc, std::string url, std::string req, int& code, string& res)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
SrsHttpUri uri; SrsHttpUri uri;
if ((ret = uri.initialize(url)) != ERROR_SUCCESS) { if ((ret = uri.initialize(url)) != ERROR_SUCCESS) {
srs_error("http: post failed. url=%s, ret=%d", url.c_str(), ret); return srs_error_new(ret, "http: post failed. url=%s", url.c_str());
return ret;
} }
if ((ret = hc->initialize(uri.get_host(), uri.get_port())) != ERROR_SUCCESS) { if ((err = hc->initialize(uri.get_host(), uri.get_port())) != srs_success) {
return ret; return srs_error_wrap(err, "http: init client");
} }
ISrsHttpMessage* msg = NULL; ISrsHttpMessage* msg = NULL;
if ((ret = hc->post(uri.get_path(), req, &msg)) != ERROR_SUCCESS) { if ((err = hc->post(uri.get_path(), req, &msg)) != srs_success) {
return ret; return srs_error_wrap(err, "http: client post");
} }
SrsAutoFree(ISrsHttpMessage, msg); SrsAutoFree(ISrsHttpMessage, msg);
code = msg->status_code(); code = msg->status_code();
if ((ret = msg->body_read_all(res)) != ERROR_SUCCESS) { if ((ret = msg->body_read_all(res)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "http: body read");
} }
// ensure the http status is ok. // ensure the http status is ok.
// https://github.com/ossrs/srs/issues/158 // https://github.com/ossrs/srs/issues/158
if (code != SRS_CONSTS_HTTP_OK && code != SRS_CONSTS_HTTP_Created) { if (code != SRS_CONSTS_HTTP_OK && code != SRS_CONSTS_HTTP_Created) {
ret = ERROR_HTTP_STATUS_INVALID; return srs_error_new(ERROR_HTTP_STATUS_INVALID, "http: status %d", code);
srs_error("invalid response status=%d. ret=%d", code, ret);
return ret;
} }
// should never be empty. // should never be empty.
if (res.empty()) { if (res.empty()) {
ret = ERROR_HTTP_DATA_INVALID; return srs_error_new(ERROR_HTTP_DATA_INVALID, "http: empty response");
srs_error("invalid empty response. ret=%d", ret);
return ret;
} }
// parse string res to json. // parse string res to json.
SrsJsonAny* info = SrsJsonAny::loads(res); SrsJsonAny* info = SrsJsonAny::loads(res);
if (!info) { if (!info) {
ret = ERROR_HTTP_DATA_INVALID; return srs_error_new(ERROR_HTTP_DATA_INVALID, "http: not json %s", res.c_str());
srs_error("invalid response %s. ret=%d", res.c_str(), ret);
return ret;
} }
SrsAutoFree(SrsJsonAny, info); SrsAutoFree(SrsJsonAny, info);
// response error code in string. // response error code in string.
if (!info->is_object()) { if (!info->is_object()) {
if (res != SRS_HTTP_RESPONSE_OK) { if (res == SRS_HTTP_RESPONSE_OK) {
ret = ERROR_HTTP_DATA_INVALID; return err;
srs_error("invalid response number %s. ret=%d", res.c_str(), ret);
return ret;
} }
return ret; return srs_error_new(ERROR_HTTP_DATA_INVALID, "http: response number code %s", res.c_str());
} }
// response standard object, format in json: {"code": 0, "data": ""} // response standard object, format in json: {"code": 0, "data": ""}
SrsJsonObject* res_info = info->to_object(); SrsJsonObject* res_info = info->to_object();
SrsJsonAny* res_code = NULL; SrsJsonAny* res_code = NULL;
if ((res_code = res_info->ensure_property_integer("code")) == NULL) { if ((res_code = res_info->ensure_property_integer("code")) == NULL) {
ret = ERROR_RESPONSE_CODE; return srs_error_new(ERROR_RESPONSE_CODE, "http: response object no code %s", res.c_str());
srs_error("invalid response without code, ret=%d", ret);
return ret;
} }
if ((res_code->to_integer()) != ERROR_SUCCESS) { if ((res_code->to_integer()) != ERROR_SUCCESS) {
ret = ERROR_RESPONSE_CODE; return srs_error_new(ERROR_RESPONSE_CODE, "http: response object code %d %s", res_code->to_integer(), res.c_str());
srs_error("error response code=%d. ret=%d", res_code->to_integer(), ret);
return ret;
} }
return ret; return err;
} }

14
trunk/src/app/srs_app_http_hooks.hpp
View file

@ -51,7 +51,7 @@ public:
* @param url the api server url, to valid the client. * @param url the api server url, to valid the client.
* ignore if empty. * ignore if empty.
*/ */
static int on_connect(std::string url, SrsRequest* req); static srs_error_t on_connect(std::string url, SrsRequest* req);
/** /**
* on_close hook, when client disconnect to srs, where client is valid by on_connect. * on_close hook, when client disconnect to srs, where client is valid by on_connect.
* @param url the api server url, to process the event. * @param url the api server url, to process the event.
@ -63,7 +63,7 @@ public:
* @param url the api server url, to valid the client. * @param url the api server url, to valid the client.
* ignore if empty. * ignore if empty.
*/ */
static int on_publish(std::string url, SrsRequest* req); static srs_error_t on_publish(std::string url, SrsRequest* req);
/** /**
* on_unpublish hook, when client(encoder) stop publish stream. * on_unpublish hook, when client(encoder) stop publish stream.
* @param url the api server url, to process the event. * @param url the api server url, to process the event.
@ -75,7 +75,7 @@ public:
* @param url the api server url, to valid the client. * @param url the api server url, to valid the client.
* ignore if empty. * ignore if empty.
*/ */
static int on_play(std::string url, SrsRequest* req); static srs_error_t on_play(std::string url, SrsRequest* req);
/** /**
* on_stop hook, when client stop to play the stream. * on_stop hook, when client stop to play the stream.
* @param url the api server url, to process the event. * @param url the api server url, to process the event.
@ -89,7 +89,7 @@ public:
* @param file the file path, can be relative or absolute path. * @param file the file path, can be relative or absolute path.
* @param cid the source connection cid, for the on_dvr is async call. * @param cid the source connection cid, for the on_dvr is async call.
*/ */
static int on_dvr(int cid, std::string url, SrsRequest* req, std::string file); static srs_error_t on_dvr(int cid, std::string url, SrsRequest* req, std::string file);
/** /**
* when hls reap segment, callback. * when hls reap segment, callback.
* @param url the api server url, to process the event. * @param url the api server url, to process the event.
@ -102,7 +102,7 @@ public:
* @param duration the segment duration in seconds. * @param duration the segment duration in seconds.
* @param cid the source connection cid, for the on_dvr is async call. * @param cid the source connection cid, for the on_dvr is async call.
*/ */
static int on_hls(int cid, std::string url, SrsRequest* req, std::string file, std::string ts_url, std::string m3u8, std::string m3u8_url, int sn, double duration); static srs_error_t on_hls(int cid, std::string url, SrsRequest* req, std::string file, std::string ts_url, std::string m3u8, std::string m3u8_url, int sn, double duration);
/** /**
* when hls reap segment, callback. * when hls reap segment, callback.
* @param url the api server url, to process the event. * @param url the api server url, to process the event.
@ -111,9 +111,9 @@ public:
* @param nb_notify the max bytes to read from notify server. * @param nb_notify the max bytes to read from notify server.
* @param cid the source connection cid, for the on_dvr is async call. * @param cid the source connection cid, for the on_dvr is async call.
*/ */
static int on_hls_notify(int cid, std::string url, SrsRequest* req, std::string ts_url, int nb_notify); static srs_error_t on_hls_notify(int cid, std::string url, SrsRequest* req, std::string ts_url, int nb_notify);
private: private:
static int do_post(SrsHttpClient* hc, std::string url, std::string req, int& code, std::string& res); static srs_error_t do_post(SrsHttpClient* hc, std::string url, std::string req, int& code, std::string& res);
}; };
#endif #endif

16
trunk/src/app/srs_app_kafka.cpp
View file

@ -155,6 +155,7 @@ string SrsKafkaPartition::hostport()
int SrsKafkaPartition::connect() int SrsKafkaPartition::connect()
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
if (transport) { if (transport) {
return ret; return ret;
@ -162,8 +163,11 @@ int SrsKafkaPartition::connect()
transport = new SrsTcpClient(host, port, SRS_KAFKA_PRODUCER_TIMEOUT); transport = new SrsTcpClient(host, port, SRS_KAFKA_PRODUCER_TIMEOUT);
kafka = new SrsKafkaClient(transport); kafka = new SrsKafkaClient(transport);
if ((ret = transport->connect()) != ERROR_SUCCESS) { if ((err = transport->connect()) != srs_success) {
disconnect(); disconnect();
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
srs_error("connect to %s partition=%d failed. ret=%d", hostport().c_str(), id, ret); srs_error("connect to %s partition=%d failed. ret=%d", hostport().c_str(), id, ret);
return ret; return ret;
} }
@ -196,14 +200,14 @@ SrsKafkaMessage::~SrsKafkaMessage()
srs_freep(obj); srs_freep(obj);
} }
int SrsKafkaMessage::call() srs_error_t SrsKafkaMessage::call()
{ {
int ret = producer->send(key, obj); int ret = producer->send(key, obj);
// the obj is manged by producer now. // the obj is manged by producer now.
obj = NULL; obj = NULL;
return ret; return srs_error_new(ret, "kafka send");
} }
string SrsKafkaMessage::to_string() string SrsKafkaMessage::to_string()
@ -552,6 +556,7 @@ srs_error_t SrsKafkaProducer::do_cycle()
int SrsKafkaProducer::request_metadata() int SrsKafkaProducer::request_metadata()
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
// ignore when disabled. // ignore when disabled.
if (!enabled) { if (!enabled) {
@ -588,7 +593,10 @@ int SrsKafkaProducer::request_metadata()
SrsAutoFree(SrsKafkaClient, kafka); SrsAutoFree(SrsKafkaClient, kafka);
// reconnect to kafka server. // reconnect to kafka server.
if ((ret = transport->connect()) != ERROR_SUCCESS) { if ((err = transport->connect()) != srs_success) {
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
srs_error("kafka connect %s:%d failed. ret=%d", server.c_str(), port, ret); srs_error("kafka connect %s:%d failed. ret=%d", server.c_str(), port, ret);
return ret; return ret;
} }

2
trunk/src/app/srs_app_kafka.hpp
View file

@ -89,7 +89,7 @@ public:
virtual ~SrsKafkaMessage(); virtual ~SrsKafkaMessage();
// interface ISrsAsyncCallTask // interface ISrsAsyncCallTask
public: public:
virtual int call(); virtual srs_error_t call();
virtual std::string to_string(); virtual std::string to_string();
}; };

70
trunk/src/app/srs_app_rtmp_conn.cpp
View file

@ -502,6 +502,7 @@ int SrsRtmpConn::service_cycle()
int SrsRtmpConn::stream_service_cycle() int SrsRtmpConn::stream_service_cycle()
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
SrsRequest* req = info->req; SrsRequest* req = info->req;
@ -563,7 +564,10 @@ int SrsRtmpConn::stream_service_cycle()
srs_error("start to play stream failed. ret=%d", ret); srs_error("start to play stream failed. ret=%d", ret);
return ret; return ret;
} }
if ((ret = http_hooks_on_play()) != ERROR_SUCCESS) { if ((err = http_hooks_on_play()) != srs_success) {
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
srs_error("http hook on_play failed. ret=%d", ret); srs_error("http hook on_play failed. ret=%d", ret);
return ret; return ret;
} }
@ -617,6 +621,7 @@ int SrsRtmpConn::stream_service_cycle()
int SrsRtmpConn::check_vhost(bool try_default_vhost) int SrsRtmpConn::check_vhost(bool try_default_vhost)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
SrsRequest* req = info->req; SrsRequest* req = info->req;
srs_assert(req != NULL); srs_assert(req != NULL);
@ -647,7 +652,11 @@ int SrsRtmpConn::check_vhost(bool try_default_vhost)
srs_verbose("check refer success."); srs_verbose("check refer success.");
} }
if ((ret = http_hooks_on_connect()) != ERROR_SUCCESS) { if ((err = http_hooks_on_connect()) != srs_success) {
srs_error("check vhost failed %s", srs_error_desc(err).c_str());
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
return ret; return ret;
} }
@ -880,6 +889,7 @@ int SrsRtmpConn::do_playing(SrsSource* source, SrsConsumer* consumer, SrsQueueRe
int SrsRtmpConn::publishing(SrsSource* source) int SrsRtmpConn::publishing(SrsSource* source)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
SrsRequest* req = info->req; SrsRequest* req = info->req;
@ -891,7 +901,10 @@ int SrsRtmpConn::publishing(SrsSource* source)
srs_verbose("check publish_refer success."); srs_verbose("check publish_refer success.");
} }
if ((ret = http_hooks_on_publish()) != ERROR_SUCCESS) { if ((err = http_hooks_on_publish()) != srs_success) {
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
srs_error("http hook on_publish failed. ret=%d", ret); srs_error("http hook on_publish failed. ret=%d", ret);
return ret; return ret;
} }
@ -1330,6 +1343,7 @@ void SrsRtmpConn::set_sock_options()
int SrsRtmpConn::check_edge_token_traverse_auth() int SrsRtmpConn::check_edge_token_traverse_auth()
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
SrsRequest* req = info->req; SrsRequest* req = info->req;
srs_assert(req); srs_assert(req);
@ -1350,7 +1364,10 @@ int SrsRtmpConn::check_edge_token_traverse_auth()
SrsTcpClient* transport = new SrsTcpClient(server, port, SRS_EDGE_TOKEN_TRAVERSE_TMMS); SrsTcpClient* transport = new SrsTcpClient(server, port, SRS_EDGE_TOKEN_TRAVERSE_TMMS);
SrsAutoFree(SrsTcpClient, transport); SrsAutoFree(SrsTcpClient, transport);
if ((ret = transport->connect()) != ERROR_SUCCESS) { if ((err = transport->connect()) != srs_success) {
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
srs_warn("Illegal edge token, tcUrl=%s to server=%s, port=%d. ret=%d", req->tcUrl.c_str(), server.c_str(), port, ret); srs_warn("Illegal edge token, tcUrl=%s to server=%s, port=%d. ret=%d", req->tcUrl.c_str(), server.c_str(), port, ret);
continue; continue;
} }
@ -1410,14 +1427,14 @@ srs_error_t SrsRtmpConn::on_disconnect()
return err; return err;
} }
int SrsRtmpConn::http_hooks_on_connect() srs_error_t SrsRtmpConn::http_hooks_on_connect()
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
SrsRequest* req = info->req; SrsRequest* req = info->req;
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) { if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
return ret; return err;
} }
// the http hooks will cause context switch, // the http hooks will cause context switch,
@ -1430,7 +1447,7 @@ int SrsRtmpConn::http_hooks_on_connect()
if (!conf) { if (!conf) {
srs_info("ignore the empty http callback: on_connect"); srs_info("ignore the empty http callback: on_connect");
return ret; return err;
} }
hooks = conf->args; hooks = conf->args;
@ -1438,13 +1455,12 @@ int SrsRtmpConn::http_hooks_on_connect()
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
if ((ret = SrsHttpHooks::on_connect(url, req)) != ERROR_SUCCESS) { if ((err = SrsHttpHooks::on_connect(url, req)) != srs_success) {
srs_error("hook client on_connect failed. url=%s, ret=%d", url.c_str(), ret); return srs_error_wrap(err, "rtmp on_connect %s", url.c_str());
return ret;
} }
} }
return ret; return err;
} }
void SrsRtmpConn::http_hooks_on_close() void SrsRtmpConn::http_hooks_on_close()
@ -1477,14 +1493,14 @@ void SrsRtmpConn::http_hooks_on_close()
} }
} }
int SrsRtmpConn::http_hooks_on_publish() srs_error_t SrsRtmpConn::http_hooks_on_publish()
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
SrsRequest* req = info->req; SrsRequest* req = info->req;
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) { if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
return ret; return err;
} }
// the http hooks will cause context switch, // the http hooks will cause context switch,
@ -1497,7 +1513,7 @@ int SrsRtmpConn::http_hooks_on_publish()
if (!conf) { if (!conf) {
srs_info("ignore the empty http callback: on_publish"); srs_info("ignore the empty http callback: on_publish");
return ret; return err;
} }
hooks = conf->args; hooks = conf->args;
@ -1505,13 +1521,12 @@ int SrsRtmpConn::http_hooks_on_publish()
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
if ((ret = SrsHttpHooks::on_publish(url, req)) != ERROR_SUCCESS) { if ((err = SrsHttpHooks::on_publish(url, req)) != srs_success) {
srs_error("hook client on_publish failed. url=%s, ret=%d", url.c_str(), ret); return srs_error_wrap(err, "rtmp on_publish %s", url.c_str());
return ret;
} }
} }
return ret; return err;
} }
void SrsRtmpConn::http_hooks_on_unpublish() void SrsRtmpConn::http_hooks_on_unpublish()
@ -1544,14 +1559,14 @@ void SrsRtmpConn::http_hooks_on_unpublish()
} }
} }
int SrsRtmpConn::http_hooks_on_play() srs_error_t SrsRtmpConn::http_hooks_on_play()
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
SrsRequest* req = info->req; SrsRequest* req = info->req;
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) { if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
return ret; return err;
} }
// the http hooks will cause context switch, // the http hooks will cause context switch,
@ -1564,7 +1579,7 @@ int SrsRtmpConn::http_hooks_on_play()
if (!conf) { if (!conf) {
srs_info("ignore the empty http callback: on_play"); srs_info("ignore the empty http callback: on_play");
return ret; return err;
} }
hooks = conf->args; hooks = conf->args;
@ -1572,13 +1587,12 @@ int SrsRtmpConn::http_hooks_on_play()
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
if ((ret = SrsHttpHooks::on_play(url, req)) != ERROR_SUCCESS) { if ((err = SrsHttpHooks::on_play(url, req)) != srs_success) {
srs_error("hook client on_play failed. url=%s, ret=%d", url.c_str(), ret); return srs_error_wrap(err, "rtmp on_play %s", url.c_str());
return ret;
} }
} }
return ret; return err;
} }
void SrsRtmpConn::http_hooks_on_stop() void SrsRtmpConn::http_hooks_on_stop()

6
trunk/src/app/srs_app_rtmp_conn.hpp
View file

@ -173,11 +173,11 @@ private:
*/ */
virtual srs_error_t on_disconnect(); virtual srs_error_t on_disconnect();
private: private:
virtual int http_hooks_on_connect(); virtual srs_error_t http_hooks_on_connect();
virtual void http_hooks_on_close(); virtual void http_hooks_on_close();
virtual int http_hooks_on_publish(); virtual srs_error_t http_hooks_on_publish();
virtual void http_hooks_on_unpublish(); virtual void http_hooks_on_unpublish();
virtual int http_hooks_on_play(); virtual srs_error_t http_hooks_on_play();
virtual void http_hooks_on_stop(); virtual void http_hooks_on_stop();
}; };

5
trunk/src/app/srs_app_rtsp.cpp
View file

@ -231,11 +231,10 @@ SrsRtspConn::~SrsRtspConn()
srs_error_t SrsRtspConn::serve() srs_error_t SrsRtspConn::serve()
{ {
int ret = ERROR_SUCCESS;
srs_error_t err = srs_success; srs_error_t err = srs_success;
if ((ret = skt->initialize(stfd)) != ERROR_SUCCESS) { if ((err = skt->initialize(stfd)) != srs_success) {
return srs_error_new(ret, "socket initialize"); return srs_error_wrap(err, "socket initialize");
} }
if ((err = trd->start()) != srs_success) { if ((err = trd->start()) != srs_success) {

17
trunk/src/main/srs_main_ingest_hls.cpp
View file

@ -366,17 +366,24 @@ int SrsIngestHlsInput::parseAac(ISrsAacHandler* handler, char* body, int nb_body
int SrsIngestHlsInput::parseM3u8(SrsHttpUri* url, double& td, double& duration) int SrsIngestHlsInput::parseM3u8(SrsHttpUri* url, double& td, double& duration)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
SrsHttpClient client; SrsHttpClient client;
srs_trace("parse input hls %s", url->get_url().c_str()); srs_trace("parse input hls %s", url->get_url().c_str());
if ((ret = client.initialize(url->get_host(), url->get_port())) != ERROR_SUCCESS) { if ((err = client.initialize(url->get_host(), url->get_port())) != srs_success) {
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
srs_error("connect to server failed. ret=%d", ret); srs_error("connect to server failed. ret=%d", ret);
return ret; return ret;
} }
ISrsHttpMessage* msg = NULL; ISrsHttpMessage* msg = NULL;
if ((ret = client.get(url->get_path(), "", &msg)) != ERROR_SUCCESS) { if ((err = client.get(url->get_path(), "", &msg)) != srs_success) {
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
srs_error("HTTP GET %s failed. ret=%d", url->get_url().c_str(), ret); srs_error("HTTP GET %s failed. ret=%d", url->get_url().c_str(), ret);
return ret; return ret;
} }
@ -575,6 +582,7 @@ void SrsIngestHlsInput::remove_dirty()
int SrsIngestHlsInput::SrsTsPiece::fetch(string m3u8) int SrsIngestHlsInput::SrsTsPiece::fetch(string m3u8)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
if (skip || sent || !body.empty()) { if (skip || sent || !body.empty()) {
return ret; return ret;
@ -598,7 +606,10 @@ int SrsIngestHlsInput::SrsTsPiece::fetch(string m3u8)
} }
ISrsHttpMessage* msg = NULL; ISrsHttpMessage* msg = NULL;
if ((ret = client.get(uri.get_path(), "", &msg)) != ERROR_SUCCESS) { if ((err = client.get(uri.get_path(), "", &msg)) != srs_success) {
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
srs_error("HTTP GET %s failed. ret=%d", uri.get_url().c_str(), ret); srs_error("HTTP GET %s failed. ret=%d", uri.get_url().c_str(), ret);
return ret; return ret;
} }

53
trunk/src/service/srs_service_http_client.cpp
View file

@ -53,16 +53,16 @@ SrsHttpClient::~SrsHttpClient()
} }
// TODO: FIXME: use ms for timeout. // TODO: FIXME: use ms for timeout.
int SrsHttpClient::initialize(string h, int p, int64_t tm) srs_error_t SrsHttpClient::initialize(string h, int p, int64_t tm)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
srs_freep(parser); srs_freep(parser);
parser = new SrsHttpParser(); parser = new SrsHttpParser();
if ((ret = parser->initialize(HTTP_RESPONSE, false)) != ERROR_SUCCESS) { if ((ret = parser->initialize(HTTP_RESPONSE, false)) != ERROR_SUCCESS) {
srs_error("initialize parser failed. ret=%d", ret); return srs_error_new(ret, "http: init parser");
return ret;
} }
// Always disconnect the transport. // Always disconnect the transport.
@ -83,7 +83,7 @@ int SrsHttpClient::initialize(string h, int p, int64_t tm)
headers["User-Agent"] = RTMP_SIG_SRS_SERVER; headers["User-Agent"] = RTMP_SIG_SRS_SERVER;
headers["Content-Type"] = "application/json"; headers["Content-Type"] = "application/json";
return ret; return err;
} }
SrsHttpClient* SrsHttpClient::set_header(string k, string v) SrsHttpClient* SrsHttpClient::set_header(string k, string v)
@ -93,18 +93,18 @@ SrsHttpClient* SrsHttpClient::set_header(string k, string v)
return this; return this;
} }
int SrsHttpClient::post(string path, string req, ISrsHttpMessage** ppmsg) srs_error_t SrsHttpClient::post(string path, string req, ISrsHttpMessage** ppmsg)
{ {
*ppmsg = NULL; *ppmsg = NULL;
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
// always set the content length. // always set the content length.
headers["Content-Length"] = srs_int2str(req.length()); headers["Content-Length"] = srs_int2str(req.length());
if ((ret = connect()) != ERROR_SUCCESS) { if ((err = connect()) != srs_success) {
srs_warn("http connect server failed. ret=%d", ret); return srs_error_wrap(err, "http: connect server");
return ret;
} }
// send POST request to uri // send POST request to uri
@ -122,14 +122,12 @@ int SrsHttpClient::post(string path, string req, ISrsHttpMessage** ppmsg)
if ((ret = transport->write((void*)data.c_str(), data.length(), NULL)) != ERROR_SUCCESS) { if ((ret = transport->write((void*)data.c_str(), data.length(), NULL)) != ERROR_SUCCESS) {
// Disconnect the transport when channel error, reconnect for next operation. // Disconnect the transport when channel error, reconnect for next operation.
disconnect(); disconnect();
srs_error("write http post failed. ret=%d", ret); return srs_error_new(ret, "http: write");
return ret;
} }
ISrsHttpMessage* msg = NULL; ISrsHttpMessage* msg = NULL;
if ((ret = parser->parse_message(transport, NULL, &msg)) != ERROR_SUCCESS) { if ((ret = parser->parse_message(transport, NULL, &msg)) != ERROR_SUCCESS) {
srs_error("parse http post response failed. ret=%d", ret); return srs_error_new(ret, "http: parse response");
return ret;
} }
srs_assert(msg); srs_assert(msg);
@ -140,21 +138,21 @@ int SrsHttpClient::post(string path, string req, ISrsHttpMessage** ppmsg)
} }
srs_info("parse http post response success."); srs_info("parse http post response success.");
return ret; return err;
} }
int SrsHttpClient::get(string path, string req, ISrsHttpMessage** ppmsg) srs_error_t SrsHttpClient::get(string path, string req, ISrsHttpMessage** ppmsg)
{ {
*ppmsg = NULL; *ppmsg = NULL;
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
// always set the content length. // always set the content length.
headers["Content-Length"] = srs_int2str(req.length()); headers["Content-Length"] = srs_int2str(req.length());
if ((ret = connect()) != ERROR_SUCCESS) { if ((err = connect()) != srs_success) {
srs_warn("http connect server failed. ret=%d", ret); return srs_error_wrap(err, "http: connect server");
return ret;
} }
// send POST request to uri // send POST request to uri
@ -172,14 +170,12 @@ int SrsHttpClient::get(string path, string req, ISrsHttpMessage** ppmsg)
if ((ret = transport->write((void*)data.c_str(), data.length(), NULL)) != ERROR_SUCCESS) { if ((ret = transport->write((void*)data.c_str(), data.length(), NULL)) != ERROR_SUCCESS) {
// Disconnect the transport when channel error, reconnect for next operation. // Disconnect the transport when channel error, reconnect for next operation.
disconnect(); disconnect();
srs_error("write http get failed. ret=%d", ret); return srs_error_new(ret, "http: write");
return ret;
} }
ISrsHttpMessage* msg = NULL; ISrsHttpMessage* msg = NULL;
if ((ret = parser->parse_message(transport, NULL, &msg)) != ERROR_SUCCESS) { if ((ret = parser->parse_message(transport, NULL, &msg)) != ERROR_SUCCESS) {
srs_error("parse http post response failed. ret=%d", ret); return srs_error_new(ret, "http: parse response");
return ret;
} }
srs_assert(msg); srs_assert(msg);
@ -190,7 +186,7 @@ int SrsHttpClient::get(string path, string req, ISrsHttpMessage** ppmsg)
} }
srs_info("parse http get response success."); srs_info("parse http get response success.");
return ret; return err;
} }
void SrsHttpClient::set_recv_timeout(int64_t tm) void SrsHttpClient::set_recv_timeout(int64_t tm)
@ -218,20 +214,19 @@ void SrsHttpClient::disconnect()
srs_freep(transport); srs_freep(transport);
} }
int SrsHttpClient::connect() srs_error_t SrsHttpClient::connect()
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
// When transport connected, ignore. // When transport connected, ignore.
if (transport) { if (transport) {
return ret; return err;
} }
transport = new SrsTcpClient(host, port, timeout); transport = new SrsTcpClient(host, port, timeout);
if ((ret = transport->connect()) != ERROR_SUCCESS) { if ((err = transport->connect()) != srs_success) {
disconnect(); disconnect();
srs_warn("http client failed, server=%s, port=%d, timeout=%" PRId64 ", ret=%d", host.c_str(), port, timeout, ret); return srs_error_wrap(err, "http: tcp connect %s:%d to=%d", host.c_str(), port, (int)timeout);
return ret;
} }
srs_info("connect to server success. server=%s, port=%d", host.c_str(), port); srs_info("connect to server success. server=%s, port=%d", host.c_str(), port);
@ -241,6 +236,6 @@ int SrsHttpClient::connect()
kbps->set_io(transport, transport); kbps->set_io(transport, transport);
return ret; return err;
} }

8
trunk/src/service/srs_service_http_client.hpp
View file

@ -75,7 +75,7 @@ public:
* @param tm The underlayer TCP transport timeout in ms. * @param tm The underlayer TCP transport timeout in ms.
* @remark we will set default values in headers, which can be override by set_header. * @remark we will set default values in headers, which can be override by set_header.
*/ */
virtual int initialize(std::string h, int p, int64_t tm = SRS_HTTP_CLIENT_TMMS); virtual srs_error_t initialize(std::string h, int p, int64_t tm = SRS_HTTP_CLIENT_TMMS);
/** /**
* Set HTTP request header in header[k]=v. * Set HTTP request header in header[k]=v.
* @return the HTTP client itself. * @return the HTTP client itself.
@ -89,7 +89,7 @@ public:
* @param ppmsg output the http message to read the response. * @param ppmsg output the http message to read the response.
* @remark user must free the ppmsg if not NULL. * @remark user must free the ppmsg if not NULL.
*/ */
virtual int post(std::string path, std::string req, ISrsHttpMessage** ppmsg); virtual srs_error_t post(std::string path, std::string req, ISrsHttpMessage** ppmsg);
/** /**
* to get data from the uri. * to get data from the uri.
* @param the path to request on. * @param the path to request on.
@ -97,14 +97,14 @@ public:
* @param ppmsg output the http message to read the response. * @param ppmsg output the http message to read the response.
* @remark user must free the ppmsg if not NULL. * @remark user must free the ppmsg if not NULL.
*/ */
virtual int get(std::string path, std::string req, ISrsHttpMessage** ppmsg); virtual srs_error_t get(std::string path, std::string req, ISrsHttpMessage** ppmsg);
private: private:
virtual void set_recv_timeout(int64_t tm); virtual void set_recv_timeout(int64_t tm);
public: public:
virtual void kbps_sample(const char* label, int64_t age); virtual void kbps_sample(const char* label, int64_t age);
private: private:
virtual void disconnect(); virtual void disconnect();
virtual int connect(); virtual srs_error_t connect();
}; };
#endif #endif

6
trunk/src/service/srs_service_rtmp_conn.cpp
View file

@ -60,6 +60,7 @@ SrsBasicRtmpClient::~SrsBasicRtmpClient()
int SrsBasicRtmpClient::connect() int SrsBasicRtmpClient::connect()
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
close(); close();
@ -67,8 +68,11 @@ int SrsBasicRtmpClient::connect()
client = new SrsRtmpClient(transport); client = new SrsRtmpClient(transport);
kbps->set_io(transport, transport); kbps->set_io(transport, transport);
if ((ret = transport->connect()) != ERROR_SUCCESS) { if ((err = transport->connect()) != srs_success) {
close(); close();
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
return ret; return ret;
} }

42
trunk/src/service/srs_service_st.cpp
View file

@ -102,9 +102,9 @@ srs_thread_t srs_thread_self()
return (srs_thread_t)st_thread_self(); return (srs_thread_t)st_thread_self();
} }
int srs_socket_connect(string server, int port, int64_t tm, srs_netfd_t* pstfd) srs_error_t srs_socket_connect(string server, int port, int64_t tm, srs_netfd_t* pstfd)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
st_utime_t timeout = ST_UTIME_NO_TIMEOUT; st_utime_t timeout = ST_UTIME_NO_TIMEOUT;
if (tm != SRS_CONSTS_NO_TMMS) { if (tm != SRS_CONSTS_NO_TMMS) {
@ -117,9 +117,7 @@ int srs_socket_connect(string server, int port, int64_t tm, srs_netfd_t* pstfd)
int sock = socket(AF_INET, SOCK_STREAM, 0); int sock = socket(AF_INET, SOCK_STREAM, 0);
if(sock == -1){ if(sock == -1){
ret = ERROR_SOCKET_CREATE; return srs_error_new(ERROR_SOCKET_CREATE, "create socket");
srs_error("create socket error. ret=%d", ret);
return ret;
} }
srs_fd_close_exec(sock); srs_fd_close_exec(sock);
@ -127,17 +125,13 @@ int srs_socket_connect(string server, int port, int64_t tm, srs_netfd_t* pstfd)
srs_assert(!stfd); srs_assert(!stfd);
stfd = st_netfd_open_socket(sock); stfd = st_netfd_open_socket(sock);
if(stfd == NULL){ if(stfd == NULL){
ret = ERROR_ST_OPEN_SOCKET; return srs_error_new(ERROR_ST_OPEN_SOCKET, "open socket");
srs_error("st_netfd_open_socket failed. ret=%d", ret);
return ret;
} }
// connect to server. // connect to server.
std::string ip = srs_dns_resolve(server); std::string ip = srs_dns_resolve(server);
if (ip.empty()) { if (ip.empty()) {
ret = ERROR_SYSTEM_IP_INVALID; return srs_error_new(ERROR_SYSTEM_IP_INVALID, "resolve server %s", server.c_str());
srs_error("dns resolve server error, ip empty. ret=%d", ret);
goto failed;
} }
addr.sin_family = AF_INET; addr.sin_family = AF_INET;
@ -145,20 +139,19 @@ int srs_socket_connect(string server, int port, int64_t tm, srs_netfd_t* pstfd)
addr.sin_addr.s_addr = inet_addr(ip.c_str()); addr.sin_addr.s_addr = inet_addr(ip.c_str());
if (st_connect((st_netfd_t)stfd, (const struct sockaddr*)&addr, sizeof(sockaddr_in), timeout) == -1){ if (st_connect((st_netfd_t)stfd, (const struct sockaddr*)&addr, sizeof(sockaddr_in), timeout) == -1){
ret = ERROR_ST_CONNECT; err = srs_error_new(ERROR_ST_CONNECT, "connect to %s:%d", ip.c_str(), port);
srs_error("connect to server error. ip=%s, port=%d, ret=%d", ip.c_str(), port, ret);
goto failed; goto failed;
} }
srs_info("connect ok. server=%s, ip=%s, port=%d", server.c_str(), ip.c_str(), port); srs_info("connect ok. server=%s, ip=%s, port=%d", server.c_str(), ip.c_str(), port);
*pstfd = stfd; *pstfd = stfd;
return ret; return err;
failed: failed:
if (stfd) { if (stfd) {
srs_close_stfd(stfd); srs_close_stfd(stfd);
} }
return ret; return err;
} }
srs_cond_t srs_cond_new() srs_cond_t srs_cond_new()
@ -252,10 +245,10 @@ SrsStSocket::~SrsStSocket()
{ {
} }
int SrsStSocket::initialize(srs_netfd_t fd) srs_error_t SrsStSocket::initialize(srs_netfd_t fd)
{ {
stfd = fd; stfd = fd;
return ERROR_SUCCESS; return srs_success;
} }
bool SrsStSocket::is_never_timeout(int64_t tm) bool SrsStSocket::is_never_timeout(int64_t tm)
@ -444,23 +437,22 @@ SrsTcpClient::~SrsTcpClient()
srs_freep(io); srs_freep(io);
} }
int SrsTcpClient::connect() srs_error_t SrsTcpClient::connect()
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
close(); close();
srs_assert(stfd == NULL); srs_assert(stfd == NULL);
if ((ret = srs_socket_connect(host, port, timeout, &stfd)) != ERROR_SUCCESS) { if ((err = srs_socket_connect(host, port, timeout, &stfd)) != srs_success) {
srs_error("connect tcp://%s:%d failed, to=%" PRId64 "ms. ret=%d", host.c_str(), port, timeout, ret); return srs_error_wrap(err, "tcp: connect %s:%d to=%d", host.c_str(), port, (int)timeout);
return ret;
} }
if ((ret = io->initialize(stfd)) != ERROR_SUCCESS) { if ((err = io->initialize(stfd)) != srs_success) {
return ret; return srs_error_wrap(err, "tcp: init socket object");
} }
return ret; return err;
} }
void SrsTcpClient::close() void SrsTcpClient::close()

8
trunk/src/service/srs_service_st.hpp
View file

@ -57,7 +57,7 @@ extern srs_thread_t srs_thread_self();
// client open socket and connect to server. // client open socket and connect to server.
// @param tm The timeout in ms. // @param tm The timeout in ms.
extern int srs_socket_connect(std::string server, int port, int64_t tm, srs_netfd_t* pstfd); extern srs_error_t srs_socket_connect(std::string server, int port, int64_t tm, srs_netfd_t* pstfd);
// Wrap for coroutine. // Wrap for coroutine.
extern srs_cond_t srs_cond_new(); extern srs_cond_t srs_cond_new();
@ -126,7 +126,7 @@ public:
virtual ~SrsStSocket(); virtual ~SrsStSocket();
public: public:
// Initialize the socket with stfd, user must manage it. // Initialize the socket with stfd, user must manage it.
virtual int initialize(srs_netfd_t fd); virtual srs_error_t initialize(srs_netfd_t fd);
public: public:
virtual bool is_never_timeout(int64_t tm); virtual bool is_never_timeout(int64_t tm);
virtual void set_recv_timeout(int64_t tm); virtual void set_recv_timeout(int64_t tm);
@ -182,14 +182,14 @@ public:
* Connect to server over TCP. * Connect to server over TCP.
* @remark We will close the exists connection before do connect. * @remark We will close the exists connection before do connect.
*/ */
virtual int connect(); virtual srs_error_t connect();
private: private:
/** /**
* Close the connection to server. * Close the connection to server.
* @remark User should never use the client when close it. * @remark User should never use the client when close it.
*/ */
virtual void close(); virtual void close();
// interface ISrsProtocolReaderWriter // interface ISrsProtocolReaderWriter
public: public:
virtual bool is_never_timeout(int64_t tm); virtual bool is_never_timeout(int64_t tm);
virtual void set_recv_timeout(int64_t tm); virtual void set_recv_timeout(int64_t tm);