SRS流媒体服务器之HTTP-FLV框架分析

当客户端推流RTMP数据发到SRS流媒体服务器，如果正确配置SRS流媒体服务器，可以输出HTTP-FLV的码流，拉流端就可以成功拉取到，那这个详细过程是怎样呢？本篇文章就来详细分析。先回顾下整体的框架：

RTMP推流端-----》SRS流媒体服务器(建立SOURCE->生成Consumer->指定封装格式endoder=FLV) 《《--------------拉流客户端拉取HTTP-FLV

 

1.简述http-flv技术

(1)在http协议中有个content-length字段，指的是http的body的长度。服务器在恢复客户端请求时，如果没有这个字段，客户端就一直接收数据，直到服务器与客户端的socket连接断开。如果有这个字段，客户端接收这个长度的数据后，就认为数据传输完毕。

http-flv直播就是利⽤了这个原理，服务器回复客户端请求的时候不加content-length字段，回复了http内容之后，紧接着发送flv数据，客户端就⼀直接收数据了。客户端就会认为一直有数据接收。

客户端发起请求，SRS流媒体服务器返回的是:

0 SrsLiveStream::SrsLiveStream (this=0xa3da40, s=0xa3bbd0, r=0xa3ad40, c=0xa3d520)
at src/App/srs_app_http_stream.cpp:514

1 0x00000000005010bb in SrsHttpStreamServer::http_mount (this=0xa11fd0, s=0xa3bbd0,
r=0xa3ad40) at src/app/srs_app_http_stream.cpp:912

2 0x00000000005620f5 in SrsHttpServer::http_mount (this=0xa11e00, s=0xa3bbd0,
r=0xa3ad40) at src/app/srs_app_http_conn.cpp:308

3 0x00000000004cd3cc in SrsServer::on_publish (this=0xa11ea0, s=0xa3bbd0, r=0xa3ad40)
at src/app/srs_app_server.cpp:1608

4 0x00000000004e6a9b in SrsSource::on_publish (this=0xa3bbd0) at src/app/srs_app_source.cpp:2466

5 0x00000000004d89f2 in SrsRtmpConn::acquire_publish (this=0xa30d00,source=0xa3bbd0) at src/app/srs_app_rtmp_conn.cpp:940

6 0x00000000004d7a74 in SrsRtmpConn::publishing (this=0xa30d00, source=0xa3bbd0) at src/app/srs_app_rtmp_conn.cpp:822
#7 0x00000000004d5229 in SrsRtmpConn::stream_service_cycle (this=0xa30d00) at
src/app/srs_app_rtmp_conn.cpp:534
#8 0x00000000004d4141 in SrsRtmpConn::service_cycle (this=0xa30d00) at
src/app/srs_app_rtmp_conn.cpp:388
#9 0x00000000004d2f09 in SrsRtmpConn::do_cycle (this=0xa30d00) at
src/app/srs_app_rtmp_conn.cpp:209
#10 0x00000000004d10fb in SrsConnection::cycle (this=0xa30d78) at
src/app/srs_app_conn.cpp:171
#11 0x0000000000509c88 in SrsSTCoroutine::cycle (this=0xa30f90) at
src/app/srs_app_st.cpp:198
#12 0x0000000000509cfd in SrsSTCoroutine::pfn (arg=0xa30f90) at
src/app/srs_app_st.cpp:213
#13 0x00000000005bdd9d in _st_thread_main () at sched.c:337
#14 0x00000000005be515 in st_thread_create (start=0x5bd719 <_st_vp_schedule+170>,
arg=0x700000001, joinable=1,
stk_size=1) at sched.c:616

(2)配置文件

主要分为两部分:

(1)配置http服务

(2)配置http-flv服务

配置⽂件如下所示：

listen 1935;
max_connections 1000; 
#srs_log_tank file; 
#srs_log_file ./objs/srs.log; 
# 前台运⾏ daemon off; 
# 打印到终端控制台 
srs_log_tank console; 
http_api { 
  enabled on; 
  listen 1985; 
}
http_server { 
  enabled on; 
  listen 8081; 
  # http监听端⼝ (1)配置的http服务器，注意端⼝，如果是云服务器⼀定要注意开 放相应端⼝ 
  dir ./objs/Nginx/html; }
