refine code for consumer to refer the rtmp connection.

@@ -116,7 +116,7 @@ int SrsStreamCache::cycle()
     // the stream cache will create consumer to cache stream,
     // the stream cache will create consumer to cache stream,
     // which will trigger to fetch stream from origin for edge.
     // which will trigger to fetch stream from origin for edge.
     SrsConsumer* consumer = NULL;
     SrsConsumer* consumer = NULL;
-    if ((ret = source->create_consumer(consumer, false, false, true)) != ERROR_SUCCESS) {
+    if ((ret = source->create_consumer(NULL, consumer, false, false, true)) != ERROR_SUCCESS) {
         srs_error("http: create consumer failed. ret=%d", ret);
         srs_error("http: create consumer failed. ret=%d", ret);
         return ret;
         return ret;
     }
     }
@@ -483,7 +483,7 @@ int SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
     
     
     // create consumer of souce, ignore gop cache, use the audio gop cache.
     // create consumer of souce, ignore gop cache, use the audio gop cache.
     SrsConsumer* consumer = NULL;
     SrsConsumer* consumer = NULL;
-    if ((ret = source->create_consumer(consumer, true, true, !enc->has_cache())) != ERROR_SUCCESS) {
+    if ((ret = source->create_consumer(NULL, consumer, true, true, !enc->has_cache())) != ERROR_SUCCESS) {
         srs_error("http: create consumer failed. ret=%d", ret);
         srs_error("http: create consumer failed. ret=%d", ret);
         return ret;
         return ret;
     }
     }

@@ -598,7 +598,7 @@ int SrsRtmpConn::playing(SrsSource* source)
     
     
     // create consumer of souce.
     // create consumer of souce.
     SrsConsumer* consumer = NULL;
     SrsConsumer* consumer = NULL;
-    if ((ret = source->create_consumer(consumer)) != ERROR_SUCCESS) {
+    if ((ret = source->create_consumer(this, consumer)) != ERROR_SUCCESS) {
         srs_error("create consumer failed. ret=%d", ret);
         srs_error("create consumer failed. ret=%d", ret);
         return ret;
         return ret;
     }
     }

@@ -418,9 +418,10 @@ ISrsWakable::~ISrsWakable()
 {
 {
 }
 }
 
 
-SrsConsumer::SrsConsumer(SrsSource* _source)
+SrsConsumer::SrsConsumer(SrsSource* s, SrsConnection* c)
 {
 {
-    source = _source;
+    source = s;
+    conn = c;
     paused = false;
     paused = false;
     jitter = new SrsRtmpJitter();
     jitter = new SrsRtmpJitter();
     queue = new SrsMessageQueue();
     queue = new SrsMessageQueue();
@@ -2143,11 +2144,11 @@ void SrsSource::on_unpublish()
     handler->on_unpublish(this, _req);
     handler->on_unpublish(this, _req);
 }
 }
 
 
-int SrsSource::create_consumer(SrsConsumer*& consumer, bool ds, bool dm, bool dg)
+int SrsSource::create_consumer(SrsConnection* conn, SrsConsumer*& consumer, bool ds, bool dm, bool dg)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
-    consumer = new SrsConsumer(this);
+    consumer = new SrsConsumer(this, conn);
     consumers.push_back(consumer);
     consumers.push_back(consumer);
     
     
     double queue_size = _srs_config->get_queue_length(_req->vhost);
     double queue_size = _srs_config->get_queue_length(_req->vhost);

@@ -51,6 +51,7 @@ class SrsStSocket;
 class SrsRtmpServer;
 class SrsRtmpServer;
 class SrsEdgeProxyContext;
 class SrsEdgeProxyContext;
 class SrsMessageArray;
 class SrsMessageArray;
+class SrsConnection;
 #ifdef SRS_AUTO_HLS
 #ifdef SRS_AUTO_HLS
 class SrsHls;
 class SrsHls;
 #endif
 #endif
@@ -224,6 +225,8 @@ private:
     SrsRtmpJitter* jitter;
     SrsRtmpJitter* jitter;
     SrsSource* source;
     SrsSource* source;
     SrsMessageQueue* queue;
     SrsMessageQueue* queue;
+    // the owner connection for debug, maybe NULL.
+    SrsConnection* conn;
     bool paused;
     bool paused;
     // when source id changed, notice all consumers
     // when source id changed, notice all consumers
     bool should_update_source_id;
     bool should_update_source_id;
@@ -236,7 +239,7 @@ private:
     int mw_duration;
     int mw_duration;
 #endif
 #endif
 public:
 public:
-    SrsConsumer(SrsSource* _source);
+    SrsConsumer(SrsSource* s, SrsConnection* c);
     virtual ~SrsConsumer();
     virtual ~SrsConsumer();
 public:
 public:
     /**
     /**
@@ -571,7 +574,7 @@ public:
     * @param dg, whether dumps the gop cache.
     * @param dg, whether dumps the gop cache.
     */
     */
     virtual int create_consumer(
     virtual int create_consumer(
-        SrsConsumer*& consumer, 
+        SrsConnection* conn, SrsConsumer*& consumer,
         bool ds = true, bool dm = true, bool dg = true
         bool ds = true, bool dm = true, bool dg = true
     );
     );
     virtual void on_consumer_destroy(SrsConsumer* consumer);
     virtual void on_consumer_destroy(SrsConsumer* consumer);

@@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 // current release version
 // current release version
 #define VERSION_MAJOR       2
 #define VERSION_MAJOR       2
 #define VERSION_MINOR       0
 #define VERSION_MINOR       0
-#define VERSION_REVISION    205
+#define VERSION_REVISION    206
 
 
 // server info.
 // server info.
 #define RTMP_SIG_SRS_KEY "SRS"
 #define RTMP_SIG_SRS_KEY "SRS"