diff --git a/README.md b/README.md index 30715d99a..ca8c02188 100755 --- a/README.md +++ b/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 * 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. * 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 diff --git a/trunk/conf/exec.conf b/trunk/conf/exec.conf new file mode 100644 index 000000000..3c8afd94f --- /dev/null +++ b/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; + } +} diff --git a/trunk/conf/full.conf b/trunk/conf/full.conf index fb45016f7..69f27557b 100644 --- a/trunk/conf/full.conf +++ b/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; } diff --git a/trunk/configure b/trunk/configure index 5ef3bd33d..0eeac78e8 100755 --- a/trunk/configure +++ b/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 diff --git a/trunk/ide/srs_upp/srs_upp.upp b/trunk/ide/srs_upp/srs_upp.upp index 116621e69..314abce10 100755 --- a/trunk/ide/srs_upp/srs_upp.upp +++ b/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, diff --git a/trunk/ide/srs_xcode/srs_xcode.xcodeproj/project.pbxproj b/trunk/ide/srs_xcode/srs_xcode.xcodeproj/project.pbxproj index 0e305b70d..db636932f 100644 --- a/trunk/ide/srs_xcode/srs_xcode.xcodeproj/project.pbxproj +++ b/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 = ""; }; 3C36DB591ABD1CB90066CCAF /* srs_librtmp.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; name = srs_librtmp.cpp; path = ../../../src/libs/srs_librtmp.cpp; sourceTree = ""; }; 3C36DB5A1ABD1CB90066CCAF /* srs_librtmp.hpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.h; name = srs_librtmp.hpp; path = ../../../src/libs/srs_librtmp.hpp; sourceTree = ""; }; + 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 = ""; }; + 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 = ""; }; + 3C4AB9341B8C9FF9006627D3 /* exec.conf */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = text; name = exec.conf; path = ../../../conf/exec.conf; sourceTree = ""; }; 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 = ""; }; 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 = ""; }; 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 = ""; }; @@ -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 */, diff --git a/trunk/src/app/srs_app_config.cpp b/trunk/src/app/srs_app_config.cpp index 871afb653..28b74f88b 100644 --- a/trunk/src/app/srs_app_config.cpp +++ b/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); @@ -1811,17 +1842,25 @@ int SrsConfig::check_config() string m = conf->at(j)->name.c_str(); if (m != "enabled" && m != "dvr_path" && m != "dvr_plan" && m != "dvr_duration" && m != "dvr_wait_keyframe" && m != "time_jitter" - ) { + ) { ret = ERROR_SYSTEM_CONFIG_INVALID; srs_error("unsupported vhost dvr directive %s, ret=%d", m.c_str(), ret); 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" - && m != "engine" - ) { + if (m != "enabled" && m != "input" && m != "ffmpeg" && 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 SrsConfig::get_exec_publishs(string vhost) +{ + vector 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 SrsConfig::get_ingesters(string vhost) { vector ingeters; diff --git a/trunk/src/app/srs_app_config.hpp b/trunk/src/app/srs_app_config.hpp index 4dd83c044..aad2ee498 100644 --- a/trunk/src/app/srs_app_config.hpp +++ b/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 get_exec_publishs(std::string vhost); // vhost ingest section public: /** diff --git a/trunk/src/app/srs_app_encoder.cpp b/trunk/src/app/srs_app_encoder.cpp index 334bfce17..fb9730b54 100644 --- a/trunk/src/app/srs_app_encoder.cpp +++ b/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", - conf->arg0().c_str()); + srs_trace("ignore the disabled transcode: %s", conf->arg0().c_str()); return ret; } diff --git a/trunk/src/app/srs_app_ng_exec.cpp b/trunk/src/app/srs_app_ng_exec.cpp new file mode 100644 index 000000000..c3afee8c4 --- /dev/null +++ b/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 + +using namespace std; + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// 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::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::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 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 argv; + for (int i = 0; i < (int)ep->args.size(); i++) { + std::string epa = ep->args.at(i); + + if (srs_string_contains(epa, ">")) { + vector 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::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; +} + diff --git a/trunk/src/app/srs_app_ng_exec.hpp b/trunk/src/app/srs_app_ng_exec.hpp new file mode 100644 index 000000000..7e1f3f057 --- /dev/null +++ b/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 +*/ +#include + +#include +#include + +#include + +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 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 + diff --git a/trunk/src/app/srs_app_pithy_print.cpp b/trunk/src/app/srs_app_pithy_print.cpp index 5edd5de44..657c4dc42 100644 --- a/trunk/src/app/srs_app_pithy_print.cpp +++ b/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); diff --git a/trunk/src/app/srs_app_pithy_print.hpp b/trunk/src/app/srs_app_pithy_print.hpp index 96c0f7055..a2532ddc5 100644 --- a/trunk/src/app/srs_app_pithy_print.hpp +++ b/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(); diff --git a/trunk/src/app/srs_app_process.cpp b/trunk/src/app/srs_app_process.cpp index 25735d3f6..93ed14f1b 100644 --- a/trunk/src/app/srs_app_process.cpp +++ b/trunk/src/app/srs_app_process.cpp @@ -62,7 +62,7 @@ int SrsProcess::initialize(string binary, vector argv) { int ret = ERROR_SUCCESS; - cli = bin = binary; + bin = binary; for (int i = 0; i < (int)argv.size(); i++) { std::string ffp = argv[i]; diff --git a/trunk/src/app/srs_app_process.hpp b/trunk/src/app/srs_app_process.hpp index aa282fd18..c619da42c 100644 --- a/trunk/src/app/srs_app_process.hpp +++ b/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 argv); public: diff --git a/trunk/src/app/srs_app_reload.cpp b/trunk/src/app/srs_app_reload.cpp index ba4ab09ad..8e86494f0 100644 --- a/trunk/src/app/srs_app_reload.cpp +++ b/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; diff --git a/trunk/src/app/srs_app_reload.hpp b/trunk/src/app/srs_app_reload.hpp index ceb32194c..d6e1648c3 100644 --- a/trunk/src/app/srs_app_reload.hpp +++ b/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); diff --git a/trunk/src/app/srs_app_source.cpp b/trunk/src/app/srs_app_source.cpp index ec00978cc..dca38d091 100755 --- a/trunk/src/app/srs_app_source.cpp +++ b/trunk/src/app/srs_app_source.cpp @@ -46,6 +46,7 @@ using namespace std; #include #include #include +#include #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; @@ -2057,6 +2078,12 @@ int SrsSource::on_publish() return ret; } #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); @@ -2089,6 +2116,8 @@ void SrsSource::on_unpublish() #ifdef SRS_AUTO_HDS hds->on_unpublish(); #endif + + ng_exec->on_unpublish(); // only clear the gop cache, // donot clear the sequence header, for it maybe not changed, diff --git a/trunk/src/app/srs_app_source.hpp b/trunk/src/app/srs_app_source.hpp index 7c626b7eb..1e2f70aff 100755 --- a/trunk/src/app/srs_app_source.hpp +++ b/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. diff --git a/trunk/src/core/srs_core.hpp b/trunk/src/core/srs_core.hpp index fd6c2df5a..c9e82f66b 100644 --- a/trunk/src/core/srs_core.hpp +++ b/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" diff --git a/trunk/src/kernel/srs_kernel_consts.hpp b/trunk/src/kernel/srs_kernel_consts.hpp index ee1142dc6..94d8ef0a3 100644 --- a/trunk/src/kernel/srs_kernel_consts.hpp +++ b/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" /////////////////////////////////////////////////////////// /////////////////////////////////////////////////////////// diff --git a/trunk/src/kernel/srs_kernel_utility.cpp b/trunk/src/kernel/srs_kernel_utility.cpp index b445e5394..c341e0c8e 100644 --- a/trunk/src/kernel/srs_kernel_utility.cpp +++ b/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 #include +#include using namespace std; #include @@ -284,6 +285,25 @@ bool srs_string_contains(string str, string flag) return str.find(flag) != string::npos; } +vector srs_string_split(string str, string flag) +{ + vector 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; diff --git a/trunk/src/kernel/srs_kernel_utility.hpp b/trunk/src/kernel/srs_kernel_utility.hpp index d6406e33d..668867936 100644 --- a/trunk/src/kernel/srs_kernel_utility.hpp +++ b/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 #include +#include 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 srs_string_split(std::string str, std::string flag); // create dir recursively extern int srs_create_dir_recursively(std::string dir); diff --git a/trunk/src/protocol/srs_rtmp_utility.cpp b/trunk/src/protocol/srs_rtmp_utility.cpp index c28d5cf32..61338bd93 100644 --- a/trunk/src/protocol/srs_rtmp_utility.cpp +++ b/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 +#include using namespace std; #include @@ -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; diff --git a/trunk/src/protocol/srs_rtmp_utility.hpp b/trunk/src/protocol/srs_rtmp_utility.hpp index 9b4c3b35a..bfbc33312 100644 --- a/trunk/src/protocol/srs_rtmp_utility.hpp +++ b/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);