stats { 
  network 0; 
  disk sda sdb xvda xvdb;
}
vhost __defaultVhost__ { 
  # 使⽤默认的vhost
  # hls 
  hls {
    enabled on; 
       hls_path ./objs/nginx/html; 
    hls_fragment 10; 
    hls_window 60;
    }
  # 使用http-flv要配置 
  http_remux {
    enabled on; 
    mount [vhost]/[app]/[stream].flv; # ⽀持flv的使⽤，flv拉流的地址 
    hstrs on; 
  }
}

 

(3)测试准备

在客户端使用ffmpeg推rtmp流，其中xxx.xxx.xxx.xxx表示IP地址，根据实际环境的ip地址去配置，命令如下:

ffmpeg -re -i xxx.flv -vcodec copy -acodec copy -f flv -y rtmp://xxx.xxx.xxx.xxx/live/livestream

在客户端拉取rtmp和http流，命令如下:

ffplay http://xxx.xxx.xxx.xxx:8081/live/livestream.flv

ffplay rtmp://xxx.xxx.xxx.xxx/live/livestream

 

2.SRS流媒体rtmp推流时的函数调用关系

RTMP推流的时候根据url，创建对应的handler，拉流的时候根据url，找到对应处理的handler。即url和handler是一一对应关系。以下流程，在RTMP推流时，创建了一个HTTP-FLV的SOURCE(函数调用关系是从下至上，即数字14到0)，关于SOURCE的详细分析，前面文章也分析过。

0 SrsLiveStream::SrsLiveStream (this=0xa3da40, s=0xa3bbd0, r=0xa3ad40, c=0xa3d520)
at src/app/srs_app_http_stream.cpp:514

1 0x00000000005010bb in SrsHttpStreamServer::http_mount (this=0xa11fd0, s=0xa3bbd0,
r=0xa3ad40) at src/app/srs_app_http_stream.cpp:912

2 0x00000000005620f5 in SrsHttpServer::http_mount (this=0xa11e00, s=0xa3bbd0,
r=0xa3ad40) at src/app/srs_app_http_conn.cpp:308

3 0x00000000004cd3cc in SrsServer::on_publish (this=0xa11ea0, s=0xa3bbd0, r=0xa3ad40)
at src/app/srs_app_server.cpp:1608

4 0x00000000004e6a9b in SrsSource::on_publish (this=0xa3bbd0) at src/app/srs_app_source.cpp:2466

5 0x00000000004d89f2 in SrsRtmpConn::acquire_publish (this=0xa30d00,source=0xa3bbd0) at src/app/srs_app_rtmp_conn.cpp:940

6 0x00000000004d7a74 in SrsRtmpConn::publishing (this=0xa30d00, source=0xa3bbd0) at src/app/srs_app_rtmp_conn.cpp:822
#7 0x00000000004d5229 in SrsRtmpConn::stream_service_cycle (this=0xa30d00) at
src/app/srs_app_rtmp_conn.cpp:534
#8 0x00000000004d4141 in SrsRtmpConn::service_cycle (this=0xa30d00) at
src/app/srs_app_rtmp_conn.cpp:388
#9 0x00000000004d2f09 in SrsRtmpConn::do_cycle (this=0xa30d00) at
src/app/srs_app_rtmp_conn.cpp:209
#10 0x00000000004d10fb in SrsConnection::cycle (this=0xa30d78) at
src/app/srs_app_conn.cpp:171
#11 0x0000000000509c88 in SrsSTCoroutine::cycle (this=0xa30f90) at
src/app/srs_app_st.cpp:198
#12 0x0000000000509cfd in SrsSTCoroutine::pfn (arg=0xa30f90) at
src/app/srs_app_st.cpp:213
#13 0x00000000005bdd9d in _st_thread_main () at sched.c:337
#14 0x00000000005be515 in st_thread_create (start=0x5bd719 <_st_vp_schedule+170>,
arg=0x700000001, joinable=1,
stk_size=1) at sched.c:616

 

3.SRS流媒体服务器源码的重要函数和类说明

RTMP不管是推流还是拉流都是对应一个连接实现，那HTTP-FLV也是一个客户端对应一个连接，如果是HLS，那client也会对应一个连接。

(1)源码中重要函数和文件说明

在SRS流媒体服务器源码中，关于处理数据的重要函数说明:

SrsLiveStream::do_serve_http：处理客户端的数据发送。

SrsHttpConn:表示每个http client或RTMP client都有这个连接。

