srs代码学习（4）-怎么转发流

publish的流和play的流怎么连接呢？这个恐怕是最绕的地方了。看了一上午的代码，淹没于各种数据结构与流程之中后，俺终于发现了连接publish和play的关键连个类是

SrsSource

SrsConsumer

负责连接着连个类实例的是

SrsRtmpConn

下面我们详细讲解连接过程

上片我们说到。在底层客户端连接上来后，会经过一系列处理，最后绕到SrsRtmpConn类的循环函数中。就是下面的函数

[cpp] view plaincopy

int SrsConnection::cycle()
{
int ret = ERROR_SUCCESS;
_srs_context->generate_id();
id = _srs_context->get_id();
ip = srs_get_peer_ip(st_netfd_fileno(stfd));
ret = do_cycle();
// if socket io error, set to closed.
if (srs_is_client_gracefully_close(ret)) {
ret = ERROR_SOCKET_CLOSED;
}
// success.
if (ret == ERROR_SUCCESS) {
srs_trace("client finished.");
}
// client close peer.
if (ret == ERROR_SOCKET_CLOSED) {
srs_warn("client disconnect peer. ret=%d", ret);
}
return ERROR_SUCCESS;
}

这个是SrsRtmpConn的基类SrsConnection的函数。在基类里，do_cycle（）是个纯虚函数。具体实现完全是靠这子类来的。

那么rtmp类型的这个子类，到底有多么的变态呢，先看看我画的一个流程图，都没有画完。一张放不下，的截好几张

够长的，这里我还只是画到了播放的时候，发布流程还没有画。因为太复杂了。

下面开始一步一步的分析

首先看do_cycle()函数这个函数主要负责握手和连命令。并在成功后。获取流的配置信息。关键代码如下

[cpp] view plaincopy

if ((ret = rtmp->handshake()) != ERROR_SUCCESS) {
srs_error("rtmp handshake failed. ret=%d", ret);
return ret;
}
srs_verbose("rtmp handshake success");
if ((ret = rtmp->connect_app(req)) != ERROR_SUCCESS) {
srs_error("rtmp connect vhost/app failed. ret=%d", ret);
return ret;
}
srs_verbose("rtmp connect app success");

注意这里有一个比较重要的数据结构

[cpp] view plaincopy

SrsRequest* req

这个主要是存储请求信息的，比如app turl streamid等等。

在各种分析后，进入下一个cycle，service_cycle()函数

service_cycly()函数在做了一些设置工作，设置比如chunk size。代码如下

[cpp] view plaincopy

if ((ret = rtmp->set_window_ack_size((int)(2.5 * 1000 * 1000))) != ERROR_SUCCESS) {
srs_error("set window acknowledgement size failed. ret=%d", ret);
return ret;
}
srs_verbose("set window acknowledgement size success");
if ((ret = rtmp->set_peer_bandwidth((int)(2.5 * 1000 * 1000), 2)) != ERROR_SUCCESS) {
srs_error("set peer bandwidth failed. ret=%d", ret);
return ret;
}

下面一段代码没有看明白。这个是一个补丁打上去的，说说为了做do token traverse。这个暂时先不研究了。

[cpp] view plaincopy

if (true) {
bool vhost_is_edge = _srs_config->get_vhost_is_edge(req->vhost);
bool edge_traverse = _srs_config->get_vhost_edge_token_traverse(req->vhost);
if (vhost_is_edge && edge_traverse) {
if ((ret = check_edge_token_traverse_auth()) != ERROR_SUCCESS) {
srs_warn("token auth failed, ret=%d", ret);
return ret;
}
}
}

接着设置chunk 的大小

[cpp] view plaincopy

int chunk_size = _srs_config->get_chunk_size(req->vhost);
if ((ret = rtmp->set_chunk_size(chunk_size)) != ERROR_SUCCESS) {
srs_error("set chunk_size=%d failed. ret=%d", chunk_size, ret);
return ret;
}

回应客户端。连接ok

[cpp] view plaincopy

if ((ret = rtmp->response_connect_app(req, local_ip.c_str())) != ERROR_SUCCESS) {
srs_error("response connect app failed. ret=%d", ret);
return ret;
}

然后连接就结束了，进入stream_service_cycle()函数，从名字上就可以看出。这个函数是开始就如流命令时代

[cpp] view plaincopy

