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, use complex error for http

Browse source
This commit is contained in:
winlin 2017-07-29 21:39:57 +08:00
parent 661eb8b37c
commit 9f5224c34a
24 changed files with 388 additions and 453 deletions
Download patch file Download diff file Expand all files Collapse all files

14
trunk/src/app/srs_app_caster_flv.cpp
View file

@ -105,7 +105,7 @@ void SrsAppCasterFlv::remove(ISrsConnection* c)
manager->remove(c); manager->remove(c);
} }
int SrsAppCasterFlv::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsAppCasterFlv::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsHttpMessage* msg = dynamic_cast<SrsHttpMessage*>(r); SrsHttpMessage* msg = dynamic_cast<SrsHttpMessage*>(r);
SrsDynamicHttpConn* conn = dynamic_cast<SrsDynamicHttpConn*>(msg->connection()); SrsDynamicHttpConn* conn = dynamic_cast<SrsDynamicHttpConn*>(msg->connection());
@ -128,7 +128,12 @@ int SrsAppCasterFlv::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
o = o.substr(0, o.length() - 4); o = o.substr(0, o.length() - 4);
} }
return conn->proxy(w, r, o); int ret = conn->proxy(w, r, o);
if (ret != ERROR_SUCCESS) {
return srs_error_new(ret, "proxy");
}
return srs_success;
} }
SrsDynamicHttpConn::SrsDynamicHttpConn(IConnectionManager* cm, srs_netfd_t fd, SrsHttpServeMux* m, string cip) SrsDynamicHttpConn::SrsDynamicHttpConn(IConnectionManager* cm, srs_netfd_t fd, SrsHttpServeMux* m, string cip)
@ -144,10 +149,9 @@ SrsDynamicHttpConn::~SrsDynamicHttpConn()
srs_freep(pprint); srs_freep(pprint);
} }
int SrsDynamicHttpConn::on_got_http_message(ISrsHttpMessage* msg) srs_error_t SrsDynamicHttpConn::on_got_http_message(ISrsHttpMessage* msg)
{ {
int ret = ERROR_SUCCESS; return srs_success;
return ret;
} }
int SrsDynamicHttpConn::proxy(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string o) int SrsDynamicHttpConn::proxy(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string o)

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

@ -72,7 +72,7 @@ public:
virtual void remove(ISrsConnection* c); virtual void remove(ISrsConnection* c);
// ISrsHttpHandler // ISrsHttpHandler
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
/** /**
@ -88,7 +88,7 @@ public:
SrsDynamicHttpConn(IConnectionManager* cm, srs_netfd_t fd, SrsHttpServeMux* m, std::string cip); SrsDynamicHttpConn(IConnectionManager* cm, srs_netfd_t fd, SrsHttpServeMux* m, std::string cip);
virtual ~SrsDynamicHttpConn(); virtual ~SrsDynamicHttpConn();
public: public:
virtual int on_got_http_message(ISrsHttpMessage* msg); virtual srs_error_t on_got_http_message(ISrsHttpMessage* msg);
public: public:
virtual int proxy(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string o); virtual int proxy(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string o);
private: private:

95
trunk/src/app/srs_app_config.cpp
View file

@ -1563,13 +1563,13 @@ srs_error_t SrsConfig::reload_conf(SrsConfig* conf)
} }
// merge config: http_api // merge config: http_api
if ((ret = reload_http_api(old_root)) != ERROR_SUCCESS) { if ((err = reload_http_api(old_root)) != srs_success) {
return srs_error_new(ret, "http api");; return srs_error_wrap(err, "http api");;
} }
// merge config: http_stream // merge config: http_stream
if ((ret = reload_http_stream(old_root)) != ERROR_SUCCESS) { if ((err = reload_http_stream(old_root)) != srs_success) {
return srs_error_new(ret, "http steram");; return srs_error_wrap(err, "http steram");;
} }
// TODO: FIXME: support reload stream_caster. // TODO: FIXME: support reload stream_caster.
@ -1583,9 +1583,10 @@ srs_error_t SrsConfig::reload_conf(SrsConfig* conf)
return err; return err;
} }
int SrsConfig::reload_http_api(SrsConfDirective* old_root) srs_error_t SrsConfig::reload_http_api(SrsConfDirective* old_root)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
// merge config. // merge config.
std::vector<ISrsReloadHandler*>::iterator it; std::vector<ISrsReloadHandler*>::iterator it;
@ -1604,13 +1605,11 @@ int SrsConfig::reload_http_api(SrsConfDirective* old_root)
for (it = subscribes.begin(); it != subscribes.end(); ++it) { for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it; ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_http_api_enabled()) != ERROR_SUCCESS) { if ((ret = subscribe->on_reload_http_api_enabled()) != ERROR_SUCCESS) {
srs_error("notify subscribes http_api disabled=>enabled failed. ret=%d", ret); return srs_error_new(ret, "http api off=>on");
return ret;
} }
} }
srs_trace("reload disabled=>enabled http_api success."); srs_trace("reload off=>on http_api success.");
return err;
return ret;
} }
// ENABLED => DISABLED // ENABLED => DISABLED
@ -1618,13 +1617,11 @@ int SrsConfig::reload_http_api(SrsConfDirective* old_root)
for (it = subscribes.begin(); it != subscribes.end(); ++it) { for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it; ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_http_api_disabled()) != ERROR_SUCCESS) { if ((ret = subscribe->on_reload_http_api_disabled()) != ERROR_SUCCESS) {
srs_error("notify subscribes http_api enabled=>disabled failed. ret=%d", ret); return srs_error_new(ret, "http api on=>off");
return ret;
} }
} }
srs_trace("reload enabled=>disabled http_api success."); srs_trace("reload http_api on=>off success.");
return err;
return ret;
} }
// ENABLED => ENABLED (modified) // ENABLED => ENABLED (modified)
@ -1634,45 +1631,41 @@ int SrsConfig::reload_http_api(SrsConfDirective* old_root)
for (it = subscribes.begin(); it != subscribes.end(); ++it) { for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it; ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_http_api_enabled()) != ERROR_SUCCESS) { if ((ret = subscribe->on_reload_http_api_enabled()) != ERROR_SUCCESS) {
srs_error("notify subscribes http_api enabled modified failed. ret=%d", ret); return srs_error_new(ret, "http api enabled");
return ret;
} }
} }
srs_trace("reload enabled modified http_api success."); srs_trace("reload http api enabled success.");
if (!srs_directive_equals(old_http_api->get("crossdomain"), new_http_api->get("crossdomain"))) { if (!srs_directive_equals(old_http_api->get("crossdomain"), new_http_api->get("crossdomain"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) { for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it; ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_http_api_crossdomain()) != ERROR_SUCCESS) { if ((err = subscribe->on_reload_http_api_crossdomain()) != srs_success) {
srs_error("notify subscribes http_api crossdomain modified failed. ret=%d", ret); return srs_error_wrap(err, "http api crossdomain");
return ret;
} }
} }
} }
srs_trace("reload crossdomain modified http_api success."); srs_trace("reload http api crossdomain success.");
if (!srs_directive_equals(old_http_api->get("raw_api"), new_http_api->get("raw_api"))) { if (!srs_directive_equals(old_http_api->get("raw_api"), new_http_api->get("raw_api"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) { for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it; ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_http_api_raw_api()) != ERROR_SUCCESS) { if ((ret = subscribe->on_reload_http_api_raw_api()) != ERROR_SUCCESS) {
srs_error("notify subscribes http_api raw_api modified failed. ret=%d", ret); return srs_error_new(ret, "http api raw_api");
return ret;
} }
} }
} }
srs_trace("reload raw_api modified http_api success."); srs_trace("reload http api raw_api success.");
return err;
return ret;
} }
srs_trace("reload http_api not changed success."); srs_trace("reload http_api success, nothing changed.");
return err;
return ret;
} }
int SrsConfig::reload_http_stream(SrsConfDirective* old_root) srs_error_t SrsConfig::reload_http_stream(SrsConfDirective* old_root)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
// merge config. // merge config.
std::vector<ISrsReloadHandler*>::iterator it; std::vector<ISrsReloadHandler*>::iterator it;
@ -1691,13 +1684,11 @@ int SrsConfig::reload_http_stream(SrsConfDirective* old_root)
for (it = subscribes.begin(); it != subscribes.end(); ++it) { for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it; ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_http_stream_enabled()) != ERROR_SUCCESS) { if ((ret = subscribe->on_reload_http_stream_enabled()) != ERROR_SUCCESS) {
srs_error("notify subscribes http_stream disabled=>enabled failed. ret=%d", ret); return srs_error_new(ret, "http stream off=>on");
return ret;
} }
} }
srs_trace("reload disabled=>enabled http_stream success."); srs_trace("reload http stream off=>on success.");
return err;
return ret;
} }
// ENABLED => DISABLED // ENABLED => DISABLED
@ -1705,13 +1696,11 @@ int SrsConfig::reload_http_stream(SrsConfDirective* old_root)
for (it = subscribes.begin(); it != subscribes.end(); ++it) { for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it; ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_http_stream_disabled()) != ERROR_SUCCESS) { if ((ret = subscribe->on_reload_http_stream_disabled()) != ERROR_SUCCESS) {
srs_error("notify subscribes http_stream enabled=>disabled failed. ret=%d", ret); return srs_error_new(ret, "http stream on=>off");
return ret;
} }
} }
srs_trace("reload enabled=>disabled http_stream success."); srs_trace("reload http stream on=>off success.");
return err;
return ret;
} }
// ENABLED => ENABLED (modified) // ENABLED => ENABLED (modified)
@ -1721,29 +1710,25 @@ int SrsConfig::reload_http_stream(SrsConfDirective* old_root)
for (it = subscribes.begin(); it != subscribes.end(); ++it) { for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it; ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_http_stream_updated()) != ERROR_SUCCESS) { if ((ret = subscribe->on_reload_http_stream_updated()) != ERROR_SUCCESS) {
srs_error("notify subscribes http_stream enabled modified failed. ret=%d", ret); return srs_error_new(ret, "http stream enabled");
return ret;
} }
} }
srs_trace("reload enabled modified http_stream success."); srs_trace("reload http stream enabled success.");
if (!srs_directive_equals(old_http_stream->get("crossdomain"), new_http_stream->get("crossdomain"))) { if (!srs_directive_equals(old_http_stream->get("crossdomain"), new_http_stream->get("crossdomain"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) { for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it; ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_http_stream_crossdomain()) != ERROR_SUCCESS) { if ((err = subscribe->on_reload_http_stream_crossdomain()) != srs_success) {
srs_error("notify subscribes http_stream crossdomain modified failed. ret=%d", ret); return srs_error_wrap(err, "http stream crossdomain");
return ret;
} }
} }
} }
srs_trace("reload crossdomain modified http_stream success."); srs_trace("reload http stream crossdomain success.");
return err;
return ret;
} }
srs_trace("reload http_stream not changed success."); srs_trace("reload http stream success, nothing changed.");
return err;
return ret;
} }
int SrsConfig::reload_transcode(SrsConfDirective* new_vhost, SrsConfDirective* old_vhost) int SrsConfig::reload_transcode(SrsConfDirective* new_vhost, SrsConfDirective* old_vhost)
@ -1920,7 +1905,7 @@ int SrsConfig::reload_ingest(SrsConfDirective* new_vhost, SrsConfDirective* old_
} }
} }
srs_trace("ingest not changed for vhost=%s", vhost.c_str()); srs_trace("ingest nothing changed for vhost=%s", vhost.c_str());
return ret; return ret;
} }
@ -2834,7 +2819,7 @@ int SrsConfig::raw_set_listen(const vector<string>& eps, bool& applied)
SrsConfDirective* conf = root->get("listen"); SrsConfDirective* conf = root->get("listen");
// not changed, ignore. // nothing changed, ignore.
if (srs_vector_actual_equals(conf->args, eps)) { if (srs_vector_actual_equals(conf->args, eps)) {
return ret; return ret;
} }

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

@ -393,12 +393,12 @@ private:
/** /**
* reload the http_api section of config. * reload the http_api section of config.
*/ */
virtual int reload_http_api(SrsConfDirective* old_root); virtual srs_error_t reload_http_api(SrsConfDirective* old_root);
/** /**
* reload the http_stream section of config. * reload the http_stream section of config.
*/ */
// TODO: FIXME: rename to http_server. // TODO: FIXME: rename to http_server.
virtual int reload_http_stream(SrsConfDirective* old_root); virtual srs_error_t reload_http_stream(SrsConfDirective* old_root);
/** /**
* reload the transcode section of vhost of config. * reload the transcode section of vhost of config.
*/ */

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

