update encoder framework

@@ -117,7 +117,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.7, 2013-11-30, add transcoding params to config.
+* v0.7, 2013-11-30, support live stream transcoding by ffmpeg.
 * v0.7, 2013-11-30, support --with/without -ffmpeg, build ffmpeg-2.1.
 * v0.7, 2013-11-30, support --with/without -ffmpeg, build ffmpeg-2.1.
 * v0.7, 2013-11-30, add ffmpeg-2.1, x264-core138, lame-3.99.5, libaacplus-2.0.2.
 * v0.7, 2013-11-30, add ffmpeg-2.1, x264-core138, lame-3.99.5, libaacplus-2.0.2.
 * v0.6, 2013-11-29, v0.6 released. 16094 lines.
 * v0.6, 2013-11-29, v0.6 released. 16094 lines.

@@ -568,7 +568,7 @@ SrsConfDirective* SrsConfig::get_vhost_enabled(std::string vhost)
 	return conf->get("enabled");
 	return conf->get("enabled");
 }
 }
 
 
-SrsConfDirective* SrsConfig::get_transcode(std::string vhost)
+SrsConfDirective* SrsConfig::get_transcode(std::string vhost, std::string scope)
 {
 {
 	SrsConfDirective* conf = get_vhost(vhost);
 	SrsConfDirective* conf = get_vhost(vhost);
 
 
@@ -576,7 +576,16 @@ SrsConfDirective* SrsConfig::get_transcode(std::string vhost)
 		return NULL;
 		return NULL;
 	}
 	}
 	
 	
-	return conf->get("transcode");
+	SrsConfDirective* transcode = conf->get("transcode");
+	if (!transcode) {
+		return NULL;
+	}
+	
+	if (transcode->arg0() == scope) {
+		return transcode;
+	}
+	
+	return NULL;
 }
 }
 
 
 SrsConfDirective* SrsConfig::get_gop_cache(std::string vhost)
 SrsConfDirective* SrsConfig::get_gop_cache(std::string vhost)

@@ -114,7 +114,7 @@ public:
 	virtual int parse_options(int argc, char** argv);
 	virtual int parse_options(int argc, char** argv);
 	virtual SrsConfDirective* get_vhost(std::string vhost);
 	virtual SrsConfDirective* get_vhost(std::string vhost);
 	virtual SrsConfDirective* get_vhost_enabled(std::string vhost);
 	virtual SrsConfDirective* get_vhost_enabled(std::string vhost);
-	virtual SrsConfDirective* get_transcode(std::string vhost);
+	virtual SrsConfDirective* get_transcode(std::string vhost, std::string scope);
 	virtual SrsConfDirective* get_gop_cache(std::string vhost);
 	virtual SrsConfDirective* get_gop_cache(std::string vhost);
 	virtual SrsConfDirective* get_forward(std::string vhost);
 	virtual SrsConfDirective* get_forward(std::string vhost);
 	virtual SrsConfDirective* get_hls(std::string vhost);
 	virtual SrsConfDirective* get_hls(std::string vhost);

@@ -25,22 +25,89 @@ 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_log.hpp>
 #include <srs_core_log.hpp>
+#include <srs_core_config.hpp>
+
+#define SRS_ENCODER_SLEEP_MS 2000
 
 
 SrsEncoder::SrsEncoder()
 SrsEncoder::SrsEncoder()
 {
 {
+	tid = NULL;
+	loop = false;
 }
 }
 
 
 SrsEncoder::~SrsEncoder()
 SrsEncoder::~SrsEncoder()
 {
 {
+	on_unpublish();
 }
 }
 
 
