Merge pull request #259 from zhengfl/master

merge from feilong: 用户连接没有断开

@@ -30,6 +30,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_kernel_utility.hpp>
 #include <srs_kernel_utility.hpp>
 #include <srs_core_performance.hpp>
 #include <srs_core_performance.hpp>
 #include <srs_app_config.hpp>
 #include <srs_app_config.hpp>
+#include <srs_app_source.hpp>
 
 
 using namespace std;
 using namespace std;
 
 
@@ -138,11 +139,12 @@ SrsQueueRecvThread::SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk, int timeout_ms)
 {
 {
     rtmp = rtmp_sdk;
     rtmp = rtmp_sdk;
     recv_error_code = ERROR_SUCCESS;
     recv_error_code = ERROR_SUCCESS;
+    _consumer = NULL;
 }
 }
 
 
 SrsQueueRecvThread::~SrsQueueRecvThread()
 SrsQueueRecvThread::~SrsQueueRecvThread()
 {
 {
-    trd.stop();
+    stop();
 
 
     // clear all messages.
     // clear all messages.
     std::vector<SrsCommonMessage*>::iterator it;
     std::vector<SrsCommonMessage*>::iterator it;
@@ -160,6 +162,7 @@ int SrsQueueRecvThread::start()
 
 
 void SrsQueueRecvThread::stop()
 void SrsQueueRecvThread::stop()
 {
 {
+    _consumer = NULL;
     trd.stop();
     trd.stop();
 }
 }
 
 
@@ -203,13 +206,22 @@ int SrsQueueRecvThread::handle(SrsCommonMessage* msg)
     // put into queue, the send thread will get and process it,
     // put into queue, the send thread will get and process it,
     // @see SrsRtmpConn::process_play_control_msg
     // @see SrsRtmpConn::process_play_control_msg
     queue.push_back(msg);
     queue.push_back(msg);
-
+#ifdef SRS_PERF_QUEUE_COND_WAIT
+    if (_consumer) {
+        _consumer->on_dispose();
+    }
+#endif
     return ERROR_SUCCESS;
     return ERROR_SUCCESS;
 }
 }
 
 
 void SrsQueueRecvThread::on_recv_error(int ret)
 void SrsQueueRecvThread::on_recv_error(int ret)
 {
 {
     recv_error_code = ret;
     recv_error_code = ret;
+#ifdef SRS_PERF_QUEUE_COND_WAIT
+    if (_consumer) {
+        _consumer->on_dispose();
+    }
+#endif
 }
 }
 
 
 void SrsQueueRecvThread::on_thread_start()
 void SrsQueueRecvThread::on_thread_start()
@@ -226,6 +238,11 @@ void SrsQueueRecvThread::on_thread_stop()
     rtmp->set_auto_response(true);
     rtmp->set_auto_response(true);
 }
 }
 
 
+void SrsQueueRecvThread::set_consumer(SrsConsumer *consumer)
+{
+    _consumer = consumer;
+}
+
 SrsPublishRecvThread::SrsPublishRecvThread(
 SrsPublishRecvThread::SrsPublishRecvThread(
     SrsRtmpServer* rtmp_sdk, 
     SrsRtmpServer* rtmp_sdk, 
     SrsRequest* _req, int mr_sock_fd, int timeout_ms, 
     SrsRequest* _req, int mr_sock_fd, int timeout_ms, 

@@ -42,6 +42,7 @@ class SrsCommonMessage;
 class SrsRtmpConn;
 class SrsRtmpConn;
 class SrsSource;
 class SrsSource;
 class SrsRequest;
 class SrsRequest;
+class SrsConsumer;
 
 
 /**
 /**
  * for the recv thread to handle the message.
  * for the recv thread to handle the message.
@@ -112,6 +113,7 @@ private:
     SrsRtmpServer* rtmp;
     SrsRtmpServer* rtmp;
     // the recv thread error code.
     // the recv thread error code.
     int recv_error_code;
     int recv_error_code;
+    SrsConsumer *_consumer;
 public:
 public:
     SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk, int timeout_ms);
     SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk, int timeout_ms);
     virtual ~SrsQueueRecvThread();
     virtual ~SrsQueueRecvThread();
@@ -130,6 +132,8 @@ public:
 public:
 public:
     virtual void on_thread_start();
     virtual void on_thread_start();
     virtual void on_thread_stop();
     virtual void on_thread_stop();
+public:
+    virtual void set_consumer(SrsConsumer *consumer);
 };
 };
 
 
 /**
 /**

@@ -595,6 +595,7 @@ int SrsRtmpConn::do_playing(SrsSource* source, SrsQueueRecvThread* trd)
     // when mw_sleep changed, resize the socket send buffer.
     // when mw_sleep changed, resize the socket send buffer.
     mw_enabled = true;
     mw_enabled = true;
     change_mw_sleep(_srs_config->get_mw_sleep_ms(req->vhost));
     change_mw_sleep(_srs_config->get_mw_sleep_ms(req->vhost));
+    trd->set_consumer(consumer);
     
     
     while (true) {
     while (true) {
         // to use isolate thread to recv, can improve about 33% performance.
         // to use isolate thread to recv, can improve about 33% performance.

@@ -515,6 +515,14 @@ void SrsConsumer::wait(int nb_msgs, int duration)
     // use cond block wait for high performance mode.
     // use cond block wait for high performance mode.
     st_cond_wait(mw_wait);
     st_cond_wait(mw_wait);
 }
 }
+
+void SrsConsumer::on_dispose()
+{
+    if (mw_waiting) {
+        st_cond_signal(mw_wait);
+        mw_waiting = false;
+    }
+}
 #endif
 #endif
 
 
 int SrsConsumer::on_play_client_pause(bool is_pause)
 int SrsConsumer::on_play_client_pause(bool is_pause)

@@ -246,6 +246,10 @@ public:
     * @param duration the messgae duration to wait.
     * @param duration the messgae duration to wait.
     */
     */
     virtual void wait(int nb_msgs, int duration);
     virtual void wait(int nb_msgs, int duration);
+    /**
+    * when waiting, a message incomming, we rouse it
+    */
+    virtual void on_dispose();
 #endif
 #endif
     /**
     /**
     * when client send the pause message.
     * when client send the pause message.