@ -98,25 +98,26 @@ srs_error_t SrsConnection::start()
srs_error_t SrsConnection::cycle() srs_error_t SrsConnection::cycle()
{ {
int ret = do_cycle(); srs_error_t err = do_cycle();
// Notify manager to remove it. // Notify manager to remove it.
manager->remove(this); manager->remove(this);
// success. // success.
if (ret == ERROR_SUCCESS) { if (err == srs_success) {
srs_trace("client finished."); srs_trace("client finished.");
return srs_success; return err;
} }
// client close peer. // client close peer.
// TODO: FIXME: Only reset the error when client closed it. // TODO: FIXME: Only reset the error when client closed it.
if (srs_is_client_gracefully_close(ret)) { if (srs_is_client_gracefully_close(srs_error_code(err))) {
srs_warn("client disconnect peer. ret=%d", ret); srs_warn("client disconnect peer. ret=%d", srs_error_code(err));
srs_freep(err);
return srs_success; return srs_success;
} }
return srs_error_new(ret, "cycle"); return srs_error_wrap(err, "cycle");
} }
int SrsConnection::srs_id() int SrsConnection::srs_id()

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

@ -121,7 +121,7 @@ protected:
/** /**
* for concrete connection to do the cycle. * for concrete connection to do the cycle.
*/ */
virtual int do_cycle() = 0; virtual srs_error_t do_cycle() = 0;
}; };
#endif #endif

148
trunk/src/app/srs_app_http_api.cpp
View file

@ -46,9 +46,10 @@ using namespace std;
#include <srs_protocol_amf0.hpp> #include <srs_protocol_amf0.hpp>
#include <srs_protocol_utility.hpp> #include <srs_protocol_utility.hpp>
int srs_api_response_jsonp(ISrsHttpResponseWriter* w, string callback, string data) srs_error_t srs_api_response_jsonp(ISrsHttpResponseWriter* w, string callback, string data)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
SrsHttpHeader* h = w->header(); SrsHttpHeader* h = w->header();
@ -56,26 +57,26 @@ int srs_api_response_jsonp(ISrsHttpResponseWriter* w, string callback, string da
h->set_content_type("text/javascript"); h->set_content_type("text/javascript");
if (!callback.empty() && (ret = w->write((char*)callback.data(), (int)callback.length())) != ERROR_SUCCESS) { if (!callback.empty() && (ret = w->write((char*)callback.data(), (int)callback.length())) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "write jsonp callback");
} }
static char* c0 = (char*)"("; static char* c0 = (char*)"(";
if ((ret = w->write(c0, 1)) != ERROR_SUCCESS) { if ((ret = w->write(c0, 1)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "write jsonp left token");
} }
if ((ret = w->write((char*)data.data(), (int)data.length())) != ERROR_SUCCESS) { if ((ret = w->write((char*)data.data(), (int)data.length())) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "write jsonp data");
} }
static char* c1 = (char*)")"; static char* c1 = (char*)")";
if ((ret = w->write(c1, 1)) != ERROR_SUCCESS) { if ((ret = w->write(c1, 1)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "write jsonp right token");
} }
return ret; return err;
} }
int srs_api_response_jsonp_code(ISrsHttpResponseWriter* w, string callback, int code) srs_error_t srs_api_response_jsonp_code(ISrsHttpResponseWriter* w, string callback, int code)
{ {
SrsJsonObject* obj = SrsJsonAny::object(); SrsJsonObject* obj = SrsJsonAny::object();
SrsAutoFree(SrsJsonObject, obj); SrsAutoFree(SrsJsonObject, obj);
@ -85,7 +86,7 @@ int srs_api_response_jsonp_code(ISrsHttpResponseWriter* w, string callback, int
return srs_api_response_jsonp(w, callback, obj->dumps()); return srs_api_response_jsonp(w, callback, obj->dumps());
} }
int srs_api_response_jsonp_code(ISrsHttpResponseWriter* w, string callback, srs_error_t err) srs_error_t srs_api_response_jsonp_code(ISrsHttpResponseWriter* w, string callback, srs_error_t err)
{ {
SrsJsonObject* obj = SrsJsonAny::object(); SrsJsonObject* obj = SrsJsonAny::object();
SrsAutoFree(SrsJsonObject, obj); SrsAutoFree(SrsJsonObject, obj);
@ -95,17 +96,24 @@ int srs_api_response_jsonp_code(ISrsHttpResponseWriter* w, string callback, srs_
return srs_api_response_jsonp(w, callback, obj->dumps()); return srs_api_response_jsonp(w, callback, obj->dumps());
} }
int srs_api_response_json(ISrsHttpResponseWriter* w, string data) srs_error_t srs_api_response_json(ISrsHttpResponseWriter* w, string data)
{ {
int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
SrsHttpHeader* h = w->header(); SrsHttpHeader* h = w->header();
h->set_content_length(data.length()); h->set_content_length(data.length());
h->set_content_type("application/json"); h->set_content_type("application/json");
return w->write((char*)data.data(), (int)data.length()); if ((ret = w->write((char*)data.data(), (int)data.length())) != ERROR_SUCCESS) {
return srs_error_new(ret, "write json");
}
return err;
} }
int srs_api_response_json_code(ISrsHttpResponseWriter* w, int code) srs_error_t srs_api_response_json_code(ISrsHttpResponseWriter* w, int code)
{ {
SrsJsonObject* obj = SrsJsonAny::object(); SrsJsonObject* obj = SrsJsonAny::object();
SrsAutoFree(SrsJsonObject, obj); SrsAutoFree(SrsJsonObject, obj);
@ -115,17 +123,17 @@ int srs_api_response_json_code(ISrsHttpResponseWriter* w, int code)
return srs_api_response_json(w, obj->dumps()); return srs_api_response_json(w, obj->dumps());
} }
int srs_api_response_json_code(ISrsHttpResponseWriter* w, srs_error_t err) srs_error_t srs_api_response_json_code(ISrsHttpResponseWriter* w, srs_error_t code)
{ {
SrsJsonObject* obj = SrsJsonAny::object(); SrsJsonObject* obj = SrsJsonAny::object();
SrsAutoFree(SrsJsonObject, obj); SrsAutoFree(SrsJsonObject, obj);
obj->set("code", SrsJsonAny::integer(srs_error_code(err))); obj->set("code", SrsJsonAny::integer(srs_error_code(code)));
return srs_api_response_json(w, obj->dumps()); return srs_api_response_json(w, obj->dumps());
} }
int srs_api_response(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string json) srs_error_t srs_api_response(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string json)
{ {
// no jsonp, directly response. // no jsonp, directly response.
if (!r->is_jsonp()) { if (!r->is_jsonp()) {
@ -137,7 +145,7 @@ int srs_api_response(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string
return srs_api_response_jsonp(w, callback, json); return srs_api_response_jsonp(w, callback, json);
} }
int srs_api_response_code(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, int code) srs_error_t srs_api_response_code(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, int code)
{ {
// no jsonp, directly response. // no jsonp, directly response.
if (!r->is_jsonp()) { if (!r->is_jsonp()) {
@ -149,24 +157,25 @@ int srs_api_response_code(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, int cod
return srs_api_response_jsonp_code(w, callback, code); return srs_api_response_jsonp_code(w, callback, code);
} }
int srs_api_response_code(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, srs_error_t err) // @remark we will free the code.
srs_error_t srs_api_response_code(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, srs_error_t code)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
// no jsonp, directly response. // no jsonp, directly response.
if (!r->is_jsonp()) { if (!r->is_jsonp()) {
ret = srs_api_response_json_code(w, err); err = srs_api_response_json_code(w, code);
} else { } else {
// jsonp, get function name from query("callback") // jsonp, get function name from query("callback")
string callback = r->query_get("callback"); string callback = r->query_get("callback");
ret = srs_api_response_jsonp_code(w, callback, err); err = srs_api_response_jsonp_code(w, callback, code);
} }
if (err != srs_success) { if (code != srs_success) {
srs_warn("error %s", srs_error_desc(err).c_str()); srs_warn("error %s", srs_error_desc(code).c_str());
srs_freep(err); srs_freep(code);
} }
return ret; return err;
} }
SrsGoApiRoot::SrsGoApiRoot() SrsGoApiRoot::SrsGoApiRoot()
@ -177,7 +186,7 @@ SrsGoApiRoot::~SrsGoApiRoot()
{ {
} }
int SrsGoApiRoot::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiRoot::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -203,7 +212,7 @@ SrsGoApiApi::~SrsGoApiApi()
{ {
} }
int SrsGoApiApi::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiApi::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -229,7 +238,7 @@ SrsGoApiV1::~SrsGoApiV1()
{ {
} }
int SrsGoApiV1::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiV1::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -275,7 +284,7 @@ SrsGoApiVersion::~SrsGoApiVersion()
{ {
} }
int SrsGoApiVersion::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiVersion::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -304,7 +313,7 @@ SrsGoApiSummaries::~SrsGoApiSummaries()
{ {
} }
int SrsGoApiSummaries::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiSummaries::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -327,7 +336,7 @@ SrsGoApiRusages::~SrsGoApiRusages()
{ {
} }
int SrsGoApiRusages::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiRusages::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -372,7 +381,7 @@ SrsGoApiSelfProcStats::~SrsGoApiSelfProcStats()
{ {
} }
int SrsGoApiSelfProcStats::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiSelfProcStats::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -449,7 +458,7 @@ SrsGoApiSystemProcStats::~SrsGoApiSystemProcStats()
{ {
} }
int SrsGoApiSystemProcStats::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiSystemProcStats::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -488,7 +497,7 @@ SrsGoApiMemInfos::~SrsGoApiMemInfos()
{ {
} }
int SrsGoApiMemInfos::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiMemInfos::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -528,7 +537,7 @@ SrsGoApiAuthors::~SrsGoApiAuthors()
{ {
} }
int SrsGoApiAuthors::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiAuthors::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -558,7 +567,7 @@ SrsGoApiFeatures::~SrsGoApiFeatures()
{ {
} }
int SrsGoApiFeatures::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiFeatures::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -656,7 +665,7 @@ SrsGoApiRequests::~SrsGoApiRequests()
{ {
} }
int SrsGoApiRequests::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiRequests::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
@ -705,7 +714,7 @@ SrsGoApiVhosts::~SrsGoApiVhosts()
{ {
} }
int SrsGoApiVhosts::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiVhosts::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -759,7 +768,7 @@ SrsGoApiStreams::~SrsGoApiStreams()
{ {
} }
int SrsGoApiStreams::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiStreams::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -813,7 +822,7 @@ SrsGoApiClients::~SrsGoApiClients()
{ {
} }
int SrsGoApiClients::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiClients::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -889,7 +898,7 @@ SrsGoApiRaw::~SrsGoApiRaw()
_srs_config->unsubscribe(this); _srs_config->unsubscribe(this);
} }
int SrsGoApiRaw::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiRaw::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success; srs_error_t err = srs_success;
@ -1304,7 +1313,7 @@ int SrsGoApiRaw::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
return srs_api_response(w, r, obj->dumps()); return srs_api_response(w, r, obj->dumps());
} }
return ret; return err;
} }
int SrsGoApiRaw::on_reload_http_api_raw_api() int SrsGoApiRaw::on_reload_http_api_raw_api()
@ -1325,7 +1334,7 @@ SrsGoApiError::~SrsGoApiError()
{ {
} }
int SrsGoApiError::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsGoApiError::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
return srs_api_response_code(w, r, 100); return srs_api_response_code(w, r, 100);
} }
@ -1370,16 +1379,16 @@ void SrsHttpApi::cleanup()
// TODO: FIXME: implements it // TODO: FIXME: implements it
} }
int SrsHttpApi::do_cycle() srs_error_t SrsHttpApi::do_cycle()
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
srs_trace("api get peer ip success. ip=%s", ip.c_str()); srs_trace("api get peer ip success. ip=%s", ip.c_str());
// initialize parser // initialize parser
if ((ret = parser->initialize(HTTP_REQUEST, true)) != ERROR_SUCCESS) { if ((ret = parser->initialize(HTTP_REQUEST, true)) != ERROR_SUCCESS) {
srs_error("api initialize http parser failed. ret=%d", ret); return srs_error_new(ret, "init parser");
return ret;
} }
// set the recv timeout, for some clients never disconnect the connection. // set the recv timeout, for some clients never disconnect the connection.
@ -1388,25 +1397,17 @@ int SrsHttpApi::do_cycle()
// initialize the cors, which will proxy to mux. // initialize the cors, which will proxy to mux.
bool crossdomain_enabled = _srs_config->get_http_api_crossdomain(); bool crossdomain_enabled = _srs_config->get_http_api_crossdomain();
if ((ret = cors->initialize(mux, crossdomain_enabled)) != ERROR_SUCCESS) { if ((err = cors->initialize(mux, crossdomain_enabled)) != srs_success) {
return ret; return srs_error_wrap(err, "init cors");
} }
// process http messages. // process http messages.
while (true) { while ((err = trd->pull()) == srs_success) {
srs_error_t err = srs_success;
if ((err = trd->pull()) != srs_success) {
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
return ret;
}
ISrsHttpMessage* req = NULL; ISrsHttpMessage* req = NULL;
// get a http message // get a http message
if ((ret = parser->parse_message(skt, this, &req)) != ERROR_SUCCESS) { if ((ret = parser->parse_message(skt, this, &req)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "parse message");
} }
// if SUCCESS, always NOT-NULL. // if SUCCESS, always NOT-NULL.
@ -1417,8 +1418,8 @@ int SrsHttpApi::do_cycle()
// ok, handle http request. // ok, handle http request.
SrsHttpResponseWriter writer(skt); SrsHttpResponseWriter writer(skt);
if ((ret = process_request(&writer, req)) != ERROR_SUCCESS) { if ((err = process_request(&writer, req)) != srs_success) {
return ret; return srs_error_wrap(err, "process request");
} }
// read all rest bytes in request body. // read all rest bytes in request body.
@ -1426,7 +1427,7 @@ int SrsHttpApi::do_cycle()
ISrsHttpResponseReader* br = req->body_reader(); ISrsHttpResponseReader* br = req->body_reader();
while (!br->eof()) { while (!br->eof()) {
if ((ret = br->read(buf, SRS_HTTP_READ_CACHE_BYTES, NULL)) != ERROR_SUCCESS) { if ((ret = br->read(buf, SRS_HTTP_READ_CACHE_BYTES, NULL)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "read response");
} }
} }
@ -1437,40 +1438,37 @@ int SrsHttpApi::do_cycle()
} }
} }
return ret; return err;
} }
int SrsHttpApi::process_request(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsHttpApi::process_request(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
SrsHttpMessage* hm = dynamic_cast<SrsHttpMessage*>(r); SrsHttpMessage* hm = dynamic_cast<SrsHttpMessage*>(r);
srs_assert(hm); srs_assert(hm);
srs_trace("HTTP API %s %s, content-length=%" PRId64 ", chunked=%d/%d", srs_trace("HTTP API %s %s, content-length=%" PRId64 ", chunked=%d/%d",
r->method_str().c_str(), r->url().c_str(), r->content_length(), r->method_str().c_str(), r->url().c_str(), r->content_length(),
hm->is_chunked(), hm->is_infinite_chunked()); hm->is_chunked(), hm->is_infinite_chunked());
// use cors server mux to serve http request, which will proxy to mux. // use cors server mux to serve http request, which will proxy to mux.
if ((ret = cors->serve_http(w, r)) != ERROR_SUCCESS) { if ((err = cors->serve_http(w, r)) != srs_success) {
if (!srs_is_client_gracefully_close(ret)) { return srs_error_wrap(err, "mux serve");
srs_error("serve http msg failed. ret=%d", ret);
}
return ret;
} }
return ret; return err;
} }
int SrsHttpApi::on_reload_http_api_crossdomain() srs_error_t SrsHttpApi::on_reload_http_api_crossdomain()
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
bool crossdomain_enabled = _srs_config->get_http_api_crossdomain(); bool crossdomain_enabled = _srs_config->get_http_api_crossdomain();
if ((ret = cors->initialize(mux, crossdomain_enabled)) != ERROR_SUCCESS) { if ((err = cors->initialize(mux, crossdomain_enabled)) != srs_success) {
return ret; return srs_error_wrap(err, "reload");
} }
return ERROR_SUCCESS; return err;
} }