SrsConsumer:每个SrsHttpConn都对应一个消费者SrsConsumer，对应RTMP或HTTP。关于SrsConsumer前面文章已经讲过，这里相当于中间数据的缓存。

(2)源码中重要类说说明

SrsBufferCache:HTTP直播流编码器的缓存。

SrsFlvStreamEncoder:将RTMP转成HTTP FLV流。

SrsTsStreamEncoder:将RTMP转成HTTP TS流。

SrsAacStreamEncoder：将RTMP含有的AAC成分转成HTTP AAC流。

SrsMp3StreamEncoder:将RTMP含有的MP3成分转成HTTP MP3流。

SrsBufferWriter:将流直接写⼊到HTTP响应的数据中。

SrsLiveStream:HTTP直播流，将RTMP转成HTTP-FLV或者其他格式，其实际是handler SrsLiveEntry 直播⼊⼝，⽤来处理HTTP 直播流。

SrsHttpStreamServer:HTTP直播流服务，服务FLV/TS/MP3/AAC流的合成。

SrsHttpResponseWriter: 负责将数据发送给客户端，本质是调⽤SrsStSocket进⾏发送

SrsHttpServeMux：HTTP请求多路复⽤器，实际就是路由，⾥⾯记录了path以及对应handler。

 

4.SRS流媒体服务器源码解析

根据源码可以得到，http和RTMP都是继承SrsConnection。源码如下:

// The http connection which request the static or stream content.
class SrsHttpConn : public SrsConnection
{
protected:
    SrsHttpParser* parser;
    ISrsHttpServeMux* http_mux;
    SrsHttpCorsMux* cors;
public:
    SrsHttpConn(IConnectionManager* cm, srs_netfd_t fd, ISrsHttpServeMux* m, std::string cip);
    virtual ~SrsHttpConn();
// Interface ISrsKbpsDelta
public:
    virtual void remark(int64_t* in, int64_t* out);
protected:
    virtual srs_error_t do_cycle();
protected:
    // When got http message,
    // for the static service or api, discard any body.
    // for the stream caster, for instance, http flv streaming, may discard the flv header or not.
    virtual srs_error_t on_got_http_message(ISrsHttpMessage* msg) = 0;
private:
    virtual srs_error_t process_request(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
    // When the connection disconnect, call this method.
    // e.g. log msg of connection and report to other system.
    // @param request: request which is converted by the last http message.
    virtual srs_error_t on_disconnect(SrsRequest* req);
// Interface ISrsReloadHandler
public:
    virtual srs_error_t on_reload_http_stream_crossdomain();
};

SrsRtmpConn继承SrsConnection，源码如下:

// The client provides the main logic control for RTMP clients.
class SrsRtmpConn : virtual public SrsConnection, virtual public ISrsReloadHandler
{
    // For the thread to directly access any field of connection.
    friend class SrsPublishRecvThread;
private:
    SrsServer* server;
    SrsRtmpServer* rtmp;
    SrsRefer* refer;
    SrsBandwidth* bandwidth;
    SrsSecurity* security;
    // The wakable handler, maybe NULL.
    // TODO: FIXME: Should refine the state for receiving thread.
    ISrsWakable* wakable;
    // The elapsed duration in srs_utime_t
    // For live play duration, for instance, rtmpdump to record.
    // @see https://github.com/ossrs/srs/issues/47
    srs_utime_t duration;
    // The MR(merged-write) sleep time in srs_utime_t.
    srs_utime_t mw_sleep;
    // The MR(merged-write) only enabled for play.
    int mw_enabled;
    // For realtime
    // @see https://github.com/ossrs/srs/issues/257
    bool realtime;
    // The minimal interval in srs_utime_t for delivery stream.
    srs_utime_t send_min_interval;
    // The publish 1st packet timeout in srs_utime_t
    srs_utime_t publish_1stpkt_timeout;
    // The publish normal packet timeout in srs_utime_t
    srs_utime_t publish_normal_timeout;
    // Whether enable the tcp_nodelay.
    bool tcp_nodelay;
    // About the rtmp client.
    SrsClientInfo* info;
public:
    SrsRtmpConn(SrsServer* svr, srs_netfd_t c, std::string cip);
    virtual ~SrsRtmpConn();
public:
    virtual void dispose();
protected:
    virtual srs_error_t do_cycle();
// Interface ISrsReloadHandler
public:
    virtual srs_error_t on_reload_vhost_removed(std::string vhost);
    virtual srs_error_t on_reload_vhost_play(std::string vhost);
    virtual srs_error_t on_reload_vhost_tcp_nodelay(std::string vhost);
    virtual srs_error_t on_reload_vhost_realtime(std::string vhost);
    virtual srs_error_t on_reload_vhost_publish(std::string vhost);
// Interface ISrsKbpsDelta
public:
    virtual void remark(int64_t* in, int64_t* out);
private:
    // When valid and connected to vhost/app, service the client.
    virtual srs_error_t service_cycle();
    // The stream(play/publish) service cycle, identify client first.
    virtual srs_error_t stream_service_cycle();
    virtual srs_error_t check_vhost(bool try_default_vhost);
    virtual srs_error_t playing(SrsSource* source);
    virtual srs_error_t do_playing(SrsSource* source, SrsConsumer* consumer, SrsQueueRecvThread* trd);
    virtual srs_error_t publishing(SrsSource* source);
    virtual srs_error_t do_publishing(SrsSource* source, SrsPublishRecvThread* trd);
    virtual srs_error_t acquire_publish(SrsSource* source);
    virtual void release_publish(SrsSource* source);
    virtual srs_error_t handle_publish_message(SrsSource* source, SrsCommonMessage* msg);
    virtual srs_error_t process_publish_message(SrsSource* source, SrsCommonMessage* msg);
    virtual srs_error_t process_play_control_msg(SrsConsumer* consumer, SrsCommonMessage* msg);
    virtual void change_mw_sleep(srs_utime_t sleep_v);
    virtual void set_sock_options();
private:
    virtual srs_error_t check_edge_token_traverse_auth();
    virtual srs_error_t do_token_traverse_auth(SrsRtmpClient* client);
private:
    // When the connection disconnect, call this method.
    // e.g. log msg of connection and report to other system.
    virtual srs_error_t on_disconnect();
private:
    virtual srs_error_t http_hooks_on_connect();
    virtual void http_hooks_on_close();
    virtual srs_error_t http_hooks_on_publish();
    virtual void http_hooks_on_unpublish();
    virtual srs_error_t http_hooks_on_play();
    virtual void http_hooks_on_stop();
};

前面的文章已经讲过了，rtmp推流的时候就会产生数据源，对应源码就是source。那http-flv client也是要从source里面拉取数据，也是要绑定一个consumer，这个思想在前面的文章中都要反复讲过。

 

5.源码调试分析

先运行SRS流媒体服务器，执行命令:

gdb ./objs/srs

如下界面:

 

 

再执行命令:

set args -c ./conf/srs.conf

r

如下界面:

 

 

在win环境，开启ffmpeg推流，推流命令在上面已经给出。

界面如下:

 

推流成功后，在win环境，用ffplay去播放。播放命令，上面已经给出了。

界面如下:

 

可以看到拉流端，关于http-flv具体的一些打印信息，如下界面:

 

使用WireShark抓http-flv包，需要设置过滤条件，http or tcp.port==8081

界面如下:

 

拉流客户端请求SRS流媒体服务器路径是/live/livestream.flv HTTP/1.1，请求方法是GET方法。

如下界面:

 

该请求数据包具体如下类型，如下界面:

 

通过WireShark抓包，也可以看到SRS流媒体服务器回应客户端消息，其中是不带有content-length。其服务端回应客户端的数据包的过程，如下界面:

 

6.http-flv在ffmpeg源码中是怎样实现呢？

这个时候客户端开启推流，经过调试分析，整个流程如下图:

 

下面的源码反应了http监听的过程(Rtmp与http类似)，也就是按照这个流程来分析:

run_master()-->SrsServer::listen()--->
SrsServer::listen_http_stream()。

(1) main函数，
src/main/srs_main_server.cpp:192行。

(2)do_main函数，
src/main/srs_main_server.cpp:184行。

(3)run函数，
src/main/srs_main_server.cpp:409行。

(4)run_master函数，
src/main/srs_main_server.cpp:469行。

(5)SrsServer::listen函数，
srs/app/srs_app_server.cpp:880行。

(6)
SrsServer::listen_http_stream，srs/app/srs_app_server.cpp:1295行。

在ffmpeg源码中搜索http_code，可以搜索到，在http.c里，有实现。源码在如下路径。

 

在SRS流媒体服务端，从各类协议总入口SrsServer::listen()开始分析。对应源码如下:

HTTP/1.1 200 OK 
Connection: Keep-Alive 
Content-Type: video/x-flv 
Server: SRS/3.0.141(OuXuli) 
Transfer-Encoding: chunked

 

如果是http协议，就会调用listen_http_stream()，到http的listen分析，对应源码如下:

 

srs_error_t SrsServer::listen_http_stream()
{
    srs_error_t err = srs_success;
    
    close_listeners(SrsListenerHttpStream);
    if (_srs_config->get_http_stream_enabled()) {
        SrsListener* listener = new SrsBufferListener(this, SrsListenerHttpStream);
        listeners.push_back(listener);
        
        std::string ep = _srs_config->get_http_stream_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, "http stream listen %s:%d", ip.c_str(), port);
        }
    }
    
    return err;
}

 

