1. 启动rtmp server 监听,为每个rtmp连接创建SrsRtmpConn对象。
run_hybrid_server 函数 执行 _srs_hybrid->register_server(new SrsServerAdapter());
执行SrsServerAdapter::run;
srs_error_t SrsServerAdapter::run(SrsWaitGroup* wg)
{
srs_error_t err = srs_success;
// Initialize the whole system, set hooks to handle server level events.
if ((err = srs->initialize(NULL)) != srs_success) {
return srs_error_wrap(err, "server initialize");
}
if ((err = srs->initialize_st()) != srs_success) {
return srs_error_wrap(err, "initialize st");
}
if ((err = srs->initialize_signal()) != srs_success) {
return srs_error_wrap(err, "initialize signal");
}
if ((err = srs->listen()) != srs_success) {
return srs_error_wrap(err, "listen");
}
if ((err = srs->register_signal()) != srs_success) {
return srs_error_wrap(err, "register signal");
}
if ((err = srs->http_handle()) != srs_success) {
return srs_error_wrap(err, "http handle");
}
if ((err = srs->ingest()) != srs_success) {
return srs_error_wrap(err, "ingest");
}
if ((err = srs->start(wg)) != srs_success) {
return srs_error_wrap(err, "start");
}
return err;
}
SrsServerAdapter::run 函数:srs->listen 函数:srs指向SrsServer ;srs->listen 指向SrsServer::listen 函数
srs_error_t SrsServerAdapter::run(SrsWaitGroup* wg)
{
srs_error_t err = srs_success;
// Initialize the whole system, set hooks to handle server level events.
if ((err = srs->initialize(NULL)) != srs_success) {
return srs_error_wrap(err, "server initialize");
}
if ((err = srs->initialize_st()) != srs_success) {
return srs_error_wrap(err, "initialize st");
}
if ((err = srs->initialize_signal()) != srs_success) {
return srs_error_wrap(err, "initialize signal");
}
if ((err = srs->listen()) != srs_success) {
return srs_error_wrap(err, "listen");
}
if ((err = srs->register_signal()) != srs_success) {
return srs_error_wrap(err, "register signal");
}
if ((err = srs->http_handle()) != srs_success) {
return srs_error_wrap(err, "http handle");
}
if ((err = srs->ingest()) != srs_success) {
return srs_error_wrap(err, "ingest");
}
if ((err = srs->start(wg)) != srs_success) {
return srs_error_wrap(err, "start");
}
return err;
}
SrsServerAdapter::SrsServerAdapter()
{
srs = new SrsServer();
}
SrsServer::listen 函数调用listen_rtmp 函数。
srs_error_t SrsServer::listen()
{
srs_error_t err = srs_success;
if ((err = listen_rtmp()) != srs_success) {
return srs_error_wrap(err, "rtmp listen");
}
if ((err = listen_http_api()) != srs_success) {
return srs_error_wrap(err, "http api listen");
}
if ((err = listen_https_api()) != srs_success) {
return srs_error_wrap(err, "https api listen");
}
if ((err = listen_http_stream()) != srs_success) {
return srs_error_wrap(err, "http stream listen");
}
if ((err = listen_https_stream()) != srs_success) {
return srs_error_wrap(err, "https stream listen");
}
if ((err = listen_stream_caster()) != srs_success) {
return srs_error_wrap(err, "stream caster listen");
}
#ifdef SRS_RTC
if (!reuse_rtc_over_server_) {
// TODO: FIXME: Refine the listeners.
close_listeners(SrsListenerTcp);
if (_srs_config->get_rtc_server_tcp_enabled()) {
SrsListener* listener = new SrsBufferListener(this, SrsListenerTcp);
listeners.push_back(listener);
std::string ep = srs_int2str(_srs_config->get_rtc_server_tcp_listen());
std::string ip;
int port;
srs_parse_endpoint(ep, ip, port);
if ((err = listener->listen(ip, port)) != srs_success) {
return srs_error_wrap(err, "tcp listen %s:%d", ip.c_str(), port);
}
}
}
#endif
if ((err = conn_manager->start()) != srs_success) {
return srs_error_wrap(err, "connection manager");
}
return err;
}
SrsServer::listen_rtmp创建SrsBufferListener 类,并传入SrsServer 对象,执行listener->listen。
listener->listen指向SrsBufferListener::listen函数。
srs_error_t SrsServer::listen_rtmp()
{
srs_error_t err = srs_success;
// stream service port.
std::vector<std::string> ip_ports = _srs_config->get_listens();
srs_assert((int)ip_ports.size() > 0);
close_listeners(SrsListenerRtmpStream);
for (int i = 0; i < (int)ip_ports.size(); i++) {
SrsListener* listener = new SrsBufferListener(this, SrsListenerRtmpStream);
listeners.push_back(listener);
int port; string ip;
srs_parse_endpoint(ip_ports[i], ip, port);
if ((err = listener->listen(ip, port)) != srs_success) { //listener = new SrsBufferListener
srs_error_wrap(err, "rtmp listen %s:%d", ip.c_str(), port);
}
}
return err;
}
SrsBufferListener::listen 函数创建SrsTcpListener 对象,并传入SrsBufferListener 对象。执行listener->listen。listener->listen 函数指向SrsTcpListener ::listen
rs_error_t SrsBufferListener::listen(string i, int p)
{
srs_error_t err = srs_success;
ip = i;
port = p;
srs_freep(listener);
listener = new SrsTcpListener(this, ip, port);
if ((err = listener->listen()) != srs_success) { //listener = new SrsTcpListener
return srs_error_wrap(err, "buffered tcp listen");
}
string v = srs_listener_type2string(type);
srs_trace("%s listen at tcp://%s:%d, fd=%d", v.c_str(), ip.c_str(), port, listener->fd());
return err;
}
SrsTcpListener::listen : 执行srs_tcp_listen 进行tcp 监听。创建SrsSTCoroutine 类,并传入SrsTcpListener对象,执行SrsSTCoroutine::start 函数。
srs_error_t SrsTcpListener::listen()
{
srs_error_t err = srs_success;
if ((err = srs_tcp_listen(ip, port, &lfd)) != srs_success) {
return srs_error_wrap(err, "listen at %s:%d", ip.c_str(), port);
}
srs_freep(trd);
trd = new SrsSTCoroutine("tcp", this); // 创建SrsSTCoroutine 并传入SrsTcpListener 对象
if ((err = trd->start()) != srs_success) { // SrsSTCoroutine::start 函数
return srs_error_wrap(err, "start coroutine");
}
return err;
}
SrsSTCoroutine::start 函数。
srs_error_t SrsSTCoroutine::start()
{
return impl_->start();
}
SrsSTCoroutine::SrsSTCoroutine(string n, ISrsCoroutineHandler* h)
{
impl_ = new SrsFastCoroutine(n, h);
}
SrsFastCoroutine::start 函数:_pfn_st_thread_create 创建协程并执行pfn 函数,最终调用handler->cycle();handler->cycle() 指向传入对象SrsTcpListener的cycle函数。
srs_error_t SrsFastCoroutine::start()
{
srs_error_t err = srs_success;
if (started || disposed) {
if (disposed) {
err = srs_error_new(ERROR_THREAD_DISPOSED, "disposed");
} else {
err = srs_error_new(ERROR_THREAD_STARTED, "started");
}
if (trd_err == srs_success) {
trd_err = srs_error_copy(err);
}
return err;
}
if ((trd = (srs_thread_t)_pfn_st_thread_create(pfn, this, 1, stack_size)) == NULL) {
err = srs_error_new(ERROR_ST_CREATE_CYCLE_THREAD, "create failed");
srs_freep(trd_err);
trd_err = srs_error_copy(err);
return err;
}
started = true;
return err;
}
void* SrsFastCoroutine::pfn(void* arg)
{
SrsFastCoroutine* p = (SrsFastCoroutine*)arg;
srs_error_t err = p->cycle();
// Set the err for function pull to fetch it.
// @see https://github.com/ossrs/srs/pull/1304#issuecomment-480484151
if (err != srs_success) {
srs_freep(p->trd_err);
// It's ok to directly use it, because it's returned by st_thread_join.
p->trd_err = err;
}
return (void*)err;
}
srs_error_t SrsFastCoroutine::cycle()
{
if (_srs_context) {
if (cid_.empty()) {
cid_ = _srs_context->generate_id();
}
_srs_context->set_id(cid_);
}
srs_error_t err = handler->cycle();
if (err != srs_success) {
return srs_error_wrap(err, "coroutine cycle");
}
// Set cycle done, no need to interrupt it.
cycle_done = true;
return err;
}
SrsFastCoroutine::SrsFastCoroutine(string n, ISrsCoroutineHandler* h, SrsContextId cid)
{
name = n;
handler = h; //handler 指向SrsTcpListener
cid_ = cid;
trd = NULL;
trd_err = srs_success;
started = interrupted = disposed = cycle_done = false;
stopping_ = false;
// 0 use default, default is 64K.
stack_size = 0;
}
SrsTcpListener::cycle :srs_accept 来一个tcp 连接有socket fd 与之对应。执行 handler->on_tcp_client, handler->on_tcp_client 指向SrsBufferListener::on_tcp_client
srs_error_t SrsTcpListener::cycle()
{
srs_error_t err = srs_success;
while (true) {
if ((err = trd->pull()) != srs_success) {
return srs_error_wrap(err, "tcp listener");
}
srs_netfd_t fd = srs_accept(lfd, NULL, NULL, SRS_UTIME_NO_TIMEOUT);
if(fd == NULL){
return srs_error_new(ERROR_SOCKET_ACCEPT, "accept at fd=%d", srs_netfd_fileno(lfd));
}
if ((err = srs_fd_closeexec(srs_netfd_fileno(fd))) != srs_success) {
return srs_error_wrap(err, "set closeexec");
}
if ((err = handler->on_tcp_client(fd)) != srs_success) { //handler 指向SrsBufferListener
return srs_error_wrap(err, "handle fd=%d", srs_netfd_fileno(fd));
}
}
return err;
}
SrsBufferListener::on_tcp_client 函数执行server->accept_client ,server->accept_client 指向SrsServer::accept_client
rs_error_t SrsBufferListener::on_tcp_client(srs_netfd_t stfd)
{
srs_error_t err = server->accept_client(type, stfd);
if (err != srs_success) {
srs_warn("accept client failed, err is %s", srs_error_desc(err).c_str());
srs_freep(err);
}
return srs_success;
}
SrsServer::accept_client 函数:
srs_error_t SrsServer::accept_client(SrsListenerType type, srs_netfd_t stfd)
{
srs_error_t err = srs_success;
ISrsResource* resource = NULL;
if ((err = fd_to_resource(type, stfd, &resource)) != srs_success) {
srs_close_stfd(stfd);
if (srs_error_code(err) == ERROR_SOCKET_GET_PEER_IP && _srs_config->empty_ip_ok()) {
srs_error_reset(err);
return srs_success;
}
return srs_error_wrap(err, "fd to resource");
}
// Ignore if no resource found.
if (!resource) {
return err;
}
// directly enqueue, the cycle thread will remove the client.
conn_manager->add(resource);
ISrsStartable* conn = dynamic_cast<ISrsStartable*>(resource);
if ((err = conn->start()) != srs_success) {
return srs_error_wrap(err, "start conn coroutine");
}
return err;
}
创建SrsRtmpConn 函数。
创建SrsRtmpConn 函数。
srs_error_t SrsServer::fd_to_resource(SrsListenerType type, srs_netfd_t& stfd, ISrsResource** pr)
{
srs_error_t err = srs_success;
int fd = srs_netfd_fileno(stfd);
string ip = srs_get_peer_ip(fd);
int port = srs_get_peer_port(fd);
// for some keep alive application, for example, the keepalived,
// will send some tcp packet which we cann't got the ip,
// we just ignore it.
if (ip.empty()) {
return srs_error_new(ERROR_SOCKET_GET_PEER_IP, "ignore empty ip, fd=%d", fd);
}
// check connection limitation.
int max_connections = _srs_config->get_max_connections();
if (handler && (err = handler->on_accept_client(max_connections, (int)conn_manager->size())) != srs_success) {
return srs_error_wrap(err, "drop client fd=%d, ip=%s:%d, max=%d, cur=%d for err: %s",
fd, ip.c_str(), port, max_connections, (int)conn_manager->size(), srs_error_desc(err).c_str());
}
if ((int)conn_manager->size() >= max_connections) {
return srs_error_new(ERROR_EXCEED_CONNECTIONS, "drop fd=%d, ip=%s:%d, max=%d, cur=%d for exceed connection limits",
fd, ip.c_str(), port, max_connections, (int)conn_manager->size());
}
// avoid fd leak when fork.
// @see https://github.com/ossrs/srs/issues/518
if (true) {
int val;
if ((val = fcntl(fd, F_GETFD, 0)) < 0) {
return srs_error_new(ERROR_SYSTEM_PID_GET_FILE_INFO, "fnctl F_GETFD error! fd=%d", fd);
}
val |= FD_CLOEXEC;
if (fcntl(fd, F_SETFD, val) < 0) {
return srs_error_new(ERROR_SYSTEM_PID_SET_FILE_INFO, "fcntl F_SETFD error! fd=%d", fd);
}
}
// We will free the stfd from now on.
srs_netfd_t fd2 = stfd;
stfd = NULL;
// The context id may change during creating the bellow objects.
SrsContextRestore(_srs_context->get_id());
#ifdef SRS_RTC
// If reuse HTTP server with WebRTC TCP, peek to detect the client.
if (reuse_rtc_over_server_ && (type == SrsListenerHttpStream || type == SrsListenerHttpsStream)) {
SrsTcpConnection* skt = new SrsTcpConnection(fd2);
SrsBufferedReadWriter* io = new SrsBufferedReadWriter(skt);
uint8_t b[10]; int nn = sizeof(b);
if ((err = io->peek((char*)b, &nn)) != srs_success) {
srs_freep(io); srs_freep(skt);
return srs_error_wrap(err, "peek");
}
// If first message is BindingRequest(00 01), prefixed with length(2B), it's WebRTC client. Generally, the frame
// length minus message length should be 20, that is the header size of STUN is 20 bytes. For example:
// 00 6c # Frame length: 0x006c = 108
// 00 01 # Message Type: Binding Request(0x0001)
// 00 58 # Message Length: 0x005 = 88
// 21 12 a4 42 # Message Cookie: 0x2112a442
// 48 32 6c 61 6b 42 35 71 42 35 4a 71 # Message Transaction ID: 12 bytes
if (nn == 10 && b[0] == 0 && b[2] == 0 && b[3] == 1 && b[1] - b[5] == 20
&& b[6] == 0x21 && b[7] == 0x12 && b[8] == 0xa4 && b[9] == 0x42
) {
*pr = new SrsRtcTcpConn(io, ip, port, this);
} else {
*pr = new SrsHttpxConn(type == SrsListenerHttpsStream, this, io, http_server, ip, port);
}
return err;
}
#endif
if (type == SrsListenerRtmpStream) {
*pr = new SrsRtmpConn(this, fd2, ip, port);
} else if (type == SrsListenerHttpApi || type == SrsListenerHttpsApi) {
*pr = new SrsHttpxConn(type == SrsListenerHttpsApi, this, new SrsTcpConnection(fd2), http_api_mux, ip, port);
} else if (type == SrsListenerHttpStream || type == SrsListenerHttpsStream) {
*pr = new SrsHttpxConn(type == SrsListenerHttpsStream, this, new SrsTcpConnection(fd2), http_server, ip, port);
#ifdef SRS_RTC
} else if (type == SrsListenerTcp) {
*pr = new SrsRtcTcpConn(new SrsTcpConnection(fd2), ip, port, this);
#endif
} else {
srs_warn("close for no service handler. fd=%d, ip=%s:%d", fd, ip.c_str(), port);
srs_close_stfd(fd2);
return err;
}
return err;
}
2. SrsRtmpConn 处理。rtmp 握手,rtmp 控制及rtmp 音视频数据传输流程等处理;
SrsRtmpConn::start 函数执行trd->start; trd 指向SrsSTCoroutine。SrsSTCoroutine 类创建协程,最终执行传入this指向SrsRtmpConn类的cycle函数
srs_error_t SrsRtmpConn::start()
{
srs_error_t err = srs_success;
if ((err = trd->start()) != srs_success) {
return srs_error_wrap(err, "coroutine");
}
return err;
}
SrsRtmpConn::SrsRtmpConn(SrsServer* svr, srs_netfd_t c, string cip, int cport)
{
// Create a identify for this client.
_srs_context->set_id(_srs_context->generate_id());
server = svr;
stfd = c;
skt = new SrsTcpConnection(c);
manager = svr;
ip = cip;
port = cport;
create_time = srsu2ms(srs_get_system_time());
trd = new SrsSTCoroutine("rtmp", this, _srs_context->get_id()); 创建协程,最终执行传入this指向SrsRtmpConn类的cycle函数
kbps = new SrsNetworkKbps();
kbps->set_io(skt, skt);
delta_ = new SrsNetworkDelta();
delta_->set_io(skt, skt);
rtmp = new SrsRtmpServer(skt);
refer = new SrsRefer();
security = new SrsSecurity();
duration = 0;
wakable = NULL;
mw_sleep = SRS_PERF_MW_SLEEP;
mw_msgs = 0;
realtime = SRS_PERF_MIN_LATENCY_ENABLED;
send_min_interval = 0;
tcp_nodelay = false;
info = new SrsClientInfo();
publish_1stpkt_timeout = 0;
publish_normal_timeout = 0;
_srs_config->subscribe(this);
}
SrsRtmpConn::cycle()
do_cycle
service_cycle
stream_service_cycle
SrsRtmpConn::stream_service_cycle 函数处理:publishing 推流处理:playing 拉流处理
srs_error_t SrsRtmpConn::stream_service_cycle()
{
srs_error_t err = srs_success;
SrsRequest* req = info->req;
if ((err = rtmp->identify_client(info->res->stream_id, info->type, req->stream, req->duration)) != srs_success) {
return srs_error_wrap(err, "rtmp: identify client");
}
srs_discovery_tc_url(req->tcUrl, req->schema, req->host, req->vhost, req->app, req->stream, req->port, req->param);
// guess stream name
if (req->stream.empty()) {
string app = req->app, param = req->param;
srs_guess_stream_by_app(req->app, req->param, req->stream);
srs_trace("Guessing by app=%s, param=%s to app=%s, param=%s, stream=%s", app.c_str(), param.c_str(), req->app.c_str(), req->param.c_str(), req->stream.c_str());
}
req->strip();
srs_trace("client identified, type=%s, vhost=%s, app=%s, stream=%s, param=%s, duration=%dms",
srs_client_type_string(info->type).c_str(), req->vhost.c_str(), req->app.c_str(), req->stream.c_str(), req->param.c_str(), srsu2msi(req->duration));
// discovery vhost, resolve the vhost from config
SrsConfDirective* parsed_vhost = _srs_config->get_vhost(req->vhost);
if (parsed_vhost) {
req->vhost = parsed_vhost->arg0();
}
if (req->schema.empty() || req->vhost.empty() || req->port == 0 || req->app.empty()) {
return srs_error_new(ERROR_RTMP_REQ_TCURL, "discovery tcUrl failed, tcUrl=%s, schema=%s, vhost=%s, port=%d, app=%s",
req->tcUrl.c_str(), req->schema.c_str(), req->vhost.c_str(), req->port, req->app.c_str());
}
// check vhost, allow default vhost.
if ((err = check_vhost(true)) != srs_success) {
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, 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) {
info->edge = _srs_config->get_vhost_is_edge(req->vhost);
bool edge_traverse = _srs_config->get_vhost_edge_token_traverse(req->vhost);
if (info->edge && edge_traverse) {
if ((err = check_edge_token_traverse_auth()) != srs_success) {
return srs_error_wrap(err, "rtmp: check token traverse");
}
}
}
// security check
if ((err = security->check(info->type, ip, req)) != srs_success) {
return srs_error_wrap(err, "rtmp: security check");
}
// Never allow the empty stream name, for HLS may write to a file with empty name.
// @see https://github.com/ossrs/srs/issues/834
if (req->stream.empty()) {
return srs_error_new(ERROR_RTMP_STREAM_NAME_EMPTY, "rtmp: empty stream");
}
// client is identified, set the timeout to service timeout.
rtmp->set_recv_timeout(SRS_CONSTS_RTMP_TIMEOUT);
rtmp->set_send_timeout(SRS_CONSTS_RTMP_TIMEOUT);
// find a source to serve.
SrsLiveSource* source = NULL;
if ((err = _srs_sources->fetch_or_create(req, server, &source)) != srs_success) {
return srs_error_wrap(err, "rtmp: fetch source");
}
srs_assert(source != NULL);
bool enabled_cache = _srs_config->get_gop_cache(req->vhost);
srs_trace("source url=%s, ip=%s, cache=%d, is_edge=%d, source_id=%s/%s",
req->get_stream_url().c_str(), ip.c_str(), enabled_cache, info->edge, source->source_id().c_str(), source->pre_source_id().c_str());
source->set_cache(enabled_cache);
switch (info->type) {
case SrsRtmpConnPlay: {
// response connection start play
if ((err = rtmp->start_play(info->res->stream_id)) != srs_success) {
return srs_error_wrap(err, "rtmp: start play");
}
// We must do stat the client before hooks, because hooks depends on it.
SrsStatistic* stat = SrsStatistic::instance();
if ((err = stat->on_client(_srs_context->get_id().c_str(), req, this, info->type)) != srs_success) {
return srs_error_wrap(err, "rtmp: stat client");
}
// We must do hook after stat, because depends on it.
if ((err = http_hooks_on_play()) != srs_success) {
return srs_error_wrap(err, "rtmp: callback on play");
}
err = playing(source);
http_hooks_on_stop();
return err;
}
case SrsRtmpConnFMLEPublish: {
if ((err = rtmp->start_fmle_publish(info->res->stream_id)) != srs_success) {
return srs_error_wrap(err, "rtmp: start FMLE publish");
}
return publishing(source);
}
case SrsRtmpConnHaivisionPublish: {
if ((err = rtmp->start_haivision_publish(info->res->stream_id)) != srs_success) {
return srs_error_wrap(err, "rtmp: start HAIVISION publish");
}
return publishing(source);
}
case SrsRtmpConnFlashPublish: {
if ((err = rtmp->start_flash_publish(info->res->stream_id)) != srs_success) {
return srs_error_wrap(err, "rtmp: start FLASH publish");
}
return publishing(source);
}
default: {
return srs_error_new(ERROR_SYSTEM_CLIENT_INVALID, "rtmp: unknown client type=%d", info->type);
}
}
return err;
}