srs代码学习(1)--listen建立过程

srs的服务侦听的建立过程。

以rtmp服务为例 srs服务侦听的建立依靠从上到下的三个类。分别是

SrsServer  

SrsStreamListener  

SrsTcpListener


端口侦听过程为

1)main函数中调用全局变量_srs_server的  listen()函数

if ((ret = _srs_server->listen()) != ERROR_SUCCESS) {
        return ret;
    }
在SrsServer的listen()函数如下
<pre name="code" class="cpp">int SrsServer::listen()
{
    int ret = ERROR_SUCCESS;
    
    if ((ret = listen_rtmp()) != ERROR_SUCCESS) {
        return ret;
    }
    
    if ((ret = listen_http_api()) != ERROR_SUCCESS) {
        return ret;
    }
    
    if ((ret = listen_http_stream()) != ERROR_SUCCESS) {
        return ret;
    }
    
    if ((ret = listen_stream_caster()) != ERROR_SUCCESS) {
        return ret;
    }
    
    return ret;
}


 
listen_rtmp()会建立一个rtmp的steamlistener
<pre name="code" class="cpp">int SrsServer::listen_rtmp()
{
    int ret = ERROR_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 SrsStreamListener(this, SrsListenerRtmpStream);
        listeners.push_back(listener);
        
        std::string ip;
        int port;
        srs_parse_endpoint(ip_ports[i], ip, port);
        
        if ((ret = listener->listen(ip, port)) != ERROR_SUCCESS) {
            srs_error("RTMP stream listen at %s:%d failed. ret=%d", ip.c_str(), port, ret);
            return ret;
        }
    }
    
    return ret;
}


 

这个streamlistener并不是真正的底层监听层。只是一个业务封装层。其类的继承顺序如下

在整个代码中,和其相识的类有SrsRtspListener  SrsHttpFlvListener两个类。这种类的主要作用是在底层链接建立有。给不同的上层放回链接信息。

代码走到这一步,侦听的socket还么米有建立起来,看SrsStreamListener的listen()函数

int SrsStreamListener::listen(string i, int p)
{
    int ret = ERROR_SUCCESS;
    
    ip = i;
    port = p;

    srs_freep(listener);
    listener = new SrsTcpListener(this, ip, port);

    if ((ret = listener->listen()) != ERROR_SUCCESS) {
        srs_error("tcp listen failed. ret=%d", ret);
        return ret;
    }
    
    srs_info("listen thread current_cid=%d, "
        "listen at port=%d, type=%d, fd=%d started success, ep=%s:%d",
        _srs_context->get_id(), p, type, listener->fd(), i.c_str(), p);

    srs_trace("%s listen at tcp://%s:%d, fd=%d", srs_listener_type2string(type).c_str(), ip.c_str(), port, listener->fd());

    return ret;
}
在这段代码里面。创建了真正的底层监听类

 listener = new SrsTcpListener(this, ip, port);
然后调用其listen()函数。代码如下

int SrsTcpListener::listen()
{
    int ret = ERROR_SUCCESS;
    
    if ((_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
        ret = ERROR_SOCKET_CREATE;
        srs_error("create linux socket error. port=%d, ret=%d", port, ret);
        return ret;
    }
    srs_verbose("create linux socket success. port=%d, fd=%d", port, _fd);
    
    int reuse_socket = 1;
    if (setsockopt(_fd, SOL_SOCKET, SO_REUSEADDR, &reuse_socket, sizeof(int)) == -1) {
        ret = ERROR_SOCKET_SETREUSE;
        srs_error("setsockopt reuse-addr error. port=%d, ret=%d", port, ret);
        return ret;
    }
    srs_verbose("setsockopt reuse-addr success. port=%d, fd=%d", port, _fd);
    
    sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.s_addr = inet_addr(ip.c_str());
    if (bind(_fd, (const sockaddr*)&addr, sizeof(sockaddr_in)) == -1) {
        ret = ERROR_SOCKET_BIND;
        srs_error("bind socket error. ep=%s:%d, ret=%d", ip.c_str(), port, ret);
        return ret;
    }
    srs_verbose("bind socket success. ep=%s:%d, fd=%d", ip.c_str(), port, _fd);
    
    if (::listen(_fd, SERVER_LISTEN_BACKLOG) == -1) {
        ret = ERROR_SOCKET_LISTEN;
        srs_error("listen socket error. ep=%s:%d, ret=%d", ip.c_str(), port, ret);
        return ret;
    }
    srs_verbose("listen socket success. ep=%s:%d, fd=%d", ip.c_str(), port, _fd);
    
    if ((_stfd = st_netfd_open_socket(_fd)) == NULL){
        ret = ERROR_ST_OPEN_SOCKET;
        srs_error("st_netfd_open_socket open socket failed. ep=%s:%d, ret=%d", ip.c_str(), port, ret);
        return ret;
    }
    srs_verbose("st open socket success. ep=%s:%d, fd=%d", ip.c_str(), port, _fd);
    
    if ((ret = pthread->start()) != ERROR_SUCCESS) {
        srs_error("st_thread_create listen thread error. ep=%s:%d, ret=%d", ip.c_str(), port, ret);
        return ret;
    }
    srs_verbose("create st listen thread success, ep=%s:%d", ip.c_str(), port);
    
    return ret;
}
到此,rtmp的监听端口就建立起来了。

在仔细看SrsTcpListener类。继承关系如下



这个类中很有趣的有了个线程类SrsReusableThread* pthread,这个线程是什么时候建立的?答案是在构造函数中

SrsTcpListener::SrsTcpListener(ISrsTcpHandler* h, string i, int p)
{
    handler = h;
    ip = i;
    port = p;

    _fd = -1;
    _stfd = NULL;

    pthread = new SrsReusableThread("tcp", this);
}
这个类的作用是什么呢?首先我们发现, SrsTcpListener 这个类继承了一个线程回调接口ISrsReusableThreadHandler。这个说明可以拥有可以在线程中运行的能力

观察线程类的结构



很奇怪,SrsReusableThread竟然也继承了一个handler类。这就表示。这个类还是一个封装类,并不是底层的真正的线程类。果然我们发现了其一个变量 internal::SrsThread* pthread,这个应该靠近点底层了吧。先放下不讲。我们是来追这个线程类也我的业务类啥关系的。这时另外一个有意思的变量ISrsReusableThreadHandler* handler,还记得
SrsTcpListener 类么,其有一个ISrsReusableThreadHandler的接口。通过这个接口,把线程和业务链接起来。看看其cycle()函数

int SrsReusableThread::cycle()
{
    return handler->cycle();
}
这个是线程这执行函数。它会调用上层类,具体到我们的例子里,就是 SrsTcpListener 的cycle().看看是什么

int SrsTcpListener::cycle()
{
    int ret = ERROR_SUCCESS;
    
    st_netfd_t client_stfd = st_accept(_stfd, NULL, NULL, ST_UTIME_NO_TIMEOUT);
    
    if(client_stfd == NULL){
        // ignore error.
        if (errno != EINTR) {
            srs_error("ignore accept thread stoppped for accept client error");
        }
        return ret;
    }
    srs_verbose("get a client. fd=%d", st_netfd_fileno(client_stfd));
    
    if ((ret = handler->on_tcp_client(client_stfd)) != ERROR_SUCCESS) {
        srs_warn("accept client error. ret=%d", ret);
        return ret;
    }
    
    return ret;
}
原来是accept函数。在有链接后直接上调给上层,具体来讲就是 SrsStreamListener类的 on_tcp_client()函数

int SrsStreamListener::on_tcp_client(st_netfd_t stfd)
{
    int ret = ERROR_SUCCESS;
    
    if ((ret = server->accept_client(type, stfd)) != ERROR_SUCCESS) {
        srs_warn("accept client error. ret=%d", ret);
        return ret;
    }

    return ret;
}
欧耶,这下调用到最上层去了。server里,建立一个链接。


到现在还有两个问题要搞明白:

1)srs的线程模型