while (!disposed) {
ret = stream_service_cycle();
// stream service must terminated with error, never success.
// when terminated with success, it's user required to stop.
if (ret == ERROR_SUCCESS) {
continue;
}
// when not system control error, fatal error, return.
if (!srs_is_system_control_error(ret)) {
if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
srs_error("stream service cycle failed. ret=%d", ret);
}
return ret;
}
// for republish, continue service
if (ret == ERROR_CONTROL_REPUBLISH) {
// set timeout to a larger value, wait for encoder to republish.
rtmp->set_send_timeout(SRS_REPUBLISH_RECV_TIMEOUT_US);
rtmp->set_recv_timeout(SRS_REPUBLISH_SEND_TIMEOUT_US);
srs_trace("control message(unpublish) accept, retry stream service.");
continue;
}
// for "some" system control error,
// logical accept and retry stream service.
if (ret == ERROR_CONTROL_RTMP_CLOSE) {
// TODO: FIXME: use ping message to anti-death of socket.
// @see: https://github.com/ossrs/srs/issues/39
// set timeout to a larger value, for user paused.
rtmp->set_recv_timeout(SRS_PAUSED_RECV_TIMEOUT_US);
rtmp->set_send_timeout(SRS_PAUSED_SEND_TIMEOUT_US);
srs_trace("control message(close) accept, retry stream service.");
continue;
}
// for other system control message, fatal error.
srs_error("control message(%d) reject as error. ret=%d", ret, ret);
return ret;
}

stream_service_cycle()函数闪亮登场

首先进行一些安全验证

[cpp] view plaincopy

f ((ret = rtmp->identify_client(res->stream_id, type, req->stream, req->duration)) != ERROR_SUCCESS) {
if (!srs_is_client_gracefully_close(ret)) {
srs_error("identify client failed. ret=%d", ret);
}
return ret;
}
req->strip();
srs_trace("client identified, type=%s, stream_name=%s, duration=%.2f",
srs_client_type_string(type).c_str(), req->stream.c_str(), req->duration);
// security check
if ((ret = security->check(type, ip, req)) != ERROR_SUCCESS) {
srs_error("security check failed. ret=%d", ret);
return ret;
}
srs_info("security check ok");

然后进入比较有意思的环节

[cpp] view plaincopy

SrsSource* source = SrsSource::fetch(req);
if (!source) {
if ((ret = SrsSource::create(req, server, server, &source)) != ERROR_SUCCESS) {
return ret;
}
}
srs_assert(source != NULL);

根据req，寻找是否有这个源，如果没有，那么久创建一个。主要creat（）是个静态函数。实现代码为

[cpp] view plaincopy

int SrsSource::create(SrsRequest* r, ISrsSourceHandler* h, ISrsHlsHandler* hh, SrsSource** pps)
{
int ret = ERROR_SUCCESS;
string stream_url = r->get_stream_url();
string vhost = r->vhost;
// should always not exists for create a source.
srs_assert (pool.find(stream_url) == pool.end());
SrsSource* source = new SrsSource();
if ((ret = source->initialize(r, h, hh)) != ERROR_SUCCESS) {
srs_freep(source);
return ret;
}
pool[stream_url] = source;
srs_info("create new source for url=%s, vhost=%s", stream_url.c_str(), vhost.c_str());
*pps = source;
return ret;
}

创建一个新的source，并且放到poo中。pool是什么

[cpp] view plaincopy

static std::map<std::string, SrsSource*> pool;

也是一个全局的静态变量，用了存储所欲的源。到此。谜底进一步解开了。

同意fetch（）函数也是静态的。

[cpp] view plaincopy

static SrsSource* fetch(SrsRequest* r);
static SrsSource* fetch(std::string vhost, std::string app, std::string stream);

ok!在绕道循环函数看看接下来该怎么办

[cpp] view plaincopy

SrsStatistic* stat = SrsStatistic::instance();
if ((ret = stat->on_client(_srs_context->get_id(), req, this, type)) != ERROR_SUCCESS) {
srs_error("stat client failed. ret=%d", ret);
return ret;
}

这个是做统计用的，没啥。

[cpp] view plaincopy

bool vhost_is_edge = _srs_config->get_vhost_is_edge(req->vhost);
bool enabled_cache = _srs_config->get_gop_cache(req->vhost);
srs_trace("source url=%s, ip=%s, cache=%d, is_edge=%d, source_id=%d[%d]",
req->get_stream_url().c_str(), ip.c_str(), enabled_cache, vhost_is_edge,
source->source_id(), source->source_id());
source->set_cache(enabled_cache);