7.拉流时HTTP连接调试

打个断点，输入如下命令，调试：

b SrsServer::listen_http_stream()

界面如下:

 

输入命令:

n

可以一行行执行。

这个时候，如果客户端开启拉流，可以看到SRS流媒体服务器的调用栈，界面如下:

 

这个http流程与前面分析的RTMP流程是类似:

(1)st_thread_create，在sched.c:616行。

(2)_st_thread_main，在sched.c:337行。

(3)函数SrsSTCoroutine::pfn，在
srs/app/srs_app_st.cpp:213行。

(4)函数SrsSTCoroutine::cycle，在
srs/app/srs_app_st.cpp:198行。

(5)函数SrsTcpListener::cycle，在
srs/app/srs_app_listener.cpp:202行。

(6)函数
SrsBufferListener::on_tcp_client，在srs/app/srs_app_server.cpp:167行。

(7)函数SrsServer::accept_client，类型是SrsListenerHttpStream，在
src/app/srs_app_server.cpp:1400行。

(8)函数SrsServer::fd2conn，类型是SrsListenerHttpStream，在
src/app/srs_app_server.cpp:1465行。

 

不同的客户端都是可以进来do_serve_http，当拉流客户端要拉取http数据时，包含真正的音视频数据，从这里可以分析，源码如下:

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");
        enc_desc = "FLV";
        enc = new SrsFlvStreamEncoder();
    } 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());
    }
    SrsAutoFree(ISrsBufferEncoder, enc);

    // Enter chunked mode, because we didn't set the content-length.
    w->write_header(SRS_CONSTS_HTTP_OK);
    
    // create consumer of souce, ignore gop cache, use the audio gop cache.
    SrsConsumer* consumer = NULL;
    if ((err = source->create_consumer(NULL, consumer, true, true, !enc->has_cache())) != srs_success) {
        return srs_error_wrap(err, "create consumer");
    }
    SrsAutoFree(SrsConsumer, consumer);
    srs_verbose("http: consumer created success.");
    
    SrsPithyPrint* pprint = SrsPithyPrint::create_http_stream();
    SrsAutoFree(SrsPithyPrint, pprint);
    
    SrsMessageArray msgs(SRS_PERF_MW_MSGS);

    // Use receive thread to accept the close event to avoid FD leak.
    // @see https://github.com/ossrs/srs/issues/636#issuecomment-298208427
    SrsHttpMessage* hr = dynamic_cast<SrsHttpMessage*>(r);
    SrsResponseonlyHttpConn* hc = dynamic_cast<SrsResponseOnlyHttpConn*>(hr->connection());
    
    // update the statistic when source disconveried.
    SrsStatistic* stat = SrsStatistic::instance();
    if ((err = stat->on_client(_srs_context->get_id(), req, hc, SrsRtmpConnPlay)) != srs_success) {
        return srs_error_wrap(err, "stat on client");
    }
    
    // the memory writer.
    SrsBufferWriter writer(w);
    if ((err = enc->initialize(&writer, cache)) != srs_success) {
        return srs_error_wrap(err, "init encoder");
    }
    
    // if gop cache enabled for encoder, dump to consumer.
    if (enc->has_cache()) {
        if ((err = enc->dump_cache(consumer, source->jitter())) != srs_success) {
            return srs_error_wrap(err, "encoder dump cache");
        }
    }
    
    SrsFlvStreamEncoder* ffe = dynamic_cast<SrsFlvStreamEncoder*>(enc);
    
    // 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");
        }
    }
    
    srs_utime_t mw_sleep = _srs_config->get_mw_sleep(req->vhost);
    if ((err = hc->set_socket_buffer(mw_sleep)) != srs_success) {
        return srs_error_wrap(err, "set mw_sleep %" PRId64, 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, srsu2msi(mw_sleep),
        enc->has_cache(), msgs.max);

    // TODO: free and erase the disabled entry after all related connections is closed.
    // TODO: FXIME: Support timeout for player, quit infinite-loop.
    while (entry->enabled) {
        // Whether client closed the FD.
        if ((err = trd->pull()) != srs_success) {
            return srs_error_wrap(err, "recv thread");
        }

        pprint->elapse();

        // get messages from consumer.
        // each msg in msgs.msgs must be free, for the SrsMessageArray never free them.
        int count = 0;
        if ((err = consumer->dump_packets(&msgs, count)) != srs_success) {
            return srs_error_wrap(err, "consumer dump packets");
        }
        
        if (count <= 0) {
            // Directly use sleep, donot use consumer wait, because we couldn't awake consumer.
            srs_usleep(mw_sleep);
            // ignore when nothing got.
            continue;
        }
        
        if (pprint->can_print()) {
            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, srsu2msi(mw_sleep));
        }
        
        // sendout all messages.
        if (ffe) {
            err = ffe->write_tags(msgs.msgs, count);
        } else {
            err = streaming_send_messages(enc, msgs.msgs, count);
        }

        // free the messages.
        for (int i = 0; i < count; i++) {
            SrsSharedPtrMessage* msg = msgs.msgs[i];
            srs_freep(msg);
        }
        
        // check send error code.
        if (err != srs_success) {
            return srs_error_wrap(err, "send messages");
        }
    }

    // Here, the entry is disabled by encoder un-publishing or reloading,
    // so we must return a io.EOF error to disconnect the client, or the client will never quit.
    return srs_error_new(ERROR_HTTP_STREAM_EOF, "Stream EOF");
}