2)如何管理各个链接。


这个下一次在摸索







### 实现流媒体服务器回调接口监听视频播放状态 为了实现在流媒体服务器上设置或实现回调以监听视频播放事件,通常可以通过配置HTTP回调机制来完成这一目标。对于SRS流媒体服务而言,在客户端执行特定操作时(如开始播放),能够触发相应的回调请求发送至预设的服务地址。 在SRS中定义了多种类型的回调事件,其中`on_play`用于表示当有客户端开始拉取并播放某个直播流时所触发的动作[^2]。具体来说,每当有一个新的播放连接建立起来之后,SRS将会向设定好的URL发起POST请求,并附带有关此次播放的信息作为JSON格式的数据体传递过去。 以下是针对此需求的一个简单示例说明: #### 配置SRS以启用`on_play`回调 编辑SRS配置文件(通常是`srs.conf`),找到对应的虚拟主机部分,并加入如下所示的相关参数来激活`on_play`回调功能: ```nginx vhost __defaultVhost__ { ... http_callback { enabled on; # 当有人开始观看某一直播流时调用该API通知外部系统 on_play http://your.server.address/api/on_play; # 可选:如果还需要处理其他类型的事件,则继续添加更多项... # on_publish http://your.server.address/api/on_publish; # on_unpublish http://your.server.address/api/on_unpublish; # on_stop http://your.server.address/api/on_stop; } } ``` 在此基础上,确保替换掉上述代码片段中的`http://your.server.address/api/on_play`为实际部署的应用程序接收这些回调消息的具体路径。 #### 处理接收到的回调信息 一旦成功设置了以上配置选项,那么每当前端用户启动一次新视频流的播放过程,位于后台运行着的Web应用就应当准备好接受到来自SRS发出的通知。假设采用Node.js框架构建这样的RESTful API端点为例: ```javascript const express = require('express'); const app = express(); app.use(express.json()); // 定义处理'play'事件的方法 app.post('/api/on_play', (req, res) => { const playInfo = req.body; console.log(`A new stream is being played: ${playInfo.stream}`); // 这里可以根据业务逻辑进一步扩展,比如更新数据库记录、统计在线人数等 res.sendStatus(200); }); // 启动Express服务器监听8081端口上的请求 app.listen(8081, () => { console.log('Server listening at port %d', 8081); }); ``` 这段脚本创建了一个简单的HTTP服务器实例用来响应来自SRS关于播放行为变化的消息推送;它会打印出每次发生的新播放活动详情到控制台日志当中去——当然这只是一个非常基础的例子而已,在真实场景下往往还会涉及到更加复杂的功能模块集成工作。
评论 2
添加红包

请填写红包祝福语或标题

红包个数最小为10个

红包金额最低5元

当前余额3.43前往充值 >
需支付:10.00
成就一亿技术人!
领取后你会自动成为博主和红包主的粉丝 规则
hope_wisdom
发出的红包
实付
使用余额支付
点击重新获取
扫码支付
钱包余额 0

抵扣说明:

1.余额是钱包充值的虚拟货币,按照1:1的比例进行支付金额的抵扣。
2.余额无法直接购买下载,可以购买VIP、付费专栏及课程。

余额充值