support set the live queue length(in seconds), drop when full.

README.md

@@ -212,7 +212,7 @@ usr sys idl wai hiq siq| read  writ| recv  send|  in   out | int   csw
 * nginx v1.5.0: 139524 lines <br/>
 
 ### History
-* v0.9, 2013-12-15, drop the old whole gop when live message queue full.
+* v0.9, 2013-12-15, support set the live queue length(in seconds), drop when full.
 * v0.9, 2013-12-15, fix the forwarder reconnect bug, feed it the sequence header.
 * v0.9, 2013-12-15, support reload the hls/forwarder/transcoder.
 * v0.9, 2013-12-14, refine the thread model for the retry threads.

trunk/conf/srs.conf

@@ -94,7 +94,7 @@ vhost dev {
     enabled         on;
     gop_cache       on;
     queue_length    10;
-    #forward         127.0.0.1:19350;
+    forward         127.0.0.1:19350;
     hls {
         enabled         off;
         hls_path        ./objs/nginx/html;

trunk/src/core/srs_core_forward.cpp

@@ -36,6 +36,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_core_rtmp.hpp>
 #include <srs_core_config.hpp>
 #include <srs_core_source.hpp>
+#include <srs_core_autofree.hpp>
 
 #define SRS_PULSE_TIMEOUT_MS 100
 #define SRS_FORWARDER_SLEEP_MS 2000
@@ -138,7 +139,7 @@ int SrsForwarder::on_meta_data(SrsSharedPtrMessage* metadata)
 	int ret = ERROR_SUCCESS;
 	
 	if ((ret = jitter->correct(metadata, 0, 0)) != ERROR_SUCCESS) {
-		srs_freep(msg);
+		srs_freep(metadata);
 		return ret;
 	}
 	
@@ -308,7 +309,7 @@ int SrsForwarder::forward()
 		// forward all messages.
 		int count = 0;
 		SrsSharedPtrMessage** msgs = NULL;
-		if ((ret = queue->get_packets(0, &msgs, count)) != ERROR_SUCCESS) {
+		if ((ret = queue->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
 			srs_error("get message to forward failed. ret=%d", ret);
 			return ret;
 		}

trunk/src/core/srs_core_source.cpp

@@ -900,11 +900,10 @@ void SrsSource::on_unpublish()
 {
 	int ret = ERROR_SUCCESS;
 	
-	double queue_size = config->get_queue_length(req->vhost);
-	
 	consumer = new SrsConsumer(this);
-	
 	consumers.push_back(consumer);
+	
+	double queue_size = config->get_queue_length(req->vhost);
 	consumer->set_queue_size(queue_size);
 
 	if (cache_metadata && (ret = consumer->enqueue(cache_metadata->copy(), sample_rate, frame_rate)) != ERROR_SUCCESS) {
@@ -959,6 +958,9 @@ int SrsSource::create_forwarders()
 		
 		SrsForwarder* forwarder = new SrsForwarder(this);
 		forwarders.push_back(forwarder);
+	
+		double queue_size = config->get_queue_length(req->vhost);
+		forwarder->set_queue_size(queue_size);
 		
 		if ((ret = forwarder->on_publish(req, forward_server)) != ERROR_SUCCESS) {
 			srs_error("start forwarder failed. "