从源码中看出，这里有个new SrsTsStreamEncoder，这个是用来合成flv数据，以供拉流端使用。

接下来打印断点调试看看。输入命令如下:

b SrsLiveStream::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)

界面如下:

 

输入命令，继续运行:

c

这时候开启拉流端，再输入命令:

bt

查看调用栈，如下界面:

 

(1)st_thread_create，在sched.c:616行。

(2)_st_thread_main，在sched.c:337行。

(3)函数SrsSTCoroutine::pfn，在
src/app/srs_app_st.cpp:213行。

(4)函数SrsSTCoroutine::cycle，在
src/app/srs_app_st.cpp:198行。

(5)函数SrsConnection::cycle，在
src/app/srs_app_conn.cpp:171行。

(6)函数SrsHttpConn::do_cycle，在
src/app/srs_app_http_conn.cpp:133行。

(7)函数
SrsHttpConn::process_request，在src/app/srs_app_http_conn.cpp:161行。

(8)函数
SrsHttpCorsMux::server_http，在src/protocol/srs_http_stack.cpp:859行。

(9)函数
SrsHttpServer::server_http，在src/app/srs_app_http_conn.cpp:300行。

(10)函数
SrsHttpServerMux::server_http，在src/protocol/srs_http_stack.cpp:711行。

(11)函数
SrsLiveStream::server_http，在src/app/srs_app_http_stream.cpp:544行。

(12)函数
SrsLiveStream::do_serve_http，在src/app/srs_app_http_stream.cpp:552行。

 

当拉流端通过onsumer->dump_packets(&msgs, count)读出消息后，然后就用ffe->write_tags(msgs.msgs, count)绑定一个Encoder(这里就指的是用flv封装)，源代码如下图:

在
SrsLiveStream::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)函数下。

int count = 0;
        if ((err = consumer->dump_packets(&msgs, count)) != srs_success) {
            return srs_error_wrap(err, "consumer dump packets");
        }
        
        if (count <= 0) {
            // Directly use sleep, donot use consumer wait, because we couldn't awake consumer.
            srs_usleep(mw_sleep);
            // ignore when nothing got.
            continue;
        }
        
        if (pprint->can_print()) {
            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, srsu2msi(mw_sleep));
        }
        
        // sendout all messages.
        if (ffe) {
            err = ffe->write_tags(msgs.msgs, count);
        } else {
            err = streaming_send_messages(enc, msgs.msgs, count);
        }

 