判断是否是边缘节点，是否需要gop缓冲。无他

[cpp] view plaincopy

switch (type) {
case SrsRtmpConnPlay: {
srs_verbose("start to play stream %s.", req->stream.c_str());
// response connection start play
if ((ret = rtmp->start_play(res->stream_id)) != ERROR_SUCCESS) {
srs_error("start to play stream failed. ret=%d", ret);
return ret;
}
if ((ret = http_hooks_on_play()) != ERROR_SUCCESS) {
srs_error("http hook on_play failed. ret=%d", ret);
return ret;
}
srs_info("start to play stream %s success", req->stream.c_str());
ret = playing(source);
http_hooks_on_stop();
return ret;
}
case SrsRtmpConnFMLEPublish: {
srs_verbose("FMLE start to publish stream %s.", req->stream.c_str());
if ((ret = rtmp->start_fmle_publish(res->stream_id)) != ERROR_SUCCESS) {
srs_error("start to publish stream failed. ret=%d", ret);
return ret;
}
return publishing(source);
}
case SrsRtmpConnFlashPublish: {
srs_verbose("flash start to publish stream %s.", req->stream.c_str());
if ((ret = rtmp->start_flash_publish(res->stream_id)) != ERROR_SUCCESS) {
srs_error("flash start to publish stream failed. ret=%d", ret);
return ret;
}
return publishing(source);
}
default: {
ret = ERROR_SYSTEM_CLIENT_INVALID;
srs_info("invalid client type=%d. ret=%d", type, ret);
return ret;
}
}

大流程看。好像是根据不同走到了发布或者播放流程里。但首先。这个type是从哪里来的。怎么没有发现呢？

[cpp] view plaincopy

int SrsRtmpServer::identify_client(int stream_id, SrsRtmpConnType& type, string& stream_name, double& duration)

在这个函数里做确认类型的。rmptserver类不在我们这次分析。

我们分析下play的流程，函数名称为

[cpp] view plaincopy

int SrsRtmpConn::playing(SrsSource* source)

关键代码

[cpp] view plaincopy

SrsConsumer* consumer = NULL;
if ((ret = source->create_consumer(this, consumer)) != ERROR_SUCCESS) {
srs_error("create consumer failed. ret=%d", ret);
return ret;
}
SrsAutoFree(SrsConsumer, consumer);
srs_verbose("consumer created success.");

利用source创建一个consumer.创建代码为

[cpp] view plaincopy

int SrsSource::create_consumer(SrsConnection* conn, SrsConsumer*& consumer, bool ds, bool dm, bool dg)
{
int ret = ERROR_SUCCESS;
consumer = new SrsConsumer(this, conn);
consumers.push_back(consumer);
double queue_size = _srs_config->get_queue_length(_req->vhost);
consumer->set_queue_size(queue_size);
// if atc, update the sequence header to gop cache time.
if (atc && !gop_cache->empty()) {
if (cache_metadata) {
cache_metadata->timestamp = gop_cache->start_time();
}
if (cache_sh_video) {
cache_sh_video->timestamp = gop_cache->start_time();
}
if (cache_sh_audio) {
cache_sh_audio->timestamp = gop_cache->start_time();
}
}
// copy metadata.
if (dm && cache_metadata && (ret = consumer->enqueue(cache_metadata, atc, jitter_algorithm)) != ERROR_SUCCESS) {
srs_error("dispatch metadata failed. ret=%d", ret);
return ret;
}
srs_info("dispatch metadata success");
// copy sequence header
// copy audio sequence first, for hls to fast parse the "right" audio codec.
// @see https://github.com/ossrs/srs/issues/301
if (ds && cache_sh_audio && (ret = consumer->enqueue(cache_sh_audio, atc, jitter_algorithm)) != ERROR_SUCCESS) {
srs_error("dispatch audio sequence header failed. ret=%d", ret);
return ret;
}
srs_info("dispatch audio sequence header success");
if (ds && cache_sh_video && (ret = consumer->enqueue(cache_sh_video, atc, jitter_algorithm)) != ERROR_SUCCESS) {
srs_error("dispatch video sequence header failed. ret=%d", ret);
return ret;
}
srs_info("dispatch video sequence header success");
// copy gop cache to client.
if (dg && (ret = gop_cache->dump(consumer, atc, jitter_algorithm)) != ERROR_SUCCESS) {
return ret;
}
// print status.
if (dg) {
srs_trace("create consumer, queue_size=%.2f, jitter=%d", queue_size, jitter_algorithm);
} else {
srs_trace("create consumer, ignore gop cache, jitter=%d", jitter_algorithm);
}
// for edge, when play edge stream, check the state
if (_srs_config->get_vhost_is_edge(_req->vhost)) {
// notice edge to start for the first client.
if ((ret = play_edge->on_client_play()) != ERROR_SUCCESS) {
srs_error("notice edge start play stream failed. ret=%d", ret);
return ret;
}
}
return ret;
}

