external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-02-15 04:42:04 +00:00
Code Issues Releases Wiki Activity

Merge branch '2.0release' into develop

Browse source
This commit is contained in:
winlin 2015-08-03 15:23:07 +08:00
commit 0e3128d3e3
18 changed files with 691 additions and 676 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
README.md
View file

@ -691,7 +691,7 @@ The publish RTMP benchmark by [SB](https://github.com/simple-rtmp-server/srs-ben
| 2014-12-03 | 2.0.12 | 1.2k(1200) | publishers | 96% | 43MB | - |
| 2014-12-03 | 2.0.47 | 1.2k(1200) | publishers | 84% | 76MB | [code][p1] |
| 2014-12-03 | 2.0.47 | 1.4k(1400) | publishers | 95% | 140MB | - |
| 2014-12-03 | 2.0.48 | 1.4k(1400 | publishers | 95% | 140MB | [code][p2] |
| 2014-12-03 | 2.0.48 | 1.4k(1400) | publishers | 95% | 140MB | [code][p2] |
| 2014-12-04 | 2.0.49 | 1.4k(1400) | publishers | 68% | 144MB | - |
| 2014-12-04 | 2.0.49 | 2.5k(2500) | publishers | 95% | 404MB | [code][p3] |
| 2014-12-04 | 2.0.51 | 2.5k(2500) | publishers | 91% | 259MB | [code][p4] |

918
trunk/src/app/srs_app_config.cpp
View file

@ -535,6 +535,290 @@ int SrsConfig::reload()
return reload_conf(&conf);
}
int SrsConfig::reload_vhost(SrsConfDirective* old_root)
{
int ret = ERROR_SUCCESS;
// merge config.
std::vector<ISrsReloadHandler*>::iterator it;
// state graph
// old_vhost new_vhost
// DISABLED => ENABLED
// ENABLED => DISABLED
// ENABLED => ENABLED (modified)
// collect all vhost names
std::vector<std::string> vhosts;
for (int i = 0; i < (int)root->directives.size(); i++) {
SrsConfDirective* vhost = root->at(i);
if (vhost->name != "vhost") {
continue;
}
vhosts.push_back(vhost->arg0());
}
for (int i = 0; i < (int)old_root->directives.size(); i++) {
SrsConfDirective* vhost = old_root->at(i);
if (vhost->name != "vhost") {
continue;
}
if (root->get("vhost", vhost->arg0())) {
continue;
}
vhosts.push_back(vhost->arg0());
}
// process each vhost
for (int i = 0; i < (int)vhosts.size(); i++) {
std::string vhost = vhosts.at(i);
SrsConfDirective* old_vhost = old_root->get("vhost", vhost);
SrsConfDirective* new_vhost = root->get("vhost", vhost);
// DISABLED => ENABLED
if (!get_vhost_enabled(old_vhost) && get_vhost_enabled(new_vhost)) {
srs_trace("vhost %s added, reload it.", vhost.c_str());
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_added(vhost)) != ERROR_SUCCESS) {
srs_error("notify subscribes added "
"vhost %s failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("reload new vhost %s success.", vhost.c_str());
continue;
}
// ENABLED => DISABLED
if (get_vhost_enabled(old_vhost) && !get_vhost_enabled(new_vhost)) {
srs_trace("vhost %s removed, reload it.", vhost.c_str());
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_removed(vhost)) != ERROR_SUCCESS) {
srs_error("notify subscribes removed "
"vhost %s failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("reload removed vhost %s success.", vhost.c_str());
continue;
}
// mode, never supports reload.
// first, for the origin and edge role change is too complex.
// second, the vhosts in origin device group normally are all origin,
// they never change to edge sometimes.
// third, the origin or upnode device can always be restart,
// edge will retry and the users connected to edge are ok.
// it's ok to add or remove edge/origin vhost.
if (get_vhost_is_edge(old_vhost) != get_vhost_is_edge(new_vhost)) {
ret = ERROR_RTMP_EDGE_RELOAD;
srs_error("reload never supports mode changed. ret=%d", ret);
return ret;
}
// ENABLED => ENABLED (modified)
if (get_vhost_enabled(new_vhost) && get_vhost_enabled(old_vhost)) {
srs_trace("vhost %s maybe modified, reload its detail.", vhost.c_str());
// atc, only one per vhost
if (!srs_directive_equals(new_vhost->get("atc"), old_vhost->get("atc"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_atc(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes atc failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_gop_cache(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes gop_cache failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_queue_length(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes queue_length failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_time_jitter(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes time_jitter failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_mix_correct(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes mix_correct failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_forward(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes forward failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_hls(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes hls failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("vhost %s reload hls success.", vhost.c_str());
}
// hds reload
if (!srs_directive_equals(new_vhost->get("hds"), old_vhost->get("hds"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_hds(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes hds failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("vhost %s reload hds success.", vhost.c_str());
}
// dvr, only one per vhost
if (!srs_directive_equals(new_vhost->get("dvr"), old_vhost->get("dvr"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_dvr(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes dvr failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("vhost %s reload hlsdvrsuccess.", 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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_mr(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes mr failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_chunk_size(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes chunk_size failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_mw(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes mw failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("vhost %s reload mw 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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_realtime(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes min_latency failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_http_updated()) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes http failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_http_updated()) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes http_static failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_http_remux_updated(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes http_remux failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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;
}
continue;
}
srs_trace("ignore reload vhost, enabled old: %d, new: %d",
get_vhost_enabled(old_vhost), get_vhost_enabled(new_vhost));
}
return ret;
}
int SrsConfig::reload_conf(SrsConfig* conf)
{
int ret = ERROR_SUCCESS;
@ -800,302 +1084,6 @@ int SrsConfig::reload_http_stream(SrsConfDirective* old_root)
return ret;
}
int SrsConfig::reload_vhost(SrsConfDirective* old_root)
{
int ret = ERROR_SUCCESS;
// merge config.
std::vector<ISrsReloadHandler*>::iterator it;
// state graph
// old_vhost new_vhost
// DISABLED => ENABLED
// ENABLED => DISABLED
// ENABLED => ENABLED (modified)
// collect all vhost names
std::vector<std::string> vhosts;
for (int i = 0; i < (int)root->directives.size(); i++) {
SrsConfDirective* vhost = root->at(i);
if (vhost->name != "vhost") {
continue;
}
vhosts.push_back(vhost->arg0());
}
for (int i = 0; i < (int)old_root->directives.size(); i++) {
SrsConfDirective* vhost = old_root->at(i);
if (vhost->name != "vhost") {
continue;
}
if (root->get("vhost", vhost->arg0())) {
continue;
}
vhosts.push_back(vhost->arg0());
}
// process each vhost
for (int i = 0; i < (int)vhosts.size(); i++) {
std::string vhost = vhosts.at(i);
SrsConfDirective* old_vhost = old_root->get("vhost", vhost);
SrsConfDirective* new_vhost = root->get("vhost", vhost);
// DISABLED => ENABLED
if (!get_vhost_enabled(old_vhost) && get_vhost_enabled(new_vhost)) {
srs_trace("vhost %s added, reload it.", vhost.c_str());
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_added(vhost)) != ERROR_SUCCESS) {
srs_error("notify subscribes added "
"vhost %s failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
// TODO: reload new http_remux in on_vhost_add
// http_remux, only one per vhost.
if (get_vhost_http_remux_enabled(vhost)) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_http_remux_updated(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes http_remux failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("vhost %s reload http_remux success.", vhost.c_str());
}
srs_trace("reload new vhost %s success.", vhost.c_str());
continue;
}
// ENABLED => DISABLED
if (get_vhost_enabled(old_vhost) && !get_vhost_enabled(new_vhost)) {
srs_trace("vhost %s removed, reload it.", vhost.c_str());
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_removed(vhost)) != ERROR_SUCCESS) {
srs_error("notify subscribes removed "
"vhost %s failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("reload removed vhost %s success.", vhost.c_str());
continue;
}
// mode, never supports reload.
// first, for the origin and edge role change is too complex.
// second, the vhosts in origin device group normally are all origin,
// they never change to edge sometimes.
// third, the origin or upnode device can always be restart,
// edge will retry and the users connected to edge are ok.
// it's ok to add or remove edge/origin vhost.
if (get_vhost_is_edge(old_vhost) != get_vhost_is_edge(new_vhost)) {
ret = ERROR_RTMP_EDGE_RELOAD;
srs_error("reload never supports mode changed. ret=%d", ret);
return ret;
}
// ENABLED => ENABLED (modified)
if (get_vhost_enabled(new_vhost) && get_vhost_enabled(old_vhost)) {
srs_trace("vhost %s maybe modified, reload its detail.", vhost.c_str());
// atc, only one per vhost
if (!srs_directive_equals(new_vhost->get("atc"), old_vhost->get("atc"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_atc(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes atc failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_gop_cache(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes gop_cache failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_queue_length(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes queue_length failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_time_jitter(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes time_jitter failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_mix_correct(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes mix_correct failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_forward(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes forward failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_hls(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes hls failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("vhost %s reload hls success.", vhost.c_str());
}
// hds reload
if (!srs_directive_equals(new_vhost->get("hds"), old_vhost->get("hds"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_hds(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes hds failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("vhost %s reload hds success.", vhost.c_str());
}
// dvr, only one per vhost
if (!srs_directive_equals(new_vhost->get("dvr"), old_vhost->get("dvr"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_dvr(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes dvr failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("vhost %s reload hlsdvrsuccess.", 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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_mr(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes mr failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_chunk_size(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes chunk_size failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_mw(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes mw failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
srs_trace("vhost %s reload mw 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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_realtime(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes min_latency failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_http_updated()) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes http failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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"))) {
for (it = subscribes.begin(); it != subscribes.end(); ++it) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_http_updated()) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes http_static failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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) {
ISrsReloadHandler* subscribe = *it;
if ((ret = subscribe->on_reload_vhost_http_remux_updated(vhost)) != ERROR_SUCCESS) {
srs_error("vhost %s notify subscribes http_remux failed. ret=%d", vhost.c_str(), ret);
return ret;
}
}
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;
}
continue;
}
srs_trace("ignore reload vhost, enabled old: %d, new: %d",
get_vhost_enabled(old_vhost), get_vhost_enabled(new_vhost));
}
return ret;
}
int SrsConfig::reload_transcode(SrsConfDirective* new_vhost, SrsConfDirective* old_vhost)
{
int ret = ERROR_SUCCESS;
@ -1494,9 +1482,6 @@ int SrsConfig::check_config()
srs_trace("srs checking config...");
vector<SrsConfDirective*> vhosts = get_vhosts();
vector<SrsConfDirective*> stream_casters = get_stream_casters();
////////////////////////////////////////////////////////////////////////
// check empty
////////////////////////////////////////////////////////////////////////
@ -1570,6 +1555,170 @@ int SrsConfig::check_config()
}
}
}
////////////////////////////////////////////////////////////////////////
// check listen for rtmp.
////////////////////////////////////////////////////////////////////////
if (true) {
vector<string> listens = get_listens();
if (listens.size() <= 0) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive \"listen\" is empty, ret=%d", ret);
return ret;
}
for (int i = 0; i < (int)listens.size(); i++) {
string port = listens[i];
if (port.empty() || ::atoi(port.c_str()) <= 0) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive listen invalid, port=%s, ret=%d", port.c_str(), ret);
return ret;
}
}
}
////////////////////////////////////////////////////////////////////////
// check max connections
////////////////////////////////////////////////////////////////////////
if (get_max_connections() <= 0) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive max_connections invalid, max_connections=%d, ret=%d", get_max_connections(), ret);
return ret;
}
// check max connections of system limits
if (true) {
int nb_consumed_fds = (int)get_listens().size();
if (!get_http_api_listen().empty()) {
nb_consumed_fds++;
}
if (!get_http_stream_listen().empty()) {
nb_consumed_fds++;
}
if (get_log_tank_file()) {
nb_consumed_fds++;
}
// 0, 1, 2 for stdin, stdout and stderr.
nb_consumed_fds += 3;
int nb_connections = get_max_connections();
int nb_total = nb_connections + nb_consumed_fds;
int max_open_files = (int)sysconf(_SC_OPEN_MAX);
int nb_canbe = max_open_files - nb_consumed_fds - 1;
// for each play connections, we open a pipe(2fds) to convert SrsConsumver to io,
// refine performance, @see: https://github.com/simple-rtmp-server/srs/issues/194
if (nb_total >= max_open_files) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("invalid max_connections=%d, required=%d, system limit to %d, "
"total=%d(max_connections=%d, nb_consumed_fds=%d), ret=%d. "
"you can change max_connections from %d to %d, or "
"you can login as root and set the limit: ulimit -HSn %d",
nb_connections, nb_total + 1, max_open_files,
nb_total, nb_connections, nb_consumed_fds,
ret, nb_connections, nb_canbe, nb_total + 1);
return ret;
}
}
////////////////////////////////////////////////////////////////////////
// check heartbeat
////////////////////////////////////////////////////////////////////////
if (get_heartbeat_interval() <= 0) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive heartbeat interval invalid, interval=%"PRId64", ret=%d",
get_heartbeat_interval(), ret);
return ret;
}
////////////////////////////////////////////////////////////////////////
// check stats
////////////////////////////////////////////////////////////////////////
if (get_stats_network() < 0) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive stats network invalid, network=%d, ret=%d",
get_stats_network(), ret);
return ret;
}
if (true) {
vector<std::string> ips = srs_get_local_ipv4_ips();
int index = get_stats_network();
if (index >= (int)ips.size()) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("stats network invalid, total local ip count=%d, index=%d, ret=%d",
(int)ips.size(), index, ret);
return ret;
}
srs_warn("stats network use index=%d, ip=%s", index, ips.at(index).c_str());
}
if (true) {
SrsConfDirective* conf = get_stats_disk_device();
if (conf == NULL || (int)conf->args.size() <= 0) {
srs_warn("stats disk not configed, disk iops disabled.");
} else {
string disks;
for (int i = 0; i < (int)conf->args.size(); i++) {
disks += conf->args.at(i);
disks += " ";
}
srs_warn("stats disk list: %s", disks.c_str());
}
}
////////////////////////////////////////////////////////////////////////
// check http api
////////////////////////////////////////////////////////////////////////
if (get_http_api_listen().empty()) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive http_api listen invalid, listen=%s, ret=%d",
get_http_api_listen().c_str(), ret);
return ret;
}
////////////////////////////////////////////////////////////////////////
// check http stream
////////////////////////////////////////////////////////////////////////
if (get_http_stream_listen().empty()) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive http_stream listen invalid, listen=%s, ret=%d",
get_http_stream_listen().c_str(), ret);
return ret;
}
////////////////////////////////////////////////////////////////////////
// check log name and level
////////////////////////////////////////////////////////////////////////
if (true) {
std::string log_filename = this->get_log_file();
if (get_log_tank_file() && log_filename.empty()) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("must specifies the file to write log to. ret=%d", ret);
return ret;
}
if (get_log_tank_file()) {
srs_trace("write log to file %s", log_filename.c_str());
srs_trace("you can: tailf %s", log_filename.c_str());
srs_trace("@see: %s", SRS_WIKI_URL_LOG);
} else {
srs_trace("write log to console");
}
}
////////////////////////////////////////////////////////////////////////
// check features
////////////////////////////////////////////////////////////////////////
#ifndef SRS_AUTO_HTTP_SERVER
if (get_http_stream_enabled()) {
srs_warn("http_stream is disabled by configure");
}
#endif
#ifndef SRS_AUTO_HTTP_API
if (get_http_api_enabled()) {
srs_warn("http_api is disabled by configure");
}
#endif
vector<SrsConfDirective*> stream_casters = get_stream_casters();
for (int n = 0; n < (int)stream_casters.size(); n++) {
SrsConfDirective* stream_caster = stream_casters[n];
for (int i = 0; stream_caster && i < (int)stream_caster->directives.size(); i++) {
@ -1584,6 +1733,9 @@ int SrsConfig::check_config()
}
}
}
vector<SrsConfDirective*> vhosts;
get_vhosts(vhosts);
for (int n = 0; n < (int)vhosts.size(); n++) {
SrsConfDirective* vhost = vhosts[n];
for (int i = 0; vhost && i < (int)vhost->directives.size(); i++) {
@ -1765,135 +1917,6 @@ int SrsConfig::check_config()
}
}
////////////////////////////////////////////////////////////////////////
// check listen for rtmp.
////////////////////////////////////////////////////////////////////////
if (true) {
vector<string> listens = get_listens();
if (listens.size() <= 0) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive \"listen\" is empty, ret=%d", ret);
return ret;
}
for (int i = 0; i < (int)listens.size(); i++) {
string port = listens[i];
if (port.empty() || ::atoi(port.c_str()) <= 0) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive listen invalid, port=%s, ret=%d", port.c_str(), ret);
return ret;
}
}
}
////////////////////////////////////////////////////////////////////////
// check max connections
////////////////////////////////////////////////////////////////////////
if (get_max_connections() <= 0) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive max_connections invalid, max_connections=%d, ret=%d", get_max_connections(), ret);
return ret;
}
// check max connections of system limits
if (true) {
int nb_consumed_fds = (int)get_listens().size();
if (!get_http_api_listen().empty()) {
nb_consumed_fds++;
}
if (!get_http_stream_listen().empty()) {
nb_consumed_fds++;
}
if (get_log_tank_file()) {
nb_consumed_fds++;
}
// 0, 1, 2 for stdin, stdout and stderr.
nb_consumed_fds += 3;
int nb_connections = get_max_connections();
int nb_total = nb_connections + nb_consumed_fds;
int max_open_files = (int)sysconf(_SC_OPEN_MAX);
int nb_canbe = max_open_files - nb_consumed_fds - 1;
// for each play connections, we open a pipe(2fds) to convert SrsConsumver to io,
// refine performance, @see: https://github.com/simple-rtmp-server/srs/issues/194
if (nb_total >= max_open_files) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("invalid max_connections=%d, required=%d, system limit to %d, "
"total=%d(max_connections=%d, nb_consumed_fds=%d), ret=%d. "
"you can change max_connections from %d to %d, or "
"you can login as root and set the limit: ulimit -HSn %d",
nb_connections, nb_total + 1, max_open_files,
nb_total, nb_connections, nb_consumed_fds,
ret, nb_connections, nb_canbe, nb_total + 1);
return ret;
}
}
////////////////////////////////////////////////////////////////////////
// check heartbeat
////////////////////////////////////////////////////////////////////////
if (get_heartbeat_interval() <= 0) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive heartbeat interval invalid, interval=%"PRId64", ret=%d",
get_heartbeat_interval(), ret);
return ret;
}
////////////////////////////////////////////////////////////////////////
// check stats
////////////////////////////////////////////////////////////////////////
if (get_stats_network() < 0) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive stats network invalid, network=%d, ret=%d",
get_stats_network(), ret);
return ret;
}
if (true) {
vector<std::string> ips = srs_get_local_ipv4_ips();
int index = get_stats_network();
if (index >= (int)ips.size()) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("stats network invalid, total local ip count=%d, index=%d, ret=%d",
(int)ips.size(), index, ret);
return ret;
}
srs_warn("stats network use index=%d, ip=%s", index, ips.at(index).c_str());
}
if (true) {
SrsConfDirective* conf = get_stats_disk_device();
if (conf == NULL || (int)conf->args.size() <= 0) {
srs_warn("stats disk not configed, disk iops disabled.");
} else {
string disks;
for (int i = 0; i < (int)conf->args.size(); i++) {
disks += conf->args.at(i);
disks += " ";
}
srs_warn("stats disk list: %s", disks.c_str());
}
}
////////////////////////////////////////////////////////////////////////
// check http api
////////////////////////////////////////////////////////////////////////
if (get_http_api_listen().empty()) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive http_api listen invalid, listen=%s, ret=%d",
get_http_api_listen().c_str(), ret);
return ret;
}
////////////////////////////////////////////////////////////////////////
// check http stream
////////////////////////////////////////////////////////////////////////
if (get_http_stream_listen().empty()) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("directive http_stream listen invalid, listen=%s, ret=%d",
get_http_stream_listen().c_str(), ret);
return ret;
}
////////////////////////////////////////////////////////////////////////
// check chunk size
////////////////////////////////////////////////////////////////////////
@ -1918,39 +1941,6 @@ int SrsConfig::check_config()
return ret;
}
}
////////////////////////////////////////////////////////////////////////
// check log name and level
////////////////////////////////////////////////////////////////////////
if (true) {
std::string log_filename = this->get_log_file();
if (get_log_tank_file() && log_filename.empty()) {
ret = ERROR_SYSTEM_CONFIG_INVALID;
srs_error("must specifies the file to write log to. ret=%d", ret);
return ret;
}
if (get_log_tank_file()) {
srs_trace("write log to file %s", log_filename.c_str());
srs_trace("you can: tailf %s", log_filename.c_str());
srs_trace("@see: %s", SRS_WIKI_URL_LOG);
} else {
srs_trace("write log to console");
}
}
////////////////////////////////////////////////////////////////////////
// check features
////////////////////////////////////////////////////////////////////////
#ifndef SRS_AUTO_HTTP_SERVER
if (get_http_stream_enabled()) {
srs_warn("http_stream is disabled by configure");
}
#endif
#ifndef SRS_AUTO_HTTP_API
if (get_http_api_enabled()) {
srs_warn("http_api is disabled by configure");
}
#endif
for (int i = 0; i < (int)vhosts.size(); i++) {
SrsConfDirective* vhost = vhosts[i];
srs_assert(vhost != NULL);
@ -2219,12 +2209,10 @@ SrsConfDirective* SrsConfig::get_vhost(string vhost)
return NULL;
}
vector<SrsConfDirective*> SrsConfig::get_vhosts()
void SrsConfig::get_vhosts(vector<SrsConfDirective*>& vhosts)
{
srs_assert(root);
std::vector<SrsConfDirective*> vhosts;
for (int i = 0; i < (int)root->directives.size(); i++) {
SrsConfDirective* conf = root->at(i);
@ -2234,8 +2222,6 @@ vector<SrsConfDirective*> SrsConfig::get_vhosts()
vhosts.push_back(conf);
}
return vhosts;
}
bool SrsConfig::get_vhost_enabled(string vhost)

11
trunk/src/app/srs_app_config.hpp
View file

@ -253,6 +253,11 @@ public:
* @remark, user can test the config before reload it.
*/
virtual int reload();
private:
/**
* reload the vhost section of config.
*/
virtual int reload_vhost(SrsConfDirective* old_root);
protected:
/**
* reload from the config.
@ -269,10 +274,6 @@ private:
*/
virtual int reload_http_stream(SrsConfDirective* old_root);
/**
* reload the vhost section of config.
*/
virtual int reload_vhost(SrsConfDirective* old_root);
/**
* reload the transcode section of vhost of config.
*/
virtual int reload_transcode(SrsConfDirective* new_vhost, SrsConfDirective* old_vhost);
@ -413,7 +414,7 @@ public:
/**
* get all vhosts in config file.
*/
virtual std::vector<SrsConfDirective*> get_vhosts();
virtual void get_vhosts(std::vector<SrsConfDirective*>& vhosts);
/**
* whether vhost is enabled
* @param vhost, the vhost name.

17
trunk/src/app/srs_app_http_api.cpp
View file

@ -543,20 +543,21 @@ int SrsHttpApi::do_cycle()
// always free it in this scope.
SrsAutoFree(ISrsHttpMessage, req);
// TODO: FIXME: use the post body.
std::string res;
// get response body.
if ((ret = req->body_read_all(res)) != ERROR_SUCCESS) {
return ret;
}
// ok, handle http request.
SrsHttpResponseWriter writer(&skt);
if ((ret = process_request(&writer, req)) != ERROR_SUCCESS) {
return ret;
}
// read all rest bytes in request body.
char buf[SRS_HTTP_READ_CACHE_BYTES];
ISrsHttpResponseReader* br = req->body_reader();
while (!br->eof()) {
if ((ret = br->read(buf, SRS_HTTP_READ_CACHE_BYTES, NULL)) != ERROR_SUCCESS) {
return ret;
}
}
// donot keep alive, disconnect it.
// @see https://github.com/simple-rtmp-server/srs/issues/399
if (!req->is_keep_alive()) {

11
trunk/src/app/srs_app_http_stream.cpp
View file

@ -863,6 +863,17 @@ void SrsHttpStreamServer::http_unmount(SrsSource* s, SrsRequest* r)
entry->stream->entry->enabled = false;
}
int SrsHttpStreamServer::on_reload_vhost_added(string vhost)
{
int ret = ERROR_SUCCESS;
if ((ret = on_reload_vhost_http_remux_updated(vhost)) != ERROR_SUCCESS) {
return ret;
}
return ret;
}
int SrsHttpStreamServer::on_reload_vhost_http_remux_updated(string vhost)
{
int ret = ERROR_SUCCESS;

1
trunk/src/app/srs_app_http_stream.hpp
View file

@ -354,6 +354,7 @@ public:
virtual void unmount_hls(SrsRequest* r);
// interface ISrsReloadHandler.
public:
virtual int on_reload_vhost_added(std::string vhost);
virtual int on_reload_vhost_http_remux_updated(std::string vhost);
virtual int on_reload_vhost_hls(std::string vhost);
// interface ISrsHttpMatchHijacker

3
trunk/src/app/srs_app_ingest.cpp
View file

@ -293,7 +293,8 @@ int SrsIngester::parse()
int ret = ERROR_SUCCESS;
// parse ingesters
std::vector<SrsConfDirective*> vhosts = _srs_config->get_vhosts();
std::vector<SrsConfDirective*> vhosts;
_srs_config->get_vhosts(vhosts);
for (int i = 0; i < (int)vhosts.size(); i++) {
SrsConfDirective* vhost = vhosts[i];

4
trunk/src/core/srs_core_autofree.hpp
View file

@ -51,8 +51,8 @@ public:
/**
* auto delete the ptr.
*/
impl__SrsAutoFree(T** _ptr) {
ptr = _ptr;
impl__SrsAutoFree(T** p) {
ptr = p;
}
virtual ~impl__SrsAutoFree() {

38
trunk/src/kernel/srs_kernel_file.cpp
View file

@ -46,50 +46,50 @@ SrsFileWriter::~SrsFileWriter()
close();
}
int SrsFileWriter::open(string file)
int SrsFileWriter::open(string p)
{
int ret = ERROR_SUCCESS;
if (fd > 0) {
ret = ERROR_SYSTEM_FILE_ALREADY_OPENED;
srs_error("file %s already opened. ret=%d", _file.c_str(), ret);
srs_error("file %s already opened. ret=%d", path.c_str(), ret);
return ret;
}
int flags = O_CREAT|O_WRONLY|O_TRUNC;
mode_t mode = S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH;
if ((fd = ::open(file.c_str(), flags, mode)) < 0) {
if ((fd = ::open(p.c_str(), flags, mode)) < 0) {
ret = ERROR_SYSTEM_FILE_OPENE;
srs_error("open file %s failed. ret=%d", file.c_str(), ret);
srs_error("open file %s failed. ret=%d", p.c_str(), ret);
return ret;
}
_file = file;
path = p;
return ret;
}
int SrsFileWriter::open_append(string file)
int SrsFileWriter::open_append(string p)
{
int ret = ERROR_SUCCESS;
if (fd > 0) {
ret = ERROR_SYSTEM_FILE_ALREADY_OPENED;
srs_error("file %s already opened. ret=%d", _file.c_str(), ret);
srs_error("file %s already opened. ret=%d", path.c_str(), ret);
return ret;
}
int flags = O_APPEND|O_WRONLY;
mode_t mode = S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH;
if ((fd = ::open(file.c_str(), flags, mode)) < 0) {
if ((fd = ::open(p.c_str(), flags, mode)) < 0) {
ret = ERROR_SYSTEM_FILE_OPENE;
srs_error("open file %s failed. ret=%d", file.c_str(), ret);
srs_error("open file %s failed. ret=%d", p.c_str(), ret);
return ret;
}
_file = file;
path = p;
return ret;
}
@ -104,7 +104,7 @@ void SrsFileWriter::close()
if (::close(fd) < 0) {
ret = ERROR_SYSTEM_FILE_CLOSE;
srs_error("close file %s failed. ret=%d", _file.c_str(), ret);
srs_error("close file %s failed. ret=%d", path.c_str(), ret);
return;
}
fd = -1;
@ -135,7 +135,7 @@ int SrsFileWriter::write(void* buf, size_t count, ssize_t* pnwrite)
// TODO: FIXME: use st_write.
if ((nwrite = ::write(fd, buf, count)) < 0) {
ret = ERROR_SYSTEM_FILE_WRITE;
srs_error("write to file %s failed. ret=%d", _file.c_str(), ret);
srs_error("write to file %s failed. ret=%d", path.c_str(), ret);
return ret;
}
@ -177,23 +177,23 @@ SrsFileReader::~SrsFileReader()
close();
}
int SrsFileReader::open(string file)
int SrsFileReader::open(string p)
{
int ret = ERROR_SUCCESS;
if (fd > 0) {
ret = ERROR_SYSTEM_FILE_ALREADY_OPENED;
srs_error("file %s already opened. ret=%d", _file.c_str(), ret);
srs_error("file %s already opened. ret=%d", path.c_str(), ret);
return ret;
}
if ((fd = ::open(file.c_str(), O_RDONLY)) < 0) {
if ((fd = ::open(p.c_str(), O_RDONLY)) < 0) {
ret = ERROR_SYSTEM_FILE_OPENE;
srs_error("open file %s failed. ret=%d", file.c_str(), ret);
srs_error("open file %s failed. ret=%d", p.c_str(), ret);
return ret;
}
_file = file;
path = p;
return ret;
}
@ -208,7 +208,7 @@ void SrsFileReader::close()
if (::close(fd) < 0) {
ret = ERROR_SYSTEM_FILE_CLOSE;
srs_error("close file %s failed. ret=%d", _file.c_str(), ret);
srs_error("close file %s failed. ret=%d", path.c_str(), ret);
return;
}
fd = -1;
@ -252,7 +252,7 @@ int SrsFileReader::read(void* buf, size_t count, ssize_t* pnread)
// TODO: FIXME: use st_read.
if ((nread = ::read(fd, buf, count)) < 0) {
ret = ERROR_SYSTEM_FILE_READ;
srs_error("read from file %s failed. ret=%d", _file.c_str(), ret);
srs_error("read from file %s failed. ret=%d", path.c_str(), ret);
return ret;
}

27
trunk/src/kernel/srs_kernel_file.hpp
View file

@ -42,20 +42,26 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
class SrsFileWriter
{
private:
std::string _file;
std::string path;
int fd;
public:
SrsFileWriter();
virtual ~SrsFileWriter();
public:
/**
* open file writer, can open then close then open...
* open file writer, in truncate mode.
* @param p a string indicates the path of file to open.
*/
virtual int open(std::string file);
virtual int open(std::string p);
/**
* open file writer in append mode.
* open file writer, in append mode.
* @param p a string indicates the path of file to open.
*/
virtual int open_append(std::string p);
/**
* close current writer.
* @remark user can reopen again.
*/
virtual int open_append(std::string file);
virtual void close();
public:
virtual bool is_open();
@ -80,16 +86,21 @@ public:
class SrsFileReader
{
private:
std::string _file;
std::string path;
int fd;
public:
SrsFileReader();
virtual ~SrsFileReader();
public:
/**
* open file reader, can open then close then open...
* open file reader.
* @param p a string indicates the path of file to open.
*/
virtual int open(std::string p);
/**
* close current reader.
* @remark user can reopen again.
*/
virtual int open(std::string file);
virtual void close();
public:
// TODO: FIXME: extract interface.

74
trunk/src/kernel/srs_kernel_flv.cpp
View file

@ -332,7 +332,7 @@ SrsSharedPtrMessage* SrsSharedPtrMessage::copy()
SrsFlvEncoder::SrsFlvEncoder()
{
_fs = NULL;
reader = NULL;
tag_stream = new SrsStream();
#ifdef SRS_PERF_FAST_FLV_ENCODER
@ -356,19 +356,19 @@ SrsFlvEncoder::~SrsFlvEncoder()
#endif
}
int SrsFlvEncoder::initialize(SrsFileWriter* fs)
int SrsFlvEncoder::initialize(SrsFileWriter* fr)
{
int ret = ERROR_SUCCESS;
srs_assert(fs);
srs_assert(fr);
if (!fs->is_open()) {
if (!fr->is_open()) {
ret = ERROR_KERNEL_FLV_STREAM_CLOSED;
srs_warn("stream is not open for encoder. ret=%d", ret);
return ret;
}
_fs = fs;
reader = fr;
return ret;
}
@ -402,14 +402,14 @@ int SrsFlvEncoder::write_header(char flv_header[9])
int ret = ERROR_SUCCESS;
// write data.
if ((ret = _fs->write(flv_header, 9, NULL)) != ERROR_SUCCESS) {
if ((ret = reader->write(flv_header, 9, NULL)) != ERROR_SUCCESS) {
srs_error("write flv header failed. ret=%d", ret);
return ret;
}
// previous tag size.
char pts[] = { (char)0x00, (char)0x00, (char)0x00, (char)0x00 };
if ((ret = _fs->write(pts, 4, NULL)) != ERROR_SUCCESS) {
if ((ret = reader->write(pts, 4, NULL)) != ERROR_SUCCESS) {
return ret;
}
@ -552,7 +552,7 @@ int SrsFlvEncoder::write_tags(SrsSharedPtrMessage** msgs, int count)
iovs += 3;
}
if ((ret = _fs->writev(iovss, nb_iovss, NULL)) != ERROR_SUCCESS) {
if ((ret = reader->writev(iovss, nb_iovss, NULL)) != ERROR_SUCCESS) {
if (!srs_is_client_gracefully_close(ret)) {
srs_error("write flv tags failed. ret=%d", ret);
}
@ -683,7 +683,7 @@ int SrsFlvEncoder::write_tag(char* header, int header_size, char* tag, int tag_s
iovs[2].iov_base = pre_size;
iovs[2].iov_len = SRS_FLV_PREVIOUS_TAG_SIZE;
if ((ret = _fs->writev(iovs, 3, NULL)) != ERROR_SUCCESS) {
if ((ret = reader->writev(iovs, 3, NULL)) != ERROR_SUCCESS) {
if (!srs_is_client_gracefully_close(ret)) {
srs_error("write flv tag failed. ret=%d", ret);
}
@ -695,7 +695,7 @@ int SrsFlvEncoder::write_tag(char* header, int header_size, char* tag, int tag_s
SrsFlvDecoder::SrsFlvDecoder()
{
_fs = NULL;
reader = NULL;
tag_stream = new SrsStream();
}
@ -704,19 +704,19 @@ SrsFlvDecoder::~SrsFlvDecoder()
srs_freep(tag_stream);
}
int SrsFlvDecoder::initialize(SrsFileReader* fs)
int SrsFlvDecoder::initialize(SrsFileReader* fr)
{
int ret = ERROR_SUCCESS;
srs_assert(fs);
srs_assert(fr);
if (!fs->is_open()) {
if (!fr->is_open()) {
ret = ERROR_KERNEL_FLV_STREAM_CLOSED;
srs_warn("stream is not open for decoder. ret=%d", ret);
return ret;
}
_fs = fs;
reader = fr;
return ret;
}
@ -727,7 +727,7 @@ int SrsFlvDecoder::read_header(char header[9])
srs_assert(header);
if ((ret = _fs->read(header, 9, NULL)) != ERROR_SUCCESS) {
if ((ret = reader->read(header, 9, NULL)) != ERROR_SUCCESS) {
return ret;
}
@ -752,7 +752,7 @@ int SrsFlvDecoder::read_tag_header(char* ptype, int32_t* pdata_size, u_int32_t*
char th[11]; // tag header
// read tag header
if ((ret = _fs->read(th, 11, NULL)) != ERROR_SUCCESS) {
if ((ret = reader->read(th, 11, NULL)) != ERROR_SUCCESS) {
if (ret != ERROR_SYSTEM_FILE_EOF) {
srs_error("read flv tag header failed. ret=%d", ret);
}
@ -789,7 +789,7 @@ int SrsFlvDecoder::read_tag_data(char* data, int32_t size)
srs_assert(data);
if ((ret = _fs->read(data, size, NULL)) != ERROR_SUCCESS) {
if ((ret = reader->read(data, size, NULL)) != ERROR_SUCCESS) {
if (ret != ERROR_SYSTEM_FILE_EOF) {
srs_error("read flv tag header failed. ret=%d", ret);
}
@ -807,7 +807,7 @@ int SrsFlvDecoder::read_previous_tag_size(char previous_tag_size[4])
srs_assert(previous_tag_size);
// ignore 4bytes tag size.
if ((ret = _fs->read(previous_tag_size, 4, NULL)) != ERROR_SUCCESS) {
if ((ret = reader->read(previous_tag_size, 4, NULL)) != ERROR_SUCCESS) {
if (ret != ERROR_SYSTEM_FILE_EOF) {
srs_error("read flv previous tag size failed. ret=%d", ret);
}
@ -819,7 +819,7 @@ int SrsFlvDecoder::read_previous_tag_size(char previous_tag_size[4])
SrsFlvVodStreamDecoder::SrsFlvVodStreamDecoder()
{
_fs = NULL;
reader = NULL;
tag_stream = new SrsStream();
}
@ -828,19 +828,19 @@ SrsFlvVodStreamDecoder::~SrsFlvVodStreamDecoder()
srs_freep(tag_stream);
}
int SrsFlvVodStreamDecoder::initialize(SrsFileReader* fs)
int SrsFlvVodStreamDecoder::initialize(SrsFileReader* fr)
{
int ret = ERROR_SUCCESS;
srs_assert(fs);
srs_assert(fr);
if (!fs->is_open()) {
if (!fr->is_open()) {
ret = ERROR_KERNEL_FLV_STREAM_CLOSED;
srs_warn("stream is not open for decoder. ret=%d", ret);
return ret;
}
_fs = fs;
reader = fr;
return ret;
}
@ -857,7 +857,7 @@ int SrsFlvVodStreamDecoder::read_header_ext(char header[13])
// 9bytes header and 4bytes first previous-tag-size
int size = 13;
if ((ret = _fs->read(header, size, NULL)) != ERROR_SUCCESS) {
if ((ret = reader->read(header, size, NULL)) != ERROR_SUCCESS) {
return ret;
}
@ -891,7 +891,7 @@ int SrsFlvVodStreamDecoder::read_sequence_header_summary(int64_t* pstart, int* p
int64_t av_sequence_offset_start = -1;
int64_t av_sequence_offset_end = -1;
for (;;) {
if ((ret = _fs->read(tag_header, SRS_FLV_TAG_HEADER_SIZE, NULL)) != ERROR_SUCCESS) {
if ((ret = reader->read(tag_header, SRS_FLV_TAG_HEADER_SIZE, NULL)) != ERROR_SUCCESS) {
return ret;
}
@ -907,7 +907,7 @@ int SrsFlvVodStreamDecoder::read_sequence_header_summary(int64_t* pstart, int* p
bool is_not_av = !is_video && !is_audio;
if (is_not_av) {
// skip body and tag size.
_fs->skip(data_size + SRS_FLV_PREVIOUS_TAG_SIZE);
reader->skip(data_size + SRS_FLV_PREVIOUS_TAG_SIZE);
continue;
}
@ -926,10 +926,10 @@ int SrsFlvVodStreamDecoder::read_sequence_header_summary(int64_t* pstart, int* p
got_video = true;
if (av_sequence_offset_start < 0) {
av_sequence_offset_start = _fs->tellg() - SRS_FLV_TAG_HEADER_SIZE;
av_sequence_offset_start = reader->tellg() - SRS_FLV_TAG_HEADER_SIZE;
}
av_sequence_offset_end = _fs->tellg() + data_size + SRS_FLV_PREVIOUS_TAG_SIZE;
_fs->skip(data_size + SRS_FLV_PREVIOUS_TAG_SIZE);
av_sequence_offset_end = reader->tellg() + data_size + SRS_FLV_PREVIOUS_TAG_SIZE;
reader->skip(data_size + SRS_FLV_PREVIOUS_TAG_SIZE);
}
// audio
@ -938,16 +938,16 @@ int SrsFlvVodStreamDecoder::read_sequence_header_summary(int64_t* pstart, int* p
got_audio = true;
if (av_sequence_offset_start < 0) {
av_sequence_offset_start = _fs->tellg() - SRS_FLV_TAG_HEADER_SIZE;
av_sequence_offset_start = reader->tellg() - SRS_FLV_TAG_HEADER_SIZE;
}
av_sequence_offset_end = _fs->tellg() + data_size + SRS_FLV_PREVIOUS_TAG_SIZE;
_fs->skip(data_size + SRS_FLV_PREVIOUS_TAG_SIZE);
av_sequence_offset_end = reader->tellg() + data_size + SRS_FLV_PREVIOUS_TAG_SIZE;
reader->skip(data_size + SRS_FLV_PREVIOUS_TAG_SIZE);
}
}
// seek to the sequence header start offset.
if (av_sequence_offset_start > 0) {
_fs->lseek(av_sequence_offset_start);
reader->lseek(av_sequence_offset_start);
*pstart = av_sequence_offset_start;
*psize = (int)(av_sequence_offset_end - av_sequence_offset_start);
}
@ -959,19 +959,19 @@ int SrsFlvVodStreamDecoder::lseek(int64_t offset)
{
int ret = ERROR_SUCCESS;
if (offset >= _fs->filesize()) {
if (offset >= reader->filesize()) {
ret = ERROR_SYSTEM_FILE_EOF;
srs_warn("flv fast decoder seek overflow file, "
"size=%"PRId64", offset=%"PRId64", ret=%d",
_fs->filesize(), offset, ret);
reader->filesize(), offset, ret);
return ret;
}
if (_fs->lseek(offset) < 0) {
if (reader->lseek(offset) < 0) {
ret = ERROR_SYSTEM_FILE_SEEK;
srs_warn("flv fast decoder seek error, "
"size=%"PRId64", offset=%"PRId64", ret=%d",
_fs->filesize(), offset, ret);
reader->filesize(), offset, ret);
return ret;
}

18
trunk/src/kernel/srs_kernel_flv.hpp
View file

@ -434,7 +434,7 @@ public:
class SrsFlvEncoder
{
private:
SrsFileWriter* _fs;
SrsFileWriter* reader;
private:
SrsStream* tag_stream;
char tag_header[SRS_FLV_TAG_HEADER_SIZE];
@ -445,9 +445,9 @@ public:
/**
* initialize the underlayer file stream.
* @remark user can initialize multiple times to encode multiple flv files.
* @remark, user must free the fs, flv encoder never close/free it.
* @remark, user must free the @param fr, flv encoder never close/free it.
*/
virtual int initialize(SrsFileWriter* fs);
virtual int initialize(SrsFileWriter* fr);
public:
/**
* write flv header.
@ -512,7 +512,7 @@ private:
class SrsFlvDecoder
{
private:
SrsFileReader* _fs;
SrsFileReader* reader;
private:
SrsStream* tag_stream;
public:
@ -522,9 +522,9 @@ public:
/**
* initialize the underlayer file stream
* @remark user can initialize multiple times to decode multiple flv files.
* @remark, user must free the fs, flv decoder never close/free it.
* @remark user must free the @param fr, flv decoder never close/free it.
*/
virtual int initialize(SrsFileReader* fs);
virtual int initialize(SrsFileReader* fr);
public:
/**
* read the flv header, donot including the 4bytes previous tag size.
@ -556,7 +556,7 @@ public:
class SrsFlvVodStreamDecoder
{
private:
SrsFileReader* _fs;
SrsFileReader* reader;
private:
SrsStream* tag_stream;
public:
@ -566,9 +566,9 @@ public:
/**
* initialize the underlayer file stream
* @remark user can initialize multiple times to decode multiple flv files.
* @remark, user must free the fs, flv decoder never close/free it.
* @remark user must free the @param fr, flv decoder never close/free it.
*/
virtual int initialize(SrsFileReader* fs);
virtual int initialize(SrsFileReader* fr);
public:
/**
* read the flv header and its size.

14
trunk/src/kernel/srs_kernel_mp3.cpp
View file

@ -40,7 +40,7 @@ using namespace std;
SrsMp3Encoder::SrsMp3Encoder()
{
_fs = NULL;
writer = NULL;
tag_stream = new SrsStream();
}
@ -49,19 +49,19 @@ SrsMp3Encoder::~SrsMp3Encoder()
srs_freep(tag_stream);
}
int SrsMp3Encoder::initialize(SrsFileWriter* fs)
int SrsMp3Encoder::initialize(SrsFileWriter* fw)
{
int ret = ERROR_SUCCESS;
srs_assert(fs);
srs_assert(fw);
if (!fs->is_open()) {
if (!fw->is_open()) {
ret = ERROR_KERNEL_MP3_STREAM_CLOSED;
srs_warn("stream is not open for encoder. ret=%d", ret);
return ret;
}
_fs = fs;
writer = fw;
return ret;
}
@ -78,7 +78,7 @@ int SrsMp3Encoder::write_header()
(char)0x00, (char)0x00, (char)0x00, (char)0x00, // FrameSize
(char)0x00, (char)0x00 // Flags
};
return _fs->write(id3, sizeof(id3), NULL);
return writer->write(id3, sizeof(id3), NULL);
}
int SrsMp3Encoder::write_audio(int64_t timestamp, char* data, int size)
@ -122,6 +122,6 @@ int SrsMp3Encoder::write_audio(int64_t timestamp, char* data, int size)
return ret;
}
return _fs->write(data + stream->pos(), size - stream->pos(), NULL);
return writer->write(data + stream->pos(), size - stream->pos(), NULL);
}

7
trunk/src/kernel/srs_kernel_mp3.hpp
View file

@ -33,7 +33,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
class SrsStream;
class SrsFileWriter;
class SrsFileReader;
/**
* encode data to aac file.
@ -41,7 +40,7 @@ class SrsFileReader;
class SrsMp3Encoder
{
private:
SrsFileWriter* _fs;
SrsFileWriter* writer;
private:
SrsStream* tag_stream;
public:
@ -51,9 +50,9 @@ public:
/**
* initialize the underlayer file stream.
* @remark user can initialize multiple times to encode multiple mp3 files.
* @remark, user must free the fs, mp3 encoder never close/free it.
* @remark, user must free the @param fw, mp3 encoder never close/free it.
*/
virtual int initialize(SrsFileWriter* fs);
virtual int initialize(SrsFileWriter* fw);
public:
/**
* write mp3 id3 v2.3 header.

42
trunk/src/kernel/srs_kernel_stream.cpp
View file

@ -31,8 +31,8 @@ using namespace std;
SrsStream::SrsStream()
{
p = _bytes = NULL;
_size = 0;
p = bytes = NULL;
nb_bytes = 0;
// TODO: support both little and big endian.
srs_assert(srs_is_little_endian());
@ -42,24 +42,24 @@ SrsStream::~SrsStream()
{
}
int SrsStream::initialize(char* bytes, int size)
int SrsStream::initialize(char* b, int nb)
{
int ret = ERROR_SUCCESS;
if (!bytes) {
if (!b) {
ret = ERROR_KERNEL_STREAM_INIT;
srs_error("stream param bytes must not be NULL. ret=%d", ret);
return ret;
}
if (size <= 0) {
if (nb <= 0) {
ret = ERROR_KERNEL_STREAM_INIT;
srs_error("stream param size must be positive. ret=%d", ret);
return ret;
}
_size = size;
p = _bytes = bytes;
nb_bytes = nb;
p = bytes = b;
srs_info("init stream ok, size=%d", size);
return ret;
@ -67,29 +67,29 @@ int SrsStream::initialize(char* bytes, int size)
char* SrsStream::data()
{
return _bytes;
return bytes;
}
int SrsStream::size()
{
return _size;
return nb_bytes;
}
int SrsStream::pos()
{
return p - _bytes;
return (int)(p - bytes);
}
bool SrsStream::empty()
{
return !_bytes || (p >= _bytes + _size);
return !bytes || (p >= bytes + nb_bytes);
}
bool SrsStream::require(int required_size)
{
srs_assert(required_size > 0);
return required_size <= _size - (p - _bytes);
return required_size <= nb_bytes - (p - bytes);
}
void SrsStream::skip(int size)
@ -111,7 +111,7 @@ int16_t SrsStream::read_2bytes()
srs_assert(require(2));
int16_t value;
pp = (char*)&value;
char* pp = (char*)&value;
pp[1] = *p++;
pp[0] = *p++;
@ -123,7 +123,7 @@ int32_t SrsStream::read_3bytes()
srs_assert(require(3));
int32_t value = 0x00;
pp = (char*)&value;
char* pp = (char*)&value;
pp[2] = *p++;
pp[1] = *p++;
pp[0] = *p++;
@ -136,7 +136,7 @@ int32_t SrsStream::read_4bytes()
srs_assert(require(4));
int32_t value;
pp = (char*)&value;
char* pp = (char*)&value;
pp[3] = *p++;
pp[2] = *p++;
pp[1] = *p++;
@ -150,7 +150,7 @@ int64_t SrsStream::read_8bytes()
srs_assert(require(8));
int64_t value;
pp = (char*)&value;
char* pp = (char*)&value;
pp[7] = *p++;
pp[6] = *p++;
pp[5] = *p++;
@ -195,7 +195,7 @@ void SrsStream::write_2bytes(int16_t value)
{
srs_assert(require(2));
pp = (char*)&value;
char* pp = (char*)&value;
*p++ = pp[1];
*p++ = pp[0];
}
@ -204,7 +204,7 @@ void SrsStream::write_4bytes(int32_t value)
{
srs_assert(require(4));
pp = (char*)&value;
char* pp = (char*)&value;
*p++ = pp[3];
*p++ = pp[2];
*p++ = pp[1];
@ -215,7 +215,7 @@ void SrsStream::write_3bytes(int32_t value)
{
srs_assert(require(3));
pp = (char*)&value;
char* pp = (char*)&value;
*p++ = pp[2];
*p++ = pp[1];
*p++ = pp[0];
@ -225,7 +225,7 @@ void SrsStream::write_8bytes(int64_t value)
{
srs_assert(require(8));
pp = (char*)&value;
char* pp = (char*)&value;
*p++ = pp[7];
*p++ = pp[6];
*p++ = pp[5];
@ -238,7 +238,7 @@ void SrsStream::write_8bytes(int64_t value)
void SrsStream::write_string(string value)
{
srs_assert(require(value.length()));
srs_assert(require((int)value.length()));
memcpy(p, value.data(), value.length());
p += value.length();

14
trunk/src/kernel/srs_kernel_stream.hpp
View file

@ -41,23 +41,25 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
class SrsStream
{
private:
// current position at bytes.
char* p;
char* pp;
char* _bytes;
int _size;
// the bytes data for stream to read or write.
char* bytes;
// the total number of bytes.
int nb_bytes;
public:
SrsStream();
virtual ~SrsStream();
public:
/**
* initialize the stream from bytes.
* @bytes, the bytes to convert from/to basic types.
* @size, the size of bytes.
* @b, the bytes to convert from/to basic types.
* @nb, the size of bytes, total number of bytes for stream.
* @remark, stream never free the bytes, user must free it.
* @remark, return error when bytes NULL.
* @remark, return error when size is not positive.
*/
virtual int initialize(char* bytes, int size);
virtual int initialize(char* b, int nb);
// get the status of stream
public:
/**

16
trunk/src/kernel/srs_kernel_ts.cpp
View file

@ -2697,11 +2697,11 @@ SrsTSMuxer::~SrsTSMuxer()
close();
}
int SrsTSMuxer::open(string _path)
int SrsTSMuxer::open(string p)
{
int ret = ERROR_SUCCESS;
path = _path;
path = p;
close();
@ -3048,7 +3048,7 @@ int SrsTsCache::do_cache_avc(SrsAvcAacCodec* codec, SrsCodecSample* sample)
SrsTsEncoder::SrsTsEncoder()
{
_fs = NULL;
writer = NULL;
codec = new SrsAvcAacCodec();
sample = new SrsCodecSample();
cache = new SrsTsCache();
@ -3065,22 +3065,22 @@ SrsTsEncoder::~SrsTsEncoder()
srs_freep(context);
}
int SrsTsEncoder::initialize(SrsFileWriter* fs)
int SrsTsEncoder::initialize(SrsFileWriter* fw)
{
int ret = ERROR_SUCCESS;
srs_assert(fs);
srs_assert(fw);
if (!fs->is_open()) {
if (!fw->is_open()) {
ret = ERROR_KERNEL_FLV_STREAM_CLOSED;
srs_warn("stream is not open for encoder. ret=%d", ret);
return ret;
}
_fs = fs;
writer = fw;
srs_freep(muxer);
muxer = new SrsTSMuxer(fs, context, SrsCodecAudioAAC, SrsCodecVideoAVC);
muxer = new SrsTSMuxer(fw, context, SrsCodecAudioAAC, SrsCodecVideoAVC);
if ((ret = muxer->open("")) != ERROR_SUCCESS) {
return ret;

8
trunk/src/kernel/srs_kernel_ts.hpp
View file

@ -1561,8 +1561,9 @@ public:
public:
/**
* open the writer, donot write the PSI of ts.
* @param p a string indicates the path of ts file to mux to.
*/
virtual int open(std::string _path);
virtual int open(std::string p);
/**
* when open ts, we donot write the header(PSI),
* for user may need to update the acodec to mp3 or others,
@ -1625,7 +1626,7 @@ private:
class SrsTsEncoder
{
private:
SrsFileWriter* _fs;
SrsFileWriter* writer;
private:
SrsAvcAacCodec* codec;
SrsCodecSample* sample;
@ -1638,8 +1639,9 @@ public:
public:
/**
* initialize the underlayer file stream.
* @param fw the writer to use for ts encoder, user must free it.
*/
virtual int initialize(SrsFileWriter* fs);
virtual int initialize(SrsFileWriter* fw);
public:
/**
* write audio/video packet.