external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

Merge branch '3.0release' of https://github.com/ossrs/srs into 3.0release

Browse source 
# Conflicts:
#	trunk/src/kernel/srs_kernel_utility.hpp
This commit is contained in:
Harlan 2018-08-15 07:37:12 +08:00
commit 22c5af62cf
49 changed files with 4667 additions and 4248 deletions
Download patch file Download diff file Expand all files Collapse all files

2
trunk/src/app/srs_app_caster_flv.cpp
View file

@ -202,7 +202,7 @@ srs_error_t SrsDynamicHttpConn::do_proxy(ISrsHttpResponseReader* rr, SrsFlvDecod
return srs_error_wrap(err, "connect %s failed, cto=%" PRId64 ", sto=%" PRId64, output.c_str(), cto, sto);
}
if ((err = sdk->publish()) != srs_success) {
if ((err = sdk->publish(SRS_CONSTS_RTMP_PROTOCOL_CHUNK_SIZE)) != srs_success) {
return srs_error_wrap(err, "publish");
}

84
trunk/src/app/srs_app_conn.cpp
View file

@ -23,6 +23,7 @@
#include <srs_app_conn.hpp>
#include <netinet/tcp.h>
using namespace std;
#include <srs_kernel_log.hpp>
@ -95,6 +96,89 @@ srs_error_t SrsConnection::start()
return err;
}
srs_error_t SrsConnection::set_tcp_nodelay(bool v)
{
srs_error_t err = srs_success;
int r0 = 0;
socklen_t nb_v = sizeof(int);
int fd = srs_netfd_fileno(stfd);
int ov = 0;
if ((r0 = getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &ov, &nb_v)) != 0) {
return srs_error_new(ERROR_SOCKET_NO_NODELAY, "getsockopt fd=%d, r0=%d", fd, r0);
}
#ifndef SRS_PERF_TCP_NODELAY
srs_warn("ignore TCP_NODELAY, fd=%d, ov=%d", fd, ov);
return err;
#endif
int iv = (v? 1:0);
if ((r0 = setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &iv, nb_v)) != 0) {
return srs_error_new(ERROR_SOCKET_NO_NODELAY, "setsockopt fd=%d, r0=%v", fd, r0);
}
if ((r0 = getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &iv, &nb_v)) != 0) {
return srs_error_new(ERROR_SOCKET_NO_NODELAY, "getsockopt fd=%d, r0=%d", fd, r0);
}
srs_trace("set fd=%d TCP_NODELAY %d=>%d", fd, ov, iv);
return err;
}
srs_error_t SrsConnection::set_socket_buffer(int buffer_ms)
{
srs_error_t err = srs_success;
int r0 = 0;
int fd = srs_netfd_fileno(stfd);
socklen_t nb_v = sizeof(int);
int ov = 0;
if ((r0 = getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &ov, &nb_v)) != 0) {
return srs_error_new(ERROR_SOCKET_SNDBUF, "getsockopt fd=%d, r0=%d", fd, r0);
}
#ifndef SRS_PERF_MW_SO_SNDBUF
srs_warn("ignore SO_SNDBUF, fd=%d, ov=%d", fd, ov);
return err;
#endif
// the bytes:
// 4KB=4096, 8KB=8192, 16KB=16384, 32KB=32768, 64KB=65536,
// 128KB=131072, 256KB=262144, 512KB=524288
// the buffer should set to sleep*kbps/8,
// for example, your system delivery stream in 1000kbps,
// sleep 800ms for small bytes, the buffer should set to:
// 800*1000/8=100000B(about 128KB).
// other examples:
// 2000*3000/8=750000B(about 732KB).
// 2000*5000/8=1250000B(about 1220KB).
int kbps = 4000;
int iv = buffer_ms * kbps / 8;
// socket send buffer, system will double it.
iv = iv / 2;
// override the send buffer by macro.
#ifdef SRS_PERF_SO_SNDBUF_SIZE
iv = SRS_PERF_SO_SNDBUF_SIZE / 2;
#endif
// set the socket send buffer when required larger buffer
if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &iv, nb_v) < 0) {
return srs_error_new(ERROR_SOCKET_SNDBUF, "setsockopt fd=%d, r0=%v", fd, r0);
}
if ((r0 = getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &iv, &nb_v)) != 0) {
return srs_error_new(ERROR_SOCKET_SNDBUF, "getsockopt fd=%d, r0=%d", fd, r0);
}
srs_trace("set fd=%d, SO_SNDBUF=%d=>%d, buffer=%dms", fd, ov, iv, buffer_ms);
return err;
}
srs_error_t SrsConnection::cycle()
{
srs_error_t err = do_cycle();

4
trunk/src/app/srs_app_conn.hpp
View file

@ -100,6 +100,10 @@ public:
* to remove the client by server->remove(this).
*/
virtual srs_error_t start();
// Set socket option TCP_NODELAY.
virtual srs_error_t set_tcp_nodelay(bool v);
// Set socket option SO_SNDBUF in ms.
virtual srs_error_t set_socket_buffer(int buffer_ms);
// interface ISrsOneCycleThreadHandler
public:
/**

9
trunk/src/app/srs_app_edge.cpp
View file

@ -110,7 +110,7 @@ srs_error_t SrsEdgeRtmpUpstream::connect(SrsRequest* r, SrsLbRoundRobin* lb)
std::string vhost = _srs_config->get_vhost_edge_transform_vhost(req->vhost);
vhost = srs_string_replace(vhost, "[vhost]", req->vhost);
url = srs_generate_rtmp_url(server, port, vhost, req->app, req->stream);
url = srs_generate_rtmp_url(server, port, req->host, vhost, req->app, req->stream, req->param);
}
srs_freep(sdk);
@ -122,7 +122,7 @@ srs_error_t SrsEdgeRtmpUpstream::connect(SrsRequest* r, SrsLbRoundRobin* lb)
return srs_error_wrap(err, "edge pull %s failed, cto=%" PRId64 ", sto=%" PRId64, url.c_str(), cto, sto);
}
if ((err = sdk->play()) != srs_success) {
if ((err = sdk->play(_srs_config->get_chunk_size(req->vhost))) != srs_success) {
return srs_error_wrap(err, "edge pull %s stream failed", url.c_str());
}
@ -469,7 +469,7 @@ srs_error_t SrsEdgeForwarder::start()
std::string vhost = _srs_config->get_vhost_edge_transform_vhost(req->vhost);
vhost = srs_string_replace(vhost, "[vhost]", req->vhost);
url = srs_generate_rtmp_url(server, port, vhost, req->app, req->stream);
url = srs_generate_rtmp_url(server, port, req->host, vhost, req->app, req->stream, req->param);
}
// open socket.
@ -482,7 +482,7 @@ srs_error_t SrsEdgeForwarder::start()
return srs_error_wrap(err, "sdk connect %s failed, cto=%" PRId64 ", sto=%" PRId64, url.c_str(), cto, sto);
}
if ((err = sdk->publish()) != srs_success) {
if ((err = sdk->publish(_srs_config->get_chunk_size(req->vhost))) != srs_success) {
return srs_error_wrap(err, "sdk publish");
}
@ -492,6 +492,7 @@ srs_error_t SrsEdgeForwarder::start()
if ((err = trd->start()) != srs_success) {
return srs_error_wrap(err, "coroutine");
}
srs_trace("edge-fwr publish url %s", url.c_str());
return err;
}

40
trunk/src/app/srs_app_forward.cpp
View file

@ -94,42 +94,6 @@ srs_error_t SrsForwarder::on_publish()
{
srs_error_t err = srs_success;
// discovery the server port and tcUrl from req and ep_forward.
std::string server;
std::string tcUrl;
int port = SRS_CONSTS_RTMP_DEFAULT_PORT;
if (true) {
// parse host:port from hostport.
srs_parse_hostport(ep_forward, server, port);
// generate tcUrl
tcUrl = srs_generate_tc_url(server, req->vhost, req->app, port, req->param);
}
// dead loop check
std::string source_ep = "rtmp://";
source_ep += req->host;
source_ep += ":";
source_ep += req->port;
source_ep += "?vhost=";
source_ep += req->vhost;
std::string dest_ep = "rtmp://";
if (ep_forward == SRS_CONSTS_LOCALHOST) {
dest_ep += req->host;
} else {
dest_ep += server;
}
dest_ep += ":";
dest_ep += port;
dest_ep += "?vhost=";
dest_ep += req->vhost;
if (source_ep == dest_ep) {
return srs_error_new(ERROR_SYSTEM_FORWARD_LOOP, "forward loop detected. src=%s, dest=%s", source_ep.c_str(), dest_ep.c_str());
}
srs_trace("start forward %s to %s, tcUrl=%s, stream=%s", source_ep.c_str(), dest_ep.c_str(), tcUrl.c_str(), req->stream.c_str());
srs_freep(trd);
trd = new SrsSTCoroutine("forward", this);
if ((err = trd->start()) != srs_success) {
@ -245,7 +209,7 @@ srs_error_t SrsForwarder::do_cycle()
srs_parse_hostport(ep_forward, server, port);
// generate url
url = srs_generate_rtmp_url(server, port, req->vhost, req->app, req->stream);
url = srs_generate_rtmp_url(server, port, req->host, req->vhost, req->app, req->stream, req->param);
}
srs_freep(sdk);
@ -257,7 +221,7 @@ srs_error_t SrsForwarder::do_cycle()
return srs_error_wrap(err, "sdk connect url=%s, cto=%" PRId64 ", sto=%" PRId64, url.c_str(), cto, sto);
}
if ((err = sdk->publish()) != srs_success) {
if ((err = sdk->publish(_srs_config->get_chunk_size(req->vhost))) != srs_success) {
return srs_error_wrap(err, "sdk publish");
}

10
trunk/src/app/srs_app_fragment.cpp
View file

@ -28,6 +28,7 @@
#include <srs_kernel_error.hpp>
#include <unistd.h>
#include <sstream>
using namespace std;
SrsFragment::SrsFragment()
@ -126,11 +127,16 @@ srs_error_t SrsFragment::rename()
string full_path = fullpath();
string tmp_file = tmppath();
int tempdur = (int)duration();
if (true) {
std::stringstream ss;
ss << tempdur;
full_path = srs_string_replace(full_path, "[duration]", ss.str());
}
if (::rename(tmp_file.c_str(), full_path.c_str()) < 0) {
return srs_error_new(ERROR_SYSTEM_FRAGMENT_RENAME, "rename %s to %s", tmp_file.c_str(), full_path.c_str());
}
filepath = full_path;
return err;
}

9
trunk/src/app/srs_app_hls.cpp
View file

@ -789,7 +789,14 @@ srs_error_t SrsHlsMuxer::_refresh_m3u8(string m3u8_file)
ss << "#EXTINF:" << segment->duration() / 1000.0 << ", no desc" << SRS_CONSTS_LF;
// {file name}\n
ss << segment->uri << SRS_CONSTS_LF;
std::string seg_uri = segment->uri;
if (true) {
std::stringstream stemp;
stemp << (int)(segment->duration());
seg_uri = srs_string_replace(seg_uri, "[duration]", stemp.str());
}
//ss << segment->uri << SRS_CONSTS_LF;
ss << seg_uri << SRS_CONSTS_LF;
}
// write m3u8 to writer.

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

@ -1404,16 +1404,19 @@ srs_error_t SrsHttpApi::do_cycle()
ISrsHttpMessage* req = NULL;
// get a http message
if ((err = parser->parse_message(skt, this, &req)) != srs_success) {
if ((err = parser->parse_message(skt, &req)) != srs_success) {
return srs_error_wrap(err, "parse message");
}
// if SUCCESS, always NOT-NULL.
srs_assert(req);
// always free it in this scope.
srs_assert(req);
SrsAutoFree(ISrsHttpMessage, req);
// Attach owner connection to message.
SrsHttpMessage* hreq = (SrsHttpMessage*)req;
hreq->set_connection(this);
// ok, handle http request.
SrsHttpResponseWriter writer(skt);
if ((err = process_request(&writer, req)) != srs_success) {

18
trunk/src/app/srs_app_http_conn.cpp
View file

@ -123,19 +123,21 @@ srs_error_t SrsHttpConn::do_cycle()
ISrsHttpMessage* req = NULL;
// get a http message
if ((err = parser->parse_message(skt, this, &req)) != srs_success) {
if ((err = parser->parse_message(skt, &req)) != srs_success) {
break;
}
// if SUCCESS, always NOT-NULL.
srs_assert(req);
// always free it in this scope.
srs_assert(req);
SrsAutoFree(ISrsHttpMessage, req);
// Attach owner connection to message.
SrsHttpMessage* hreq = (SrsHttpMessage*)req;
hreq->set_connection(this);
// copy request to last request object.
srs_freep(last_req);
SrsHttpMessage* hreq = dynamic_cast<SrsHttpMessage*>(req);
last_req = hreq->to_request(hreq->host());
// may should discard the body.
@ -218,10 +220,14 @@ srs_error_t SrsResponseOnlyHttpConn::pop_message(ISrsHttpMessage** preq)
return srs_error_wrap(err, "init socket");
}
if ((err = parser->parse_message(&skt, this, preq)) != srs_success) {
if ((err = parser->parse_message(&skt, preq)) != srs_success) {
return srs_error_wrap(err, "parse message");
}
// Attach owner connection to message.
SrsHttpMessage* hreq = (SrsHttpMessage*)(*preq);
hreq->set_connection(this);
return err;
}
@ -237,8 +243,8 @@ srs_error_t SrsResponseOnlyHttpConn::on_got_http_message(ISrsHttpMessage* msg)
}
// drop all request body.
char body[4096];
while (!br->eof()) {
char body[4096];
if ((err = br->read(body, 4096, NULL)) != srs_success) {
return srs_error_wrap(err, "read response");
}

7
trunk/src/app/srs_app_http_hooks.cpp
View file

@ -141,6 +141,7 @@ srs_error_t SrsHttpHooks::on_publish(string url, SrsRequest* req)
obj->set("app", SrsJsonAny::str(req->app.c_str()));
obj->set("tcUrl", SrsJsonAny::str(req->tcUrl.c_str()));
obj->set("stream", SrsJsonAny::str(req->stream.c_str()));
obj->set("param", SrsJsonAny::str(req->param.c_str()));
std::string data = obj->dumps();
std::string res;
@ -173,6 +174,7 @@ void SrsHttpHooks::on_unpublish(string url, SrsRequest* req)
obj->set("vhost", SrsJsonAny::str(req->vhost.c_str()));
obj->set("app", SrsJsonAny::str(req->app.c_str()));
obj->set("stream", SrsJsonAny::str(req->stream.c_str()));
obj->set("param", SrsJsonAny::str(req->param.c_str()));
std::string data = obj->dumps();
std::string res;
@ -208,6 +210,7 @@ srs_error_t SrsHttpHooks::on_play(string url, SrsRequest* req)
obj->set("vhost", SrsJsonAny::str(req->vhost.c_str()));
obj->set("app", SrsJsonAny::str(req->app.c_str()));
obj->set("stream", SrsJsonAny::str(req->stream.c_str()));
obj->set("param", SrsJsonAny::str(req->param.c_str()));
obj->set("pageUrl", SrsJsonAny::str(req->pageUrl.c_str()));
std::string data = obj->dumps();
@ -241,6 +244,7 @@ void SrsHttpHooks::on_stop(string url, SrsRequest* req)
obj->set("vhost", SrsJsonAny::str(req->vhost.c_str()));
obj->set("app", SrsJsonAny::str(req->app.c_str()));
obj->set("stream", SrsJsonAny::str(req->stream.c_str()));
obj->set("param", SrsJsonAny::str(req->param.c_str()));
std::string data = obj->dumps();
std::string res;
@ -277,6 +281,7 @@ srs_error_t SrsHttpHooks::on_dvr(int cid, string url, SrsRequest* req, string fi
obj->set("vhost", SrsJsonAny::str(req->vhost.c_str()));
obj->set("app", SrsJsonAny::str(req->app.c_str()));
obj->set("stream", SrsJsonAny::str(req->stream.c_str()));
obj->set("param", SrsJsonAny::str(req->param.c_str()));
obj->set("cwd", SrsJsonAny::str(cwd.c_str()));
obj->set("file", SrsJsonAny::str(file.c_str()));
@ -318,6 +323,7 @@ srs_error_t SrsHttpHooks::on_hls(int cid, string url, SrsRequest* req, string fi
obj->set("vhost", SrsJsonAny::str(req->vhost.c_str()));
obj->set("app", SrsJsonAny::str(req->app.c_str()));
obj->set("stream", SrsJsonAny::str(req->stream.c_str()));
obj->set("param", SrsJsonAny::str(req->param.c_str()));
obj->set("duration", SrsJsonAny::number(duration));
obj->set("cwd", SrsJsonAny::str(cwd.c_str()));
obj->set("file", SrsJsonAny::str(file.c_str()));
@ -355,6 +361,7 @@ srs_error_t SrsHttpHooks::on_hls_notify(int cid, std::string url, SrsRequest* re
url = srs_string_replace(url, "[app]", req->app);
url = srs_string_replace(url, "[stream]", req->stream);
url = srs_string_replace(url, "[ts_url]", ts_url);
url = srs_string_replace(url, "[param]", req->param);
int64_t starttime = srs_update_system_time_ms();

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

@ -54,10 +54,11 @@ using namespace std;
#include <srs_app_server.hpp>
#include <srs_app_statistic.hpp>
#include <srs_app_recv_thread.hpp>
#include <srs_app_http_hooks.hpp>
SrsBufferCache::SrsBufferCache(SrsSource* s, SrsRequest* r)
{
req = r->copy();
req = r->copy()->as_http();
source = s;
queue = new SrsMessageQueue(true);
trd = new SrsSTCoroutine("http-stream", this);
@ -466,7 +467,7 @@ SrsLiveStream::SrsLiveStream(SrsSource* s, SrsRequest* r, SrsBufferCache* c)
{
source = s;
cache = c;
req = r->copy();
req = r->copy()->as_http();
}
SrsLiveStream::~SrsLiveStream()
@ -476,9 +477,10 @@ SrsLiveStream::~SrsLiveStream()
srs_error_t SrsLiveStream::update(SrsSource* s, SrsRequest* r)
{
srs_freep(req);
source = s;
req = r->copy();
srs_freep(req);
req = r->copy()->as_http();
return srs_success;
}
@ -487,24 +489,51 @@ srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage
{
srs_error_t err = srs_success;
if ((err = http_hooks_on_play()) != srs_success) {
return srs_error_wrap(err, "http hook");
}
err = do_serve_http(w, r);
http_hooks_on_stop();
return err;
}
srs_error_t SrsLiveStream::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{
srs_error_t err = srs_success;
string enc_desc;
ISrsBufferEncoder* enc = NULL;
srs_assert(entry);
if (srs_string_ends_with(entry->pattern, ".flv")) {
w->header()->set_content_type("video/x-flv");
#ifdef SRS_PERF_FAST_FLV_ENCODER
enc = new SrsFastFlvStreamEncoder();
bool realtime = _srs_config->get_realtime_enabled(req->vhost);
if (realtime) {
enc_desc = "FLV";
enc = new SrsFlvStreamEncoder();
} else {
enc_desc = "FastFLV";
enc = new SrsFastFlvStreamEncoder();
}
#else
enc_desc = "FLV";
enc = new SrsFlvStreamEncoder();
#endif
} else if (srs_string_ends_with(entry->pattern, ".aac")) {
w->header()->set_content_type("audio/x-aac");
enc_desc = "AAC";
enc = new SrsAacStreamEncoder();
} else if (srs_string_ends_with(entry->pattern, ".mp3")) {
w->header()->set_content_type("audio/mpeg");
enc_desc = "MP3";
enc = new SrsMp3StreamEncoder();
} else if (srs_string_ends_with(entry->pattern, ".ts")) {
w->header()->set_content_type("video/MP2T");
enc_desc = "TS";
enc = new SrsTsStreamEncoder();
} else {
return srs_error_new(ERROR_HTTP_LIVE_STREAM_EXT, "invalid pattern=%s", entry->pattern.c_str());
@ -552,12 +581,28 @@ srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage
SrsHttpMessage* hr = dynamic_cast<SrsHttpMessage*>(r);
SrsResponseOnlyHttpConn* hc = dynamic_cast<SrsResponseOnlyHttpConn*>(hr->connection());
// Set the socket options for transport.
bool tcp_nodelay = _srs_config->get_tcp_nodelay(req->vhost);
if (tcp_nodelay) {
if ((err = hc->set_tcp_nodelay(tcp_nodelay)) != srs_success) {
return srs_error_wrap(err, "set tcp nodelay");
}
}
int mw_sleep = _srs_config->get_mw_sleep_ms(req->vhost);
if ((err = hc->set_socket_buffer(mw_sleep)) != srs_success) {
return srs_error_wrap(err, "set mw_sleep");
}
SrsHttpRecvThread* trd = new SrsHttpRecvThread(hc);
SrsAutoFree(SrsHttpRecvThread, trd);
if ((err = trd->start()) != srs_success) {
return srs_error_wrap(err, "start recv thread");
}
srs_trace("FLV %s, encoder=%s, nodelay=%d, mw_sleep=%dms, cache=%d, msgs=%d",
entry->pattern.c_str(), enc_desc.c_str(), tcp_nodelay, mw_sleep, enc->has_cache(), msgs.max);
// TODO: free and erase the disabled entry after all related connections is closed.
while (entry->enabled) {
@ -576,17 +621,15 @@ srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage
}
if (count <= 0) {
srs_info("http: sleep %dms for no msg", SRS_CONSTS_RTMP_PULSE_TMMS);
// directly use sleep, donot use consumer wait.
srs_usleep(SRS_CONSTS_RTMP_PULSE_TMMS * 1000);
// Directly use sleep, donot use consumer wait, because we couldn't awake consumer.
srs_usleep(mw_sleep * 1000);
// ignore when nothing got.
continue;
}
if (pprint->can_print()) {
srs_info("-> " SRS_CONSTS_LOG_HTTP_STREAM " http: got %d msgs, age=%d, min=%d, mw=%d",
count, pprint->age(), SRS_PERF_MW_MIN_MSGS, SRS_CONSTS_RTMP_PULSE_TMMS);
srs_trace("-> " SRS_CONSTS_LOG_HTTP_STREAM " http: got %d msgs, age=%d, min=%d, mw=%d",
count, pprint->age(), SRS_PERF_MW_MIN_MSGS, mw_sleep);
}
// sendout all messages.
@ -615,6 +658,69 @@ srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage
return err;
}
srs_error_t SrsLiveStream::http_hooks_on_play()
{
srs_error_t err = srs_success;
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
return err;
}
// the http hooks will cause context switch,
// so we must copy all hooks for the on_connect may freed.
// @see https://github.com/ossrs/srs/issues/475
vector<string> hooks;
if (true) {
SrsConfDirective* conf = _srs_config->get_vhost_on_play(req->vhost);
if (!conf) {
return err;
}
hooks = conf->args;
}
for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i);
if ((err = SrsHttpHooks::on_play(url, req)) != srs_success) {
return srs_error_wrap(err, "rtmp on_play %s", url.c_str());
}
}
return err;
}
void SrsLiveStream::http_hooks_on_stop()
{
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
return;
}
// the http hooks will cause context switch,
// so we must copy all hooks for the on_connect may freed.
// @see https://github.com/ossrs/srs/issues/475
vector<string> hooks;
if (true) {
SrsConfDirective* conf = _srs_config->get_vhost_on_stop(req->vhost);
if (!conf) {
srs_info("ignore the empty http callback: on_stop");
return;
}
hooks = conf->args;
}
for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i);
SrsHttpHooks::on_stop(url, req);
}
return;
}
srs_error_t SrsLiveStream::streaming_send_messages(ISrsBufferEncoder* enc, SrsSharedPtrMessage** msgs, int nb_msgs)
{
srs_error_t err = srs_success;
@ -760,7 +866,7 @@ srs_error_t SrsHttpStreamServer::http_mount(SrsSource* s, SrsRequest* r)
srs_freep(tmpl->req);
tmpl->source = s;
tmpl->req = r->copy();
tmpl->req = r->copy()->as_http();
sflvs[sid] = entry;
@ -776,7 +882,7 @@ srs_error_t SrsHttpStreamServer::http_mount(SrsSource* s, SrsRequest* r)
if ((err = entry->cache->start()) != srs_success) {
return srs_error_wrap(err, "http: start stream cache failed");
}
srs_trace("http: mount flv stream for vhost=%s, mount=%s", sid.c_str(), mount.c_str());
srs_trace("http: mount flv stream for sid=%s, mount=%s", sid.c_str(), mount.c_str());
} else {
entry = sflvs[sid];
entry->stream->update(s, r);

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

@ -34,9 +34,7 @@ class SrsFlvTransmuxer;
class SrsTsTransmuxer;
/**
* for the srs http stream cache,
* for example, the audio stream cache to make android(weixin) happy.
* we start a thread to shrink the queue.
* A cache for HTTP Live Streaming encoder, to make android(weixin) happy.
*/
class SrsBufferCache : public ISrsCoroutineHandler
{
@ -60,7 +58,7 @@ public:
};
/**
* the stream encoder in some codec, for example, flv or aac.
* The encoder to transmux RTMP stream.
*/
class ISrsBufferEncoder
{
@ -94,7 +92,7 @@ public:
};
/**
* the flv stream encoder, remux rtmp stream to flv stream.
* Transmux RTMP to HTTP Live Streaming.
*/
class SrsFlvStreamEncoder : public ISrsBufferEncoder
{
@ -115,7 +113,7 @@ public:
#ifdef SRS_PERF_FAST_FLV_ENCODER
/**
* the fast flv stream encoder.
* A Fast HTTP FLV Live Streaming, to write multiple tags by writev.
* @see https://github.com/ossrs/srs/issues/405
*/
class SrsFastFlvStreamEncoder : public SrsFlvStreamEncoder
@ -132,7 +130,7 @@ public:
#endif
/**
* the ts stream encoder, remux rtmp stream to ts stream.
* Transmux RTMP to HTTP TS Streaming.
*/
class SrsTsStreamEncoder : public ISrsBufferEncoder
{
@ -152,7 +150,7 @@ public:
};
/**
* the aac stream encoder, remux rtmp stream to aac stream.
* Transmux RTMP with AAC stream to HTTP AAC Streaming.
*/
class SrsAacStreamEncoder : public ISrsBufferEncoder
{
@ -173,7 +171,7 @@ public:
};
/**
* the mp3 stream encoder, remux rtmp stream to mp3 stream.
* Transmux RTMP with MP3 stream to HTTP MP3 Streaming.
*/
class SrsMp3StreamEncoder : public ISrsBufferEncoder
{
@ -215,8 +213,7 @@ public:
};
/**
* the flv live stream supports access rtmp in flv over http.
* srs will remux rtmp to flv streaming.
* HTTP Live Streaming, to transmux RTMP to HTTP FLV or other format.
*/
class SrsLiveStream : public ISrsHttpHandler
{
@ -231,11 +228,14 @@ public:
public:
virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
private:
virtual srs_error_t do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
virtual srs_error_t http_hooks_on_play();
virtual void http_hooks_on_stop();
virtual srs_error_t streaming_send_messages(ISrsBufferEncoder* enc, SrsSharedPtrMessage** msgs, int nb_msgs);
};
/**
* the srs live entry
* The Live Entry, to handle HTTP Live Streaming.
*/
struct SrsLiveEntry
{
@ -264,8 +264,7 @@ public:
};
/**
* the http stream server instance,
* serve http stream, for example, flv/ts/mp3/aac live stream.
* The HTTP Live Streaming Server, to serve FLV/TS/MP3/AAC stream.
*/
// TODO: Support multiple stream.
class SrsHttpStreamServer : virtual public ISrsReloadHandler

2
trunk/src/app/srs_app_mpegts_udp.cpp
View file

@ -626,7 +626,7 @@ srs_error_t SrsMpegtsOverUdp::connect()
return srs_error_wrap(err, "connect %s failed, cto=%" PRId64 ", sto=%" PRId64, output.c_str(), cto, sto);
}
if ((err = sdk->publish()) != srs_success) {
if ((err = sdk->publish(SRS_CONSTS_RTMP_PROTOCOL_CHUNK_SIZE)) != srs_success) {
close();
return srs_error_wrap(err, "publish");
}

2
trunk/src/app/srs_app_ng_exec.cpp
View file

@ -221,7 +221,7 @@ string SrsNgExec::parse(SrsRequest* req, string tmpl)
output = srs_string_replace(output, "[pageUrl]", req->pageUrl);
if (output.find("[url]") != string::npos) {
string url = srs_generate_rtmp_url(req->host, req->port, req->vhost, req->app, req->stream);
string url = srs_generate_rtmp_url(req->host, req->port, req->host, req->vhost, req->app, req->stream, req->param);
output = srs_string_replace(output, "[url]", url);
}

79
trunk/src/app/srs_app_rtmp_conn.cpp
View file

@ -470,9 +470,9 @@ srs_error_t SrsRtmpConn::stream_service_cycle()
srs_discovery_tc_url(req->tcUrl, req->schema, req->host, req->vhost, req->app, req->stream, req->port, req->param);
req->strip();
srs_trace("client identified, type=%s, vhost=%s, app=%s, stream_name=%s, duration=%.2f",
srs_client_type_string(info->type).c_str(), req->vhost.c_str(), req->app.c_str(), req->stream.c_str(), req->duration);
srs_trace("client identified, type=%s, vhost=%s, app=%s, stream=%s, param=%s, duration=%.2f",
srs_client_type_string(info->type).c_str(), req->vhost.c_str(), req->app.c_str(), req->stream.c_str(), req->param.c_str(), req->duration);
// discovery vhost, resolve the vhost from config
SrsConfDirective* parsed_vhost = _srs_config->get_vhost(req->vhost);
if (parsed_vhost) {
@ -489,11 +489,10 @@ srs_error_t SrsRtmpConn::stream_service_cycle()
return srs_error_wrap(err, "check vhost");
}
srs_trace("connected stream, tcUrl=%s, pageUrl=%s, swfUrl=%s, schema=%s, vhost=%s, port=%d, app=%s, stream=%s, args=%s",
req->tcUrl.c_str(), req->pageUrl.c_str(), req->swfUrl.c_str(),
req->schema.c_str(), req->vhost.c_str(), req->port,
req->app.c_str(), req->stream.c_str(), (req->args? "(obj)":"null"));
srs_trace("connected stream, tcUrl=%s, pageUrl=%s, swfUrl=%s, schema=%s, vhost=%s, port=%d, app=%s, stream=%s, param=%s, args=%s",
req->tcUrl.c_str(), req->pageUrl.c_str(), req->swfUrl.c_str(), req->schema.c_str(), req->vhost.c_str(), req->port,
req->app.c_str(), req->stream.c_str(), req->param.c_str(), (req->args? "(obj)":"null"));
// do token traverse before serve it.
// @see https://github.com/ossrs/srs/pull/239
if (true) {
@ -1137,48 +1136,7 @@ void SrsRtmpConn::change_mw_sleep(int sleep_ms)
return;
}
// get the sock buffer size.
int fd = srs_netfd_fileno(stfd);
int onb_sbuf = 0;
socklen_t sock_buf_size = sizeof(int);
getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &onb_sbuf, &sock_buf_size);
#ifdef SRS_PERF_MW_SO_SNDBUF
// the bytes:
// 4KB=4096, 8KB=8192, 16KB=16384, 32KB=32768, 64KB=65536,
// 128KB=131072, 256KB=262144, 512KB=524288
// the buffer should set to sleep*kbps/8,
// for example, your system delivery stream in 1000kbps,
// sleep 800ms for small bytes, the buffer should set to:
// 800*1000/8=100000B(about 128KB).
// other examples:
// 2000*3000/8=750000B(about 732KB).
// 2000*5000/8=1250000B(about 1220KB).
int kbps = 5000;
int socket_buffer_size = sleep_ms * kbps / 8;
// socket send buffer, system will double it.
int nb_sbuf = socket_buffer_size / 2;
// override the send buffer by macro.
#ifdef SRS_PERF_SO_SNDBUF_SIZE
nb_sbuf = SRS_PERF_SO_SNDBUF_SIZE / 2;
#endif
// set the socket send buffer when required larger buffer
if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &nb_sbuf, sock_buf_size) < 0) {
srs_warn("set sock SO_SENDBUF=%d failed.", nb_sbuf);
}
getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &nb_sbuf, &sock_buf_size);
srs_trace("mw changed sleep %d=>%d, max_msgs=%d, esbuf=%d, sbuf %d=>%d, realtime=%d",
mw_sleep, sleep_ms, SRS_PERF_MW_MSGS, socket_buffer_size,
onb_sbuf, nb_sbuf, realtime);
#else
srs_trace("mw changed sleep %d=>%d, max_msgs=%d, sbuf %d, realtime=%d",
mw_sleep, sleep_ms, SRS_PERF_MW_MSGS, onb_sbuf, realtime);
#endif
set_socket_buffer(sleep_ms);
mw_sleep = sleep_ms;
}
@ -1189,25 +1147,12 @@ void SrsRtmpConn::set_sock_options()
bool nvalue = _srs_config->get_tcp_nodelay(req->vhost);
if (nvalue != tcp_nodelay) {
tcp_nodelay = nvalue;
#ifdef SRS_PERF_TCP_NODELAY
int fd = srs_netfd_fileno(stfd);
socklen_t nb_v = sizeof(int);
int ov = 0;
getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &ov, &nb_v);
int v = tcp_nodelay;
// set the socket send buffer when required larger buffer
if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &v, nb_v) < 0) {
srs_warn("set sock TCP_NODELAY=%d failed.", v);
srs_error_t err = set_tcp_nodelay(tcp_nodelay);
if (err != srs_success) {
srs_warn("ignore err %s", srs_error_desc(err).c_str());
srs_freep(err);
}
getsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &v, &nb_v);
srs_trace("set TCP_NODELAY %d=>%d", ov, v);
#else
srs_warn("SRS_PERF_TCP_NODELAY is disabled but tcp_nodelay configed.");
#endif
}
}