40
trunk/src/app/srs_app_http_api.hpp
View file

@ -44,7 +44,7 @@ public:
SrsGoApiRoot(); SrsGoApiRoot();
virtual ~SrsGoApiRoot(); virtual ~SrsGoApiRoot();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiApi : public ISrsHttpHandler class SrsGoApiApi : public ISrsHttpHandler
@ -53,7 +53,7 @@ public:
SrsGoApiApi(); SrsGoApiApi();
virtual ~SrsGoApiApi(); virtual ~SrsGoApiApi();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiV1 : public ISrsHttpHandler class SrsGoApiV1 : public ISrsHttpHandler
@ -62,7 +62,7 @@ public:
SrsGoApiV1(); SrsGoApiV1();
virtual ~SrsGoApiV1(); virtual ~SrsGoApiV1();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiVersion : public ISrsHttpHandler class SrsGoApiVersion : public ISrsHttpHandler
@ -71,7 +71,7 @@ public:
SrsGoApiVersion(); SrsGoApiVersion();
virtual ~SrsGoApiVersion(); virtual ~SrsGoApiVersion();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiSummaries : public ISrsHttpHandler class SrsGoApiSummaries : public ISrsHttpHandler
@ -80,7 +80,7 @@ public:
SrsGoApiSummaries(); SrsGoApiSummaries();
virtual ~SrsGoApiSummaries(); virtual ~SrsGoApiSummaries();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiRusages : public ISrsHttpHandler class SrsGoApiRusages : public ISrsHttpHandler
@ -89,7 +89,7 @@ public:
SrsGoApiRusages(); SrsGoApiRusages();
virtual ~SrsGoApiRusages(); virtual ~SrsGoApiRusages();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiSelfProcStats : public ISrsHttpHandler class SrsGoApiSelfProcStats : public ISrsHttpHandler
@ -98,7 +98,7 @@ public:
SrsGoApiSelfProcStats(); SrsGoApiSelfProcStats();
virtual ~SrsGoApiSelfProcStats(); virtual ~SrsGoApiSelfProcStats();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiSystemProcStats : public ISrsHttpHandler class SrsGoApiSystemProcStats : public ISrsHttpHandler
@ -107,7 +107,7 @@ public:
SrsGoApiSystemProcStats(); SrsGoApiSystemProcStats();
virtual ~SrsGoApiSystemProcStats(); virtual ~SrsGoApiSystemProcStats();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiMemInfos : public ISrsHttpHandler class SrsGoApiMemInfos : public ISrsHttpHandler
@ -116,7 +116,7 @@ public:
SrsGoApiMemInfos(); SrsGoApiMemInfos();
virtual ~SrsGoApiMemInfos(); virtual ~SrsGoApiMemInfos();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiAuthors : public ISrsHttpHandler class SrsGoApiAuthors : public ISrsHttpHandler
@ -125,7 +125,7 @@ public:
SrsGoApiAuthors(); SrsGoApiAuthors();
virtual ~SrsGoApiAuthors(); virtual ~SrsGoApiAuthors();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiFeatures : public ISrsHttpHandler class SrsGoApiFeatures : public ISrsHttpHandler
@ -134,7 +134,7 @@ public:
SrsGoApiFeatures(); SrsGoApiFeatures();
virtual ~SrsGoApiFeatures(); virtual ~SrsGoApiFeatures();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiRequests : public ISrsHttpHandler class SrsGoApiRequests : public ISrsHttpHandler
@ -143,7 +143,7 @@ public:
SrsGoApiRequests(); SrsGoApiRequests();
virtual ~SrsGoApiRequests(); virtual ~SrsGoApiRequests();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiVhosts : public ISrsHttpHandler class SrsGoApiVhosts : public ISrsHttpHandler
@ -152,7 +152,7 @@ public:
SrsGoApiVhosts(); SrsGoApiVhosts();
virtual ~SrsGoApiVhosts(); virtual ~SrsGoApiVhosts();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiStreams : public ISrsHttpHandler class SrsGoApiStreams : public ISrsHttpHandler
@ -161,7 +161,7 @@ public:
SrsGoApiStreams(); SrsGoApiStreams();
virtual ~SrsGoApiStreams(); virtual ~SrsGoApiStreams();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiClients : public ISrsHttpHandler class SrsGoApiClients : public ISrsHttpHandler
@ -170,7 +170,7 @@ public:
SrsGoApiClients(); SrsGoApiClients();
virtual ~SrsGoApiClients(); virtual ~SrsGoApiClients();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsGoApiRaw : virtual public ISrsHttpHandler, virtual public ISrsReloadHandler class SrsGoApiRaw : virtual public ISrsHttpHandler, virtual public ISrsReloadHandler
@ -186,7 +186,7 @@ public:
SrsGoApiRaw(SrsServer* svr); SrsGoApiRaw(SrsServer* svr);
virtual ~SrsGoApiRaw(); virtual ~SrsGoApiRaw();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
// interface ISrsReloadHandler // interface ISrsReloadHandler
public: public:
virtual int on_reload_http_api_raw_api(); virtual int on_reload_http_api_raw_api();
@ -198,7 +198,7 @@ public:
SrsGoApiError(); SrsGoApiError();
virtual ~SrsGoApiError(); virtual ~SrsGoApiError();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
class SrsHttpApi : virtual public SrsConnection, virtual public ISrsReloadHandler class SrsHttpApi : virtual public SrsConnection, virtual public ISrsReloadHandler
@ -217,12 +217,12 @@ public:
virtual int64_t get_recv_bytes_delta(); virtual int64_t get_recv_bytes_delta();
virtual void cleanup(); virtual void cleanup();
protected: protected:
virtual int do_cycle(); virtual srs_error_t do_cycle();
private: private:
virtual int process_request(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t process_request(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
// interface ISrsReloadHandler // interface ISrsReloadHandler
public: public:
virtual int on_reload_http_api_crossdomain(); virtual srs_error_t on_reload_http_api_crossdomain();
}; };
#endif #endif

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

