fix the forwarder reconnect bug, feed it the sequence header.

@@ -212,6 +212,7 @@ usr sys idl wai hiq siq| read  writ| recv  send|  in   out | int   csw
 * nginx v1.5.0: 139524 lines <br/>
 * nginx v1.5.0: 139524 lines <br/>
 
 
 ### History
 ### History
+* 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-15, support reload the hls/forwarder/transcoder.
 * v0.9, 2013-12-14, refine the thread model for the retry threads.
 * v0.9, 2013-12-14, refine the thread model for the retry threads.
 * v0.9, 2013-12-10, auto install depends tools/libs on centos/ubuntu.
 * v0.9, 2013-12-10, auto install depends tools/libs on centos/ubuntu.

@@ -59,7 +59,9 @@ else
     echo "build x264"
     echo "build x264"
     cd $ff_current_dir &&
     cd $ff_current_dir &&
     rm -rf x264-snapshot-20131129-2245-stable && unzip -q ${ff_src_dir}/x264-snapshot-20131129-2245-stable.zip &&
     rm -rf x264-snapshot-20131129-2245-stable && unzip -q ${ff_src_dir}/x264-snapshot-20131129-2245-stable.zip &&
-    cd x264-snapshot-20131129-2245-stable && ./configure --prefix=${ff_release_dir} --disable-opencl --bit-depth=8 --enable-static && make && make install
+    cd x264-snapshot-20131129-2245-stable && 
+    ./configure --prefix=${ff_release_dir} --disable-opencl --bit-depth=8 --enable-static && 
+    make && make install
     ret=$?; if [[ 0 -ne ${ret} ]]; then echo "build x264 failed"; exit 1; fi
     ret=$?; if [[ 0 -ne ${ret} ]]; then echo "build x264 failed"; exit 1; fi
 fi
 fi
 
 

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

@@ -35,14 +35,17 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_core_pithy_print.hpp>
 #include <srs_core_pithy_print.hpp>
 #include <srs_core_rtmp.hpp>
 #include <srs_core_rtmp.hpp>
 #include <srs_core_config.hpp>
 #include <srs_core_config.hpp>
+#include <srs_core_source.hpp>
 
 
 #define SRS_PULSE_TIMEOUT_MS 100
 #define SRS_PULSE_TIMEOUT_MS 100
 #define SRS_FORWARDER_SLEEP_MS 2000
 #define SRS_FORWARDER_SLEEP_MS 2000
 #define SRS_SEND_TIMEOUT_US 3000000L
 #define SRS_SEND_TIMEOUT_US 3000000L
 #define SRS_RECV_TIMEOUT_US SRS_SEND_TIMEOUT_US
 #define SRS_RECV_TIMEOUT_US SRS_SEND_TIMEOUT_US
 
 
-SrsForwarder::SrsForwarder()
+SrsForwarder::SrsForwarder(SrsSource* _source)
 {
 {
+	source = _source;
+	
 	client = NULL;
 	client = NULL;
 	stfd = NULL;
 	stfd = NULL;
 	stream_id = 0;
 	stream_id = 0;
@@ -182,14 +185,17 @@ int SrsForwarder::cycle()
 		return ret;
 		return ret;
 	}
 	}
 	
 	
-	// TODO: FIXME: need to cache the metadata and sequence header when reconnect.
-	
 	if ((ret = client->publish(stream_name, stream_id)) != ERROR_SUCCESS) {
 	if ((ret = client->publish(stream_name, stream_id)) != ERROR_SUCCESS) {
 		srs_error("connect with server failed, stream_name=%s, stream_id=%d. ret=%d", 
 		srs_error("connect with server failed, stream_name=%s, stream_id=%d. ret=%d", 
 			stream_name.c_str(), stream_id, ret);
 			stream_name.c_str(), stream_id, ret);
 		return ret;
 		return ret;
 	}
 	}
 	
 	
+	if ((ret = source->on_forwarder_start(this)) != ERROR_SUCCESS) {
+		srs_error("callback the source to feed the sequence header failed. ret=%d", ret);
+		return ret;
+	}
+	
 	if ((ret = forward()) != ERROR_SUCCESS) {
 	if ((ret = forward()) != ERROR_SUCCESS) {
 		return ret;
 		return ret;
 	}
 	}