2
trunk/src/app/srs_app_rtsp.cpp
View file

@ -661,7 +661,7 @@ srs_error_t SrsRtspConn::connect()
}
// publish.
if ((err = sdk->publish()) != srs_success) {
if ((err = sdk->publish(SRS_CONSTS_RTMP_PROTOCOL_CHUNK_SIZE)) != srs_success) {
close();
return srs_error_wrap(err, "publish %s failed", url.c_str());
}

1
trunk/src/app/srs_app_server.cpp
View file

@ -653,6 +653,7 @@ srs_error_t SrsServer::acquire_pid_file()
if (fcntl(fd, F_SETLK, &lock) == -1) {
if(errno == EACCES || errno == EAGAIN) {
::close(fd);
srs_error("srs is already running!");
return srs_error_new(ERROR_SYSTEM_PID_ALREADY_RUNNING, "srs is already running");
}

4
trunk/src/app/srs_app_source.cpp
View file

@ -1482,8 +1482,8 @@ srs_error_t SrsOriginHub::create_forwarders()
}
// TODO: FIXME: support queue size.
//double queue_size = _srs_config->get_queue_length(req->vhost);
//forwarder->set_queue_size(queue_size);
double queue_size = _srs_config->get_queue_length(req->vhost);
forwarder->set_queue_size(queue_size);
if ((err = forwarder->on_publish()) != srs_success) {
return srs_error_wrap(err, "start forwarder failed, vhost=%s, app=%s, stream=%s, forward-to=%s",