对应源码文件为Srs_app_source.cpp，拉流端通过SrsConsumer::dump_packets(SrsMessageArray* msgs, int& count)，读取消息。关于这个函数的调用，在前面文章有更详细的分析。

srs_error_t SrsConsumer::dump_packets(SrsMessageArray* msgs, int& count)
{
    srs_error_t err = srs_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 err;
    }
    
    // pump msgs from queue.
    if ((err = queue->dump_packets(max, msgs->msgs, count)) != srs_success) {
        return srs_error_wrap(err, "dump packets");
    }
    
    return err;
}SrsConsumer::dump_packets(SrsMessageArray* msgs, int& count)

在源码Srs_app_http_stream.cpp，调用函数ffe->write_tags(msgs.msgs, count)(包括写头和数据)，绑定Encoder，这里指的是封装格式。详细分析，会在后面的文章继续分析，源码如下:

srs_error_t SrsFlvStreamEncoder::write_tags(SrsSharedPtrMessage** msgs, int count)
{
    srs_error_t err = srs_success;

    // For https://github.com/ossrs/srs/issues/939
    if (!header_written) {
        bool has_video = false;
        bool has_audio = false;

        for (int i = 0; i < count && (!has_video || !has_audio); i++) {
            SrsSharedPtrMessage* msg = msgs[i];
            if (msg->is_video()) {
                has_video = true;
            } else if (msg->is_audio()) {
                has_audio = true;
            }
        }

        // Drop data if no A+V.
        if (!has_video && !has_audio) {
            return err;
        }

        if ((err = write_header(has_video, has_audio))  != srs_success) {
            return srs_error_wrap(err, "write header");
        }
    }

    return enc->write_tags(msgs, count);
}

 

8.拉流时，SRS流媒体服务器发送数据给客户端

调试界面如下:

 

在SRS流媒体服务器给客户端发送数据的函数，打断点，跟踪函数调用流程，输入命令如下:

b SrsHttpResponseWriter::writev

(1)包括了前面连接过程的连接，这些流程就反应了函数调用的关系(调用关系是从下至上，即从15到0)，跟踪流程如下:

0 SrsHttpResponseWriter::writev (this=0x7ffff7f1ebd0, iov=0xaeaa80, iovcnt=240,
pnwrite=0x0) at src/service/srs_service_http_conn.cpp:784

1 0x00000000004fde62 in SrsBufferWriter::writev (this=0x7ffff7f1e860, iov=0xaeaa80,
iovcnt=240, pnwrite=0x0) at src/app/srs_app_http_stream.cpp:511

2 0x000000000040f109 in SrsFlvTransmuxer::write_tags (this=0xb92fb0, msgs=0xaea310,
count=80) at src/kernel/srs_kernel_flv.cpp:538

3 0x00000000004fd0b1 in SrsFlvStreamEncoder::write_tags (this=0xb51490, msgs=0xaea310,
count=80) at src/app/srs_app_http_stream.cpp:345

4 0x00000000004ff0dc in SrsLiveStream::do_serve_http (this=0xa3d9f0, w=0x7ffff7f1ebd0,
r=0xb92840) at src/app/srs_app_http_stream.cpp:677

5 0x00000000004fe108 in SrsLiveStream::serve_http (this=0xa3d9f0, w=0x7ffff7f1ebd0,
r=0xb92840) at src/app/srs_app_http_stream.cpp:544

6 0x000000000049c86f in SrsHttpServeMux::serve_http (this=0xa11fe0, w=0x7ffff7f1ebd0,
r=0xb92840) at src/protocol/srs_http_stack.cpp:711
7 0x0000000000562080 in SrsHttpServer::serve_http (this=0xa11e00, w=0x7ffff7f1ebd0,
r=0xb92840) at src/app/srs_app_http_conn.cpp:300
8 0x000000000049d6be in SrsHttpCorsMux::serve_http (this=0xb37440, w=0x7ffff7f1ebd0,
r=0xb92840) at src/protocol/srs_http_stack.cpp:859
9 0x0000000000561086 in SrsHttpConn::process_request (this=0xb93ff0, w=0x7ffff7f1ebd0,
r=0xb92840) at src/app/srs_app_http_conn.cpp:161
10 0x0000000000560ce8 in SrsHttpConn::do_cycle (this=0xb93ff0) at src/app/srs_app_http_conn.cpp:133 ---Type <return> to continue, or q <return> to quit---
11 0x00000000004d10fb in SrsConnection::cycle (this=0xb93ff0) at src/app/srs_app_conn.cpp:171
12 0x0000000000509c88 in SrsSTCoroutine::cycle (this=0xb93f10) at src/app/srs_app_st.cpp:198
13 0x0000000000509cfd in SrsSTCoroutine::pfn (arg=0xb93f10) at src/app/srs_app_st.cpp:213
14 0x00000000005bdd9d in _st_thread_main () at sched.c:337
15 0x00000000005be515 in st_thread_create (start=0x5bd719 <_st_vp_schedule+170>,
arg=0x900000001, joinable=1,stk_size=1) at sched.c:616

 

