cleanup connections when terminate server.

@@ -40,6 +40,7 @@ SrsConnection::SrsConnection(IConnectionManager* cm, st_netfd_t c)
     id = 0;
     id = 0;
     manager = cm;
     manager = cm;
     stfd = c;
     stfd = c;
+    disposed = false;
     
     
     // the client thread should reap itself, 
     // the client thread should reap itself, 
     // so we never use joinable.
     // so we never use joinable.
@@ -50,12 +51,24 @@ SrsConnection::SrsConnection(IConnectionManager* cm, st_netfd_t c)
 
 
 SrsConnection::~SrsConnection()
 SrsConnection::~SrsConnection()
 {
 {
+    dispose();
+    
+    srs_freep(pthread);
+}
+
+void SrsConnection::dispose()
+{
+    if (disposed) {
+        return;
+    }
+    
+    disposed = true;
+    
     /**
     /**
      * when delete the connection, stop the connection,
      * when delete the connection, stop the connection,
      * close the underlayer socket, delete the thread.
      * close the underlayer socket, delete the thread.
      */
      */
     srs_close_stfd(stfd);
     srs_close_stfd(stfd);
-    srs_freep(pthread);
 }
 }
 
 
 int SrsConnection::start()
 int SrsConnection::start()

@@ -83,11 +83,20 @@ protected:
     * the ip of client.
     * the ip of client.
     */
     */
     std::string ip;
     std::string ip;
+    /**
+     * whether the connection is disposed,
+     * when disposed, connection should stop cycle and cleanup itself.
+     */;
+    bool disposed;
 public:
 public:
     SrsConnection(IConnectionManager* cm, st_netfd_t c);
     SrsConnection(IConnectionManager* cm, st_netfd_t c);
     virtual ~SrsConnection();
     virtual ~SrsConnection();
 public:
 public:
     /**
     /**
+     * to dipose the connection.
+     */
+    virtual void dispose();
+    /**
     * start the client green thread.
     * start the client green thread.
     * when server get a client from listener, 
     * when server get a client from listener, 
     * 1. server will create an concrete connection(for instance, RTMP connection),
     * 1. server will create an concrete connection(for instance, RTMP connection),

@@ -529,7 +529,7 @@ int SrsHttpApi::do_cycle()
     skt.set_recv_timeout(SRS_HTTP_RECV_TIMEOUT_US);
     skt.set_recv_timeout(SRS_HTTP_RECV_TIMEOUT_US);
     
     
     // process http messages.
     // process http messages.
-    for (;;) {
+    while(!disposed) {
         ISrsHttpMessage* req = NULL;
         ISrsHttpMessage* req = NULL;
         
         
         // get a http message
         // get a http message

@@ -2533,7 +2533,7 @@ int SrsHttpConn::do_cycle()
     skt.set_recv_timeout(SRS_HTTP_RECV_TIMEOUT_US);
     skt.set_recv_timeout(SRS_HTTP_RECV_TIMEOUT_US);
     
     
     // process http messages.
     // process http messages.
-    for (;;) {
+    while (!disposed) {
         ISrsHttpMessage* req = NULL;
         ISrsHttpMessage* req = NULL;
         
         
         // get a http message
         // get a http message

@@ -320,7 +320,7 @@ int SrsRtmpConn::service_cycle()
     }
     }
     srs_verbose("on_bw_done success");
     srs_verbose("on_bw_done success");
     
     
-    while (true) {
+    while (!disposed) {
         ret = stream_service_cycle();
         ret = stream_service_cycle();
         
         
         // stream service must terminated with error, never success.
         // stream service must terminated with error, never success.
@@ -361,6 +361,8 @@ int SrsRtmpConn::service_cycle()
         srs_error("control message(%d) reject as error. ret=%d", ret, ret);
         srs_error("control message(%d) reject as error. ret=%d", ret, ret);
         return ret;
         return ret;
     }
     }
+    
+    return ret;
 }
 }
 
 
 int SrsRtmpConn::stream_service_cycle()
 int SrsRtmpConn::stream_service_cycle()
@@ -635,7 +637,7 @@ int SrsRtmpConn::do_playing(SrsSource* source, SrsConsumer* consumer, SrsQueueRe
     // set the sock options.
     // set the sock options.
     play_set_sock_options();
     play_set_sock_options();
     
     
-    while (true) {
+    while (!disposed) {
         // collect elapse for pithy print.
         // collect elapse for pithy print.
         pprint->elapse();
         pprint->elapse();
 
 
@@ -865,7 +867,7 @@ int SrsRtmpConn::do_publishing(SrsSource* source, SrsPublishRecvThread* trd)
     }
     }
 
 
     int64_t nb_msgs = 0;
     int64_t nb_msgs = 0;
-    while (true) {
+    while (!disposed) {
         pprint->elapse();
         pprint->elapse();
 
 
         // cond wait for error.
         // cond wait for error.

@@ -514,17 +514,7 @@ void SrsServer::destroy()
 {
 {
     srs_warn("start destroy server");
     srs_warn("start destroy server");
     
     
-    _srs_config->unsubscribe(this);
-    
-    close_listeners(SrsListenerRtmpStream);
-    close_listeners(SrsListenerHttpApi);
-    close_listeners(SrsListenerHttpStream);
-
-#ifdef SRS_AUTO_INGEST
-    ingester->dispose();
-#endif
-    
-    SrsSource::dispose_all();
+    dispose();
     
     
 #ifdef SRS_AUTO_HTTP_API
 #ifdef SRS_AUTO_HTTP_API
     srs_freep(http_api_mux);
     srs_freep(http_api_mux);
@@ -550,32 +540,35 @@ void SrsServer::destroy()
     srs_freep(signal_manager);
     srs_freep(signal_manager);
     
     
     srs_freep(handler);
     srs_freep(handler);
-    
-    // @remark never destroy the connections, 
-    // for it's still alive.
-
-    // @remark never destroy the source, 
-    // when we free all sources, the fmle publish may retry
-    // and segment fault.
-    
-#ifdef SRS_MEM_WATCH
-    srs_memory_report();
-#endif
 }
 }
 
 
 void SrsServer::dispose()
 void SrsServer::dispose()
 {
 {
     _srs_config->unsubscribe(this);
     _srs_config->unsubscribe(this);
     
     
+    // prevent fresh clients.
+    close_listeners(SrsListenerRtmpStream);
+    close_listeners(SrsListenerHttpApi);
+    close_listeners(SrsListenerHttpStream);
+    close_listeners(SrsListenerMpegTsOverUdp);
+    close_listeners(SrsListenerRtsp);
+    close_listeners(SrsListenerFlv);
+    
 #ifdef SRS_AUTO_INGEST
 #ifdef SRS_AUTO_INGEST
     ingester->dispose();
     ingester->dispose();
-    srs_trace("gracefully dispose ingesters");
 #endif
 #endif
     
     
     SrsSource::dispose_all();
     SrsSource::dispose_all();
-    srs_trace("gracefully dispose sources");
     
     
-    srs_trace("terminate server");
+    while (!conns.empty()) {
+        std::vector<SrsConnection*>::iterator it;
+        for (it = conns.begin(); it != conns.end(); ++it) {
+            SrsConnection* conn = *it;
+            conn->dispose();
+        }
+        
+        st_usleep(100 * 1000);
+    }
     
     
 #ifdef SRS_MEM_WATCH
 #ifdef SRS_MEM_WATCH
     srs_memory_report();
     srs_memory_report();