fix #442, support kickoff client.

@@ -41,6 +41,7 @@ SrsConnection::SrsConnection(IConnectionManager* cm, st_netfd_t c)
     manager = cm;
     manager = cm;
     stfd = c;
     stfd = c;
     disposed = false;
     disposed = false;
+    expired = false;
     
     
     // the client thread should reap itself, 
     // the client thread should reap itself, 
     // so we never use joinable.
     // so we never use joinable.
@@ -116,4 +117,9 @@ int SrsConnection::srs_id()
     return id;
     return id;
 }
 }
 
 
+void SrsConnection::expire()
+{
+    expired = true;
+}
+
 
 

@@ -88,6 +88,11 @@ protected:
      * when disposed, connection should stop cycle and cleanup itself.
      * when disposed, connection should stop cycle and cleanup itself.
      */
      */
     bool disposed;
     bool disposed;
+    /**
+     * whether connection is expired, application definition.
+     * when expired, the connection must never be served and quit ASAP.
+     */
+    bool expired;
 public:
 public:
     SrsConnection(IConnectionManager* cm, st_netfd_t c);
     SrsConnection(IConnectionManager* cm, st_netfd_t c);
     virtual ~SrsConnection();
     virtual ~SrsConnection();
@@ -125,6 +130,10 @@ public:
     * get the srs id which identify the client.
     * get the srs id which identify the client.
     */
     */
     virtual int srs_id();
     virtual int srs_id();
+    /**
+     * set connection to expired.
+     */
+    virtual void expire();
 protected:
 protected:
     /**
     /**
     * for concrete connection to do the cycle.
     * for concrete connection to do the cycle.

@@ -660,22 +660,8 @@ int SrsGoApiStreams::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
         srs_error("stream stream_id=%d not found. ret=%d", sid, ret);
         srs_error("stream stream_id=%d not found. ret=%d", sid, ret);
         return srs_http_response_code(w, 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;
         std::stringstream data;
         
         
         if (!stream) {
         if (!stream) {
@@ -726,10 +712,15 @@ int SrsGoApiClients::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
         ret = ERROR_RTMP_STREAM_NOT_FOUND;
         ret = ERROR_RTMP_STREAM_NOT_FOUND;
         srs_error("stream client_id=%d not found. ret=%d", cid, ret);
         srs_error("stream client_id=%d not found. ret=%d", cid, ret);
         return srs_http_response_code(w, 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;
         std::stringstream data;
         
         
         if (!client) {
         if (!client) {
@@ -751,6 +742,8 @@ int SrsGoApiClients::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
         }
         }
         
         
         return srs_http_response_json(w, ss.str());
         return srs_http_response_json(w, ss.str());
+    } else {
+        return srs_go_http_error(w, SRS_CONSTS_HTTP_MethodNotAllowed);
     }
     }
 
 
     return ret;
     return ret;

@@ -495,7 +495,7 @@ int SrsRtmpConn::stream_service_cycle()
     
     
     // 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)) != ERROR_SUCCESS) {
+    if ((ret = stat->on_client(_srs_context->get_id(), req, this)) != ERROR_SUCCESS) {
         srs_error("stat client failed. ret=%d", ret);
         srs_error("stat client failed. ret=%d", ret);
         return ret;
         return ret;
     }
     }
@@ -671,6 +671,13 @@ int SrsRtmpConn::do_playing(SrsSource* source, SrsConsumer* consumer, SrsQueueRe
     while (!disposed) {
     while (!disposed) {
         // collect elapse for pithy print.
         // collect elapse for pithy print.
         pprint->elapse();
         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.
         // to use isolate thread to recv, can improve about 33% performance.
         // @see: https://github.com/simple-rtmp-server/srs/issues/196
         // @see: https://github.com/simple-rtmp-server/srs/issues/196
@@ -875,9 +882,9 @@ int SrsRtmpConn::do_publishing(SrsSource* source, SrsPublishRecvThread* trd)
         pprint->elapse();
         pprint->elapse();
         
         
         // when source is set to expired, disconnect it.
         // when source is set to expired, disconnect it.
-        if (source->expired()) {
+        if (expired) {
             ret = ERROR_USER_DISCONNECT;
             ret = ERROR_USER_DISCONNECT;
-            srs_error("source is expired. ret=%d", ret);
+            srs_error("connection expired. ret=%d", ret);
             return ret;
             return ret;
         }
         }
 
 

@@ -898,7 +898,6 @@ SrsSource::SrsSource()
     jitter_algorithm = SrsRtmpJitterAlgorithmOFF;
     jitter_algorithm = SrsRtmpJitterAlgorithmOFF;
     mix_correct = false;
     mix_correct = false;
     mix_queue = new SrsMixQueue();
     mix_queue = new SrsMixQueue();
-    is_expired = false;
     
     
 #ifdef SRS_AUTO_HLS
 #ifdef SRS_AUTO_HLS
     hls = new SrsHls();
     hls = new SrsHls();
@@ -2101,7 +2100,6 @@ void SrsSource::on_unpublish()
     
     
     _can_publish = true;
     _can_publish = true;
     _source_id = -1;
     _source_id = -1;
-    is_expired = false;
 
 
     // notify the handler.
     // notify the handler.
     srs_assert(handler);
     srs_assert(handler);
@@ -2260,13 +2258,3 @@ void SrsSource::destroy_forwarders()
     forwarders.clear();
     forwarders.clear();
 }
 }
 
 
-bool SrsSource::expired()
-{
-    return is_expired;
-}
-
-void SrsSource::set_expired()
-{
-    is_expired = true;
-}
-

@@ -453,8 +453,6 @@ private:
     // whether use interlaced/mixed algorithm to correct timestamp.
     // whether use interlaced/mixed algorithm to correct timestamp.
     bool mix_correct;
     bool mix_correct;
     SrsMixQueue* mix_queue;
     SrsMixQueue* mix_queue;
-    // the flag of source expired or not.
-    bool is_expired;
     // whether stream is monotonically increase.
     // whether stream is monotonically increase.
     bool is_monotonically_increase;
     bool is_monotonically_increase;
     int64_t last_packet_time;
     int64_t last_packet_time;
@@ -586,12 +584,6 @@ public:
 private:
 private:
     virtual int create_forwarders();
     virtual int create_forwarders();
     virtual void destroy_forwarders();
     virtual void destroy_forwarders();
-public:
-    virtual bool expired();
-    /**
-    * set source expired.
-    */
-    virtual void set_expired();
 };
 };
 
 
 #endif
 #endif