(2)客户端拉取HTTP—FLV播放流程

当RTMP推流成功后，这里通过调试，跟踪客户端拉流时，SRS流媒体服务器的播放流程。当拉取HTTP-FLV流时，每个播放的SrsFlvStreamEncoder是独⽴，互不影响。调用关系是从下至上，即11到0。如下：

0 SrsFlvStreamEncoder::SrsFlvStreamEncoder (this=0xa57820) at src/app/srs_app_http_stream.cpp:250

1 0x00000000004fe2fd in SrsLiveStream::do_serve_http (this=0xa3da20, w=0x7ffff7eb5bd0,
r=0xa5d7c0) at src/app/srs_app_http_stream.cpp:562

2 0x00000000004fe108 in SrsLiveStream::serve_http (this=0xa3da20, w=0x7ffff7eb5bd0,
r=0xa5d7c0) at src/app/srs_app_http_stream.cpp:544

3 0x000000000049c86f in SrsHttpServeMux::serve_http (this=0xa11fe0, w=0x7ffff7eb5bd0,
r=0xa5d7c0) at src/protocol/srs_http_stack.cpp:711

4 0x0000000000562080 in SrsHttpServer::serve_http (this=0xa11e00, w=0x7ffff7eb5bd0,
r=0xa5d7c0) at src/app/srs_app_http_conn.cpp:300

5 0x000000000049d6be in SrsHttpCorsMux::serve_http (this=0xa52930, w=0x7ffff7eb5bd0,
r=0xa5d7c0) at src/protocol/srs_http_stack.cpp:859

6 0x0000000000561086 in SrsHttpConn::process_request (this=0xa5d120,
w=0x7ffff7eb5bd0, r=0xa5d7c0) at src/app/srs_app_http_conn.cpp:161

7 0x0000000000560ce8 in SrsHttpConn::do_cycle (this=0xa5d120) at
src/app/srs_app_http_conn.cpp:133

8 0x00000000004d10fb in SrsConnection::cycle (this=0xa5d120) at
src/app/srs_app_conn.cpp:171

9 0x0000000000509c88 in SrsSTCoroutine::cycle (this=0xa5d1c0) at
src/app/srs_app_st.cpp:198

10 0x0000000000509cfd in SrsSTCoroutine::pfn (arg=0xa5d1c0) at
src/app/srs_app_st.cpp:213

11 0x00000000005bdd9d in _st_thread_main () at sched.c:337

 

(3)客户端拉取HTTP_FLV流过程中，部分日志如下:

[Trace][10457][554] HTTP client ip=175.0.54.116, request=0, to=15000ms

[Trace][10457][554] HTTP GET http://111.229.231.225:8081/live/livestream.flv, content-length=-1

[Trace][10457][554] http: mount flv stream for sid=/live/livestream, mount=/live/livestream.flv

[Trace][10457][554] flv: source url=/live/livestream, is_edge=0, source_id=-1[-1]

[Trace][10457][554] create consumer, active=0, queue_size=0.00,jitter=30000000

[Trace][10457][554] set fd=10, SO_SNDBUF=46080=>175000,buffer=350ms

[Trace][10457][554] FLV /live/livestream.flv, encoder=FLV,nodelay=0, mw_sleep=350ms, cache=0, msgs=128

 

9.总结

本篇文章在前面文章的基础上，讲解从客户端RTMP推流到SRS流媒体服务器，拉流端拉取HTTP_FLV数据的过程，并通过调试的方法，跟踪SRS流媒体服务器的函数调用流程。能够帮助大家理清其中错综复杂的关系，对于源码分析非常有帮助。

本篇文章就分析到这里，欢迎关注，转发，点赞，收藏，分享，评论区讨论。