refine the thread model for the retry threads

@@ -81,11 +81,11 @@ vhost __defaultVhost__ {
 vhost dev {
     enabled         on;
     gop_cache       on;
-    hls             on;
+    hls             off;
     hls_path        ./objs/nginx/html;
     hls_fragment    5;
     hls_window      30;
-    #forward         127.0.0.1:19350;
+    forward         127.0.0.1:19350;
     http_hooks {
         enabled         off;
         on_connect      http://127.0.0.1:8085/api/v1/clients;
@@ -96,7 +96,7 @@ vhost dev {
         on_stop         http://127.0.0.1:8085/api/v1/sessions;
     }
     transcode {
-        enabled     off;
+        enabled     on;
         ffmpeg      ./objs/ffmpeg/bin/ffmpeg;
         engine dev {
             enabled         on;

@@ -116,7 +116,7 @@ MODULE_FILES=("srs_core" "srs_core_log" "srs_core_server"
         "srs_core_handshake" "srs_core_pithy_print"
         "srs_core_config" "srs_core_refer" "srs_core_reload"
         "srs_core_hls" "srs_core_forward" "srs_core_encoder"
-        "srs_core_http")
+        "srs_core_http" "srs_core_thread")
 MODULE_DIR="src/core" . auto/modules.sh
 CORE_OBJS="${MODULE_OBJS[@]}"
 

@@ -111,3 +111,17 @@ void srs_vhost_resolve(std::string& vhost, std::string& app)
 		}
 	}
 }
+
+void srs_close_stfd(st_netfd_t& stfd)
+{
+	if (stfd) {
+		int fd = st_netfd_fileno(stfd);
+		st_netfd_close(stfd);
+		stfd = NULL;
+		
+		// st does not close it sometimes, 
+		// close it manually.
+		close(fd);
+	}
+}
+

@@ -46,6 +46,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <stddef.h>
 #include <sys/types.h>
 
+#include <st.h>
+
 // generated by configure.
 #include <srs_auto_headers.hpp>
 
@@ -102,4 +104,7 @@ extern std::string srs_dns_resolve(std::string host);
 //		app...vhost...request_vhost
 extern void srs_vhost_resolve(std::string& vhost, std::string& app);
 
+// close the netfd, and close the underlayer fd.
+extern void srs_close_stfd(st_netfd_t& stfd);
+
 #endif
\ No newline at end of file

@@ -36,15 +36,7 @@ SrsConnection::SrsConnection(SrsServer* srs_server, st_netfd_t client_stfd)
 
 SrsConnection::~SrsConnection()
 {
-	if (stfd) {
-		int fd = st_netfd_fileno(stfd);
-		st_netfd_close(stfd);
-		stfd = NULL;
-		
-		// st does not close it sometimes, 
-		// close it manually.
-		close(fd);
-	}
+	srs_close_stfd(stfd);
 }
 
 int SrsConnection::start()

@@ -30,8 +30,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 #include <srs_core.hpp>
 
-#include <st.h>
-
 class SrsServer;
 class SrsConnection
 {

@@ -483,52 +483,15 @@ void SrsFFMPEG::stop()
 
 SrsEncoder::SrsEncoder()
 {
-	tid = NULL;
-	loop = false;
+	pthread = new SrsThread(this, SRS_ENCODER_SLEEP_MS);
+	pithy_print = new SrsPithyPrint(SRS_STAGE_ENCODER);
 }
 
 SrsEncoder::~SrsEncoder()
 {
 	on_unpublish();
-}
-
-int SrsEncoder::parse_scope_engines(SrsRequest* req)
-{
-	int ret = ERROR_SUCCESS;
-	
-	// parse all transcode engines.
-	SrsConfDirective* conf = NULL;
-	
-	// parse vhost scope engines
-	std::string scope = "";
-	if ((conf = config->get_transcode(req->vhost, scope)) != NULL) {
-		if ((ret = parse_transcode(req, conf)) != ERROR_SUCCESS) {
-			srs_error("parse vhost scope=%s transcode engines failed. "
-				"ret=%d", scope.c_str(), ret);
-			return ret;
-		}
-	}
-	// parse app scope engines
-	scope = req->app;
-	if ((conf = config->get_transcode(req->vhost, scope)) != NULL) {
-		if ((ret = parse_transcode(req, conf)) != ERROR_SUCCESS) {
-			srs_error("parse app scope=%s transcode engines failed. "
-				"ret=%d", scope.c_str(), ret);
-			return ret;
-		}
-	}
-	// parse stream scope engines
-	scope += "/";
-	scope += req->stream;
-	if ((conf = config->get_transcode(req->vhost, scope)) != NULL) {
-		if ((ret = parse_transcode(req, conf)) != ERROR_SUCCESS) {
-			srs_error("parse stream scope=%s transcode engines failed. "
-				"ret=%d", scope.c_str(), ret);
-			return ret;
-		}
-	}
 	
-	return ret;
+	srs_freep(pthread);
 }
 
 int SrsEncoder::on_publish(SrsRequest* req)
@@ -539,6 +502,7 @@ int SrsEncoder::on_publish(SrsRequest* req)
 	
 	// ignore the loop encoder
 	if (ret == ERROR_ENCODER_LOOP) {
+		clear_engines();
 		ret = ERROR_SUCCESS;
 	}
 	
@@ -548,9 +512,7 @@ int SrsEncoder::on_publish(SrsRequest* req)
 	}
     
     // start thread to run all encoding engines.
-    srs_assert(!tid);
-    if((tid = st_thread_create(encoder_thread, this, 1, 0)) == NULL) {
-		ret = ERROR_ST_CREATE_FORWARD_THREAD;
+    if ((ret = pthread->start()) != ERROR_SUCCESS) {
         srs_error("st_thread_create failed. ret=%d", ret);
         return ret;
     }
@@ -560,23 +522,58 @@ int SrsEncoder::on_publish(SrsRequest* req)
 
 void SrsEncoder::on_unpublish()
 {
-	if (tid) {
-		loop = false;
-		st_thread_interrupt(tid);
-		st_thread_join(tid, NULL);
-		tid = NULL;
+	pthread->stop();
+	clear_engines();
+}
+
+int SrsEncoder::cycle()
+{
+	int ret = ERROR_SUCCESS;
+	
+	std::vector<SrsFFMPEG*>::iterator it;
+	for (it = ffmpegs.begin(); it != ffmpegs.end(); ++it) {
+		SrsFFMPEG* ffmpeg = *it;
+		
+		// start all ffmpegs.
+		if ((ret = ffmpeg->start()) != ERROR_SUCCESS) {
+			srs_error("ffmpeg start failed. ret=%d", ret);
+			return ret;
 		}
 
-	clear_engines();
+		// check ffmpeg status.
+		if ((ret = ffmpeg->cycle()) != ERROR_SUCCESS) {
+			srs_error("ffmpeg cycle failed. ret=%d", ret);
+			return ret;
+		}
+	}
+
+	// pithy print
+	encoder();
+	pithy_print->elapse(SRS_ENCODER_SLEEP_MS);
+	
+	return ret;
+}
+
+void SrsEncoder::on_leave_loop()
+{
+	// kill ffmpeg when finished and it alive
+	std::vector<SrsFFMPEG*>::iterator it;
+
+	for (it = ffmpegs.begin(); it != ffmpegs.end(); ++it) {
+		SrsFFMPEG* ffmpeg = *it;
+		ffmpeg->stop();
+	}
 }
 
 void SrsEncoder::clear_engines()
 {
 	std::vector<SrsFFMPEG*>::iterator it;
+	
 	for (it = ffmpegs.begin(); it != ffmpegs.end(); ++it) {
 		SrsFFMPEG* ffmpeg = *it;
 		srs_freep(ffmpeg);
 	}
+
 	ffmpegs.clear();
 }
 
@@ -585,6 +582,45 @@ SrsFFMPEG* SrsEncoder::at(int index)
 	return ffmpegs[index];
 }
 
+int SrsEncoder::parse_scope_engines(SrsRequest* req)
+{
+	int ret = ERROR_SUCCESS;
+	
+	// parse all transcode engines.
+	SrsConfDirective* conf = NULL;
+	
+	// parse vhost scope engines
+	std::string scope = "";
+	if ((conf = config->get_transcode(req->vhost, scope)) != NULL) {
+		if ((ret = parse_transcode(req, conf)) != ERROR_SUCCESS) {
+			srs_error("parse vhost scope=%s transcode engines failed. "
+				"ret=%d", scope.c_str(), ret);
+			return ret;
+		}
+	}
+	// parse app scope engines
+	scope = req->app;
+	if ((conf = config->get_transcode(req->vhost, scope)) != NULL) {
+		if ((ret = parse_transcode(req, conf)) != ERROR_SUCCESS) {
+			srs_error("parse app scope=%s transcode engines failed. "
+				"ret=%d", scope.c_str(), ret);
+			return ret;
+		}
+	}
+	// parse stream scope engines
+	scope += "/";
+	scope += req->stream;
+	if ((conf = config->get_transcode(req->vhost, scope)) != NULL) {
+		if ((ret = parse_transcode(req, conf)) != ERROR_SUCCESS) {
+			srs_error("parse stream scope=%s transcode engines failed. "
+				"ret=%d", scope.c_str(), ret);
+			return ret;
+		}
+	}
+	
+	return ret;
+}
+
 int SrsEncoder::parse_transcode(SrsRequest* req, SrsConfDirective* conf)
 {
 	int ret = ERROR_SUCCESS;
@@ -631,7 +667,6 @@ int SrsEncoder::parse_transcode(SrsRequest* req, SrsConfDirective* conf)
 			
 			// if got a loop, donot transcode the whole stream.
 			if (ret == ERROR_ENCODER_LOOP) {
-				clear_engines();
 				break;
 			}
 			
@@ -646,85 +681,14 @@ int SrsEncoder::parse_transcode(SrsRequest* req, SrsConfDirective* conf)
 	return ret;
 }
 
-int SrsEncoder::cycle()
-{
-	int ret = ERROR_SUCCESS;
-	
-	std::vector<SrsFFMPEG*>::iterator it;
-	for (it = ffmpegs.begin(); it != ffmpegs.end(); ++it) {
-		SrsFFMPEG* ffmpeg = *it;
-		
-		// start all ffmpegs.
-		if ((ret = ffmpeg->start()) != ERROR_SUCCESS) {
-			srs_error("ffmpeg start failed. ret=%d", ret);
-			return ret;
-		}
-
-		// check ffmpeg status.
-		if ((ret = ffmpeg->cycle()) != ERROR_SUCCESS) {
-			srs_error("ffmpeg cycle failed. ret=%d", ret);
-			return ret;
-		}
-	}
-	
-	return ret;
-}
-
-void SrsEncoder::encoder_cycle()
-{
-	int ret = ERROR_SUCCESS;
-	
-	log_context->generate_id();
-	srs_trace("encoder cycle start");
-	
-	SrsPithyPrint pithy_print(SRS_STAGE_ENCODER);
-	
-	while (loop) {
-		if ((ret = cycle()) != ERROR_SUCCESS) {
-			srs_warn("encoder cycle failed, ignored and retry, ret=%d", ret);
-		} else {
-			srs_info("encoder cycle success, retry");
-		}
-		
-		if (!loop) {
-			break;
-		}
-
-		encoder(&pithy_print);
-		pithy_print.elapse(SRS_ENCODER_SLEEP_MS);
-		
-		st_usleep(SRS_ENCODER_SLEEP_MS * 1000);
-	}
-	
-	// kill ffmpeg when finished and it alive
-	std::vector<SrsFFMPEG*>::iterator it;
-	for (it = ffmpegs.begin(); it != ffmpegs.end(); ++it) {
-		SrsFFMPEG* ffmpeg = *it;
-		ffmpeg->stop();
-	}
-	
-	srs_trace("encoder cycle finished");
-}
-
-void SrsEncoder::encoder(SrsPithyPrint* pithy_print)
+void SrsEncoder::encoder()
 {
 	// reportable
 	if (pithy_print->can_print()) {
-		srs_trace("-> time=%"PRId64", encoders=%d",
-			pithy_print->get_age(), (int)ffmpegs.size());
+		// TODO: FIXME: show more info.
+		srs_trace("-> time=%"PRId64", encoders=%d", pithy_print->get_age(), (int)ffmpegs.size());
 	}
 }
 
-void* SrsEncoder::encoder_thread(void* arg)
-{
-	SrsEncoder* obj = (SrsEncoder*)arg;
-	srs_assert(obj != NULL);
-	
-	obj->loop = true;
-	obj->encoder_cycle();
-	
-	return NULL;
-}
-
 #endif
 

@@ -32,7 +32,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <string>
 #include <vector>
 
-#include <st.h>
+#include <srs_core_thread.hpp>
 
 class SrsConfDirective;
 class SrsRequest;
@@ -85,28 +85,29 @@ public:
 * the encoder for a stream,
 * may use multiple ffmpegs to transcode the specified stream.
 */
-class SrsEncoder
+class SrsEncoder : public ISrsThreadHandler
 {
 private:
 	std::vector<SrsFFMPEG*> ffmpegs;
 private:
-	st_thread_t tid;
-	bool loop;
+	SrsThread* pthread;
+	SrsPithyPrint* pithy_print;
 public:
 	SrsEncoder();
 	virtual ~SrsEncoder();
 public:
 	virtual int on_publish(SrsRequest* req);
 	virtual void on_unpublish();
+// interface ISrsThreadHandler.
+public:
+	virtual int cycle();
+	virtual void on_leave_loop();
 private:
-	virtual int parse_scope_engines(SrsRequest* req);
 	virtual void clear_engines();
 	virtual SrsFFMPEG* at(int index);
+	virtual int parse_scope_engines(SrsRequest* req);
 	virtual int parse_transcode(SrsRequest* req, SrsConfDirective* conf);
-	virtual int cycle();
-	virtual void encoder_cycle();
-	virtual void encoder(SrsPithyPrint* pithy_print);
-	static void* encoder_thread(void* arg);
+	virtual void encoder();
 };
 
 #endif

@@ -37,8 +37,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #define ERROR_ST_OPEN_SOCKET			102
 #define ERROR_ST_CREATE_LISTEN_THREAD	103
 #define ERROR_ST_CREATE_CYCLE_THREAD	104
-#define ERROR_ST_CREATE_FORWARD_THREAD	105
-#define ERROR_ST_CONNECT				106
+#define ERROR_ST_CONNECT				105
 
 #define ERROR_SOCKET_CREATE 			200
 #define ERROR_SOCKET_SETREUSE 			201

@@ -47,8 +47,7 @@ SrsForwarder::SrsForwarder()
 	stfd = NULL;
 	stream_id = 0;
 
-	tid = NULL;
-	loop = false;
+	pthread = new SrsThread(this, SRS_FORWARDER_SLEEP_MS);
 }
 
 SrsForwarder::~SrsForwarder()
@@ -61,6 +60,8 @@ SrsForwarder::~SrsForwarder()
 		srs_freep(msg);
 	}
 	msgs.clear();
