support forward stream to origin/edge

@@ -47,14 +47,16 @@ m3u8 url: http://127.0.0.1:80/live/livestream.m3u8
 8. support cache last gop for flash player to fast startup.<br/>
 8. support cache last gop for flash player to fast startup.<br/>
 9. support listen at multiple ports.<br/>
 9. support listen at multiple ports.<br/>
 10. support long time(>4.6hours) publish/play.<br/>
 10. support long time(>4.6hours) publish/play.<br/>
-11. [dev] support forward publish stream to build active-standby cluster.<br/>
-12. [plan] support live stream transcoding by ffmpeg.<br/>
-13. [plan] support full http callback api.<br/>
-14. [plan] support network based cli and json result.<br/>
-15. [plan] support bandwidth test api and flash client.<br/>
-16. no edge server, origin server only.<br/>
-17. no vod streaming, live streaming only.<br/>
-18. no multiple processes, single process only.<br/>
+11. high performace, 1800 connections(500kbps), 900Mbps, CPU 90.2%, 41MB<br/>
+12. support forward publish stream to build active-standby cluster.<br/>
+13. support broadcast by forward the stream to other servers(origin/edge).<br/>
+14. [plan] support live stream transcoding by ffmpeg.<br/>
+15. [plan] support full http callback api.<br/>
+16. [plan] support network based cli and json result.<br/>
+17. [plan] support bandwidth test api and flash client.<br/>
+18. no edge server, origin server only.<br/>
+19. no vod streaming, live streaming only.<br/>
+20. no multiple processes, single process only.<br/>
 
 
 ### Performance
 ### Performance
 1. 300 connections, 150Mbps, 500kbps, CPU 18.8%, 5956KB.
 1. 300 connections, 150Mbps, 500kbps, CPU 18.8%, 5956KB.

@@ -15,7 +15,7 @@ vhost __defaultVhost__ {
     hls_path        ./objs/nginx/html;
     hls_path        ./objs/nginx/html;
     hls_fragment    5;
     hls_fragment    5;
     hls_window      30;
     hls_window      30;
-    forward         192.168.1.50;
+    forward         127.0.0.1:1936;
 }
 }
 # the vhost which forward publish streams.
 # the vhost which forward publish streams.
 vhost forward.vhost.com {
 vhost forward.vhost.com {
@@ -104,5 +104,8 @@ pithy_print {
     # shared print interval for all play clients, in milliseconds.
     # shared print interval for all play clients, in milliseconds.
     # if not specified, set to 1300.
     # if not specified, set to 1300.
     play            3000;
     play            3000;
+    # shared print interval for all forwarders, in milliseconds.
+    # if not specified, set to 2000.
+    forwarder       3000;
 }
 }
 
 

@@ -681,6 +681,16 @@ SrsConfDirective* SrsConfig::get_pithy_print_publish()
 	return pithy->get("publish");
 	return pithy->get("publish");
 }
 }
 
 