@@ -322,7 +322,7 @@ void SrsStatistic::on_stream_close(SrsRequest* req)
     stream->close();
     stream->close();
 }
 }
 
 
-int SrsStatistic::on_client(int id, SrsRequest* req)
+int SrsStatistic::on_client(int id, SrsRequest* req, SrsConnection* conn)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
@@ -341,6 +341,7 @@ int SrsStatistic::on_client(int id, SrsRequest* req)
     }
     }
     
     
     // got client.
     // got client.
+    client->conn = conn;
     stream->nb_clients++;
     stream->nb_clients++;
     vhost->nb_clients++;
     vhost->nb_clients++;
 
 

@@ -112,6 +112,7 @@ struct SrsStatisticClient
 {
 {
 public:
 public:
     SrsStatisticStream* stream;
     SrsStatisticStream* stream;
+    SrsConnection* conn;
     int id;
     int id;
 public:
 public:
     SrsStatisticClient();
     SrsStatisticClient();
@@ -181,8 +182,9 @@ public:
      * when got a client to publish/play stream,
      * when got a client to publish/play stream,
      * @param id, the client srs id.
      * @param id, the client srs id.
      * @param req, the client request object.
      * @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
      * client disconnect
      * @remark the on_disconnect always call, while the on_client is call when
      * @remark the on_disconnect always call, while the on_client is call when

@@ -264,6 +264,11 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #define ERROR_HTTP_REQUEST_EOF              4029
 #define ERROR_HTTP_REQUEST_EOF              4029
 
 
 ///////////////////////////////////////////////////////
 ///////////////////////////////////////////////////////
+// HTTP API error.
+///////////////////////////////////////////////////////
+//#define ERROR_API_METHOD_NOT_ALLOWD
+
+///////////////////////////////////////////////////////
 // user-define error.
 // user-define error.
 ///////////////////////////////////////////////////////
 ///////////////////////////////////////////////////////
 #define ERROR_USER_START                    9000
 #define ERROR_USER_START                    9000

@@ -120,8 +120,11 @@ string srs_go_http_detect(char* data, int size)
     return "application/octet-stream"; // fallback
     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 srs_go_http_error(ISrsHttpResponseWriter* w, int code, string error)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
@@ -287,7 +290,7 @@ bool SrsHttpNotFoundHandler::is_not_found()
 
 
 int SrsHttpNotFoundHandler::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
 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)
 SrsHttpFileServer::SrsHttpFileServer(string root_dir)

@@ -76,6 +76,11 @@ class ISrsHttpResponseWriter;
 #define SRS_CONSTS_HTTP_PUT HTTP_PUT
 #define SRS_CONSTS_HTTP_PUT HTTP_PUT
 #define SRS_CONSTS_HTTP_DELETE HTTP_DELETE
 #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.
 // helper function: response in json format.
 extern int srs_http_response_json(ISrsHttpResponseWriter* w, std::string data);
 extern int srs_http_response_json(ISrsHttpResponseWriter* w, std::string data);
 /**
 /**