+	
+	srs_freep(pthread);
 }
 
 int SrsForwarder::on_publish(SrsRequest* req, std::string forward_server)
@@ -110,17 +111,8 @@ int SrsForwarder::on_publish(SrsRequest* req, std::string forward_server)
 		source_ep.c_str(), dest_ep.c_str(), tc_url.c_str(), 
 		stream_name.c_str());
 	
-	// TODO: seems bug when republish and reforward.
-	
-	// start forward
-	if ((ret = open_socket()) != ERROR_SUCCESS) {
-		return ret;
-	}
-    
-    srs_assert(!tid);
-    if((tid = st_thread_create(forward_thread, this, 1, 0)) == NULL){
-		ret = ERROR_ST_CREATE_FORWARD_THREAD;
-        srs_error("st_thread_create failed. ret=%d", ret);
+	if ((ret = pthread->start()) != ERROR_SUCCESS) {
+        srs_error("start srs thread failed. ret=%d", ret);
         return ret;
 	}
 	
@@ -129,22 +121,9 @@ int SrsForwarder::on_publish(SrsRequest* req, std::string forward_server)
 
 void SrsForwarder::on_unpublish()
 {
-	if (tid) {
-		loop = false;
-		st_thread_interrupt(tid);
-		st_thread_join(tid, NULL);
-		tid = NULL;
-	}
-	
-	if (stfd) {
-		int fd = st_netfd_fileno(stfd);
-		st_netfd_close(stfd);
-		stfd = NULL;
+	pthread->stop();
 	
-		// st does not close it sometimes, 
-		// close it manually.
-		close(fd);
-	}
+	close_underlayer_socket();
 	
 	srs_freep(client);
 }
@@ -178,10 +157,59 @@ int SrsForwarder::on_video(SrsSharedPtrMessage* msg)
 	return ret;
 }
 
