add forward st thread

@@ -23,19 +23,53 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 
 #include <srs_core_forward.hpp>
 #include <srs_core_forward.hpp>
 
 
+#include <stdlib.h>
+
 #include <srs_core_error.hpp>
 #include <srs_core_error.hpp>
+#include <srs_core_rtmp.hpp>
+#include <srs_core_log.hpp>
 
 
 SrsForwarder::SrsForwarder()
 SrsForwarder::SrsForwarder()
 {
 {
+	client = new SrsRtmpClient();
+	port = 1935;
+	tid = NULL;
+	loop = false;
 }
 }
 
 
 SrsForwarder::~SrsForwarder()
 SrsForwarder::~SrsForwarder()
 {
 {
+	srs_freep(client);
+	
+	if (tid) {
+		loop = false;
+		st_thread_interrupt(tid);
+		st_thread_join(tid, NULL);
+		tid = NULL;
+	}
 }
 }
 
 
 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)
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
+	
+	tc_url = "rtmp://";
+	tc_url += vhost;
+	tc_url += "/";
+	tc_url += app;
+	
+	stream_name = stream;
+	server = forward_server;
+	
+	size_t pos = forward_server.find(":");
+	if (pos != std::string::npos) {
+		port = ::atoi(forward_server.substr(pos + 1).c_str());
+		server = forward_server.substr(0, pos);
+	}
+	
+	srs_trace("forward stream=%s, tcUrl=%s to server=%s, port=%d",
+		stream_name.c_str(), tc_url.c_str(), server.c_str(), port);
+	
 	return ret;
 	return ret;
 }
 }
 
 

@@ -33,12 +33,21 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 
 class SrsSharedPtrMessage;
 class SrsSharedPtrMessage;
 class SrsOnMetaDataPacket;
 class SrsOnMetaDataPacket;
+class SrsRtmpClient;
 
 
 /**
 /**
 * forward the stream to other servers.
 * forward the stream to other servers.
 */
 */
 class SrsForwarder
 class SrsForwarder
 {
 {
+private:
+	std::string tc_url;
+	std::string stream_name;
+	std::string server;
+	int port;
+	SrsRtmpClient* client;
+	st_thread_t tid;
+	bool loop;
 public:
 public:
 	SrsForwarder();
 	SrsForwarder();
 	virtual ~SrsForwarder();
 	virtual ~SrsForwarder();

@@ -163,6 +163,24 @@ SrsResponse::~SrsResponse()
 {
 {
 }
 }
 
 
+SrsRtmpClient::SrsRtmpClient()
+{
+	stfd = NULL;
+}
+
+SrsRtmpClient::~SrsRtmpClient()
+{
+	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);
+	}
+}
+
 SrsRtmp::SrsRtmp(st_netfd_t client_stfd)
 SrsRtmp::SrsRtmp(st_netfd_t client_stfd)
 {
 {
 	protocol = new SrsProtocol(client_stfd);
 	protocol = new SrsProtocol(client_stfd);

@@ -93,6 +93,18 @@ enum SrsClientType
 };
 };
 
 
 /**
 /**
+* implements the client role protocol.
+*/
+class SrsRtmpClient
+{
+private:
+	st_netfd_t stfd;
+public:
+	SrsRtmpClient();
+	virtual ~SrsRtmpClient();
+};
+
+/**
 * the rtmp provices rtmp-command-protocol services,
 * the rtmp provices rtmp-command-protocol services,
 * a high level protocol, media stream oriented services,
 * a high level protocol, media stream oriented services,
 * such as connect to vhost/app, play stream, get audio/video data.
 * such as connect to vhost/app, play stream, get audio/video data.