+SrsConfDirective* SrsConfig::get_pithy_print_forwarder()
+{
+	SrsConfDirective* pithy = root->get("pithy_print");
+	if (!pithy) {
+		return NULL;
+	}
+	
+	return pithy->get("forwarder");
+}
+
 SrsConfDirective* SrsConfig::get_pithy_print_play()
 SrsConfDirective* SrsConfig::get_pithy_print_play()
 {
 {
 	SrsConfDirective* pithy = root->get("pithy_print");
 	SrsConfDirective* pithy = root->get("pithy_print");

@@ -126,6 +126,7 @@ public:
 	virtual SrsConfDirective* get_listen();
 	virtual SrsConfDirective* get_listen();
 	virtual SrsConfDirective* get_chunk_size();
 	virtual SrsConfDirective* get_chunk_size();
 	virtual SrsConfDirective* get_pithy_print_publish();
 	virtual SrsConfDirective* get_pithy_print_publish();
+	virtual SrsConfDirective* get_pithy_print_forwarder();
 	virtual SrsConfDirective* get_pithy_print_play();
 	virtual SrsConfDirective* get_pithy_print_play();
 private:
 private:
 	virtual int parse_file(const char* filename);
 	virtual int parse_file(const char* filename);

@@ -32,7 +32,10 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_core_error.hpp>
 #include <srs_core_error.hpp>
 #include <srs_core_rtmp.hpp>
 #include <srs_core_rtmp.hpp>
 #include <srs_core_log.hpp>
 #include <srs_core_log.hpp>
+#include <srs_core_protocol.hpp>
+#include <srs_core_pithy_print.hpp>
 
 
+#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
@@ -49,6 +52,13 @@ SrsForwarder::SrsForwarder()
 SrsForwarder::~SrsForwarder()
 SrsForwarder::~SrsForwarder()
 {
 {
 	on_unpublish();
 	on_unpublish();
+	
+	std::vector<SrsSharedPtrMessage*>::iterator it;
+	for (it = msgs.begin(); it != msgs.end(); ++it) {
+		SrsSharedPtrMessage* msg = *it;
+		srs_freep(msg);
+	}
+	msgs.clear();
 }
 }
 
 
 int SrsForwarder::on_publish(std::string vhost, std::string _app, std::string stream, std::string forward_server)
 int SrsForwarder::on_publish(std::string vhost, std::string _app, std::string stream, std::string forward_server)
@@ -111,18 +121,27 @@ void SrsForwarder::on_unpublish()
 int SrsForwarder::on_meta_data(SrsSharedPtrMessage* metadata)
 int SrsForwarder::on_meta_data(SrsSharedPtrMessage* metadata)
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
+	
+	msgs.push_back(metadata);
+	
 	return ret;
 	return ret;
 }
 }
 
 
 int SrsForwarder::on_audio(SrsSharedPtrMessage* msg)
 int SrsForwarder::on_audio(SrsSharedPtrMessage* msg)
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
+	
+	msgs.push_back(msg);
+	
 	return ret;
 	return ret;
 }
 }
 
 
 int SrsForwarder::on_video(SrsSharedPtrMessage* msg)
 int SrsForwarder::on_video(SrsSharedPtrMessage* msg)
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
+	
+	msgs.push_back(msg);
+	
 	return ret;
 	return ret;
 }
 }
 
 
@@ -232,6 +251,60 @@ int SrsForwarder::forward_cycle_imp()
 		return ret;
 		return ret;
 	}
 	}
 	
 	
+	if ((ret = forward()) != ERROR_SUCCESS) {
+		return ret;
+	}
+	
+	return ret;
+}
+
+int SrsForwarder::forward()
+{
+	int ret = ERROR_SUCCESS;
+	
+	client->set_recv_timeout(SRS_PULSE_TIMEOUT_MS * 1000);
+	
+	SrsPithyPrint pithy_print(SRS_STAGE_FORWARDER);
+
+	while (loop) {
+		pithy_print.elapse(SRS_PULSE_TIMEOUT_MS);
+		
+		// switch to other st-threads.
+		st_usleep(0);
+
+		// read from client.
+		if (true) {
+			SrsCommonMessage* msg = NULL;
+			ret = client->recv_message(&msg);
+			
+			srs_verbose("play loop recv message. ret=%d", ret);
+			if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) {
+				srs_error("recv server control message failed. ret=%d", ret);
+				return ret;
+			}
+		}
+		
+		int count = (int)msgs.size();
+
+		// reportable
+		if (pithy_print.can_print()) {
+			srs_trace("-> clock=%u, time=%"PRId64", msgs=%d, obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d", 
+				(int)(srs_get_system_time_ms()/1000), pithy_print.get_age(), count, client->get_send_bytes(), client->get_recv_bytes(), client->get_send_kbps(), client->get_recv_kbps());
+		}
+	
+		// all msgs to forward.
+		for (int i = 0; i < count; i++) {
+			SrsSharedPtrMessage* msg = msgs[i];
+			msgs[i] = NULL;
+			
+			if ((ret = client->send_message(msg)) != ERROR_SUCCESS) {
+				srs_error("forwarder send message to server failed. ret=%d", ret);
+				return ret;
+			}
+		}
+		msgs.clear();
+	}
+	
 	return ret;
 	return ret;
 }
 }
 
 

@@ -30,6 +30,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_core.hpp>
 #include <srs_core.hpp>
 
 
 #include <string>
 #include <string>
+#include <vector>
 
 
 #include <st.h>
 #include <st.h>
 
 
@@ -55,6 +56,7 @@ private:
 	bool loop;
 	bool loop;
 private:
 private:
 	SrsRtmpClient* client;
 	SrsRtmpClient* client;
+	std::vector<SrsSharedPtrMessage*> msgs;
 public:
 public:
 	SrsForwarder();
 	SrsForwarder();
 	virtual ~SrsForwarder();
 	virtual ~SrsForwarder();