-int SrsEncoder::on_publish(std::string vhost, std::string app, std::string stream)
+int SrsEncoder::on_publish(std::string _vhost, std::string _app, std::string _stream)
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
+
+	vhost = _vhost;
+	app = _app;
+	stream = _stream;
+    
+    srs_assert(!tid);
+    if((tid = st_thread_create(encoder_thread, this, 1, 0)) == NULL){
+		ret = ERROR_ST_CREATE_FORWARD_THREAD;
+        srs_error("st_thread_create failed. ret=%d", ret);
+        return ret;
+    }
+    
 	return ret;
 	return ret;
 }
 }
 
 
 void SrsEncoder::on_unpublish()
 void SrsEncoder::on_unpublish()
 {
 {
+	if (tid) {
+		loop = false;
+		st_thread_interrupt(tid);
+		st_thread_join(tid, NULL);
+		tid = NULL;
+	}
+}
+
+int SrsEncoder::cycle()
+{
+	int ret = ERROR_SUCCESS;
+	return ret;
+}
+
+void SrsEncoder::encoder_cycle()
+{
+	int ret = ERROR_SUCCESS;
+	
+	log_context->generate_id();
+	srs_trace("encoder cycle start");
+	
+	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;
+		}
+		
+		st_usleep(SRS_ENCODER_SLEEP_MS * 1000);
+	}
+	
+	// TODO: kill ffmpeg when finished and it alive
+	
+	srs_trace("encoder cycle finished");
+}
+
+void* SrsEncoder::encoder_thread(void* arg)
+{
+	SrsEncoder* obj = (SrsEncoder*)arg;
+	srs_assert(obj != NULL);
+	
+	obj->loop = true;
+	obj->encoder_cycle();
+	
+	return NULL;
 }
 }
 
 

@@ -31,14 +31,27 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 
 #include <string>
 #include <string>
 
 
+#include <st.h>
+
 class SrsEncoder
 class SrsEncoder
 {
 {
+private:
+	std::string vhost;
+	std::string app;
+	std::string stream;
+private:
+	st_thread_t tid;
+	bool loop;
 public:
 public:
 	SrsEncoder();
 	SrsEncoder();
 	virtual ~SrsEncoder();
 	virtual ~SrsEncoder();
 public:
 public:
 	virtual int on_publish(std::string vhost, std::string app, std::string stream);
 	virtual int on_publish(std::string vhost, std::string app, std::string stream);
 	virtual void on_unpublish();
 	virtual void on_unpublish();
+private:
+	virtual int cycle();
+	virtual void encoder_cycle();
+	static void* encoder_thread(void* arg);
 };
 };
 
 
 #endif
 #endif

@@ -43,10 +43,11 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 SrsForwarder::SrsForwarder()
 SrsForwarder::SrsForwarder()
 {
 {
 	client = NULL;
 	client = NULL;
-	tid = NULL;
 	stfd = NULL;
 	stfd = NULL;
-	loop = false;
 	stream_id = 0;
 	stream_id = 0;
+	
+	tid = NULL;
+	loop = false;
 }
 }
 
 
 SrsForwarder::~SrsForwarder()
 SrsForwarder::~SrsForwarder()
@@ -221,7 +222,7 @@ std::string SrsForwarder::parse_server(std::string host)
     return ipv4;
     return ipv4;
 }
 }
 
 
-int SrsForwarder::forward_cycle_imp()
+int SrsForwarder::cycle()
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
 
 
@@ -316,7 +317,7 @@ void SrsForwarder::forward_cycle()
 	srs_trace("forward cycle start");
 	srs_trace("forward cycle start");
 	
 	
 	while (loop) {
 	while (loop) {
-		if ((ret = forward_cycle_imp()) != ERROR_SUCCESS) {
+		if ((ret = cycle()) != ERROR_SUCCESS) {
 			srs_warn("forward cycle failed, ignored and retry, ret=%d", ret);
 			srs_warn("forward cycle failed, ignored and retry, ret=%d", ret);
 		} else {
 		} else {
 			srs_info("forward cycle success, retry");
 			srs_info("forward cycle success, retry");

@@ -71,7 +71,7 @@ private:
 	virtual int connect_server();
 	virtual int connect_server();
 	std::string parse_server(std::string host);
 	std::string parse_server(std::string host);
 private:
 private:
-	virtual int forward_cycle_imp();
+	virtual int cycle();
 	virtual int forward();
 	virtual int forward();
 	virtual void forward_cycle();
 	virtual void forward_cycle();
 	static void* forward_thread(void* arg);
 	static void* forward_thread(void* arg);