refine code, the consumer always alive longer than queue recv thread.

@@ -134,9 +134,10 @@ void SrsRecvThread::on_thread_stop()
     handler->on_thread_stop();
     handler->on_thread_stop();
 }
 }
 
 
-SrsQueueRecvThread::SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk, int timeout_ms)
+SrsQueueRecvThread::SrsQueueRecvThread(SrsConsumer* consumer, SrsRtmpServer* rtmp_sdk, int timeout_ms)
     : trd(this, rtmp_sdk, timeout_ms)
     : trd(this, rtmp_sdk, timeout_ms)
 {
 {
+    _consumer = consumer;
     rtmp = rtmp_sdk;
     rtmp = rtmp_sdk;
     recv_error_code = ERROR_SUCCESS;
     recv_error_code = ERROR_SUCCESS;
     _consumer = NULL;
     _consumer = NULL;
@@ -237,11 +238,6 @@ 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, 

@@ -113,9 +113,9 @@ 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;
+    SrsConsumer* _consumer;
 public:
 public:
-    SrsQueueRecvThread(SrsRtmpServer* rtmp_sdk, int timeout_ms);
+    SrsQueueRecvThread(SrsConsumer* consumer, SrsRtmpServer* rtmp_sdk, int timeout_ms);
     virtual ~SrsQueueRecvThread();
     virtual ~SrsQueueRecvThread();
 public:
 public:
     virtual int start();
     virtual int start();
@@ -132,8 +132,6 @@ 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);
 };
 };
 
 
 /**
 /**

@@ -560,9 +560,18 @@ int SrsRtmpConn::playing(SrsSource* source)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
+    // create consumer of souce.
+    SrsConsumer* consumer = NULL;
+    if ((ret = source->create_consumer(consumer)) != ERROR_SUCCESS) {
+        srs_error("create consumer failed. ret=%d", ret);
+        return ret;
+    }
+    SrsAutoFree(SrsConsumer, consumer);
+    srs_verbose("consumer created success.");
+    
     // use isolate thread to recv, 
     // use isolate thread to recv, 
     // @see: https://github.com/winlinvip/simple-rtmp-server/issues/217
     // @see: https://github.com/winlinvip/simple-rtmp-server/issues/217
-    SrsQueueRecvThread trd(rtmp, SRS_PERF_MW_SLEEP);
+    SrsQueueRecvThread trd(consumer, rtmp, SRS_PERF_MW_SLEEP);
     
     
     // start isolate recv thread.
     // start isolate recv thread.
     if ((ret = trd.start()) != ERROR_SUCCESS) {
     if ((ret = trd.start()) != ERROR_SUCCESS) {
@@ -571,7 +580,7 @@ int SrsRtmpConn::playing(SrsSource* source)
     }
     }
     
     
     // delivery messages for clients playing stream.
     // delivery messages for clients playing stream.
-    ret = do_playing(source, &trd);
+    ret = do_playing(source, consumer, &trd);
     
     
     // stop isolate recv thread
     // stop isolate recv thread
     trd.stop();
     trd.stop();
@@ -584,27 +593,18 @@ int SrsRtmpConn::playing(SrsSource* source)
     return ret;
     return ret;
 }
 }
 
 
-int SrsRtmpConn::do_playing(SrsSource* source, SrsQueueRecvThread* trd)
+int SrsRtmpConn::do_playing(SrsSource* source, SrsConsumer* consumer, SrsQueueRecvThread* trd)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
+    srs_assert(consumer != NULL);
+    
     if ((ret = refer->check(req->pageUrl, _srs_config->get_refer_play(req->vhost))) != ERROR_SUCCESS) {
     if ((ret = refer->check(req->pageUrl, _srs_config->get_refer_play(req->vhost))) != ERROR_SUCCESS) {
         srs_error("check play_refer failed. ret=%d", ret);
         srs_error("check play_refer failed. ret=%d", ret);
         return ret;
         return ret;
     }
     }
     srs_verbose("check play_refer success.");
     srs_verbose("check play_refer success.");
     
     
-    SrsConsumer* consumer = NULL;
-    if ((ret = source->create_consumer(consumer)) != ERROR_SUCCESS) {
-        srs_error("create consumer failed. ret=%d", ret);
-        return ret;
-    }
-    
-    srs_assert(consumer != NULL);
-    SrsAutoFree(SrsConsumer, consumer);
-    trd->set_consumer(consumer);
-    srs_verbose("consumer created success.");
-    
     // initialize other components
     // initialize other components
     SrsPithyPrint pithy_print(SRS_CONSTS_STAGE_PLAY_USER);
     SrsPithyPrint pithy_print(SRS_CONSTS_STAGE_PLAY_USER);
     SrsMessageArray msgs(SRS_PERF_MW_MSGS);
     SrsMessageArray msgs(SRS_PERF_MW_MSGS);

@@ -103,7 +103,7 @@ private:
     virtual int stream_service_cycle();
     virtual int stream_service_cycle();
     virtual int check_vhost();
     virtual int check_vhost();
     virtual int playing(SrsSource* source);
     virtual int playing(SrsSource* source);
-    virtual int do_playing(SrsSource* source, SrsQueueRecvThread* trd);
+    virtual int do_playing(SrsSource* source, SrsConsumer* consumer, SrsQueueRecvThread* trd);
     virtual int fmle_publishing(SrsSource* source);
     virtual int fmle_publishing(SrsSource* source);
     virtual int flash_publishing(SrsSource* source);
     virtual int flash_publishing(SrsSource* source);
     virtual int do_publishing(SrsSource* source, SrsPublishRecvThread* trd);
     virtual int do_publishing(SrsSource* source, SrsPublishRecvThread* trd);