@@ -70,6 +72,7 @@ private:
 	std::string parse_server(std::string host);
 	std::string parse_server(std::string host);
 private:
 private:
 	virtual int forward_cycle_imp();
 	virtual int forward_cycle_imp();
+	virtual int forward();
 	virtual void forward_cycle();
 	virtual void forward_cycle();
 	static void* forward_thread(void* arg);
 	static void* forward_thread(void* arg);
 };
 };

@@ -34,6 +34,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #define SRS_STAGE_DEFAULT_INTERVAL_MS 1200
 #define SRS_STAGE_DEFAULT_INTERVAL_MS 1200
 #define SRS_STAGE_PLAY_USER_INTERVAL_MS 1300
 #define SRS_STAGE_PLAY_USER_INTERVAL_MS 1300
 #define SRS_STAGE_PUBLISH_USER_INTERVAL_MS 1100
 #define SRS_STAGE_PUBLISH_USER_INTERVAL_MS 1100
+#define SRS_STAGE_FORWARDER_INTERVAL_MS 2000
 
 
 struct SrsStageInfo : public SrsReloadHandler
 struct SrsStageInfo : public SrsReloadHandler
 {
 {
@@ -73,6 +74,14 @@ struct SrsStageInfo : public SrsReloadHandler
 				}
 				}
 				break;
 				break;
 			}
 			}
+			case SRS_STAGE_FORWARDER: {
+				pithy_print_time_ms = SRS_STAGE_FORWARDER_INTERVAL_MS;
+				SrsConfDirective* conf = config->get_pithy_print_forwarder();
+				if (conf && !conf->arg0().empty()) {
+					pithy_print_time_ms = ::atoi(conf->arg0().c_str());
+				}
+				break;
+			}
 			default: {
 			default: {
 				pithy_print_time_ms = SRS_STAGE_DEFAULT_INTERVAL_MS;
 				pithy_print_time_ms = SRS_STAGE_DEFAULT_INTERVAL_MS;
 				break;
 				break;

@@ -34,6 +34,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #define SRS_STAGE_PLAY_USER 1
 #define SRS_STAGE_PLAY_USER 1
 // the pithy stage for all publish clients.
 // the pithy stage for all publish clients.
 #define SRS_STAGE_PUBLISH_USER 2
 #define SRS_STAGE_PUBLISH_USER 2
+// the pithy stage for all forward clients.
+#define SRS_STAGE_FORWARDER 3
 
 
 /**
 /**
 * the stage is used for a collection of object to do print,
 * the stage is used for a collection of object to do print,

@@ -184,6 +184,36 @@ void SrsRtmpClient::set_send_timeout(int64_t timeout_us)
 	protocol->set_send_timeout(timeout_us);
 	protocol->set_send_timeout(timeout_us);
 }
 }
 
 
+int64_t SrsRtmpClient::get_recv_bytes()
+{
+	return protocol->get_recv_bytes();
+}
+
+int64_t SrsRtmpClient::get_send_bytes()
+{
+	return protocol->get_send_bytes();
+}
+
+int SrsRtmpClient::get_recv_kbps()
+{
+	return protocol->get_recv_kbps();
+}
+
+int SrsRtmpClient::get_send_kbps()
+{
+	return protocol->get_send_kbps();
+}
+
+int SrsRtmpClient::recv_message(SrsCommonMessage** pmsg)
+{
+	return protocol->recv_message(pmsg);
+}
+
+int SrsRtmpClient::send_message(ISrsMessage* msg)
+{
+	return protocol->send_message(msg);
+}
+
 int SrsRtmpClient::handshake()
 int SrsRtmpClient::handshake()
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;

@@ -108,6 +108,12 @@ public:
 public:
 public:
 	virtual void set_recv_timeout(int64_t timeout_us);
 	virtual void set_recv_timeout(int64_t timeout_us);
 	virtual void set_send_timeout(int64_t timeout_us);
 	virtual void set_send_timeout(int64_t timeout_us);
+	virtual int64_t get_recv_bytes();
+	virtual int64_t get_send_bytes();
+	virtual int get_recv_kbps();
+	virtual int get_send_kbps();
+	virtual int recv_message(SrsCommonMessage** pmsg);
+	virtual int send_message(ISrsMessage* msg);
 public:
 public:
 	virtual int handshake();
 	virtual int handshake();
 	virtual int connect_app(std::string app, std::string tc_url);
 	virtual int connect_app(std::string app, std::string tc_url);