refine the consumer, use srs message queue to shrink message when overflow.

@@ -93,7 +93,7 @@ vhost __defaultVhost__ {
 vhost dev {
 vhost dev {
     enabled         on;
     enabled         on;
     gop_cache       on;
     gop_cache       on;
-    queue_length    30;
+    queue_length    10;
     #forward         127.0.0.1:19350;
     #forward         127.0.0.1:19350;
     hls {
     hls {
         enabled         off;
         enabled         off;

@@ -1303,7 +1303,7 @@ double SrsConfig::get_queue_length(string vhost)
 	}
 	}
 	
 	
 	conf = conf->get("queue_length");
 	conf = conf->get("queue_length");
-	if (conf || conf->arg0().empty()) {
+	if (!conf || conf->arg0().empty()) {
 		return SRS_CONF_DEFAULT_QUEUE_LENGTH;
 		return SRS_CONF_DEFAULT_QUEUE_LENGTH;
 	}
 	}
 	
 	

@@ -171,6 +171,10 @@ int SrsMessageQueue::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, in
 	} else {
 	} else {
 		count = srs_min(max_count, (int)msgs.size());
 		count = srs_min(max_count, (int)msgs.size());
 	}
 	}
+
+	if (count <= 0) {
+		return ret;
+	}
 	
 	
 	pmsgs = new SrsSharedPtrMessage*[count];
 	pmsgs = new SrsSharedPtrMessage*[count];
 	
 	
@@ -178,6 +182,9 @@ int SrsMessageQueue::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, in
 		pmsgs[i] = msgs[i];
 		pmsgs[i] = msgs[i];
 	}
 	}
 	
 	
+	SrsSharedPtrMessage* last = msgs[count - 1];
+	av_start_time = last->header.timestamp;
+	
 	if (count == (int)msgs.size()) {
 	if (count == (int)msgs.size()) {
 		msgs.clear();
 		msgs.clear();
 	} else {
 	} else {
@@ -217,15 +224,15 @@ void SrsMessageQueue::shrink()
 		return;
 		return;
 	}
 	}
 	
 	
+	srs_trace("shrink the cache queue, size=%d, removed=%d, max=%.2f", 
+		(int)msgs.size(), iframe_index, queue_size_ms / 1000.0);
+	
 	// remove the first gop from the front
 	// remove the first gop from the front
 	for (int i = 0; i < iframe_index; i++) {
 	for (int i = 0; i < iframe_index; i++) {
 		SrsSharedPtrMessage* msg = msgs[i];
 		SrsSharedPtrMessage* msg = msgs[i];
 		srs_freep(msg);
 		srs_freep(msg);
 	}
 	}
 	msgs.erase(msgs.begin(), msgs.begin() + iframe_index);
 	msgs.erase(msgs.begin(), msgs.begin() + iframe_index);
-	
-	srs_trace("shrink the cache queue, "
-		"size=%d, removed=%d", (int)msgs.size(), iframe_index);
 }
 }
 
 
 void SrsMessageQueue::clear()
 void SrsMessageQueue::clear()
@@ -893,9 +900,12 @@ void SrsSource::on_unpublish()
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
 	
 	
+	double queue_size = config->get_queue_length(req->vhost);
+	
 	consumer = new SrsConsumer(this);
 	consumer = new SrsConsumer(this);
-	consumer->set_queue_size(config->get_queue_length(req->vhost));
+	
 	consumers.push_back(consumer);
 	consumers.push_back(consumer);
+	consumer->set_queue_size(queue_size);
 
 
 	if (cache_metadata && (ret = consumer->enqueue(cache_metadata->copy(), sample_rate, frame_rate)) != ERROR_SUCCESS) {
 	if (cache_metadata && (ret = consumer->enqueue(cache_metadata->copy(), sample_rate, frame_rate)) != ERROR_SUCCESS) {
 		srs_error("dispatch metadata failed. ret=%d", ret);
 		srs_error("dispatch metadata failed. ret=%d", ret);
@@ -919,6 +929,8 @@ void SrsSource::on_unpublish()
 		return ret;
 		return ret;
 	}
 	}
 	
 	
+	srs_trace("create consumer, queue_size=%.2f, tba=%d, tbv=%d", queue_size, sample_rate, frame_rate);
+	
 	return ret;
 	return ret;
 }
 }