代码好长。主要是创建放进数据结构中，并拷贝一些metadata进去，对于edge的处理，还没有看明白。

之后的动作

[cpp] view plaincopy

SrsQueueRecvThread trd(consumer, rtmp, SRS_PERF_MW_SLEEP);
// start isolate recv thread.
if ((ret = trd.start()) != ERROR_SUCCESS) {
srs_error("start isolate recv thread failed. ret=%d", ret);
return ret;
}

什么？单独创建了一个接受线程，实现了recv never send,send never recv,据说这样效率提高了33%

在绕回去

[cpp] view plaincopy

// delivery messages for clients playing stream.
wakable = consumer;
ret = do_playing(source, consumer, &trd);
wakable = NULL;

进入下一个循环体do_playing()

在分析下一个函数之前。让我总结下缩做的工作

1）创建或者获取了一个source

2）创建一个consumer

3) 创建一个接受线程

下面开始看函数的关键代码

[cpp] view plaincopy

// setup the realtime.
realtime = _srs_config->get_realtime_enabled(req->vhost);
// setup the mw config.
// when mw_sleep changed, resize the socket send buffer.
mw_enabled = true;
change_mw_sleep(_srs_config->get_mw_sleep_ms(req->vhost));
// initialize the send_min_interval
send_min_interval = _srs_config->get_send_min_interval(req->vhost);

做实时性 merge write 的设置

[cpp] view plaincopy

while (!trd->empty()) {
SrsCommonMessage* msg = trd->pump();
srs_verbose("pump client message to process.");
if ((ret = process_play_control_msg(consumer, msg)) != ERROR_SUCCESS) {
if (!srs_is_system_control_error(ret) && !srs_is_client_gracefully_close(ret)) {
srs_error("process play control message failed. ret=%d", ret);
}
return ret;
}
}

首先处理接受消息。主要是暂停的消息。

下面进入核心代码

[cpp] view plaincopy

<strong><span style="color:#ff6666;"> int count = (send_min_interval > 0)? 1 : 0;
if ((ret = consumer->dump_packets(&msgs, count)) != ERROR_SUCCESS) {
srs_error("get messages from consumer failed. ret=%d", ret);
return ret;
}</span></strong>

这段代码的作用就是，把消息，从consumer里，拷贝到本地的msg队列里。当然。这个拷贝是浅拷贝，只是指针过来了。

首先看msgs的定义

[cpp] view plaincopy

SrsMessageArray msgs(SRS_PERF_MW_MSGS)

这个类里有个核心变量

[cpp] view plaincopy

SrsSharedPtrMessage** msgs;

可以看到它保存的是一个指向指针的指针。

那么dump_packets是怎么实现的呢？

[cpp] view plaincopy

int SrsConsumer::dump_packets(SrsMessageArray* msgs, int& count)
{
int ret =ERROR_SUCCESS;
srs_assert(count >= 0);
srs_assert(msgs->max > 0);
// the count used as input to reset the max if positive.
int max = count? srs_min(count, msgs->max) : msgs->max;
// the count specifies the max acceptable count,
// here maybe 1+, and we must set to 0 when got nothing.
count = 0;
if (should_update_source_id) {
srs_trace("update source_id=%d[%d]", source->source_id(), source->source_id());
should_update_source_id = false;
}
// paused, return nothing.
if (paused) {
return ret;
}
// pump msgs from queue.
if ((ret = queue->dump_packets(max, msgs->msgs, count)) != ERROR_SUCCESS) {
return ret;
}
return ret;
}

[cpp] view plaincopy