@@ -38,6 +38,7 @@ class SrsSharedPtrMessage;
 class SrsOnMetaDataPacket;
 class SrsOnMetaDataPacket;
 class SrsRtmpClient;
 class SrsRtmpClient;
 class SrsRequest;
 class SrsRequest;
+class SrsSource;
 
 
 /**
 /**
 * forward the stream to other servers.
 * forward the stream to other servers.
@@ -55,10 +56,11 @@ private:
 	st_netfd_t stfd;
 	st_netfd_t stfd;
 	SrsThread* pthread;
 	SrsThread* pthread;
 private:
 private:
+	SrsSource* source;
 	SrsRtmpClient* client;
 	SrsRtmpClient* client;
 	std::vector<SrsSharedPtrMessage*> msgs;
 	std::vector<SrsSharedPtrMessage*> msgs;
 public:
 public:
-	SrsForwarder();
+	SrsForwarder(SrsSource* _source);
 	virtual ~SrsForwarder();
 	virtual ~SrsForwarder();
 public:
 public:
 	virtual int on_publish(SrsRequest* req, std::string forward_server);
 	virtual int on_publish(SrsRequest* req, std::string forward_server);

@@ -450,7 +450,7 @@ int SrsSource::on_reload_forward(string vhost)
 		srs_error("create forwarders failed. ret=%d", ret);
 		srs_error("create forwarders failed. ret=%d", ret);
 		return ret;
 		return ret;
 	}
 	}
-	// TODO: FIXME: must feed it the sequence header.
+
 	srs_trace("vhost %s forwarders reload success", vhost.c_str());
 	srs_trace("vhost %s forwarders reload success", vhost.c_str());
 	
 	
 	return ret;
 	return ret;
@@ -498,6 +498,28 @@ int SrsSource::on_reload_transcode(string vhost)
 	return ret;
 	return ret;
 }
 }
 
 
+int SrsSource::on_forwarder_start(SrsForwarder* forwarder)
+{
+	int ret = ERROR_SUCCESS;
+		
+	// feed the forwarder the metadata/sequence header,
+	// when reload to enable the forwarder.
+	if (cache_metadata && (ret = forwarder->on_meta_data(cache_metadata->copy())) != ERROR_SUCCESS) {
+		srs_error("forwarder process onMetaData message failed. ret=%d", ret);
+		return ret;
+	}
+	if (cache_sh_video && (ret = forwarder->on_video(cache_sh_video->copy())) != ERROR_SUCCESS) {
+		srs_error("forwarder process video sequence header message failed. ret=%d", ret);
+		return ret;
+	}
+	if (cache_sh_audio && (ret = forwarder->on_audio(cache_sh_audio->copy())) != ERROR_SUCCESS) {
+		srs_error("forwarder process audio sequence header message failed. ret=%d", ret);
+		return ret;
+	}
+	
+	return ret;
+}
+
 bool SrsSource::can_publish()
 bool SrsSource::can_publish()
 {
 {
 	return _can_publish;
 	return _can_publish;
@@ -837,7 +859,7 @@ int SrsSource::create_forwarders()
 	for (int i = 0; conf && i < (int)conf->args.size(); i++) {
 	for (int i = 0; conf && i < (int)conf->args.size(); i++) {
 		std::string forward_server = conf->args.at(i);
 		std::string forward_server = conf->args.at(i);
 		
 		
-		SrsForwarder* forwarder = new SrsForwarder();
+		SrsForwarder* forwarder = new SrsForwarder(this);
 		forwarders.push_back(forwarder);
 		forwarders.push_back(forwarder);
 		
 		
 		if ((ret = forwarder->on_publish(req, forward_server)) != ERROR_SUCCESS) {
 		if ((ret = forwarder->on_publish(req, forward_server)) != ERROR_SUCCESS) {

@@ -221,6 +221,9 @@ public:
 	virtual int on_reload_forward(std::string vhost);
 	virtual int on_reload_forward(std::string vhost);
 	virtual int on_reload_hls(std::string vhost);
 	virtual int on_reload_hls(std::string vhost);
 	virtual int on_reload_transcode(std::string vhost);
 	virtual int on_reload_transcode(std::string vhost);
+// for the SrsForwarder to callback to request the sequence headers.
+public:
+	virtual int on_forwarder_start(SrsForwarder* forwarder);
 public:
 public:
 	virtual bool can_publish();
 	virtual bool can_publish();
 	virtual int on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata);
 	virtual int on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata);