fix #367, support nginx-rtmp exec. 3.0.1

README.md

@@ -219,6 +219,7 @@ Supported operating systems and hardware:
 1. [experiment] Support push flv stream over HTTP POST to SRS, read [wiki]([CN][v2_CN_Streamer2], [EN][v2_EN_Streamer2]).
 1. [experiment] Support [srs-dolphin][srs-dolphin], the multiple-process SRS.
 1. [experiment] Support [remote console](http://ossrs.net:1985/console), read [srs-ngb][srs-ngb].
+1. Support nginx-rtmp style exec, read [bug #367][bug #367].
 1. [no-plan] Support <500ms latency, FRSC(Fast RTMP-compatible Stream Channel tech).
 1. [no-plan] Support RTMP 302 redirect [bug #92][bug #92].
 1. [no-plan] Support multiple processes, for both origin and edge
@@ -344,6 +345,7 @@ Remark:
 
 ## History
 
+* v3.0, 2015-08-25, fix [#367](https://github.com/simple-rtmp-server/srs/issues/367), support nginx-rtmp exec. 3.0.1
 * <strong>v2.0, 2015-08-23, [2.0 alpha(2.0.185)](https://github.com/simple-rtmp-server/srs/releases/tag/2.0a0) released. 89022 lines.</strong>
 * v2.0, 2015-08-22, HTTP API support JSONP by specifies the query string callback=xxx.
 * v2.0, 2015-08-20, fix [#380](https://github.com/simple-rtmp-server/srs/issues/380), srs-librtmp send sequence header when sps or pps changed.
@@ -998,6 +1000,7 @@ Winlin
 [bug #304]: https://github.com/simple-rtmp-server/srs/issues/304
 [bug #133]: https://github.com/simple-rtmp-server/srs/issues/133
 [bug #92]: https://github.com/simple-rtmp-server/srs/issues/92
+[bug #367]: https://github.com/simple-rtmp-server/srs/issues/367
 
 
 [contact]: https://github.com/simple-rtmp-server/srs/wiki/v1_CN_Contact

trunk/conf/exec.conf

@@ -0,0 +1,12 @@
+# the config for srs to support nginx-rtmp exec.
+# @see https://github.com/simple-rtmp-server/srs/wiki/v2_CN_NgExec
+# @see full.conf for detail config.
+
+listen              1935;
+max_connections     1000;
+vhost __defaultVhost__ {
+    exec {
+        enabled     on;
+        publish     ./objs/ffmpeg/bin/ffmpeg -f flv -i [url] -c copy -y ./[stream].flv;
+    }
+}

trunk/conf/full.conf

@@ -934,6 +934,9 @@ vhost exec.srs.com {
         #       [engine] the tanscode engine name.
         # other variables for exec only:
         #       [url] the rtmp url which trigger the publish.
+        #       [tcUrl] the client request tcUrl.
+        #       [swfUrl] the client request swfUrl.
+        #       [pageUrl] the client request pageUrl.
         # @remark empty to ignore this exec.
         publish     ./objs/ffmpeg/bin/ffmpeg -f flv -i [url] -c copy -y ./[stream].flv;
     }

trunk/configure

@@ -176,7 +176,7 @@ if [ $SRS_EXPORT_LIBRTMP_PROJECT = NO ]; then
             "srs_app_heartbeat" "srs_app_empty" "srs_app_http_client" "srs_app_http_static"
             "srs_app_recv_thread" "srs_app_security" "srs_app_statistic" "srs_app_hds"
             "srs_app_mpegts_udp" "srs_app_rtsp" "srs_app_listener" "srs_app_async_call"
-            "srs_app_caster_flv" "srs_app_process")
+            "srs_app_caster_flv" "srs_app_process" "srs_app_ng_exec")
     DEFINES=""
     # add each modules for app
     for SRS_MODULE in ${SRS_MODULES[*]}; do

trunk/ide/srs_upp/srs_upp.upp

@@ -117,6 +117,8 @@ file
 	../../src/app/srs_app_log.cpp,
 	../../src/app/srs_app_mpegts_udp.hpp,
 	../../src/app/srs_app_mpegts_udp.cpp,
+        ../../src/app/srs_app_ng_exec.hpp,
+        ../../src/app/srs_app_ng_exec.cpp,
         ../../src/app/srs_app_process.hpp,
         ../../src/app/srs_app_process.cpp,
 	../../src/app/srs_app_recv_thread.hpp,

trunk/ide/srs_xcode/srs_xcode.xcodeproj/project.pbxproj

@@ -79,6 +79,7 @@
 		3C36DB5B1ABD1CB90066CCAF /* srs_lib_bandwidth.cpp in Sources */ = {isa = PBXBuildFile; fileRef = 3C36DB551ABD1CB90066CCAF /* srs_lib_bandwidth.cpp */; };
 		3C36DB5C1ABD1CB90066CCAF /* srs_lib_simple_socket.cpp in Sources */ = {isa = PBXBuildFile; fileRef = 3C36DB571ABD1CB90066CCAF /* srs_lib_simple_socket.cpp */; };
 		3C36DB5D1ABD1CB90066CCAF /* srs_librtmp.cpp in Sources */ = {isa = PBXBuildFile; fileRef = 3C36DB591ABD1CB90066CCAF /* srs_librtmp.cpp */; };
+		3C4AB9331B8C9148006627D3 /* srs_app_ng_exec.cpp in Sources */ = {isa = PBXBuildFile; fileRef = 3C4AB9311B8C9148006627D3 /* srs_app_ng_exec.cpp */; };
 		3C4F97121B8B466D00FF0E46 /* srs_app_process.cpp in Sources */ = {isa = PBXBuildFile; fileRef = 3C4F97101B8B466D00FF0E46 /* srs_app_process.cpp */; };
 		3C5265B41B241BF0009CA186 /* srs_core_mem_watch.cpp in Sources */ = {isa = PBXBuildFile; fileRef = 3C5265B21B241BF0009CA186 /* srs_core_mem_watch.cpp */; };
 		3C663F0F1AB0155100286D8B /* srs_aac_raw_publish.c in Sources */ = {isa = PBXBuildFile; fileRef = 3C663F021AB0155100286D8B /* srs_aac_raw_publish.c */; };
@@ -329,6 +330,9 @@
 		3C36DB581ABD1CB90066CCAF /* srs_lib_simple_socket.hpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.h; name = srs_lib_simple_socket.hpp; path = ../../../src/libs/srs_lib_simple_socket.hpp; sourceTree = "<group>"; };
 		3C36DB591ABD1CB90066CCAF /* srs_librtmp.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; name = srs_librtmp.cpp; path = ../../../src/libs/srs_librtmp.cpp; sourceTree = "<group>"; };
 		3C36DB5A1ABD1CB90066CCAF /* srs_librtmp.hpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.h; name = srs_librtmp.hpp; path = ../../../src/libs/srs_librtmp.hpp; sourceTree = "<group>"; };
+		3C4AB9311B8C9148006627D3 /* srs_app_ng_exec.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; name = srs_app_ng_exec.cpp; path = ../../../src/app/srs_app_ng_exec.cpp; sourceTree = "<group>"; };
+		3C4AB9321B8C9148006627D3 /* srs_app_ng_exec.hpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.h; name = srs_app_ng_exec.hpp; path = ../../../src/app/srs_app_ng_exec.hpp; sourceTree = "<group>"; };
+		3C4AB9341B8C9FF9006627D3 /* exec.conf */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = text; name = exec.conf; path = ../../../conf/exec.conf; sourceTree = "<group>"; };
 		3C4F97101B8B466D00FF0E46 /* srs_app_process.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; name = srs_app_process.cpp; path = ../../../src/app/srs_app_process.cpp; sourceTree = "<group>"; };
 		3C4F97111B8B466D00FF0E46 /* srs_app_process.hpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.h; name = srs_app_process.hpp; path = ../../../src/app/srs_app_process.hpp; sourceTree = "<group>"; };
 		3C5265B21B241BF0009CA186 /* srs_core_mem_watch.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; name = srs_core_mem_watch.cpp; path = ../../../src/core/srs_core_mem_watch.cpp; sourceTree = "<group>"; };
@@ -574,6 +578,8 @@
 				3C1232751AAE81D900CE8F6C /* srs_app_log.hpp */,
 				3C1232761AAE81D900CE8F6C /* srs_app_mpegts_udp.cpp */,
 				3C1232771AAE81D900CE8F6C /* srs_app_mpegts_udp.hpp */,
+				3C4AB9311B8C9148006627D3 /* srs_app_ng_exec.cpp */,
+				3C4AB9321B8C9148006627D3 /* srs_app_ng_exec.hpp */,
 				3C1232781AAE81D900CE8F6C /* srs_app_pithy_print.cpp */,
 				3C1232791AAE81D900CE8F6C /* srs_app_pithy_print.hpp */,
 				3C4F97101B8B466D00FF0E46 /* srs_app_process.cpp */,
@@ -676,6 +682,7 @@
 		3C1EE6AF1AB107EE00576EE9 /* conf */ = {
 			isa = PBXGroup;
 			children = (
+				3C4AB9341B8C9FF9006627D3 /* exec.conf */,
 				3C1EE6B01AB1080900576EE9 /* bandwidth.conf */,
 				3C1EE6B11AB1080900576EE9 /* console.conf */,
 				3C1EE6B21AB1080900576EE9 /* demo.19350.conf */,
@@ -902,6 +909,7 @@
 				3C689F981AB6AAAC00C9CEEE /* key.c in Sources */,
 				3C12329B1AAE81D900CE8F6C /* srs_app_ffmpeg.cpp in Sources */,
 				3C1232421AAE81A400CE8F6C /* srs_rtmp_amf0.cpp in Sources */,
+				3C4AB9331B8C9148006627D3 /* srs_app_ng_exec.cpp in Sources */,
 				3C1232AA1AAE81D900CE8F6C /* srs_app_pithy_print.cpp in Sources */,
 				3C12329C1AAE81D900CE8F6C /* srs_app_forward.cpp in Sources */,
 				3C1232251AAE814D00CE8F6C /* srs_kernel_file.cpp in Sources */,

trunk/src/app/srs_app_config.cpp

@@ -633,6 +633,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload atc success.", vhost.c_str());
             }
+            
             // gop_cache, only one per vhost
             if (!srs_directive_equals(new_vhost->get("gop_cache"), old_vhost->get("gop_cache"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -644,6 +645,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload gop_cache success.", vhost.c_str());
             }
+            
             // queue_length, only one per vhost
             if (!srs_directive_equals(new_vhost->get("queue_length"), old_vhost->get("queue_length"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -655,6 +657,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload queue_length success.", vhost.c_str());
             }
+            
             // time_jitter, only one per vhost
             if (!srs_directive_equals(new_vhost->get("time_jitter"), old_vhost->get("time_jitter"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -666,6 +669,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload time_jitter success.", vhost.c_str());
             }
+            
             // mix_correct, only one per vhost
             if (!srs_directive_equals(new_vhost->get("mix_correct"), old_vhost->get("mix_correct"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -677,6 +681,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload mix_correct success.", vhost.c_str());
             }
+            
             // forward, only one per vhost
             if (!srs_directive_equals(new_vhost->get("forward"), old_vhost->get("forward"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -688,6 +693,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload forward success.", vhost.c_str());
             }
+            
             // hls, only one per vhost
             // @remark, the hls_on_error directly support reload.
             if (!srs_directive_equals(new_vhost->get("hls"), old_vhost->get("hls"))) {
@@ -722,8 +728,21 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                         return ret;
                     }
                 }
-                srs_trace("vhost %s reload hlsdvrsuccess.", vhost.c_str());
+                srs_trace("vhost %s reload dvr success.", vhost.c_str());
             }
+            
+            // exec, only one per vhost
+            if (!srs_directive_equals(new_vhost->get("exec"), old_vhost->get("exec"))) {
+                for (it = subscribes.begin(); it != subscribes.end(); ++it) {
+                    ISrsReloadHandler* subscribe = *it;
+                    if ((ret = subscribe->on_reload_vhost_exec(vhost)) != ERROR_SUCCESS) {
+                        srs_error("vhost %s notify subscribes exec failed. ret=%d", vhost.c_str(), ret);
+                        return ret;
+                    }
+                }
+                srs_trace("vhost %s reload exec success.", vhost.c_str());
+            }
+            
             // mr, only one per vhost
             if (!srs_directive_equals(new_vhost->get("mr"), old_vhost->get("mr"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -735,6 +754,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload mr success.", vhost.c_str());
             }
+            
             // chunk_size, only one per vhost.
             if (!srs_directive_equals(new_vhost->get("chunk_size"), old_vhost->get("chunk_size"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -746,6 +766,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload chunk_size success.", vhost.c_str());
             }
+            
             // mw, only one per vhost
             if (!srs_directive_equals(new_vhost->get("mw_latency"), old_vhost->get("mw_latency"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -757,6 +778,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload mw success.", vhost.c_str());
             }
+            
             // smi(send_min_interval), only one per vhost
             if (!srs_directive_equals(new_vhost->get("send_min_interval"), old_vhost->get("send_min_interval"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -768,6 +790,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload smi success.", vhost.c_str());
             }
+            
             // tcp_nodelay, only one per vhost
             if (!srs_directive_equals(new_vhost->get("tcp_nodelay"), old_vhost->get("tcp_nodelay"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -779,6 +802,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload tcp_nodelay success.", vhost.c_str());
             }
+            
             // publish_1stpkt_timeout, only one per vhost
             if (!srs_directive_equals(new_vhost->get("publish_1stpkt_timeout"), old_vhost->get("publish_1stpkt_timeout"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -790,6 +814,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload p1stpt success.", vhost.c_str());
             }
+            
             // publish_normal_timeout, only one per vhost
             if (!srs_directive_equals(new_vhost->get("publish_normal_timeout"), old_vhost->get("publish_normal_timeout"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -801,6 +826,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload pnt success.", vhost.c_str());
             }
+            
             // min_latency, only one per vhost
             if (!srs_directive_equals(new_vhost->get("min_latency"), old_vhost->get("min_latency"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -812,6 +838,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload min_latency success.", vhost.c_str());
             }
+            
             // http, only one per vhost.
             if (!srs_directive_equals(new_vhost->get("http"), old_vhost->get("http"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -823,6 +850,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload http success.", vhost.c_str());
             }
+            
             // http_static, only one per vhost.
             // @remark, http_static introduced as alias of http.
             if (!srs_directive_equals(new_vhost->get("http_static"), old_vhost->get("http_static"))) {
@@ -835,6 +863,7 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload http_static success.", vhost.c_str());
             }
+            
             // http_remux, only one per vhost.
             if (!srs_directive_equals(new_vhost->get("http_remux"), old_vhost->get("http_remux"))) {
                 for (it = subscribes.begin(); it != subscribes.end(); ++it) {
@@ -846,10 +875,12 @@ int SrsConfig::reload_vhost(SrsConfDirective* old_root)
                 }
                 srs_trace("vhost %s reload http_remux success.", vhost.c_str());
             }
+            
             // transcode, many per vhost.
             if ((ret = reload_transcode(new_vhost, old_vhost)) != ERROR_SUCCESS) {
                 return ret;
             }
+            
             // ingest, many per vhost.
             if ((ret = reload_ingest(new_vhost, old_vhost)) != ERROR_SUCCESS) {
                 return ret;
@@ -1799,7 +1830,7 @@ int SrsConfig::check_config()
                 && n != "publish_1stpkt_timeout" && n != "publish_normal_timeout"
                 && n != "security" && n != "http_remux"
                 && n != "http" && n != "http_static"
-                && n != "hds"
+                && n != "hds" && n != "exec"
             ) {
                 ret = ERROR_SYSTEM_CONFIG_INVALID;
                 srs_error("unsupported vhost directive %s, ret=%d", n.c_str(), ret);
@@ -1817,11 +1848,19 @@ int SrsConfig::check_config()
                         return ret;
                     }
                 }
+            } else if (n == "exec") {
+                for (int j = 0; j < (int)conf->directives.size(); j++) {
+                    string m = conf->at(j)->name.c_str();
+                    if (m != "enabled" && m != "publish") {
+                        ret = ERROR_SYSTEM_CONFIG_INVALID;
+                        srs_error("unsupported vhost exec directive %s, ret=%d", m.c_str(), ret);
+                        return ret;
+                    }
+                }
             } else if (n == "mr") {
                 for (int j = 0; j < (int)conf->directives.size(); j++) {
                     string m = conf->at(j)->name.c_str();
-                    if (m != "enabled" && m != "latency"
+                    if (m != "enabled" && m != "latency") {
-                        ) {
                         ret = ERROR_SYSTEM_CONFIG_INVALID;
                         srs_error("unsupported vhost mr directive %s, ret=%d", m.c_str(), ret);
                         return ret;
@@ -1830,9 +1869,7 @@ int SrsConfig::check_config()
             } else if (n == "ingest") {
                 for (int j = 0; j < (int)conf->directives.size(); j++) {
                     string m = conf->at(j)->name.c_str();
-                    if (m != "enabled" && m != "input" && m != "ffmpeg"
+                    if (m != "enabled" && m != "input" && m != "ffmpeg" && m != "engine") {
-                        && m != "engine"
-                        ) {
                         ret = ERROR_SYSTEM_CONFIG_INVALID;
                         srs_error("unsupported vhost ingest directive %s, ret=%d", m.c_str(), ret);
                         return ret;
@@ -3347,6 +3384,52 @@ string SrsConfig::get_engine_output(SrsConfDirective* engine)
     return conf->arg0();
 }
 
+SrsConfDirective* SrsConfig::get_exec(string vhost)
+{
+    SrsConfDirective* conf = get_vhost(vhost);
+    if (!conf) {
+        return NULL;
+    }
+    
+    return conf->get("exec");
+}
+
+bool SrsConfig::get_exec_enabled(string vhost)
+{
+    static bool DEFAULT = false;
+    
+    SrsConfDirective* conf = get_exec(vhost);
+    if (!conf) {
+        return DEFAULT;
+    }
+    
+    conf = conf->get("enabled");
+    if (!conf || conf->arg0().empty()) {
+        return DEFAULT;
+    }
+    
+    return SRS_CONF_PERFER_FALSE(conf->arg0());
+}
+
+vector<SrsConfDirective*> SrsConfig::get_exec_publishs(string vhost)
+{
+    vector<SrsConfDirective*> eps;
+    
+    SrsConfDirective* conf = get_exec(vhost);
+    if (!conf) {
+        return eps;
+    }
+    
+    for (int i = 0; i < (int)conf->directives.size(); i++) {
+        SrsConfDirective* ep = conf->at(i);
+        if (ep->name == "publish") {
+            eps.push_back(ep);
+        }
+    }
+    
+    return eps;
+}
+
 vector<SrsConfDirective*> SrsConfig::get_ingesters(string vhost)
 {
     vector<SrsConfDirective*> ingeters;

trunk/src/app/srs_app_config.hpp

@@ -793,6 +793,21 @@ public:
     * @remark, we will use some variable, for instance, [vhost] to substitude with vhost.
     */
     virtual std::string         get_engine_output(SrsConfDirective* engine);
+// vhost exec secion
+private:
+    /**
+     * get the exec directive of vhost.
+     */
+    virtual SrsConfDirective*   get_exec(std::string vhost);
+public:
+    /**
+     * whether the exec is enabled of vhost.
+     */
+    virtual bool                get_exec_enabled(std::string vhost);
+    /**
+     * get all exec publish directives of vhost.
+     */
+    virtual std::vector<SrsConfDirective*>      get_exec_publishs(std::string vhost);
 // vhost ingest section
 public:
     /**

trunk/src/app/srs_app_encoder.cpp

@@ -208,8 +208,7 @@ int SrsEncoder::parse_ffmpeg(SrsRequest* req, SrsConfDirective* conf)
     
     // enabled
     if (!_srs_config->get_transcode_enabled(conf)) {
-        srs_trace("ignore the disabled transcode: %s", 
+        srs_trace("ignore the disabled transcode: %s", conf->arg0().c_str());
-            conf->arg0().c_str());
         return ret;
     }
     

trunk/src/app/srs_app_ng_exec.cpp

@@ -0,0 +1,214 @@
+/*
+ The MIT License (MIT)
+ 
+ Copyright (c) 2013-2015 SRS(simple-rtmp-server)
+ 
+ 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_app_ng_exec.hpp>
+
+using namespace std;
+
+#include <srs_kernel_error.hpp>
+#include <srs_kernel_log.hpp>
+#include <srs_app_config.hpp>
+#include <srs_rtmp_stack.hpp>
+#include <srs_app_pithy_print.hpp>
+#include <srs_app_process.hpp>
+#include <srs_kernel_utility.hpp>
+#include <srs_kernel_consts.hpp>
+#include <srs_rtmp_utility.hpp>
+
+// when error, ng-exec sleep for a while and retry.
+#define SRS_RTMP_EXEC_SLEEP_US (int64_t)(3*1000*1000LL)
+
+SrsNgExec::SrsNgExec()
+{
+    pthread = new SrsReusableThread("encoder", this, SRS_RTMP_EXEC_SLEEP_US);
+    pprint = SrsPithyPrint::create_exec();
+}
+
+SrsNgExec::~SrsNgExec()
+{
+    on_unpublish();
+    
+    srs_freep(pthread);
+    srs_freep(pprint);
+}
+
+int SrsNgExec::on_publish(SrsRequest* req)
+{
+    int ret = ERROR_SUCCESS;
+    
+    // when publish, parse the exec_publish.
+    if ((ret = parse_exec_publish(req)) != ERROR_SUCCESS) {
+        return ret;
+    }
+    
+    // start thread to run all processes.
+    if ((ret = pthread->start()) != ERROR_SUCCESS) {
+        srs_error("st_thread_create failed. ret=%d", ret);
+        return ret;
+    }
+    srs_trace("exec thread cid=%d, current_cid=%d", pthread->cid(), _srs_context->get_id());
+    
+    return ret;
+}
+
+void SrsNgExec::on_unpublish()
+{
+    pthread->stop();
+    clear_exec_publish();
+}
+
+int SrsNgExec::cycle()
+{
+    int ret = ERROR_SUCCESS;
+    
+    // ignore when no exec.
+    if (exec_publishs.empty()) {
+        return ret;
+    }
+    
+    std::vector<SrsProcess*>::iterator it;
+    for (it = exec_publishs.begin(); it != exec_publishs.end(); ++it) {
+        SrsProcess* process = *it;
+        
+        // start all processes.
+        if ((ret = process->start()) != ERROR_SUCCESS) {
+            srs_error("exec publish start failed. ret=%d", ret);
+            return ret;
+        }
+        
+        // check process status.
+        if ((ret = process->cycle()) != ERROR_SUCCESS) {
+            srs_error("exec publish cycle failed. ret=%d", ret);
+            return ret;
+        }
+    }
+    
+    // pithy print
+    show_exec_log_message();
+    
+    return ret;
+}
+
+void SrsNgExec::on_thread_stop()
+{
+    std::vector<SrsProcess*>::iterator it;
+    for (it = exec_publishs.begin(); it != exec_publishs.end(); ++it) {
+        SrsProcess* ep = *it;
+        ep->stop();
+    }
+}
+
+int SrsNgExec::parse_exec_publish(SrsRequest* req)
+{
+    int ret = ERROR_SUCCESS;
+    
+    if (!_srs_config->get_exec_enabled(req->vhost)) {
+        srs_trace("ignore disabled exec for vhost=%s", req->vhost.c_str());
+        return ret;
+    }
+    
+    // stream name: vhost/app/stream for print
+    input_stream_name = req->vhost;
+    input_stream_name += "/";
+    input_stream_name += req->app;
+    input_stream_name += "/";
+    input_stream_name += req->stream;
+    
+    std::vector<SrsConfDirective*> eps = _srs_config->get_exec_publishs(req->vhost);
+    for (int i = 0; i < (int)eps.size(); i++) {
+        SrsConfDirective* ep = eps.at(i);
+        SrsProcess* process = new SrsProcess();
+        
+        std::string binary = ep->arg0();
+        std::vector<std::string> argv;
+        for (int i = 0; i < (int)ep->args.size(); i++) {
+            std::string epa = ep->args.at(i);
+            
+            if (srs_string_contains(epa, ">")) {
+                vector<string> epas = srs_string_split(epa, ">");
+                for (int j = 0; j < (int)epas.size(); j++) {
+                    argv.push_back(parse(req, epas.at(j)));
+                    if (j == 0) {
+                        argv.push_back(">");
+                    }
+                }
+                continue;
+            }
+            
+            argv.push_back(parse(req, epa));
+        }
+        
+        if ((ret = process->initialize(binary, argv)) != ERROR_SUCCESS) {
+            srs_freep(process);
+            srs_error("initialize process failed, binary=%s, vhost=%s, ret=%d", binary.c_str(), req->vhost.c_str(), ret);
+            return ret;
+        }
+        
+        exec_publishs.push_back(process);
+    }
+    
+    return ret;
+}
+
+void SrsNgExec::clear_exec_publish()
+{
+    std::vector<SrsProcess*>::iterator it;
+    for (it = exec_publishs.begin(); it != exec_publishs.end(); ++it) {
+        SrsProcess* ep = *it;
+        srs_freep(ep);
+    }
+}
+
+void SrsNgExec::show_exec_log_message()
+{
+    pprint->elapse();
+    
+    // reportable
+    if (pprint->can_print()) {
+        // TODO: FIXME: show more info.
+        srs_trace("-> "SRS_CONSTS_LOG_EXEC" time=%"PRId64", publish=%d, input=%s",
+                  pprint->age(), (int)exec_publishs.size(), input_stream_name.c_str());
+    }
+}
+
+string SrsNgExec::parse(SrsRequest* req, string tmpl)
+{
+    string output = tmpl;
+    
+    output = srs_string_replace(output, "[vhost]", req->vhost);
+    output = srs_string_replace(output, "[port]", req->port);
+    output = srs_string_replace(output, "[app]", req->app);
+    output = srs_string_replace(output, "[stream]", req->stream);
+    
+    output = srs_string_replace(output, "[tcUrl]", req->tcUrl);
+    output = srs_string_replace(output, "[swfUrl]", req->swfUrl);
+    output = srs_string_replace(output, "[pageUrl]", req->pageUrl);
+    
+    if (output.find("[url]") != string::npos) {
+        string url = srs_generate_rtmp_url(req->host, ::atoi(req->port.c_str()), req->vhost, req->app, req->stream);
+        output = srs_string_replace(output, "[url]", url);
+    }
+    
+    return output;
+}
+

trunk/src/app/srs_app_ng_exec.hpp

@@ -0,0 +1,71 @@
+/*
+ The MIT License (MIT)
+ 
+ Copyright (c) 2013-2015 SRS(simple-rtmp-server)
+ 
+ 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_APP_NG_EXEC_HPP
+#define SRS_APP_NG_EXEC_HPP
+
+/*
+#include <srs_app_ng_exec.hpp>
+*/
+#include <srs_core.hpp>
+
+#include <vector>
+#include <string>
+
+#include <srs_app_thread.hpp>
+
+class SrsRequest;
+class SrsPithyPrint;
+class SrsProcess;
+
+/**
+ * the ng-exec is the exec feature introduced by nginx-rtmp,
+ * @see https://github.com/arut/nginx-rtmp-module/wiki/Directives#exec_push
+ * @see https://github.com/simple-rtmp-server/srs/issues/367
+ */
+class SrsNgExec : public ISrsReusableThreadHandler
+{
+private:
+    SrsReusableThread* pthread;
+    SrsPithyPrint* pprint;
+    std::string input_stream_name;
+    std::vector<SrsProcess*> exec_publishs;
+public:
+    SrsNgExec();
+    virtual ~SrsNgExec();
+public:
+    virtual int on_publish(SrsRequest* req);
+    virtual void on_unpublish();
+// interface ISrsReusableThreadHandler.
+public:
+    virtual int cycle();
+    virtual void on_thread_stop();
+private:
+    virtual int parse_exec_publish(SrsRequest* req);
+    virtual void clear_exec_publish();
+    virtual void show_exec_log_message();
+    virtual std::string parse(SrsRequest* req, std::string tmpl);
+};
+
+#endif
+

trunk/src/app/srs_app_pithy_print.cpp

@@ -108,6 +108,8 @@ SrsPithyPrint::SrsPithyPrint(int _stage_id)
 #define SRS_CONSTS_STAGE_HTTP_STREAM 9
 // the pithy stage for all http stream cache.
 #define SRS_CONSTS_STAGE_HTTP_STREAM_CACHE 10
+// for the ng-exec stage.
+#define SRS_CONSTS_STAGE_EXEC 11
 
 SrsPithyPrint* SrsPithyPrint::create_rtmp_play()
 {
@@ -134,6 +136,11 @@ SrsPithyPrint* SrsPithyPrint::create_encoder()
     return new SrsPithyPrint(SRS_CONSTS_STAGE_ENCODER);
 }
 
+SrsPithyPrint* SrsPithyPrint::create_exec()
+{
+    return new SrsPithyPrint(SRS_CONSTS_STAGE_EXEC);
+}
+
 SrsPithyPrint* SrsPithyPrint::create_ingester()
 {
     return new SrsPithyPrint(SRS_CONSTS_STAGE_INGESTER);

trunk/src/app/srs_app_pithy_print.hpp

@@ -90,6 +90,7 @@ public:
     static SrsPithyPrint* create_hls();
     static SrsPithyPrint* create_forwarder();
     static SrsPithyPrint* create_encoder();
+    static SrsPithyPrint* create_exec();
     static SrsPithyPrint* create_ingester();
     static SrsPithyPrint* create_edge();
     static SrsPithyPrint* create_caster();

trunk/src/app/srs_app_process.cpp

@@ -62,7 +62,7 @@ int SrsProcess::initialize(string binary, vector<string> argv)
 {
     int ret = ERROR_SUCCESS;
     
-    cli = bin = binary;
+    bin = binary;
     
     for (int i = 0; i < (int)argv.size(); i++) {
         std::string ffp = argv[i];

trunk/src/app/srs_app_process.hpp

@@ -38,7 +38,7 @@
  *      // the binary is the process to fork.
  *      binary = "./objs/ffmpeg/bin/ffmpeg";
  *      // where argv is a array contains each params.
- *      argv = ["-i", "in.flv", "1", ">", "/dev/null", "2", ">", "/dev/null"];
+ *      argv = ["./objs/ffmpeg/bin/ffmpeg", "-i", "in.flv", "1", ">", "/dev/null", "2", ">", "/dev/null"];
  *
  *      process = new SrsProcess();
  *      if ((ret = process->initialize(binary, argv)) != ERROR_SUCCESS) { return ret; }
@@ -71,6 +71,8 @@ public:
     virtual bool started();
     /**
      * initialize the process with binary and argv.
+     * @param binary the binary path to exec.
+     * @param argv the argv for binary path, the argv[0] generally is the binary.
      */
     virtual int initialize(std::string binary, std::vector<std::string> argv);
 public:

trunk/src/app/srs_app_reload.cpp

@@ -205,6 +205,11 @@ int ISrsReloadHandler::on_reload_vhost_transcode(string /*vhost*/)
     return ERROR_SUCCESS;
 }
 
+int ISrsReloadHandler::on_reload_vhost_exec(string /*vhost*/)
+{
+    return ERROR_SUCCESS;
+}
+
 int ISrsReloadHandler::on_reload_ingest_removed(string /*vhost*/, string /*ingest_id*/)
 {
     return ERROR_SUCCESS;

trunk/src/app/srs_app_reload.hpp

@@ -80,6 +80,7 @@ public:
     virtual int on_reload_vhost_pnt(std::string vhost);
     virtual int on_reload_vhost_chunk_size(std::string vhost);
     virtual int on_reload_vhost_transcode(std::string vhost);
+    virtual int on_reload_vhost_exec(std::string vhost);
     virtual int on_reload_ingest_removed(std::string vhost, std::string ingest_id);
     virtual int on_reload_ingest_added(std::string vhost, std::string ingest_id);
     virtual int on_reload_ingest_updated(std::string vhost, std::string ingest_id);

trunk/src/app/srs_app_source.cpp

@@ -46,6 +46,7 @@ using namespace std;
 #include <srs_app_statistic.hpp>
 #include <srs_core_autofree.hpp>
 #include <srs_rtmp_utility.hpp>
+#include <srs_app_ng_exec.hpp>
 
 #define CONST_MAX_JITTER_MS         250
 #define CONST_MAX_JITTER_MS_NEG         -250
@@ -921,6 +922,7 @@ SrsSource::SrsSource()
     publish_edge = new SrsPublishEdge();
     gop_cache = new SrsGopCache();
     aggregate_stream = new SrsStream();
+    ng_exec = new SrsNgExec();
     
     is_monotonically_increase = false;
     last_packet_time = 0;
@@ -955,6 +957,7 @@ SrsSource::~SrsSource()
     srs_freep(publish_edge);
     srs_freep(gop_cache);
     srs_freep(aggregate_stream);
+    srs_freep(ng_exec);
     
 #ifdef SRS_AUTO_HLS
     srs_freep(hls);
@@ -1259,6 +1262,24 @@ int SrsSource::on_reload_vhost_transcode(string vhost)
     return ret;
 }
 
+int SrsSource::on_reload_vhost_exec(string vhost)
+{
+    int ret = ERROR_SUCCESS;
+    
+    if (_req->vhost != vhost) {
+        return ret;
+    }
+    
+    ng_exec->on_unpublish();
+    if ((ret = ng_exec->on_publish(_req)) != ERROR_SUCCESS) {
+        srs_error("start exec failed. ret=%d", ret);
+        return ret;
+    }
+    srs_trace("vhost %s exec reload success", vhost.c_str());
+    
+    return ret;
+}
+
 int SrsSource::on_forwarder_start(SrsForwarder* forwarder)
 {
     int ret = ERROR_SUCCESS;
@@ -2058,6 +2079,12 @@ int SrsSource::on_publish()
     }
 #endif
     
+    // TODO: FIXME: use initialize to set req.
+    if ((ret = ng_exec->on_publish(_req)) != ERROR_SUCCESS) {
+        srs_error("start exec failed. ret=%d", ret);
+        return ret;
+    }
+
     // notify the handler.
     srs_assert(handler);
     if ((ret = handler->on_publish(this, _req)) != ERROR_SUCCESS) {
@@ -2090,6 +2117,8 @@ void SrsSource::on_unpublish()
     hds->on_unpublish();
 #endif
     
+    ng_exec->on_unpublish();
+
     // only clear the gop cache,
     // donot clear the sequence header, for it maybe not changed,
     // when drop dup sequence header, drop the metadata also.

trunk/src/app/srs_app_source.hpp

@@ -51,6 +51,7 @@ class SrsStSocket;
 class SrsRtmpServer;
 class SrsEdgeProxyContext;
 class SrsMessageArray;
+class SrsNgExec;
 #ifdef SRS_AUTO_HLS
 class SrsHls;
 #endif
@@ -469,8 +470,11 @@ private:
     SrsEncoder* encoder;
 #endif
 #ifdef SRS_AUTO_HDS
+    // adobe hds(http dynamic streaming).
     SrsHds *hds;
 #endif
+    // nginx-rtmp exec feature.
+    SrsNgExec* ng_exec;
     // edge control service
     SrsPlayEdge* play_edge;
     SrsPublishEdge* publish_edge;
@@ -524,6 +528,7 @@ public:
     virtual int on_reload_vhost_hds(std::string vhost);
     virtual int on_reload_vhost_dvr(std::string vhost);
     virtual int on_reload_vhost_transcode(std::string vhost);
+    virtual int on_reload_vhost_exec(std::string vhost);
 // for the tools callback
 public:
     // for the SrsForwarder to callback to request the sequence headers.

trunk/src/core/srs_core.hpp

@@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 // current release version
 #define VERSION_MAJOR       3
 #define VERSION_MINOR       0
-#define VERSION_REVISION    0
+#define VERSION_REVISION    1
 
 // server info.
 #define RTMP_SIG_SRS_KEY "SRS"

trunk/src/kernel/srs_kernel_consts.hpp

@@ -168,6 +168,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #define SRS_CONSTS_LOG_HTTP_STREAM_CACHE "HTC"
 // stream caster log id.
 #define SRS_CONSTS_LOG_STREAM_CASTER "SCS"
+// the nginx exec log id.
+#define SRS_CONSTS_LOG_EXEC "EXE"
 
 ///////////////////////////////////////////////////////////
 ///////////////////////////////////////////////////////////

trunk/src/kernel/srs_kernel_utility.cpp

@@ -35,6 +35,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <sys/stat.h>
 #include <fcntl.h>
 
+#include <vector>
 using namespace std;
 
 #include <srs_kernel_log.hpp>
@@ -284,6 +285,25 @@ bool srs_string_contains(string str, string flag)
     return str.find(flag) != string::npos;
 }
 
+vector<string> srs_string_split(string str, string flag)
+{
+    vector<string> arr;
+    
+    size_t pos;
+    string s = str;
+    
+    while ((pos = s.find(flag)) != string::npos) {
+        arr.push_back(s.substr(0, pos));
+        s = s.substr(pos + 1);
+    }
+    
+    if (!s.empty()) {
+        arr.push_back(s);
+    }
+    
+    return arr;
+}
+
 int srs_do_create_dir_recursively(string dir)
 {
     int ret = ERROR_SUCCESS;

trunk/src/kernel/srs_kernel_utility.hpp

@@ -31,6 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_core.hpp>
 
 #include <string>
+#include <vector>
 
 class SrsStream;
 class SrsBitStream;
@@ -69,6 +70,8 @@ extern bool srs_string_ends_with(std::string str, std::string flag);
 extern bool srs_string_starts_with(std::string str, std::string flag);
 // whether string contains with
 extern bool srs_string_contains(std::string str, std::string flag);
+// split the string by flag to array.
+extern std::vector<std::string> srs_string_split(std::string str, std::string flag);
 
 // create dir recursively
 extern int srs_create_dir_recursively(std::string dir);

trunk/src/protocol/srs_rtmp_utility.cpp

@@ -29,6 +29,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #endif
 
 #include <stdlib.h>
+#include <sstream>
 using namespace std;
 
 #include <srs_kernel_log.hpp>
@@ -239,6 +240,22 @@ std::string srs_generate_stream_url(std::string vhost, std::string app, std::str
     return url;
 }
 
+string srs_generate_rtmp_url(string server, int port, string vhost, string app, string stream)
+{
+    std::stringstream ss;
+    
+    ss << "rtmp://" << server << ":" << std::dec << port << "/" << app;
+    
+    // when default or server is vhost, donot specifies the vhost in params.
+    if (SRS_CONSTS_RTMP_DEFAULT_VHOST != vhost && server != vhost) {
+        ss << "...vhost..." << vhost;
+    }
+    
+    ss << "/" << stream;
+    
+    return ss.str();
+}
+
 int srs_write_large_iovs(ISrsProtocolReaderWriter* skt, iovec* iovs, int size, ssize_t* pnwrite)
 {
     int ret = ERROR_SUCCESS;

trunk/src/protocol/srs_rtmp_utility.hpp

@@ -106,6 +106,9 @@ extern int srs_rtmp_create_msg(char type, u_int32_t timestamp, char* data, int s
 // get the stream identify, vhost/app/stream.
 extern std::string srs_generate_stream_url(std::string vhost, std::string app, std::string stream);
 
+// genereate the rtmp url, for instance, rtmp://server:port/app...vhost...vhost/stream
+extern std::string srs_generate_rtmp_url(std::string server, int port, std::string vhost, std::string app, std::string stream);
+
 // write large numbers of iovs.
 extern int srs_write_large_iovs(ISrsProtocolReaderWriter* skt, iovec* iovs, int size, ssize_t* pnwrite = NULL);