From 94641c812b2ac230cc5da093641b69c9dc2b877c Mon Sep 17 00:00:00 2001 From: winlin Date: Sat, 22 Aug 2015 13:36:15 +0800 Subject: [PATCH] fix #442, support kickoff client. --- trunk/src/app/srs_app_conn.cpp | 6 ++++++ trunk/src/app/srs_app_conn.hpp | 9 +++++++++ trunk/src/app/srs_app_http_api.cpp | 29 ++++++++++----------------- trunk/src/app/srs_app_rtmp_conn.cpp | 13 +++++++++--- trunk/src/app/srs_app_source.cpp | 12 ----------- trunk/src/app/srs_app_source.hpp | 8 -------- trunk/src/app/srs_app_statistic.cpp | 3 ++- trunk/src/app/srs_app_statistic.hpp | 4 +++- trunk/src/kernel/srs_kernel_error.hpp | 5 +++++ trunk/src/protocol/srs_http_stack.cpp | 9 ++++++--- trunk/src/protocol/srs_http_stack.hpp | 5 +++++ 11 files changed, 57 insertions(+), 46 deletions(-) diff --git a/trunk/src/app/srs_app_conn.cpp b/trunk/src/app/srs_app_conn.cpp index 8cedcfec3..3aa88fbdb 100644 --- a/trunk/src/app/srs_app_conn.cpp +++ b/trunk/src/app/srs_app_conn.cpp @@ -41,6 +41,7 @@ SrsConnection::SrsConnection(IConnectionManager* cm, st_netfd_t c) manager = cm; stfd = c; disposed = false; + expired = false; // the client thread should reap itself, // so we never use joinable. @@ -116,4 +117,9 @@ int SrsConnection::srs_id() return id; } +void SrsConnection::expire() +{ + expired = true; +} + diff --git a/trunk/src/app/srs_app_conn.hpp b/trunk/src/app/srs_app_conn.hpp index d5130bf71..ae48421ba 100644 --- a/trunk/src/app/srs_app_conn.hpp +++ b/trunk/src/app/srs_app_conn.hpp @@ -88,6 +88,11 @@ protected: * when disposed, connection should stop cycle and cleanup itself. */ bool disposed; + /** + * whether connection is expired, application definition. + * when expired, the connection must never be served and quit ASAP. + */ + bool expired; public: SrsConnection(IConnectionManager* cm, st_netfd_t c); virtual ~SrsConnection(); @@ -125,6 +130,10 @@ public: * get the srs id which identify the client. */ virtual int srs_id(); + /** + * set connection to expired. + */ + virtual void expire(); protected: /** * for concrete connection to do the cycle. diff --git a/trunk/src/app/srs_app_http_api.cpp b/trunk/src/app/srs_app_http_api.cpp index fd9eb5449..46e862ab9 100755 --- a/trunk/src/app/srs_app_http_api.cpp +++ b/trunk/src/app/srs_app_http_api.cpp @@ -660,22 +660,8 @@ int SrsGoApiStreams::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) srs_error("stream stream_id=%d not found. ret=%d", sid, ret); return srs_http_response_code(w, ret); } - - if (r->is_http_delete()) { - srs_assert(stream); - - SrsSource* source = SrsSource::fetch(stream->vhost->vhost, stream->app, stream->stream); - if (!source) { - ret = ERROR_SOURCE_NOT_FOUND; - srs_warn("source not found for sid=%d", sid); - return srs_http_response_code(w, ret); - } - - source->set_expired(); - srs_warn("disconnent stream=%d successfully. vhost=%s, app=%s, stream=%s.", - sid, stream->vhost->vhost.c_str(), stream->app.c_str(), stream->stream.c_str()); - return srs_http_response_code(w, ret); - } else if (r->is_http_get()) { + + if (r->is_http_get()) { std::stringstream data; if (!stream) { @@ -726,10 +712,15 @@ int SrsGoApiClients::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) ret = ERROR_RTMP_STREAM_NOT_FOUND; srs_error("stream client_id=%d not found. ret=%d", cid, ret); return srs_http_response_code(w, ret); - } - if (r->is_http_get()) { + if (r->is_http_delete()) { + srs_assert(client); + + client->conn->expire(); + srs_warn("delete client=%d ok", cid); + return srs_http_response_code(w, ret); + } else if (r->is_http_get()) { std::stringstream data; if (!client) { @@ -751,6 +742,8 @@ int SrsGoApiClients::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) } return srs_http_response_json(w, ss.str()); + } else { + return srs_go_http_error(w, SRS_CONSTS_HTTP_MethodNotAllowed); } return ret; diff --git a/trunk/src/app/srs_app_rtmp_conn.cpp b/trunk/src/app/srs_app_rtmp_conn.cpp index e4651eecd..08423517e 100755 --- a/trunk/src/app/srs_app_rtmp_conn.cpp +++ b/trunk/src/app/srs_app_rtmp_conn.cpp @@ -495,7 +495,7 @@ int SrsRtmpConn::stream_service_cycle() // update the statistic when source disconveried. SrsStatistic* stat = SrsStatistic::instance(); - if ((ret = stat->on_client(_srs_context->get_id(), req)) != ERROR_SUCCESS) { + if ((ret = stat->on_client(_srs_context->get_id(), req, this)) != ERROR_SUCCESS) { srs_error("stat client failed. ret=%d", ret); return ret; } @@ -671,6 +671,13 @@ int SrsRtmpConn::do_playing(SrsSource* source, SrsConsumer* consumer, SrsQueueRe while (!disposed) { // collect elapse for pithy print. pprint->elapse(); + + // when source is set to expired, disconnect it. + if (expired) { + ret = ERROR_USER_DISCONNECT; + srs_error("connection expired. ret=%d", ret); + return ret; + } // to use isolate thread to recv, can improve about 33% performance. // @see: https://github.com/simple-rtmp-server/srs/issues/196 @@ -875,9 +882,9 @@ int SrsRtmpConn::do_publishing(SrsSource* source, SrsPublishRecvThread* trd) pprint->elapse(); // when source is set to expired, disconnect it. - if (source->expired()) { + if (expired) { ret = ERROR_USER_DISCONNECT; - srs_error("source is expired. ret=%d", ret); + srs_error("connection expired. ret=%d", ret); return ret; } diff --git a/trunk/src/app/srs_app_source.cpp b/trunk/src/app/srs_app_source.cpp index b86a04490..ec00978cc 100755 --- a/trunk/src/app/srs_app_source.cpp +++ b/trunk/src/app/srs_app_source.cpp @@ -898,7 +898,6 @@ SrsSource::SrsSource() jitter_algorithm = SrsRtmpJitterAlgorithmOFF; mix_correct = false; mix_queue = new SrsMixQueue(); - is_expired = false; #ifdef SRS_AUTO_HLS hls = new SrsHls(); @@ -2101,7 +2100,6 @@ void SrsSource::on_unpublish() _can_publish = true; _source_id = -1; - is_expired = false; // notify the handler. srs_assert(handler); @@ -2260,13 +2258,3 @@ void SrsSource::destroy_forwarders() forwarders.clear(); } -bool SrsSource::expired() -{ - return is_expired; -} - -void SrsSource::set_expired() -{ - is_expired = true; -} - diff --git a/trunk/src/app/srs_app_source.hpp b/trunk/src/app/srs_app_source.hpp index 18cdab86c..7c626b7eb 100755 --- a/trunk/src/app/srs_app_source.hpp +++ b/trunk/src/app/srs_app_source.hpp @@ -453,8 +453,6 @@ private: // whether use interlaced/mixed algorithm to correct timestamp. bool mix_correct; SrsMixQueue* mix_queue; - // the flag of source expired or not. - bool is_expired; // whether stream is monotonically increase. bool is_monotonically_increase; int64_t last_packet_time; @@ -586,12 +584,6 @@ public: private: virtual int create_forwarders(); virtual void destroy_forwarders(); -public: - virtual bool expired(); - /** - * set source expired. - */ - virtual void set_expired(); }; #endif diff --git a/trunk/src/app/srs_app_statistic.cpp b/trunk/src/app/srs_app_statistic.cpp index 386404680..dd6983a7e 100755 --- a/trunk/src/app/srs_app_statistic.cpp +++ b/trunk/src/app/srs_app_statistic.cpp @@ -322,7 +322,7 @@ void SrsStatistic::on_stream_close(SrsRequest* req) stream->close(); } -int SrsStatistic::on_client(int id, SrsRequest* req) +int SrsStatistic::on_client(int id, SrsRequest* req, SrsConnection* conn) { int ret = ERROR_SUCCESS; @@ -341,6 +341,7 @@ int SrsStatistic::on_client(int id, SrsRequest* req) } // got client. + client->conn = conn; stream->nb_clients++; vhost->nb_clients++; diff --git a/trunk/src/app/srs_app_statistic.hpp b/trunk/src/app/srs_app_statistic.hpp index 27ea06daa..3abfc561e 100755 --- a/trunk/src/app/srs_app_statistic.hpp +++ b/trunk/src/app/srs_app_statistic.hpp @@ -112,6 +112,7 @@ struct SrsStatisticClient { public: SrsStatisticStream* stream; + SrsConnection* conn; int id; public: SrsStatisticClient(); @@ -181,8 +182,9 @@ public: * when got a client to publish/play stream, * @param id, the client srs id. * @param req, the client request object. + * @param conn, the physical absract connection object. */ - virtual int on_client(int id, SrsRequest* req); + virtual int on_client(int id, SrsRequest* req, SrsConnection* conn); /** * client disconnect * @remark the on_disconnect always call, while the on_client is call when diff --git a/trunk/src/kernel/srs_kernel_error.hpp b/trunk/src/kernel/srs_kernel_error.hpp index 47c68c9b1..bb7fddcef 100755 --- a/trunk/src/kernel/srs_kernel_error.hpp +++ b/trunk/src/kernel/srs_kernel_error.hpp @@ -263,6 +263,11 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. #define ERROR_AAC_BYTES_INVALID 4028 #define ERROR_HTTP_REQUEST_EOF 4029 +/////////////////////////////////////////////////////// +// HTTP API error. +/////////////////////////////////////////////////////// +//#define ERROR_API_METHOD_NOT_ALLOWD + /////////////////////////////////////////////////////// // user-define error. /////////////////////////////////////////////////////// diff --git a/trunk/src/protocol/srs_http_stack.cpp b/trunk/src/protocol/srs_http_stack.cpp index 75c55e5e3..8f767279b 100755 --- a/trunk/src/protocol/srs_http_stack.cpp +++ b/trunk/src/protocol/srs_http_stack.cpp @@ -120,8 +120,11 @@ string srs_go_http_detect(char* data, int size) return "application/octet-stream"; // fallback } -// Error replies to the request with the specified error message and HTTP code. -// The error message should be plain text. +int srs_go_http_error(ISrsHttpResponseWriter* w, int 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) { int ret = ERROR_SUCCESS; @@ -287,7 +290,7 @@ bool SrsHttpNotFoundHandler::is_not_found() int SrsHttpNotFoundHandler::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) { - return srs_go_http_error(w, SRS_CONSTS_HTTP_NotFound, SRS_CONSTS_HTTP_NotFound_str); + return srs_go_http_error(w, SRS_CONSTS_HTTP_NotFound); } SrsHttpFileServer::SrsHttpFileServer(string root_dir) diff --git a/trunk/src/protocol/srs_http_stack.hpp b/trunk/src/protocol/srs_http_stack.hpp index 2888ce186..4e4fed7a2 100644 --- a/trunk/src/protocol/srs_http_stack.hpp +++ b/trunk/src/protocol/srs_http_stack.hpp @@ -76,6 +76,11 @@ class ISrsHttpResponseWriter; #define SRS_CONSTS_HTTP_PUT HTTP_PUT #define SRS_CONSTS_HTTP_DELETE HTTP_DELETE +// Error replies to the request with the specified error message and HTTP code. +// The error message should be plain text. +extern int srs_go_http_error(ISrsHttpResponseWriter* w, int code); +extern int srs_go_http_error(ISrsHttpResponseWriter* w, int code, std::string error); + // helper function: response in json format. extern int srs_http_response_json(ISrsHttpResponseWriter* w, std::string data); /**