@ -95,16 +95,16 @@ void SrsHttpConn::cleanup()
// TODO: FIXME: implements it // TODO: FIXME: implements it
} }
int SrsHttpConn::do_cycle() srs_error_t SrsHttpConn::do_cycle()
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
srs_trace("HTTP client ip=%s", ip.c_str()); srs_trace("HTTP client ip=%s", ip.c_str());
// initialize parser // initialize parser
if ((ret = parser->initialize(HTTP_REQUEST, false)) != ERROR_SUCCESS) { if ((ret = parser->initialize(HTTP_REQUEST, false)) != ERROR_SUCCESS) {
srs_error("http initialize http parser failed. ret=%d", ret); return srs_error_new(ret, "init parser");
return ret;
} }
// set the recv timeout, for some clients never disconnect the connection. // set the recv timeout, for some clients never disconnect the connection.
@ -116,20 +116,12 @@ int SrsHttpConn::do_cycle()
// initialize the cors, which will proxy to mux. // initialize the cors, which will proxy to mux.
bool crossdomain_enabled = _srs_config->get_http_stream_crossdomain(); bool crossdomain_enabled = _srs_config->get_http_stream_crossdomain();
if ((ret = cors->initialize(http_mux, crossdomain_enabled)) != ERROR_SUCCESS) { if ((err = cors->initialize(http_mux, crossdomain_enabled)) != srs_success) {
return ret; return srs_error_wrap(err, "init cors");
} }
// process http messages. // process http messages.
while (true) { while ((err = trd->pull()) == srs_success) {
srs_error_t err = srs_success;
if ((err = trd->pull()) != srs_success) {
// TODO: FIXME: Use error
ret = srs_error_code(err);
srs_freep(err);
return ret;
}
ISrsHttpMessage* req = NULL; ISrsHttpMessage* req = NULL;
// get a http message // get a http message
@ -143,19 +135,19 @@ int SrsHttpConn::do_cycle()
// always free it in this scope. // always free it in this scope.
SrsAutoFree(ISrsHttpMessage, req); SrsAutoFree(ISrsHttpMessage, req);
// get the last request, for report the info of request on connection disconnect. // copy request to last request object.
delete last_req; srs_freep(last_req);
SrsHttpMessage* hreq = dynamic_cast<SrsHttpMessage*>(req); SrsHttpMessage* hreq = dynamic_cast<SrsHttpMessage*>(req);
last_req = hreq->to_request(hreq->host()); last_req = hreq->to_request(hreq->host());
// may should discard the body. // may should discard the body.
if ((ret = on_got_http_message(req)) != ERROR_SUCCESS) { if ((err = on_got_http_message(req)) != srs_success) {
break; break;
} }
// ok, handle http request. // ok, handle http request.
SrsHttpResponseWriter writer(skt); SrsHttpResponseWriter writer(skt);
if ((ret = process_request(&writer, req)) != ERROR_SUCCESS) { if ((err = process_request(&writer, req)) != srs_success) {
break; break;
} }
@ -166,50 +158,47 @@ int SrsHttpConn::do_cycle()
} }
} }
int disc_ret = ERROR_SUCCESS; srs_error_t r0 = srs_success;
if ((disc_ret = on_disconnect(last_req)) != ERROR_SUCCESS) { if ((r0 = on_disconnect(last_req)) != srs_success) {
srs_warn("connection on disconnect peer failed, but ignore this error. disc_ret=%d, ret=%d", disc_ret, ret); err = srs_error_wrap(err, "on disconnect %s", srs_error_desc(r0).c_str());
srs_freep(r0);
} }
return ret; return err;
} }
int SrsHttpConn::process_request(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsHttpConn::process_request(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
srs_trace("HTTP %s %s, content-length=%" PRId64 "", srs_trace("HTTP %s %s, content-length=%" PRId64 "",
r->method_str().c_str(), r->url().c_str(), r->content_length()); r->method_str().c_str(), r->url().c_str(), r->content_length());
// use cors server mux to serve http request, which will proxy to http_remux. // use cors server mux to serve http request, which will proxy to http_remux.
if ((ret = cors->serve_http(w, r)) != ERROR_SUCCESS) { if ((err = cors->serve_http(w, r)) != srs_success) {
if (!srs_is_client_gracefully_close(ret)) { return srs_error_wrap(err, "mux serve");
srs_error("serve http msg failed. ret=%d", ret);
}
return ret;
} }
return ret; return err;
} }
int SrsHttpConn::on_disconnect(SrsRequest* req) srs_error_t SrsHttpConn::on_disconnect(SrsRequest* req)
{ {
int ret = ERROR_SUCCESS; // TODO: FIXME: Implements it.
// TODO: implements it.s return srs_success;
return ret;
} }
int SrsHttpConn::on_reload_http_stream_crossdomain() srs_error_t SrsHttpConn::on_reload_http_stream_crossdomain()
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
// initialize the cors, which will proxy to mux. // initialize the cors, which will proxy to mux.
bool crossdomain_enabled = _srs_config->get_http_stream_crossdomain(); bool crossdomain_enabled = _srs_config->get_http_stream_crossdomain();
if ((ret = cors->initialize(http_mux, crossdomain_enabled)) != ERROR_SUCCESS) { if ((err = cors->initialize(http_mux, crossdomain_enabled)) != srs_success) {
return ret; return srs_error_wrap(err, "init mux");
} }
return ret; return err;
} }
SrsResponseOnlyHttpConn::SrsResponseOnlyHttpConn(IConnectionManager* cm, srs_netfd_t fd, ISrsHttpServeMux* m, string cip) SrsResponseOnlyHttpConn::SrsResponseOnlyHttpConn(IConnectionManager* cm, srs_netfd_t fd, ISrsHttpServeMux* m, string cip)
@ -221,43 +210,45 @@ SrsResponseOnlyHttpConn::~SrsResponseOnlyHttpConn()
{ {
} }
int SrsResponseOnlyHttpConn::pop_message(ISrsHttpMessage** preq) srs_error_t SrsResponseOnlyHttpConn::pop_message(ISrsHttpMessage** preq)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
SrsStSocket skt; SrsStSocket skt;
if ((ret = skt.initialize(stfd)) != ERROR_SUCCESS) { if ((ret = skt.initialize(stfd)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "init socket");
} }
if ((ret = parser->parse_message(&skt, this, preq)) != ERROR_SUCCESS) { if ((ret = parser->parse_message(&skt, this, preq)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "parse message");
} }
return ret; return err;
} }
int SrsResponseOnlyHttpConn::on_got_http_message(ISrsHttpMessage* msg) srs_error_t SrsResponseOnlyHttpConn::on_got_http_message(ISrsHttpMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
ISrsHttpResponseReader* br = msg->body_reader(); ISrsHttpResponseReader* br = msg->body_reader();
// when not specified the content length, ignore. // when not specified the content length, ignore.
if (msg->content_length() == -1) { if (msg->content_length() == -1) {
return ret; return err;
} }
// drop all request body. // drop all request body.
while (!br->eof()) { while (!br->eof()) {
char body[4096]; char body[4096];
if ((ret = br->read(body, 4096, NULL)) != ERROR_SUCCESS) { if ((ret = br->read(body, 4096, NULL)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "read response");
} }
} }
return ret; return err;
} }
SrsHttpServer::SrsHttpServer(SrsServer* svr) SrsHttpServer::SrsHttpServer(SrsServer* svr)
@ -293,10 +284,16 @@ srs_error_t SrsHttpServer::initialize()
return err; return err;
} }
int SrsHttpServer::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsHttpServer::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
srs_error_t err = srs_success;
// try http stream first. // try http stream first.
if (http_stream->mux.can_serve(r)) { ISrsHttpHandler* h = NULL;
if ((err = http_stream->mux.find_handler(r, &h)) != srs_success) {
return srs_error_wrap(err, "find handler");
}
if (!h->is_not_found()) {
return http_stream->mux.serve_http(w, r); return http_stream->mux.serve_http(w, r);
} }

16
trunk/src/app/srs_app_http_conn.hpp
View file