-int SrsForwarder::open_socket()
+int SrsForwarder::cycle()
 {
 	int ret = ERROR_SUCCESS;
 	
+	if ((ret = connect_server()) != ERROR_SUCCESS) {
+		return ret;
+	}
+	srs_assert(client);
+
+	client->set_recv_timeout(SRS_RECV_TIMEOUT_US);
+	client->set_send_timeout(SRS_SEND_TIMEOUT_US);
+	
+	if ((ret = client->handshake()) != ERROR_SUCCESS) {
+		srs_error("handshake with server failed. ret=%d", ret);
+		return ret;
+	}
+	if ((ret = client->connect_app(app, tc_url)) != ERROR_SUCCESS) {
+		srs_error("connect with server failed, tcUrl=%s. ret=%d", tc_url.c_str(), ret);
+		return ret;
+	}
+	if ((ret = client->create_stream(stream_id)) != ERROR_SUCCESS) {
+		srs_error("connect with server failed, stream_id=%d. ret=%d", stream_id, 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) {
+		srs_error("connect with server failed, stream_name=%s, stream_id=%d. ret=%d", 
+			stream_name.c_str(), stream_id, ret);
+		return ret;
+	}
+	
+	if ((ret = forward()) != ERROR_SUCCESS) {
+		return ret;
+	}
+	
+	return ret;
+}
+
+void SrsForwarder::close_underlayer_socket()
+{
+	srs_close_stfd(stfd);
+}
+
+int SrsForwarder::connect_server()
+{
+	int ret = ERROR_SUCCESS;
+	
+	// reopen
+	close_underlayer_socket();
+	
+	// open socket.
 	srs_trace("forward stream=%s, tcUrl=%s to server=%s, port=%d",
 		stream_name.c_str(), tc_url.c_str(), server.c_str(), port);
 
@@ -192,6 +220,7 @@ int SrsForwarder::open_socket()
         return ret;
     }
     
+    srs_assert(!stfd);
     stfd = st_netfd_open_socket(sock);
     if(stfd == NULL){
         ret = ERROR_ST_OPEN_SOCKET;
@@ -202,13 +231,7 @@ int SrsForwarder::open_socket()
 	srs_freep(client);
 	client = new SrsRtmpClient(stfd);
 	
-	return ret;
-}
-
-int SrsForwarder::connect_server()
-{
-	int ret = ERROR_SUCCESS;
-	
+	// connect to server.
 	std::string ip = srs_dns_resolve(server);
 	if (ip.empty()) {
 		ret = ERROR_SYSTEM_IP_INVALID;
@@ -231,46 +254,6 @@ int SrsForwarder::connect_server()
 	return ret;
 }
 
-int SrsForwarder::cycle()
-{
-	int ret = ERROR_SUCCESS;
-
-	client->set_recv_timeout(SRS_RECV_TIMEOUT_US);
-	client->set_send_timeout(SRS_SEND_TIMEOUT_US);
-	
-	if ((ret = connect_server()) != ERROR_SUCCESS) {
-		return ret;
-	}
-	srs_assert(client);
-	
-	if ((ret = client->handshake()) != ERROR_SUCCESS) {
-		srs_error("handshake with server failed. ret=%d", ret);
-		return ret;
-	}
-	if ((ret = client->connect_app(app, tc_url)) != ERROR_SUCCESS) {
-		srs_error("connect with server failed, tcUrl=%s. ret=%d", tc_url.c_str(), ret);
-		return ret;
-	}
-	if ((ret = client->create_stream(stream_id)) != ERROR_SUCCESS) {
-		srs_error("connect with server failed, stream_id=%d. ret=%d", stream_id, 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) {
-		srs_error("connect with server failed, stream_name=%s, stream_id=%d. ret=%d", 
-			stream_name.c_str(), stream_id, ret);
-		return ret;
-	}
-	
-	if ((ret = forward()) != ERROR_SUCCESS) {
-		return ret;
-	}
-	
-	return ret;
-}
-
 int SrsForwarder::forward()
 {
 	int ret = ERROR_SUCCESS;
@@ -279,9 +262,7 @@ int SrsForwarder::forward()
 	
 	SrsPithyPrint pithy_print(SRS_STAGE_FORWARDER);
 
-	while (loop) {
-		pithy_print.elapse(SRS_PULSE_TIMEOUT_MS);
-		
+	while (true) {
 		// switch to other st-threads.
 		st_usleep(0);
 
@@ -303,7 +284,8 @@ int SrsForwarder::forward()
 			continue;
 		}
 
-		// reportable
+		// pithy print
+		pithy_print.elapse(SRS_PULSE_TIMEOUT_MS);
 		if (pithy_print.can_print()) {
 			srs_trace("-> time=%"PRId64", msgs=%d, obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d", 
 				pithy_print.get_age(), count, client->get_send_bytes(), client->get_recv_bytes(), client->get_send_kbps(), client->get_recv_kbps());
@@ -345,43 +327,3 @@ int SrsForwarder::forward()
 	return ret;
 }
 
-void SrsForwarder::forward_cycle()
-{
-	int ret = ERROR_SUCCESS;
-	
-	log_context->generate_id();
-	srs_trace("forward cycle start");
-	
-	while (loop) {
-		if ((ret = cycle()) != ERROR_SUCCESS) {
-			srs_warn("forward cycle failed, ignored and retry, ret=%d", ret);
-		} else {
-			srs_info("forward cycle success, retry");
-		}
-		
-		if (!loop) {
-			break;
-		}
-		
-		st_usleep(SRS_FORWARDER_SLEEP_MS * 1000);
-
-		if ((ret = open_socket()) != ERROR_SUCCESS) {
-			srs_warn("forward cycle reopen failed, ignored and retry, ret=%d", ret);
-		} else {
-			srs_info("forward cycle reopen success");
-		}
-	}
-	srs_trace("forward cycle finished");
-}
-
-void* SrsForwarder::forward_thread(void* arg)
-{
-	SrsForwarder* obj = (SrsForwarder*)arg;
-	srs_assert(obj != NULL);
-	
-	obj->loop = true;
-	obj->forward_cycle();
-	
-	return NULL;
-}
-

@@ -32,7 +32,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <string>
 #include <vector>
 
-#include <st.h>
+#include <srs_core_thread.hpp>
 
 class SrsSharedPtrMessage;
 class SrsOnMetaDataPacket;
@@ -42,7 +42,7 @@ class SrsRequest;
 /**
 * forward the stream to other servers.
 */
-class SrsForwarder
+class SrsForwarder : public ISrsThreadHandler
 {
 private:
 	std::string app;
@@ -53,8 +53,7 @@ private:
 	int port;
 private:
 	st_netfd_t stfd;
-	st_thread_t tid;
-	bool loop;
+	SrsThread* pthread;
 private:
 	SrsRtmpClient* client;
 	std::vector<SrsSharedPtrMessage*> msgs;
@@ -67,14 +66,13 @@ public:
 	virtual int on_meta_data(SrsSharedPtrMessage* metadata);
 	virtual int on_audio(SrsSharedPtrMessage* msg);
 	virtual int on_video(SrsSharedPtrMessage* msg);
+// interface ISrsThreadHandler.
+public:
+	virtual int cycle();
 private:
-	virtual int open_socket();
+	virtual void close_underlayer_socket();
 	virtual int connect_server();
-private:
-	virtual int cycle();
 	virtual int forward();
-	virtual void forward_cycle();
-	static void* forward_thread(void* arg);
 };
 
 #endif

@@ -184,15 +184,7 @@ void SrsHttpClient::disconnect()
 {
 	connected = false;
 	
-	if (stfd) {
-		int fd = st_netfd_fileno(stfd);
-		st_netfd_close(stfd);
-		stfd = NULL;
-		
-		// st does not close it sometimes, 
-		// close it manually.
-		::close(fd);
-	}
+	srs_close_stfd(stfd);
 }
 
 int SrsHttpClient::connect(SrsHttpUri* uri)

@@ -36,7 +36,6 @@ class SrsSocket;
 
 #include <string>
 
-#include <st.h>
 #include <http_parser.h>
 
 #define SRS_HTTP_HEADER_BUFFER		1024

@@ -29,8 +29,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <string>
 #include <map>
 
-#include <st.h>
-
 ILogContext::ILogContext()
 {
 }

@@ -33,8 +33,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <map>
 #include <string>
 
-#include <st.h>
-
 #include <srs_core_log.hpp>
 #include <srs_core_error.hpp>
 

@@ -32,8 +32,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 #include <string>
 
-#include <st.h>
-
 class SrsProtocol;
 class ISrsMessage;
 class SrsCommonMessage;

@@ -30,8 +30,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 #include <algorithm>
 
-#include <st.h>
-
 #include <srs_core_log.hpp>
 #include <srs_core_error.hpp>
 #include <srs_core_client.hpp>
@@ -49,23 +47,15 @@ SrsListener::SrsListener(SrsServer* _server, SrsListenerType _type)
 	server = _server;
 	type = _type;
 
-	tid = NULL;
-	loop = false;
+	pthread = new SrsThread(this, 0);
 }
 
 SrsListener::~SrsListener()
 {
-	if (stfd) {
-		st_netfd_close(stfd);
-		stfd = NULL;
-	}
+	srs_close_stfd(stfd);
 	
-	if (tid) {
-		loop = false;
-		st_thread_interrupt(tid);
-		st_thread_join(tid, NULL);
-		tid = NULL;
-	}
+	pthread->stop();
+	srs_freep(pthread);
 	
 	// st does not close it sometimes, 
 	// close it manually.
@@ -118,8 +108,7 @@ int SrsListener::listen(int _port)
     }
     srs_verbose("st open socket success. fd=%d", fd);
     
-    if ((tid = st_thread_create(listen_thread, this, 1, 0)) == NULL) {
-        ret = ERROR_ST_CREATE_LISTEN_THREAD;
+    if ((ret = pthread->start()) != ERROR_SUCCESS) {
         srs_error("st_thread_create listen thread error. ret=%d", ret);
         return ret;
     }
@@ -130,41 +119,32 @@ int SrsListener::listen(int _port)
 	return ret;
 }
 
-void SrsListener::listen_cycle()
+void SrsListener::on_enter_loop()
 {
-	int ret = ERROR_SUCCESS;
-	
-	log_context->generate_id();
 	srs_trace("listen cycle start, port=%d, type=%d, fd=%d", port, type, fd);
+}
+
+int SrsListener::cycle()
+{
+	int ret = ERROR_SUCCESS;
 	
-	while (loop) {
     st_netfd_t client_stfd = st_accept(stfd, NULL, NULL, ST_UTIME_NO_TIMEOUT);
     
     if(client_stfd == NULL){
         // ignore error.
         srs_warn("ignore accept thread stoppped for accept client error");
-	        continue;
+        return ret;
     }
     srs_verbose("get a client. fd=%d", st_netfd_fileno(client_stfd));
 	
 	if ((ret = server->accept_client(type, client_stfd)) != ERROR_SUCCESS) {
 		srs_warn("accept client error. ret=%d", ret);
-			continue;
+		return ret;
 	}
 	
 	srs_verbose("accept client finished. conns=%d, ret=%d", (int)conns.size(), ret);
-	}
-}
 	
-void* SrsListener::listen_thread(void* arg)
-{
-	SrsListener* obj = (SrsListener*)arg;
-	srs_assert(obj != NULL);
-	
-	obj->loop = true;
-	obj->listen_cycle();
-	
-	return NULL;
+	return ret;
 }
 
 SrsServer::SrsServer()
@@ -312,8 +292,7 @@ int SrsServer::accept_client(SrsListenerType type, st_netfd_t client_stfd)
 		srs_error("exceed the max connections, drop client: "
 			"clients=%d, max=%d, fd=%d", (int)conns.size(), max_connections, fd);
 			
-		st_netfd_close(client_stfd);
-		::close(fd);
+		srs_close_stfd(client_stfd);
 		
 		return ret;
 	}

@@ -32,9 +32,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 #include <vector>
 
-#include <st.h>
-
 #include <srs_core_reload.hpp>
+#include <srs_core_thread.hpp>
 
 class SrsServer;
 class SrsConnection;
@@ -45,7 +44,7 @@ enum SrsListenerType
 	SrsListenerApi
 };
 
-class SrsListener
+class SrsListener : public ISrsThreadHandler
 {
 public:
 	SrsListenerType type;
@@ -54,16 +53,16 @@ private:
 	st_netfd_t stfd;
 	int port;
 	SrsServer* server;
-	st_thread_t tid;
-	bool loop;
+	SrsThread* pthread;
 public:
 	SrsListener(SrsServer* _server, SrsListenerType _type);
 	virtual ~SrsListener();
 public:
 	virtual int listen(int port);
-private:
-	virtual void listen_cycle();
-	static void* listen_thread(void* arg);
+// interface ISrsThreadHandler.
+public:
+	virtual void on_enter_loop();
+	virtual int cycle();
 };
 
 class SrsServer : public SrsReloadHandler

@@ -30,8 +30,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 #include <srs_core.hpp>
 
-#include <st.h>
-
 /**
 * the socket provides TCP socket over st,
 * that is, the sync socket mechanism.

@@ -425,6 +425,8 @@ int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata
 	
 	metadata->metadata->set("server", new SrsAmf0String(
 		RTMP_SIG_SRS_KEY" "RTMP_SIG_SRS_VERSION" ("RTMP_SIG_SRS_URL_SHORT")"));
+	metadata->metadata->set("contributor", 
+		new SrsAmf0String(RTMP_SIG_SRS_CONTRIBUTOR));
 	
 	SrsAmf0Any* prop = NULL;
 	if ((prop = metadata->metadata->get_property("audiosamplerate")) != NULL) {

+/*
+The MIT License (MIT)
+
+Copyright (c) 2013 winlin
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+*/
+
+#include <srs_core_thread.hpp>
+
+#include <srs_core_error.hpp>
+#include <srs_core_log.hpp>
+
+ISrsThreadHandler::ISrsThreadHandler()
+{
+}
+
+ISrsThreadHandler::~ISrsThreadHandler()
+{
+}
+
+void ISrsThreadHandler::on_enter_loop()
+{
+}
+
+int ISrsThreadHandler::on_before_cycle()
+{
+	int ret = ERROR_SUCCESS;
+	return ret;
+}
+
+int ISrsThreadHandler::on_end_cycle()
+{
+	int ret = ERROR_SUCCESS;
+	return ret;
+}
+
+void ISrsThreadHandler::on_leave_loop()
+{
+}
+
+SrsThread::SrsThread(ISrsThreadHandler* thread_handler, int64_t interval_ms)
+{
+	handler = thread_handler;
+	cycle_interval_milliseconds = interval_ms;
+	
+	tid = NULL;
+	loop = false;
+}
+
+SrsThread::~SrsThread()
+{
+	stop();
+}
+
+int SrsThread::start()
+{
+	int ret = ERROR_SUCCESS;
+	
+    if(tid) {
+        srs_info("thread already running.");
+        return ret;
+    }
+    
+    if((tid = st_thread_create(thread_fun, this, 1, 0)) == NULL){
+		ret = ERROR_ST_CREATE_CYCLE_THREAD;
+        srs_error("st_thread_create failed. ret=%d", ret);
+        return ret;
+    }
+    
+    return ret;
+}
+
+void SrsThread::stop()
+{
+	if (tid) {
+		loop = false;
+		
+		// the interrupt will cause the socket to read/write error,
+		// which will terminate the cycle thread.
+		st_thread_interrupt(tid);
+		
+		// wait the thread to exit.
+		st_thread_join(tid, NULL);
+		
+		tid = NULL;
+	}
+}
+
+void SrsThread::thread_cycle()
+{
+	int ret = ERROR_SUCCESS;
+	
+	srs_assert(handler);
+	
+	log_context->generate_id();
+	srs_trace("thread cycle start");
+
+	handler->on_end_cycle();
+	
+	loop = true;
+	while (loop) {
+		if ((ret = handler->on_before_cycle()) != ERROR_SUCCESS) {
+			srs_warn("thread on before cycle failed, ignored and retry, ret=%d", ret);
+			goto failed;
+		}
+		srs_info("thread on before cycle success");
+		
+		if ((ret = handler->cycle()) != ERROR_SUCCESS) {
+			srs_warn("thread cycle failed, ignored and retry, ret=%d", ret);
+			goto failed;
+		}
+		srs_info("thread cycle success");
+		
+		if ((ret = handler->on_end_cycle()) != ERROR_SUCCESS) {
+			srs_warn("thread on end cycle failed, ignored and retry, ret=%d", ret);
+			goto failed;
+		}
+		srs_info("thread on end cycle success");
+
+failed:
+		if (!loop) {
+			break;
+		}
+		
+		st_usleep(cycle_interval_milliseconds * 1000);
+	}
+	
+	handler->on_leave_loop();
+	srs_trace("thread cycle finished");
+}
+
+void* SrsThread::thread_fun(void* arg)
+{
+	SrsThread* obj = (SrsThread*)arg;
+	srs_assert(obj);
+	
+	obj->thread_cycle();
+	
+	return NULL;
+}
\ No newline at end of file

+/*
+The MIT License (MIT)
+
+Copyright (c) 2013 winlin
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+*/
+
+#ifndef SRS_CORE_THREAD_HPP
+#define SRS_CORE_THREAD_HPP
+
+/*
+#include <srs_core_thread.hpp>
+*/
+#include <srs_core.hpp>
+
+/**
+* the handler for the thread, callback interface.
+* the thread model defines as:
+* 	handler->on_enter_loop()
+* 	while loop:
+*		handler->on_before_cycle()
+*		handler->cycle()
+*		handler->on_end_cycle()
+*		if !loop then break for user stop thread.
+*		sleep(CycleIntervalMilliseconds)
+* 	handler->on_leave_loop()
+* when stop, the thread will interrupt the st_thread,
+* which will cause the socket to return error and 
+* terminate the cycle thread.
+*/
+class ISrsThreadHandler
+{
+public:
+	ISrsThreadHandler();
+	virtual ~ISrsThreadHandler();
+public:
+	virtual void on_enter_loop();
+	virtual int on_before_cycle();
+	virtual int cycle() = 0;
+	virtual int on_end_cycle();
+	virtual void on_leave_loop();
+};
+
+/**
+* provides servies from st_thread_t,
+* for common thread usage.
+*/
+class SrsThread
+{
+private:
+	st_thread_t tid;
+	bool loop;
+private:
+	ISrsThreadHandler* handler;
+	int64_t cycle_interval_milliseconds;
+public:
+	/**
+	* initialize the thread.
+	* @param thread_handler, the cycle handler for the thread.
+	* @param interval_ms, the sleep interval when cycle finished.
+	*/
+	SrsThread(ISrsThreadHandler* thread_handler, int64_t interval_ms);
+	virtual ~SrsThread();
+public:
+	/**
+	* start the thread, invoke the cycle of handler util
+	* user stop the thread.
+	* @remark ignore any error of cycle of handler.
+	* @remark user can start multiple times, ignore if already started.
+	*/
+	virtual int start();
+	/**
+	* stop the thread, wait for the thread to terminate.
+	* @remark user can stop multiple times, ignore if already stopped.
+	*/
+	virtual void stop();
+private:
+	virtual void thread_cycle();
+	static void* thread_fun(void* arg);
+};
+
+#endif

@@ -6,52 +6,54 @@ file
     core readonly separator,
     ..\core\srs_core.hpp,
     ..\core\srs_core.cpp,
-	..\core\srs_core_error.hpp,
-	..\core\srs_core_error.cpp,
+    ..\core\srs_core_amf0.hpp,
+    ..\core\srs_core_amf0.cpp,
     ..\core\srs_core_autofree.hpp,
     ..\core\srs_core_autofree.cpp,
-	..\core\srs_core_server.hpp,
-	..\core\srs_core_server.cpp,
-	..\core\srs_core_reload.hpp,
-	..\core\srs_core_reload.cpp,
+    ..\core\srs_core_buffer.hpp,
+    ..\core\srs_core_buffer.cpp,
+    ..\core\srs_core_client.hpp,
+    ..\core\srs_core_client.cpp,
+    ..\core\srs_core_codec.hpp,
+    ..\core\srs_core_codec.cpp,
     ..\core\srs_core_config.hpp,
     ..\core\srs_core_config.cpp,
-	..\core\srs_core_refer.hpp,
-	..\core\srs_core_refer.cpp,
     ..\core\srs_core_conn.hpp,
     ..\core\srs_core_conn.cpp,
-	..\core\srs_core_client.hpp,
-	..\core\srs_core_client.cpp,
-        ..\core\srs_core_http.hpp,
-        ..\core\srs_core_http.cpp,
-	..\core\srs_core_source.hpp,
-	..\core\srs_core_source.cpp,
-	..\core\srs_core_forward.hpp,
-	..\core\srs_core_forward.cpp,
     ..\core\srs_core_encoder.hpp,
     ..\core\srs_core_encoder.cpp,
-	..\core\srs_core_hls.hpp,
-	..\core\srs_core_hls.cpp,
-	..\core\srs_core_codec.hpp,
-	..\core\srs_core_codec.cpp,
-	..\core\srs_core_rtmp.hpp,
-	..\core\srs_core_rtmp.cpp,
+    ..\core\srs_core_error.hpp,
+    ..\core\srs_core_error.cpp,
+    ..\core\srs_core_forward.hpp,
+    ..\core\srs_core_forward.cpp,
     ..\core\srs_core_handshake.hpp,
     ..\core\srs_core_handshake.cpp,
+    ..\core\srs_core_hls.hpp,
+    ..\core\srs_core_hls.cpp,
+    ..\core\srs_core_http.hpp,
+    ..\core\srs_core_http.cpp,
+    ..\core\srs_core_log.hpp,
+    ..\core\srs_core_log.cpp,
+    ..\core\srs_core_pithy_print.hpp,
+    ..\core\srs_core_pithy_print.cpp,
     ..\core\srs_core_protocol.hpp,
     ..\core\srs_core_protocol.cpp,
-	..\core\srs_core_amf0.hpp,
-	..\core\srs_core_amf0.cpp,
+    ..\core\srs_core_refer.hpp,
+    ..\core\srs_core_refer.cpp,
+    ..\core\srs_core_reload.hpp,
+    ..\core\srs_core_reload.cpp,
+    ..\core\srs_core_rtmp.hpp,
+    ..\core\srs_core_rtmp.cpp,
+    ..\core\srs_core_thread.hpp,
+    ..\core\srs_core_thread.cpp,
+    ..\core\srs_core_server.hpp,
+    ..\core\srs_core_server.cpp,
     ..\core\srs_core_stream.hpp,
     ..\core\srs_core_stream.cpp,
     ..\core\srs_core_socket.hpp,
     ..\core\srs_core_socket.cpp,
-	..\core\srs_core_buffer.hpp,
-	..\core\srs_core_buffer.cpp,
-	..\core\srs_core_pithy_print.hpp,
-	..\core\srs_core_pithy_print.cpp,
-	..\core\srs_core_log.hpp,
-	..\core\srs_core_log.cpp,
+    ..\core\srs_core_source.hpp,
+    ..\core\srs_core_source.cpp,
     research readonly separator,
     ..\..\research\ts_info.cc;