int SrsMessageQueue::dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count)
{
int ret = ERROR_SUCCESS;
int nb_msgs = (int)msgs.size();
if (nb_msgs <= 0) {
return ret;
}
srs_assert(max_count > 0);
count = srs_min(max_count, nb_msgs);
SrsSharedPtrMessage** omsgs = msgs.data();
for (int i = 0; i < count; i++) {
pmsgs[i] = omsgs[i];
}
SrsSharedPtrMessage* last = omsgs[count - 1];
av_start_time = last->timestamp;
if (count >= nb_msgs) {
// the pmsgs is big enough and clear msgs at most time.
msgs.clear();
} else {
// erase some vector elements may cause memory copy,
// maybe can use more efficient vector.swap to avoid copy.
// @remark for the pmsgs is big enough, for instance, SRS_PERF_MW_MSGS 128,
// the rtmp play client will get 128msgs once, so this branch rarely execute.
msgs.erase(msgs.begin(), msgs.begin() + count);
}
return ret;
}

ok代码我就不想分析了。只是个指针拷贝。

下一个问题。consumer的数据是怎么来的呢？

看source的代码,比如音频数据

[cpp] view plaincopy

int SrsSource::on_audio_imp(SrsSharedPtrMessage* msg)

代码里有这么一段

[cpp] view plaincopy

// copy to all consumer
if (!drop_for_reduce) {
for (int i = 0; i < (int)consumers.size(); i++) {
SrsConsumer* consumer = consumers.at(i);
if ((ret = consumer->enqueue(msg, atc, jitter_algorithm)) != ERROR_SUCCESS) {
srs_error("dispatch the audio failed. ret=%d", ret);
return ret;
}
}
srs_info("dispatch audio success.");
}

到此，一个循环就结束了。

srs代码学习（4）-怎么转发流相关推荐

srs代码学习（2）- 线程模型
代码阅读到现在.发现srs有两大类线程.一个是主线程的逻辑. 一个是监听线程簇.结构图如下一定还有第三种线程模型,负责底层的多路分发.今天还没有发现. 2016.08.25--21:00 仔细阅读了 ...
srs代码学习（1）--listen建立过程
srs的服务侦听的建立过程. 以rtmp服务为例 srs服务侦听的建立依靠从上到下的三个类.分别是 SrsServer SrsStreamListener SrsTcpListener 端口侦听过程为 ...
2016年大数据Spark“蘑菇云”行动代码学习之AdClickedStreamingStats模块分析
2016年大数据Spark"蘑菇云"行动代码学习之AdClickedStreamingStats模块分析系统背景:用户使用终端设备(IPAD.手机.浏览器)等登录系统,系 ...
SRS 代码分析【mpeg-ts解析】
SRS 代码分析[mpeg-ts解析] 1.SrsTsContext的decode接口定义如下: int SrsTsContext::decode(SrsBuffer* stream, ISrsTsH ...
VTM10.0代码学习5：coding_unit()cu_pred_data()
此系列是为了记录自己学习VTM10.0的过程和锻炼表达能力,主要是从解码端进行入手.由于本人水平有限,出现的错误恳请大家指正,欢迎与大家一起交流进步. 上一篇博客(VTM10.0代码学习4)讲述了将语 ...
VTM10.0代码学习10：EncGOP_compressGOP()
此系列是为了记录自己学习VTM10.0的过程,目前正在看编码端.主要的参考文档有JVET-S2001-vH和JVET-S2002-v1.由于本人水平有限,出现的错误恳请大家指正,欢迎与大家一起交流进步 ...
社区圆桌分享：代码学习无止境，程序员如何规划自己的职业生涯发展？
CloudWeGo Study Group 是由 CloudWeGo 社区发起的学习小组,开展以 30 天为一期的源码解读和学习活动,帮助新成员融入社区圈子,和社区 Committer 互动交流,并学 ...
Android第一行代码学习思考笔记(碎片、广播、持久化技术和Android数据库)
Android第一行代码学习思考笔记(碎片.广播.持久化技术和Android数据库第四章手机平板要兼顾--探究碎片 4.1碎片是什么(Fragment) 4.2碎片的使用方式 4.2.1碎片的简单 ...
VTM10.0代码学习7：decompressCtu()xReconIntraQT()
此系列是为了记录自己学习VTM10.0的过程和锻炼表达能力,主要是从解码端进行入手.由于本人水平有限,出现的错误恳请大家指正,欢迎与大家一起交流进步. 之前的博文(VTM10.0代码学习3)提到两个重 ...

srs代码学习（4）-怎么转发流

srs代码学习（4）-怎么转发流相关推荐

最新文章

热门文章