@ -74,23 +74,23 @@ public:
virtual int64_t get_recv_bytes_delta(); virtual int64_t get_recv_bytes_delta();
virtual void cleanup(); virtual void cleanup();
protected: protected:
virtual int do_cycle(); virtual srs_error_t do_cycle();
protected: protected:
// when got http message, // when got http message,
// for the static service or api, discard any body. // for the static service or api, discard any body.
// for the stream caster, for instance, http flv streaming, may discard the flv header or not. // for the stream caster, for instance, http flv streaming, may discard the flv header or not.
virtual int on_got_http_message(ISrsHttpMessage* msg) = 0; virtual srs_error_t on_got_http_message(ISrsHttpMessage* msg) = 0;
private: private:
virtual int process_request(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t process_request(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
/** /**
* when the connection disconnect, call this method. * when the connection disconnect, call this method.
* e.g. log msg of connection and report to other system. * e.g. log msg of connection and report to other system.
* @param request: request which is converted by the last http message. * @param request: request which is converted by the last http message.
*/ */
virtual int on_disconnect(SrsRequest* req); virtual srs_error_t on_disconnect(SrsRequest* req);
// interface ISrsReloadHandler // interface ISrsReloadHandler
public: public:
virtual int on_reload_http_stream_crossdomain(); virtual srs_error_t on_reload_http_stream_crossdomain();
}; };
/** /**
@ -107,9 +107,9 @@ public:
// serving it, but we need to start a thread to read message to detect whether FD is closed. // serving it, but we need to start a thread to read message to detect whether FD is closed.
// @see https://github.com/ossrs/srs/issues/636#issuecomment-298208427 // @see https://github.com/ossrs/srs/issues/636#issuecomment-298208427
// @remark Should only used in HTTP-FLV streaming connection. // @remark Should only used in HTTP-FLV streaming connection.
virtual int pop_message(ISrsHttpMessage** preq); virtual srs_error_t pop_message(ISrsHttpMessage** preq);
public: public:
virtual int on_got_http_message(ISrsHttpMessage* msg); virtual srs_error_t on_got_http_message(ISrsHttpMessage* msg);
}; };
/** /**
@ -128,7 +128,7 @@ public:
virtual srs_error_t initialize(); virtual srs_error_t initialize();
// ISrsHttpServeMux // ISrsHttpServeMux
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
// http flv/ts/mp3/aac stream // http flv/ts/mp3/aac stream
public: public:
virtual int http_mount(SrsSource* s, SrsRequest* r); virtual int http_mount(SrsSource* s, SrsRequest* r);

54
trunk/src/app/srs_app_http_static.cpp
View file

@ -60,29 +60,28 @@ SrsVodStream::~SrsVodStream()
{ {
} }
int SrsVodStream::serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath, int offset) srs_error_t SrsVodStream::serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath, int offset)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
SrsFileReader fs; SrsFileReader fs;
// open flv file // open flv file
if ((ret = fs.open(fullpath)) != ERROR_SUCCESS) { if ((ret = fs.open(fullpath)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "open file");
} }
if (offset > fs.filesize()) { if (offset > fs.filesize()) {
ret = ERROR_HTTP_REMUX_OFFSET_OVERFLOW; return srs_error_new(ERROR_HTTP_REMUX_OFFSET_OVERFLOW, "http flv streaming %s overflow. size=%" PRId64 ", offset=%d",
srs_warn("http flv streaming %s overflow. size=%" PRId64 ", offset=%d, ret=%d", fullpath.c_str(), fs.filesize(), offset);
fullpath.c_str(), fs.filesize(), offset, ret);
return ret;
} }
SrsFlvVodStreamDecoder ffd; SrsFlvVodStreamDecoder ffd;
// open fast decoder // open fast decoder
if ((ret = ffd.initialize(&fs)) != ERROR_SUCCESS) { if ((ret = ffd.initialize(&fs)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "init ffd");
} }
// save header, send later. // save header, send later.
@ -90,7 +89,7 @@ int SrsVodStream::serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r
// send flv header // send flv header
if ((ret = ffd.read_header_ext(flv_header)) != ERROR_SUCCESS) { if ((ret = ffd.read_header_ext(flv_header)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "ffd read header");
} }
// save sequence header, send later // save sequence header, send later
@ -101,18 +100,16 @@ int SrsVodStream::serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r
// send sequence header // send sequence header
int64_t start = 0; int64_t start = 0;
if ((ret = ffd.read_sequence_header_summary(&start, &sh_size)) != ERROR_SUCCESS) { if ((ret = ffd.read_sequence_header_summary(&start, &sh_size)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "ffd read sps");
} }
if (sh_size <= 0) { if (sh_size <= 0) {
ret = ERROR_HTTP_REMUX_SEQUENCE_HEADER; return srs_error_new(ERROR_HTTP_REMUX_SEQUENCE_HEADER, "no sequence, size=%d", sh_size);
srs_warn("http flv streaming no sequence header. size=%d, ret=%d", sh_size, ret);
return ret;
} }
} }
sh_data = new char[sh_size]; sh_data = new char[sh_size];
SrsAutoFreeA(char, sh_data); SrsAutoFreeA(char, sh_data);
if ((ret = fs.read(sh_data, sh_size, NULL)) != ERROR_SUCCESS) { if ((ret = fs.read(sh_data, sh_size, NULL)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "fs read");
} }
// seek to data offset // seek to data offset
@ -124,29 +121,29 @@ int SrsVodStream::serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r
// write flv header and sequence header. // write flv header and sequence header.
if ((ret = w->write(flv_header, sizeof(flv_header))) != ERROR_SUCCESS) { if ((ret = w->write(flv_header, sizeof(flv_header))) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "write flv header");
} }
if (sh_size > 0 && (ret = w->write(sh_data, sh_size)) != ERROR_SUCCESS) { if (sh_size > 0 && (ret = w->write(sh_data, sh_size)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "write sequence");
} }
// write body. // write body.
if ((ret = ffd.seek2(offset)) != ERROR_SUCCESS) { if ((ret = ffd.seek2(offset)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "ffd seek");
} }
// send data // send data
if ((ret = copy(w, &fs, r, (int)left)) != ERROR_SUCCESS) { if ((err = copy(w, &fs, r, (int)left)) != srs_success) {
srs_warn("read flv=%s size=%d failed, ret=%d", fullpath.c_str(), left, ret); return srs_error_wrap(err, "read flv=%s size=%d", fullpath.c_str(), left);
return ret;
} }
return ret; return err;
} }
int SrsVodStream::serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath, int start, int end) srs_error_t SrsVodStream::serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath, int start, int end)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
srs_assert(start >= 0); srs_assert(start >= 0);
srs_assert(end == -1 || end >= 0); srs_assert(end == -1 || end >= 0);
@ -155,7 +152,7 @@ int SrsVodStream::serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r
// open flv file // open flv file
if ((ret = fs.open(fullpath)) != ERROR_SUCCESS) { if ((ret = fs.open(fullpath)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "fs open");
} }
// parse -1 to whole file. // parse -1 to whole file.
@ -164,10 +161,8 @@ int SrsVodStream::serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r
} }
if (end > fs.filesize() || start > end) { if (end > fs.filesize() || start > end) {
ret = ERROR_HTTP_REMUX_OFFSET_OVERFLOW; return srs_error_new(ERROR_HTTP_REMUX_OFFSET_OVERFLOW, "http mp4 streaming %s overflow. size=%" PRId64 ", offset=%d",
srs_warn("http mp4 streaming %s overflow. size=%" PRId64 ", offset=%d, ret=%d", fullpath.c_str(), fs.filesize(), start);
fullpath.c_str(), fs.filesize(), start, ret);
return ret;
} }
// seek to data offset, [start, end] for range. // seek to data offset, [start, end] for range.
@ -189,12 +184,11 @@ int SrsVodStream::serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r
fs.seek2(start); fs.seek2(start);
// send data // send data
if ((ret = copy(w, &fs, r, (int)left)) != ERROR_SUCCESS) { if ((err = copy(w, &fs, r, (int)left)) != srs_success) {
srs_warn("read mp4=%s size=%d failed, ret=%d", fullpath.c_str(), left, ret); return srs_error_wrap(err, "read mp4=%s size=%d", fullpath.c_str(), left);
return ret;
} }
return ret; return err;
} }
SrsHttpStaticServer::SrsHttpStaticServer(SrsServer* svr) SrsHttpStaticServer::SrsHttpStaticServer(SrsServer* svr)

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

@ -40,8 +40,8 @@ public:
SrsVodStream(std::string root_dir); SrsVodStream(std::string root_dir);
virtual ~SrsVodStream(); virtual ~SrsVodStream();
protected: protected:
virtual int serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int offset); virtual srs_error_t serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int offset);
virtual int serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int start, int end); virtual srs_error_t serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int start, int end);
}; };
/** /**

73
trunk/src/app/srs_app_http_stream.cpp
View file

@ -478,7 +478,7 @@ int SrsLiveStream::update(SrsSource* s, SrsRequest* r)
return ret; return ret;
} }
int SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success; srs_error_t err = srs_success;
@ -503,17 +503,14 @@ int SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
w->header()->set_content_type("video/MP2T"); w->header()->set_content_type("video/MP2T");
enc = new SrsTsStreamEncoder(); enc = new SrsTsStreamEncoder();
} else { } else {
ret = ERROR_HTTP_LIVE_STREAM_EXT; return srs_error_new(ERROR_HTTP_LIVE_STREAM_EXT, "invalid pattern=%s", entry->pattern.c_str());
srs_error("http: unsupported pattern %s", entry->pattern.c_str());
return ret;
} }
SrsAutoFree(ISrsBufferEncoder, enc); SrsAutoFree(ISrsBufferEncoder, enc);
// create consumer of souce, ignore gop cache, use the audio gop cache. // create consumer of souce, ignore gop cache, use the audio gop cache.
SrsConsumer* consumer = NULL; SrsConsumer* consumer = NULL;
if ((ret = source->create_consumer(NULL, consumer, true, true, !enc->has_cache())) != ERROR_SUCCESS) { if ((ret = source->create_consumer(NULL, consumer, true, true, !enc->has_cache())) != ERROR_SUCCESS) {
srs_error("http: create consumer failed. ret=%d", ret); return srs_error_new(ret, "create consumer");
return ret;
} }
SrsAutoFree(SrsConsumer, consumer); SrsAutoFree(SrsConsumer, consumer);
srs_verbose("http: consumer created success."); srs_verbose("http: consumer created success.");
@ -526,22 +523,19 @@ int SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
// update the statistic when source disconveried. // update the statistic when source disconveried.
SrsStatistic* stat = SrsStatistic::instance(); SrsStatistic* stat = SrsStatistic::instance();
if ((ret = stat->on_client(_srs_context->get_id(), req, NULL, SrsRtmpConnPlay)) != ERROR_SUCCESS) { if ((ret = stat->on_client(_srs_context->get_id(), req, NULL, SrsRtmpConnPlay)) != ERROR_SUCCESS) {
srs_error("stat client failed. ret=%d", ret); return srs_error_new(ret, "stat on client");
return ret;
} }
// the memory writer. // the memory writer.
SrsBufferWriter writer(w); SrsBufferWriter writer(w);
if ((ret = enc->initialize(&writer, cache)) != ERROR_SUCCESS) { if ((ret = enc->initialize(&writer, cache)) != ERROR_SUCCESS) {
srs_error("http: initialize stream encoder failed. ret=%d", ret); return srs_error_new(ret, "init encoder");
return ret;
} }
// if gop cache enabled for encoder, dump to consumer. // if gop cache enabled for encoder, dump to consumer.
if (enc->has_cache()) { if (enc->has_cache()) {
if ((ret = enc->dump_cache(consumer, source->jitter())) != ERROR_SUCCESS) { if ((ret = enc->dump_cache(consumer, source->jitter())) != ERROR_SUCCESS) {
srs_error("http: dump cache to consumer failed. ret=%d", ret); return srs_error_new(ret, "encoder dump cache");
return ret;
} }
} }
@ -558,12 +552,7 @@ int SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
SrsAutoFree(SrsHttpRecvThread, trd); SrsAutoFree(SrsHttpRecvThread, trd);
if ((err = trd->start()) != srs_success) { if ((err = trd->start()) != srs_success) {
// TODO: FIXME: Use error return srs_error_wrap(err, "start recv thread");
ret = srs_error_code(err);
srs_freep(err);
srs_error("http: start notify thread failed, ret=%d", ret);
return ret;
} }
// TODO: free and erase the disabled entry after all related connections is closed. // TODO: free and erase the disabled entry after all related connections is closed.
@ -571,16 +560,15 @@ int SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
pprint->elapse(); pprint->elapse();
// Whether client closed the FD. // Whether client closed the FD.
if ((ret = trd->error_code()) != ERROR_SUCCESS) { if ((err = trd->pull()) != srs_success) {
return ret; return srs_error_wrap(err, "recv thread");
} }
// get messages from consumer. // get messages from consumer.
// each msg in msgs.msgs must be free, for the SrsMessageArray never free them. // each msg in msgs.msgs must be free, for the SrsMessageArray never free them.
int count = 0; int count = 0;
if ((ret = consumer->dump_packets(&msgs, count)) != ERROR_SUCCESS) { if ((ret = consumer->dump_packets(&msgs, count)) != ERROR_SUCCESS) {
srs_error("http: get messages from consumer failed. ret=%d", ret); return srs_error_new(ret, "consumer dump packets");
return ret;
} }
if (count <= 0) { if (count <= 0) {
@ -594,7 +582,7 @@ int SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
if (pprint->can_print()) { if (pprint->can_print()) {
srs_info("-> " SRS_CONSTS_LOG_HTTP_STREAM " http: got %d msgs, age=%d, min=%d, mw=%d", srs_info("-> " SRS_CONSTS_LOG_HTTP_STREAM " http: got %d msgs, age=%d, min=%d, mw=%d",
count, pprint->age(), SRS_PERF_MW_MIN_MSGS, SRS_CONSTS_RTMP_PULSE_TMMS); count, pprint->age(), SRS_PERF_MW_MIN_MSGS, SRS_CONSTS_RTMP_PULSE_TMMS);
} }
// sendout all messages. // sendout all messages.
@ -616,14 +604,11 @@ int SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
// check send error code. // check send error code.
if (ret != ERROR_SUCCESS) { if (ret != ERROR_SUCCESS) {
if (!srs_is_client_gracefully_close(ret)) { return srs_error_new(ret, "send messages");
srs_error("http: send messages to client failed. ret=%d", ret);
}
return ret;
} }
} }
return ret; return err;
} }
int SrsLiveStream::streaming_send_messages(ISrsBufferEncoder* enc, SrsSharedPtrMessage** msgs, int nb_msgs) int SrsLiveStream::streaming_send_messages(ISrsBufferEncoder* enc, SrsSharedPtrMessage** msgs, int nb_msgs)
@ -895,26 +880,27 @@ int SrsHttpStreamServer::on_reload_vhost_http_remux_updated(string vhost)
return ret; return ret;
} }
int SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph) srs_error_t SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
// when handler not the root, we think the handler is ok. // when handler not the root, we think the handler is ok.
ISrsHttpHandler* h = *ph? *ph : NULL; ISrsHttpHandler* h = *ph? *ph : NULL;
if (h && h->entry && h->entry->pattern != "/") { if (h && h->entry && h->entry->pattern != "/") {
return ret; return err;
} }
// only hijack for http streaming, http-flv/ts/mp3/aac. // only hijack for http streaming, http-flv/ts/mp3/aac.
std::string ext = request->ext(); std::string ext = request->ext();
if (ext.empty()) { if (ext.empty()) {
return ret; return err;
} }
// find the actually request vhost. // find the actually request vhost.
SrsConfDirective* vhost = _srs_config->get_vhost(request->host()); SrsConfDirective* vhost = _srs_config->get_vhost(request->host());
if (!vhost || !_srs_config->get_vhost_enabled(vhost)) { if (!vhost || !_srs_config->get_vhost_enabled(vhost)) {
return ret; return err;
} }
// find the entry template for the stream. // find the entry template for the stream.
@ -923,7 +909,7 @@ int SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph)
// no http streaming on vhost, ignore. // no http streaming on vhost, ignore.
std::map<std::string, SrsLiveEntry*>::iterator it = tflvs.find(vhost->arg0()); std::map<std::string, SrsLiveEntry*>::iterator it = tflvs.find(vhost->arg0());
if (it == tflvs.end()) { if (it == tflvs.end()) {
return ret; return err;
} }
// hstrs always enabled. // hstrs always enabled.
@ -936,22 +922,22 @@ int SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph)
// check entry and request extension. // check entry and request extension.
if (entry->is_flv()) { if (entry->is_flv()) {
if (ext != ".flv") { if (ext != ".flv") {
return ret; return err;
} }
} else if (entry->is_ts()) { } else if (entry->is_ts()) {
if (ext != ".ts") { if (ext != ".ts") {
return ret; return err;
} }
} else if (entry->is_mp3()) { } else if (entry->is_mp3()) {
if (ext != ".mp3") { if (ext != ".mp3") {
return ret; return err;
} }
} else if (entry->is_aac()) { } else if (entry->is_aac()) {
if (ext != ".aac") { if (ext != ".aac") {
return ret; return err;
} }
} else { } else {
return ret; return err;
} }
} }
@ -973,21 +959,20 @@ int SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph)
// for the http flv edge use hijack to trigger the edge ingester, we always mount it // for the http flv edge use hijack to trigger the edge ingester, we always mount it
// eventhough the origin does not exists the specified stream. // eventhough the origin does not exists the specified stream.
if (!_srs_config->get_vhost_http_remux_enabled(r->vhost)) { if (!_srs_config->get_vhost_http_remux_enabled(r->vhost)) {
srs_error("stream is disabled, hijack failed. ret=%d", ret); return srs_error_new(ERROR_HTTP_HIJACK, "stream disabled");
return ret;
} }
} }
} }
SrsSource* s = NULL; SrsSource* s = NULL;
if ((ret = SrsSource::fetch_or_create(r, server, &s)) != ERROR_SUCCESS) { if ((ret = SrsSource::fetch_or_create(r, server, &s)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "source create");
} }
srs_assert(s != NULL); srs_assert(s != NULL);
// create http streaming handler. // create http streaming handler.
if ((ret = http_mount(s, r)) != ERROR_SUCCESS) { if ((ret = http_mount(s, r)) != ERROR_SUCCESS) {
return ret; return srs_error_new(ret, "http mount");
} }
// use the handler if exists. // use the handler if exists.
@ -1001,9 +986,9 @@ int SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph)
// trigger edge to fetch from origin. // trigger edge to fetch from origin.
bool vhost_is_edge = _srs_config->get_vhost_is_edge(r->vhost); bool vhost_is_edge = _srs_config->get_vhost_is_edge(r->vhost);
srs_trace("flv: source url=%s, is_edge=%d, source_id=%d[%d]", srs_trace("flv: source url=%s, is_edge=%d, source_id=%d[%d]",
r->get_stream_url().c_str(), vhost_is_edge, s->source_id(), s->source_id()); r->get_stream_url().c_str(), vhost_is_edge, s->source_id(), s->source_id());
return ret; return err;
} }
int SrsHttpStreamServer::initialize_flv_streaming() int SrsHttpStreamServer::initialize_flv_streaming()

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

@ -229,7 +229,7 @@ public:
virtual ~SrsLiveStream(); virtual ~SrsLiveStream();
virtual int update(SrsSource* s, SrsRequest* r); virtual int update(SrsSource* s, SrsRequest* r);
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
private: private:
virtual int streaming_send_messages(ISrsBufferEncoder* enc, SrsSharedPtrMessage** msgs, int nb_msgs); virtual int streaming_send_messages(ISrsBufferEncoder* enc, SrsSharedPtrMessage** msgs, int nb_msgs);
}; };
@ -294,7 +294,7 @@ public:
virtual int on_reload_vhost_http_remux_updated(std::string vhost); virtual int on_reload_vhost_http_remux_updated(std::string vhost);
// interface ISrsHttpMatchHijacker // interface ISrsHttpMatchHijacker
public: public:
virtual int hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph); virtual srs_error_t hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph);
private: private:
virtual int initialize_flv_streaming(); virtual int initialize_flv_streaming();
virtual int initialize_flv_entry(std::string vhost); virtual int initialize_flv_entry(std::string vhost);

18
trunk/src/app/srs_app_recv_thread.cpp
View file

@ -560,7 +560,6 @@ void SrsPublishRecvThread::set_socket_buffer(int sleep_ms)
SrsHttpRecvThread::SrsHttpRecvThread(SrsResponseOnlyHttpConn* c) SrsHttpRecvThread::SrsHttpRecvThread(SrsResponseOnlyHttpConn* c)
{ {
conn = c; conn = c;
error = ERROR_SUCCESS;
trd = new SrsSTCoroutine("http-receive", this, _srs_context->get_id()); trd = new SrsSTCoroutine("http-receive", this, _srs_context->get_id());
} }
@ -580,28 +579,21 @@ srs_error_t SrsHttpRecvThread::start()
return err; return err;
} }
int SrsHttpRecvThread::error_code() srs_error_t SrsHttpRecvThread::pull()
{ {
return error; return trd->pull();
} }
srs_error_t SrsHttpRecvThread::cycle() srs_error_t SrsHttpRecvThread::cycle()
{ {
int ret = ERROR_SUCCESS;
srs_error_t err = srs_success; srs_error_t err = srs_success;
while (true) { while ((err = trd->pull()) == srs_success) {
if ((err = trd->pull()) != srs_success) {
return srs_error_wrap(err, "http recv thread");
}
ISrsHttpMessage* req = NULL; ISrsHttpMessage* req = NULL;
SrsAutoFree(ISrsHttpMessage, req); SrsAutoFree(ISrsHttpMessage, req);
if ((ret = conn->pop_message(&req)) != ERROR_SUCCESS) { if ((err = conn->pop_message(&req)) != srs_success) {
err = srs_error_new(ret, "pop message"); return srs_error_wrap(err, "pop message");
error = ret;
break;
} }
} }

3
trunk/src/app/srs_app_recv_thread.hpp
View file

@ -234,14 +234,13 @@ class SrsHttpRecvThread : public ISrsCoroutineHandler
private: private:
SrsResponseOnlyHttpConn* conn; SrsResponseOnlyHttpConn* conn;
SrsCoroutine* trd; SrsCoroutine* trd;
int error;
public: public:
SrsHttpRecvThread(SrsResponseOnlyHttpConn* c); SrsHttpRecvThread(SrsResponseOnlyHttpConn* c);
virtual ~SrsHttpRecvThread(); virtual ~SrsHttpRecvThread();
public: public:
virtual srs_error_t start(); virtual srs_error_t start();
public: public:
virtual int error_code(); virtual srs_error_t pull();
// interface ISrsOneCycleThreadHandler // interface ISrsOneCycleThreadHandler
public: public:
virtual srs_error_t cycle(); virtual srs_error_t cycle();

8
trunk/src/app/srs_app_reload.cpp
View file

@ -85,9 +85,9 @@ int ISrsReloadHandler::on_reload_http_api_disabled()
return ERROR_SUCCESS; return ERROR_SUCCESS;
} }
int ISrsReloadHandler::on_reload_http_api_crossdomain() srs_error_t ISrsReloadHandler::on_reload_http_api_crossdomain()
{ {
return ERROR_SUCCESS; return srs_success;
} }
int ISrsReloadHandler::on_reload_http_api_raw_api() int ISrsReloadHandler::on_reload_http_api_raw_api()
@ -110,9 +110,9 @@ int ISrsReloadHandler::on_reload_http_stream_updated()
return ERROR_SUCCESS; return ERROR_SUCCESS;
} }
int ISrsReloadHandler::on_reload_http_stream_crossdomain() srs_error_t ISrsReloadHandler::on_reload_http_stream_crossdomain()
{ {
return ERROR_SUCCESS; return srs_success;
} }
int ISrsReloadHandler::on_reload_vhost_http_updated() int ISrsReloadHandler::on_reload_vhost_http_updated()

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

@ -51,12 +51,12 @@ public:
virtual int on_reload_pithy_print(); virtual int on_reload_pithy_print();
virtual int on_reload_http_api_enabled(); virtual int on_reload_http_api_enabled();
virtual int on_reload_http_api_disabled(); virtual int on_reload_http_api_disabled();
virtual int on_reload_http_api_crossdomain(); virtual srs_error_t on_reload_http_api_crossdomain();
virtual int on_reload_http_api_raw_api(); virtual int on_reload_http_api_raw_api();
virtual int on_reload_http_stream_enabled(); virtual int on_reload_http_stream_enabled();
virtual int on_reload_http_stream_disabled(); virtual int on_reload_http_stream_disabled();
virtual int on_reload_http_stream_updated(); virtual int on_reload_http_stream_updated();
virtual int on_reload_http_stream_crossdomain(); virtual srs_error_t on_reload_http_stream_crossdomain();
public: public:
// TODO: FIXME: should rename to http_static // TODO: FIXME: should rename to http_static
virtual int on_reload_vhost_http_updated(); virtual int on_reload_vhost_http_updated();

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

@ -157,17 +157,17 @@ void SrsRtmpConn::dispose()
} }
// TODO: return detail message when error for client. // TODO: return detail message when error for client.
int SrsRtmpConn::do_cycle() srs_error_t SrsRtmpConn::do_cycle()
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
srs_trace("RTMP client ip=%s, fd=%d", ip.c_str(), srs_netfd_fileno(stfd)); srs_trace("RTMP client ip=%s, fd=%d", ip.c_str(), srs_netfd_fileno(stfd));
// notify kafka cluster. // notify kafka cluster.
#ifdef SRS_AUTO_KAFKA #ifdef SRS_AUTO_KAFKA
if ((ret = _srs_kafka->on_client(srs_id(), SrsListenerRtmpStream, ip)) != ERROR_SUCCESS) { if ((ret = _srs_kafka->on_client(srs_id(), SrsListenerRtmpStream, ip)) != ERROR_SUCCESS) {
srs_error("kafka handler on_client failed. ret=%d", ret); return srs_error_new(ret, "kafka on client");
return ret;
} }
#endif #endif
@ -175,17 +175,13 @@ int SrsRtmpConn::do_cycle()
rtmp->set_send_timeout(SRS_CONSTS_RTMP_TMMS); rtmp->set_send_timeout(SRS_CONSTS_RTMP_TMMS);
if ((ret = rtmp->handshake()) != ERROR_SUCCESS) { if ((ret = rtmp->handshake()) != ERROR_SUCCESS) {
srs_error("rtmp handshake failed. ret=%d", ret); return srs_error_new(ret, "rtmp handshake");
return ret;
} }
srs_verbose("rtmp handshake success");
SrsRequest* req = info->req; SrsRequest* req = info->req;
if ((ret = rtmp->connect_app(req)) != ERROR_SUCCESS) { if ((ret = rtmp->connect_app(req)) != ERROR_SUCCESS) {
srs_error("rtmp connect vhost/app failed. ret=%d", ret); return srs_error_new(ret, "rtmp connect tcUrl");
return ret;
} }
srs_verbose("rtmp connect app success");
// set client ip to request. // set client ip to request.
req->ip = ip; req->ip = ip;
@ -197,28 +193,22 @@ int SrsRtmpConn::do_cycle()
} }
srs_info("discovery app success. schema=%s, vhost=%s, port=%d, app=%s", srs_info("discovery app success. schema=%s, vhost=%s, port=%d, app=%s",
req->schema.c_str(), req->vhost.c_str(), req->port, req->app.c_str()); req->schema.c_str(), req->vhost.c_str(), req->port, req->app.c_str());
if (req->schema.empty() || req->vhost.empty() || req->port == 0 || req->app.empty()) { if (req->schema.empty() || req->vhost.empty() || req->port == 0 || req->app.empty()) {
ret = ERROR_RTMP_REQ_TCURL; return srs_error_new(ERROR_RTMP_REQ_TCURL, "discovery tcUrl failed, tcUrl=%s, schema=%s, vhost=%s, port=%d, app=%s",
srs_error("discovery tcUrl failed. " req->tcUrl.c_str(), req->schema.c_str(), req->vhost.c_str(), req->port, req->app.c_str());
"tcUrl=%s, schema=%s, vhost=%s, port=%d, app=%s, ret=%d",
req->tcUrl.c_str(), req->schema.c_str(), req->vhost.c_str(), req->port, req->app.c_str(), ret);
return ret;
} }
// check vhost, allow default vhost. // check vhost, allow default vhost.
if ((ret = check_vhost(true)) != ERROR_SUCCESS) { if ((ret = check_vhost(true)) != ERROR_SUCCESS) {
srs_error("check vhost failed. ret=%d", ret); return srs_error_new(ret, "check vhost");
return ret;
} }
srs_verbose("check vhost success.");
srs_trace("connect app, " srs_trace("connect app, tcUrl=%s, pageUrl=%s, swfUrl=%s, schema=%s, vhost=%s, port=%d, app=%s, args=%s",
"tcUrl=%s, pageUrl=%s, swfUrl=%s, schema=%s, vhost=%s, port=%d, app=%s, args=%s", req->tcUrl.c_str(), req->pageUrl.c_str(), req->swfUrl.c_str(),
req->tcUrl.c_str(), req->pageUrl.c_str(), req->swfUrl.c_str(), req->schema.c_str(), req->vhost.c_str(), req->port,
req->schema.c_str(), req->vhost.c_str(), req->port, req->app.c_str(), (req->args? "(obj)":"null"));
req->app.c_str(), (req->args? "(obj)":"null"));
// show client identity // show client identity
if(req->args) { if(req->args) {
@ -242,21 +232,24 @@ int SrsRtmpConn::do_cycle()
} }
srs_info("edge-srs ip=%s, version=%s, pid=%d, id=%d", srs_info("edge-srs ip=%s, version=%s, pid=%d, id=%d",
srs_server_ip.c_str(), srs_version.c_str(), srs_pid, srs_id); srs_server_ip.c_str(), srs_version.c_str(), srs_pid, srs_id);
if (srs_pid > 0) { if (srs_pid > 0) {
srs_trace("edge-srs ip=%s, version=%s, pid=%d, id=%d", srs_trace("edge-srs ip=%s, version=%s, pid=%d, id=%d",
srs_server_ip.c_str(), srs_version.c_str(), srs_pid, srs_id); srs_server_ip.c_str(), srs_version.c_str(), srs_pid, srs_id);
} }
} }
ret = service_cycle(); if ((ret = service_cycle()) != ERROR_SUCCESS) {
err = srs_error_new(ret, "service cycle");
int disc_ret = ERROR_SUCCESS;
if ((disc_ret = on_disconnect()) != ERROR_SUCCESS) {
srs_warn("connection on disconnect peer failed, but ignore this error. disc_ret=%d, ret=%d", disc_ret, ret);
} }
return ret; srs_error_t r0 = srs_success;
if ((r0 = on_disconnect()) != srs_success) {
err = srs_error_wrap(err, "on disconnect %s", srs_error_desc(r0).c_str());
srs_freep(r0);
}
return err;
} }
int SrsRtmpConn::on_reload_vhost_removed(string vhost) int SrsRtmpConn::on_reload_vhost_removed(string vhost)
@ -1399,22 +1392,22 @@ int SrsRtmpConn::do_token_traverse_auth(SrsRtmpClient* client)
return ret; return ret;
} }
int SrsRtmpConn::on_disconnect() srs_error_t SrsRtmpConn::on_disconnect()
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
http_hooks_on_close(); http_hooks_on_close();
#ifdef SRS_AUTO_KAFKA #ifdef SRS_AUTO_KAFKA
if ((ret = _srs_kafka->on_close(srs_id())) != ERROR_SUCCESS) { if ((ret = _srs_kafka->on_close(srs_id())) != ERROR_SUCCESS) {
srs_error("notify kafka failed. ret=%d", ret); return srs_error_new(ret, "kafka on close");
return ret;
} }
#endif #endif
// TODO: implements it. // TODO: FIXME: Implements it.
return ret; return err;
} }
int SrsRtmpConn::http_hooks_on_connect() int SrsRtmpConn::http_hooks_on_connect()

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

@ -132,7 +132,7 @@ public:
public: public:
virtual void dispose(); virtual void dispose();
protected: protected:
virtual int do_cycle(); virtual srs_error_t do_cycle();
// interface ISrsReloadHandler // interface ISrsReloadHandler
public: public:
virtual int on_reload_vhost_removed(std::string vhost); virtual int on_reload_vhost_removed(std::string vhost);
@ -171,7 +171,7 @@ private:
* when the connection disconnect, call this method. * when the connection disconnect, call this method.
* e.g. log msg of connection and report to other system. * e.g. log msg of connection and report to other system.
*/ */
virtual int on_disconnect(); virtual srs_error_t on_disconnect();
private: private:
virtual int http_hooks_on_connect(); virtual int http_hooks_on_connect();
virtual void http_hooks_on_close(); virtual void http_hooks_on_close();

4
trunk/src/kernel/srs_kernel_error.cpp
View file

@ -121,7 +121,9 @@ SrsCplxError* SrsCplxError::wrap(const char* func, const char* file, int line, S
err->func = func; err->func = func;
err->file = file; err->file = file;
err->line = line; err->line = line;
err->code = v->code; if (v) {
err->code = v->code;
}
err->rerrno = rerrno; err->rerrno = rerrno;
err->msg = buffer; err->msg = buffer;
err->wrapped = v; err->wrapped = v;

1
trunk/src/kernel/srs_kernel_error.hpp
View file

@ -167,6 +167,7 @@
#define ERROR_RTMP_CLIENT_NOT_FOUND 2049 #define ERROR_RTMP_CLIENT_NOT_FOUND 2049
#define ERROR_OpenSslCreateHMAC 2050 #define ERROR_OpenSslCreateHMAC 2050
#define ERROR_RTMP_STREAM_NAME_EMPTY 2051 #define ERROR_RTMP_STREAM_NAME_EMPTY 2051
#define ERROR_HTTP_HIJACK 2052
// //
// system control message, // system control message,
// not an error, but special control logic. // not an error, but special control logic.

124
trunk/src/protocol/srs_http_stack.cpp
View file

@ -122,21 +122,23 @@ string srs_go_http_detect(char* data, int size)
return "application/octet-stream"; // fallback return "application/octet-stream"; // fallback
} }
int srs_go_http_error(ISrsHttpResponseWriter* w, int code) srs_error_t srs_go_http_error(ISrsHttpResponseWriter* w, int code)
{ {
return srs_go_http_error(w, code, srs_generate_http_status_text(code)); return srs_go_http_error(w, code, srs_generate_http_status_text(code));
} }
int srs_go_http_error(ISrsHttpResponseWriter* w, int code, string error) srs_error_t srs_go_http_error(ISrsHttpResponseWriter* w, int code, string error)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
w->header()->set_content_type("text/plain; charset=utf-8"); w->header()->set_content_type("text/plain; charset=utf-8");
w->header()->set_content_length(error.length()); w->header()->set_content_length(error.length());
w->write_header(code); w->write_header(code);
w->write((char*)error.data(), (int)error.length()); if ((ret = w->write((char*)error.data(), (int)error.length())) != ERROR_SUCCESS) {
return srs_error_new(ret, "http write");
}
return ret; return srs_success;
} }
SrsHttpHeader::SrsHttpHeader() SrsHttpHeader::SrsHttpHeader()
@ -237,10 +239,8 @@ SrsHttpRedirectHandler::~SrsHttpRedirectHandler()
{ {
} }
int SrsHttpRedirectHandler::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsHttpRedirectHandler::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
int ret = ERROR_SUCCESS;
string location = url; string location = url;
if (!r->query().empty()) { if (!r->query().empty()) {
location += "?" + r->query(); location += "?" + r->query();
@ -257,7 +257,7 @@ int SrsHttpRedirectHandler::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessag
w->final_request(); w->final_request();
srs_info("redirect to %s.", location.c_str()); srs_info("redirect to %s.", location.c_str());
return ret; return srs_success;
} }
SrsHttpNotFoundHandler::SrsHttpNotFoundHandler() SrsHttpNotFoundHandler::SrsHttpNotFoundHandler()
@ -273,7 +273,7 @@ bool SrsHttpNotFoundHandler::is_not_found()
return true; return true;
} }
int SrsHttpNotFoundHandler::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsHttpNotFoundHandler::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
return srs_go_http_error(w, SRS_CONSTS_HTTP_NotFound); return srs_go_http_error(w, SRS_CONSTS_HTTP_NotFound);
} }
@ -287,7 +287,7 @@ SrsHttpFileServer::~SrsHttpFileServer()
{ {
} }
int SrsHttpFileServer::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsHttpFileServer::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
string upath = r->path(); string upath = r->path();
@ -328,16 +328,16 @@ int SrsHttpFileServer::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
return serve_file(w, r, fullpath); return serve_file(w, r, fullpath);
} }
int SrsHttpFileServer::serve_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath) srs_error_t SrsHttpFileServer::serve_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_error_t err = srs_success;
// open the target file. // open the target file.
SrsFileReader fs; SrsFileReader fs;
if ((ret = fs.open(fullpath)) != ERROR_SUCCESS) { if ((ret = fs.open(fullpath)) != ERROR_SUCCESS) {
srs_warn("open file %s failed, ret=%d", fullpath.c_str(), ret); return srs_error_new(ret, "open file %s", fullpath.c_str());
return ret;
} }
int64_t length = fs.filesize(); int64_t length = fs.filesize();
@ -393,17 +393,18 @@ int SrsHttpFileServer::serve_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r,
// write body. // write body.
int64_t left = length; int64_t left = length;
if ((ret = copy(w, &fs, r, (int)left)) != ERROR_SUCCESS) { if ((err = copy(w, &fs, r, (int)left)) != srs_success) {
if (!srs_is_client_gracefully_close(ret)) { return srs_error_wrap(err, "copy file=%s size=%d", fullpath.c_str(), left);
srs_error("read file=%s size=%d failed, ret=%d", fullpath.c_str(), left, ret);
}
return ret;
} }
return w->final_request(); if ((ret = w->final_request()) != ERROR_SUCCESS) {
return srs_error_new(ret, "final request");
}
return err;
} }
int SrsHttpFileServer::serve_flv_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath) srs_error_t SrsHttpFileServer::serve_flv_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath)
{ {
std::string start = r->query_get("start"); std::string start = r->query_get("start");
if (start.empty()) { if (start.empty()) {
@ -418,7 +419,7 @@ int SrsHttpFileServer::serve_flv_file(ISrsHttpResponseWriter* w, ISrsHttpMessage
return serve_flv_stream(w, r, fullpath, offset); return serve_flv_stream(w, r, fullpath, offset);
} }
int SrsHttpFileServer::serve_mp4_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath) srs_error_t SrsHttpFileServer::serve_mp4_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath)
{ {
// for flash to request mp4 range in query string. // for flash to request mp4 range in query string.
// for example, http://digitalprimates.net/dash/DashTest.html?url=http://dashdemo.edgesuite.net/digitalprimates/nexus/oops-20120802-manifest.mpd // for example, http://digitalprimates.net/dash/DashTest.html?url=http://dashdemo.edgesuite.net/digitalprimates/nexus/oops-20120802-manifest.mpd
@ -455,17 +456,17 @@ int SrsHttpFileServer::serve_mp4_file(ISrsHttpResponseWriter* w, ISrsHttpMessage
return serve_mp4_stream(w, r, fullpath, start, end); return serve_mp4_stream(w, r, fullpath, start, end);
} }
int SrsHttpFileServer::serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath, int offset) srs_error_t SrsHttpFileServer::serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath, int offset)
{ {
return serve_file(w, r, fullpath); return serve_file(w, r, fullpath);
} }
int SrsHttpFileServer::serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath, int start, int end) srs_error_t SrsHttpFileServer::serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath, int start, int end)
{ {
return serve_file(w, r, fullpath); return serve_file(w, r, fullpath);
} }
int SrsHttpFileServer::copy(ISrsHttpResponseWriter* w, SrsFileReader* fs, ISrsHttpMessage* r, int size) srs_error_t SrsHttpFileServer::copy(ISrsHttpResponseWriter* w, SrsFileReader* fs, ISrsHttpMessage* r, int size)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -485,7 +486,11 @@ int SrsHttpFileServer::copy(ISrsHttpResponseWriter* w, SrsFileReader* fs, ISrsHt
} }
} }
return ret; if (ret != ERROR_SUCCESS) {
return srs_error_new(ret, "copy");
}
return srs_success;
} }
SrsHttpMuxEntry::SrsHttpMuxEntry() SrsHttpMuxEntry::SrsHttpMuxEntry()
@ -629,54 +634,34 @@ srs_error_t SrsHttpServeMux::handle(std::string pattern, ISrsHttpHandler* handle
return srs_success; return srs_success;
} }
bool SrsHttpServeMux::can_serve(ISrsHttpMessage* r) srs_error_t SrsHttpServeMux::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
ISrsHttpHandler* h = NULL; ISrsHttpHandler* h = NULL;
if ((ret = find_handler(r, &h)) != ERROR_SUCCESS) { if ((err = find_handler(r, &h)) != srs_success) {
return false; return srs_error_wrap(err, "find handler");
} }
srs_assert(h); srs_assert(h);
return !h->is_not_found(); if ((err = h->serve_http(w, r)) != srs_success) {
} return srs_error_wrap(err, "serve http");
int SrsHttpServeMux::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{
int ret = ERROR_SUCCESS;
ISrsHttpHandler* h = NULL;
if ((ret = find_handler(r, &h)) != ERROR_SUCCESS) {
srs_error("find handler failed. ret=%d", ret);
return ret;
} }
srs_assert(h); return err;
if ((ret = h->serve_http(w, r)) != ERROR_SUCCESS) {
if (!srs_is_client_gracefully_close(ret)) {
srs_error("handler serve http failed. ret=%d", ret);
}
return ret;
}
return ret;
} }
int SrsHttpServeMux::find_handler(ISrsHttpMessage* r, ISrsHttpHandler** ph) srs_error_t SrsHttpServeMux::find_handler(ISrsHttpMessage* r, ISrsHttpHandler** ph)
{ {
int ret = ERROR_SUCCESS; srs_error_t err = srs_success;
// TODO: FIXME: support the path . and .. // TODO: FIXME: support the path . and ..
if (r->url().find("..") != std::string::npos) { if (r->url().find("..") != std::string::npos) {
ret = ERROR_HTTP_URL_NOT_CLEAN; return srs_error_new(ERROR_HTTP_URL_NOT_CLEAN, "url %s not canonical", r->url().c_str());
srs_error("htt url not canonical, url=%s. ret=%d", r->url().c_str(), ret);
return ret;
} }
if ((ret = match(r, ph)) != ERROR_SUCCESS) { if ((err = match(r, ph)) != srs_success) {
srs_error("http match handler failed. ret=%d", ret); return srs_error_wrap(err, "http match");
return ret;
} }
// always hijack. // always hijack.
@ -685,9 +670,8 @@ int SrsHttpServeMux::find_handler(ISrsHttpMessage* r, ISrsHttpHandler** ph)
std::vector<ISrsHttpMatchHijacker*>::iterator it; std::vector<ISrsHttpMatchHijacker*>::iterator it;
for (it = hijackers.begin(); it != hijackers.end(); ++it) { for (it = hijackers.begin(); it != hijackers.end(); ++it) {
ISrsHttpMatchHijacker* hijacker = *it; ISrsHttpMatchHijacker* hijacker = *it;
if ((ret = hijacker->hijack(r, ph)) != ERROR_SUCCESS) { if ((err = hijacker->hijack(r, ph)) != srs_success) {
srs_error("hijacker match failed. ret=%d", ret); return srs_error_wrap(err, "http hijack");
return ret;
} }
} }
} }
@ -697,13 +681,11 @@ int SrsHttpServeMux::find_handler(ISrsHttpMessage* r, ISrsHttpHandler** ph)
*ph = h404; *ph = h404;
} }
return ret; return err;
} }
int SrsHttpServeMux::match(ISrsHttpMessage* r, ISrsHttpHandler** ph) srs_error_t SrsHttpServeMux::match(ISrsHttpMessage* r, ISrsHttpHandler** ph)
{ {
int ret = ERROR_SUCCESS;
std::string path = r->path(); std::string path = r->path();
// Host-specific pattern takes precedence over generic ones // Host-specific pattern takes precedence over generic ones
@ -735,7 +717,7 @@ int SrsHttpServeMux::match(ISrsHttpMessage* r, ISrsHttpHandler** ph)
*ph = h; *ph = h;
return ret; return srs_success;
} }
bool SrsHttpServeMux::path_match(string pattern, string path) bool SrsHttpServeMux::path_match(string pattern, string path)
@ -773,16 +755,18 @@ SrsHttpCorsMux::~SrsHttpCorsMux()
{ {
} }
int SrsHttpCorsMux::initialize(ISrsHttpServeMux* worker, bool cros_enabled) srs_error_t SrsHttpCorsMux::initialize(ISrsHttpServeMux* worker, bool cros_enabled)
{ {
next = worker; next = worker;
enabled = cros_enabled; enabled = cros_enabled;
return ERROR_SUCCESS; return srs_success;
} }
int SrsHttpCorsMux::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error_t SrsHttpCorsMux::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{ {
int ret = ERROR_SUCCESS;
// If CORS enabled, and there is a "Origin" header, it's CORS. // If CORS enabled, and there is a "Origin" header, it's CORS.
if (enabled) { if (enabled) {
for (int i = 0; i < r->request_header_count(); i++) { for (int i = 0; i < r->request_header_count(); i++) {
@ -811,7 +795,9 @@ int SrsHttpCorsMux::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
} else { } else {
w->write_header(SRS_CONSTS_HTTP_MethodNotAllowed); w->write_header(SRS_CONSTS_HTTP_MethodNotAllowed);
} }
return w->final_request(); if ((ret = w->final_request()) != ERROR_SUCCESS) {
return srs_error_new(ret, "final request");
}
} }
srs_assert(next); srs_assert(next);

42
trunk/src/protocol/srs_http_stack.hpp
View file

@ -77,8 +77,8 @@ class ISrsHttpResponseWriter;
// Error replies to the request with the specified error message and HTTP code. // Error replies to the request with the specified error message and HTTP code.
// The error message should be plain text. // The error message should be plain text.
extern int srs_go_http_error(ISrsHttpResponseWriter* w, int code); extern srs_error_t srs_go_http_error(ISrsHttpResponseWriter* w, int code);
extern int srs_go_http_error(ISrsHttpResponseWriter* w, int code, std::string error); extern srs_error_t srs_go_http_error(ISrsHttpResponseWriter* w, int code, std::string error);
// get the status text of code. // get the status text of code.
extern std::string srs_generate_http_status_text(int status); extern std::string srs_generate_http_status_text(int status);
@ -255,7 +255,7 @@ public:
virtual ~ISrsHttpHandler(); virtual ~ISrsHttpHandler();
public: public:
virtual bool is_not_found(); virtual bool is_not_found();
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) = 0; virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) = 0;
}; };
// Redirect to a fixed URL // Redirect to a fixed URL
@ -268,7 +268,7 @@ public:
SrsHttpRedirectHandler(std::string u, int c); SrsHttpRedirectHandler(std::string u, int c);
virtual ~SrsHttpRedirectHandler(); virtual ~SrsHttpRedirectHandler();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
// NotFound replies to the request with an HTTP 404 not found error. // NotFound replies to the request with an HTTP 404 not found error.
@ -279,7 +279,7 @@ public:
virtual ~SrsHttpNotFoundHandler(); virtual ~SrsHttpNotFoundHandler();
public: public:
virtual bool is_not_found(); virtual bool is_not_found();
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
// FileServer returns a handler that serves HTTP requests // FileServer returns a handler that serves HTTP requests
@ -298,31 +298,31 @@ public:
SrsHttpFileServer(std::string root_dir); SrsHttpFileServer(std::string root_dir);
virtual ~SrsHttpFileServer(); virtual ~SrsHttpFileServer();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
private: private:
/** /**
* serve the file by specified path * serve the file by specified path
*/ */
virtual int serve_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath); virtual srs_error_t serve_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
virtual int serve_flv_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath); virtual srs_error_t serve_flv_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
virtual int serve_mp4_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath); virtual srs_error_t serve_mp4_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
protected: protected:
/** /**
* when access flv file with x.flv?start=xxx * when access flv file with x.flv?start=xxx
*/ */
virtual int serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int offset); virtual srs_error_t serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int offset);
/** /**
* when access mp4 file with x.mp4?range=start-end * when access mp4 file with x.mp4?range=start-end
* @param start the start offset in bytes. * @param start the start offset in bytes.
* @param end the end offset in bytes. -1 to end of file. * @param end the end offset in bytes. -1 to end of file.
* @remark response data in [start, end]. * @remark response data in [start, end].
*/ */
virtual int serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int start, int end); virtual srs_error_t serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int start, int end);
protected: protected:
/** /**
* copy the fs to response writer in size bytes. * copy the fs to response writer in size bytes.
*/ */
virtual int copy(ISrsHttpResponseWriter* w, SrsFileReader* fs, ISrsHttpMessage* r, int size); virtual srs_error_t copy(ISrsHttpResponseWriter* w, SrsFileReader* fs, ISrsHttpMessage* r, int size);
}; };
// the mux entry for server mux. // the mux entry for server mux.
@ -353,7 +353,7 @@ public:
* @param request the http request message to match the handler. * @param request the http request message to match the handler.
* @param ph the already matched handler, hijack can rewrite it. * @param ph the already matched handler, hijack can rewrite it.
*/ */
virtual int hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph) = 0; virtual srs_error_t hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph) = 0;
}; };
/** /**
@ -365,7 +365,7 @@ public:
ISrsHttpServeMux(); ISrsHttpServeMux();
virtual ~ISrsHttpServeMux(); virtual ~ISrsHttpServeMux();
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) = 0; virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) = 0;
}; };
// ServeMux is an HTTP request multiplexer. // ServeMux is an HTTP request multiplexer.
@ -427,15 +427,13 @@ public:
// Handle registers the handler for the given pattern. // Handle registers the handler for the given pattern.
// If a handler already exists for pattern, Handle panics. // If a handler already exists for pattern, Handle panics.
virtual srs_error_t handle(std::string pattern, ISrsHttpHandler* handler); virtual srs_error_t handle(std::string pattern, ISrsHttpHandler* handler);
// whether the http muxer can serve the specified message,
// if not, user can try next muxer.
virtual bool can_serve(ISrsHttpMessage* r);
// interface ISrsHttpServeMux // interface ISrsHttpServeMux
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
public:
virtual srs_error_t find_handler(ISrsHttpMessage* r, ISrsHttpHandler** ph);
private: private:
virtual int find_handler(ISrsHttpMessage* r, ISrsHttpHandler** ph); virtual srs_error_t match(ISrsHttpMessage* r, ISrsHttpHandler** ph);
virtual int match(ISrsHttpMessage* r, ISrsHttpHandler** ph);
virtual bool path_match(std::string pattern, std::string path); virtual bool path_match(std::string pattern, std::string path);
}; };
@ -453,10 +451,10 @@ public:
SrsHttpCorsMux(); SrsHttpCorsMux();
virtual ~SrsHttpCorsMux(); virtual ~SrsHttpCorsMux();
public: public:
virtual int initialize(ISrsHttpServeMux* worker, bool cros_enabled); virtual srs_error_t initialize(ISrsHttpServeMux* worker, bool cros_enabled);
// interface ISrsHttpServeMux // interface ISrsHttpServeMux
public: public:
virtual int serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
// for http header. // for http header.