[网络协议]10、SRS4.0源代码分析之WebRTC推流端处理

目标：

上一节分析了SRS4.0中WebRTC模块的总体架构和软件处理流程。接下来分析SRS4.0 WebRTC模块针对客户端推流连接上各种协议报文的软件处理逻辑。

内容：

WebRTC模块在启动过程中：
1、创建SrsUdpMuxListener监听对象，监听指定的UDP端口(默认配置8000端口)。

srs_error_t SrsRtcServer::listen_udp() {  // 创建监听对象
    SrsUdpMuxListener* listener = new SrsUdpMuxListener(this, ip, port);
    listener->listen();  
}

srs_error_t SrsUdpMuxListener::listen() {  // bind监听端口，并启动监听协程
    srs_udp_listen(ip, port, &lfd);  
    trd = new SrsSTCoroutine("udp", this, cid);
    trd->start(); 
}

2、启动此对象内部的SrsUdpMuxListener::cycle()协程，从UDP监听端口接收数据。

srs_error_t SrsUdpMuxListener::cycle() { // 此协程从监听端口读取数据
    ......
    SrsUdpMuxSocket skt(lfd);
    while (true) {
        // 以阻塞方式从监听端口读取数据，
        // 在skt.recvfrom内部使用对端地址的port(16bit)+ipv4(32bit)构成一个fast_id_(64bit)
        // 同理，使用对端地址的ipv4+port构成一个字符串类型的peer_id_
        skt.recvfrom(SRS_UTIME_NO_TIMEOUT); 

        handler->on_udp_packet(&skt); // 这里实际是调用SrsRtcServer::on_udp_packet()
    }
}

int SrsUdpMuxSocket::recvfrom(srs_utime_t timeout){
    nread = srs_recvfrom(lfd, buf, nb_buf, (sockaddr*)&from, &fromlen, timeout);
}

根据上一节的介绍，推流客户端通过API接口完成SDP交换，再从服务器的SDP信息中，获取服务器的IP地址+端口号，并按照WebRTC协议的要求，向服务器端口依次发送各种协议报文，完成客户端与服务器的连接建立、安全认证和RTP报文加密传输。

所以，WebRTC客户端与服务器的连接建立过程中大概涉及四种主要的协议处理

1. 客户端和服务端通过STUN协议和ICE机制建立连接
2. 客户端和服务端通过DTLS协议报文完成安全认证并生成SRTP加解密所需的密钥
3. 客户端和服务端之间通过SRTP算法实现RTP报文的加解密
4. 客户端和服务端之间通过RTCP报文完成音视频数据的Qos处理

srs_error_t SrsRtcServer::on_udp_packet(SrsUdpMuxSocket* skt) {
    // 查找udp客户端对应的SrsRtcConnection
    session = (SrsRtcConnection*)_srs_rtc_manager->find_by_fast_id(fast_id);
    session = (SrsRtcConnection*)_srs_rtc_manager->find_by_id(peer_id);
    
    // STUN协议报文处理
    if (srs_is_stun((uint8_t*)data, size)) { 
        // ping.decode()内部检查接收到的必须是合法STUN报文，否则返回错误信息
        if ((err = ping.decode(data, size)) != srs_success) {
            return srs_error_wrap(err, "decode stun packet failed");
        }
        return session->on_stun(skt, &ping); 
    }
    
    // RTP协议报文处理
    if (is_rtp_or_rtcp && !is_rtcp) { return session->on_rtp(data, size); }
    
    // RTCP协议报文处理
    if (is_rtp_or_rtcp && is_rtcp) { return session->on_rtcp(data, size); }
    
    // DTLS协议报文处理
    if (srs_is_dtls((uint8_t*)data, size)) { return session->on_dtls(data, size); }
}

1、STUN报文格式与Lite-ICE协商

由于IPv4地址不足以及网络架构的原因，一般用户的电脑或手机总是在一个局域网中，通过连接NAT网关接入公网Internet网络。

一般情况下，同一个局域网的设备之间通过私网IP地址进行通信，局域网设备通过NAT网关获取一个公网IP+端口实现与公网服务器之间的通信。

不同局域网中的设备，因为互相之间不知道对端设备的公网IP，所以一般情况下，无法直接通信。

STUN协议简单说，就是让一种私网设备获取自身公网IP地址的方法，它的运行原理很简单，如下所示：

????1、处于局域网的私网设备向公网STUN服务器发送STUN Binding Request请求报文。

????2、请求报文经过NAT网关时，请求报文中的源IP和源端口号被NAT网关修改为网关出口的公网IP+端口号。

????3、STUN服务器接收到请求报文后，返回一个STUN Binding Response 响应报文，并将服务器所看到的设备公网IP地址+端口信息(这个地址也被称为服务器反射地址server reflex address)，放到响应报文的净荷中一起返回给私网设备。

所以，如上过程所示，SRS4.0的WebRTC模块首先要实现一个简单的STUN服务，即接收客户端发送的STUN Binding Request请求报文，并返回一个STUN Binding Response 响应报文。代码如下：

1、SrsStunPacket::decode()函数用于校验客户端发送的Binding Request请求报文是否正确，并得到报文各字段信息

srs_error_t SrsStunPacket::decode(const char* buf, const int nb_buf)
{
    SrsBuffer* stream = new SrsBuffer(const_cast<char*>(buf), nb_buf);
    
    if (stream->left() < 20) { // 校验STUN报文长度一定不能少于20个字节 
        return srs_error_new(ERROR_RTC_STUN, "invalid stun packet, size=%d", stream->size());
    }
    // 按STUN报文格式依次读取各个字段，用于报文格式校验
    message_type = stream->read_2bytes();
    uint16_t message_len = stream->read_2bytes(); // STUN报文除去头部以后的净荷长度
    string magic_cookie = stream->read_string(4);
    transcation_id = stream->read_string(12);
    
    // 如果STUN报文的净荷长度+20字节的STUN报文头不等于UDP数据包长度，则数据包不是STUN报文，直接丢弃
    if (nb_buf != 20 + message_len) { 
        return srs_error_new(ERROR_RTC_STUN, "invalid stun packet, message_len=%d, nb_buf=%d", message_len, nb_buf);
    }

    // 按照TLV方式，依次解析STUN报文净荷部分的各个Attributes信息
    while (stream->left() >= 4) {
        uint16_t type = stream->read_2bytes();
        uint16_t len = stream->read_2bytes();
        ......
    }
}

STUN报文格式总是由一个20字节的STUN报文头+若干个TLV格式的STUN属性（attributes）字段组成，如下：
STUN报文头格式(固定20个字节，所以软件校验时首先判断报文长度小于20字节的都不是STUN报文)

 0                   1                   2                   3
 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|     STUN Message Type         |        Message Length         |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|              Magic Cookie 固定值0x2112A442                     |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|                                                               |
|                  Transaction ID 12Byte                        |
|                                                               |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

根据RFC5766，常用的 STUN Message Type定义如下:
  BINDING REQUEST(0x0001)       / RESPONSE(0x0101) / ERROR_RESPONSE(0x0111) / INDICATION(0x0011)
  SHARED_SECRET REQUEST(0x0002) / RESPONSE(0x0102) / ERROR_RESPONSE(0x0112) 
  ALLOCATE REQUEST(0x0003)      / RESPONSE(0x0103) / ERROR_RESPONSE(0x0113)
  REFRESH REQUEST(0x0004)       / RESPONSE(0x0104) / ERROR_RESPONSE(0x0114)
  SEND INDICATION(0x0016) 
  DATA INDICATION(0x0017)
  CREATE_PERM REQUEST(0x0008)   / RESPONSE(0x0108) / ERROR_RESPONSE(0x0118)
  CHANNEL_BIND REQUEST(0x0009)  / RESPONSE(0x0109) / ERROR_RESPONSE(0x0119)

STUN报文属性（attributes）字段格式

 0                   1                   2                   3
 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|             Type              |             Length            |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|                       Value(variable)                         |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
常见的属性类型:
  STUN_ATTR_MAPPED_ADDR       = 0x0001,/**MAPPED-ADDRESS */
  STUN_ATTR_USERNAME          = 0x0006,/**USERNAME attribute */
  STUN_ATTR_PASSWORD          = 0x0007,/**PASSWORD attribute */
  STUN_ATTR_MESSAGE_INTEGRITY = 0x0008,/**MESSAGE-INTEGRITY */  
  STUN_ATTR_ERROR_CODE        = 0x0009,/**ERROR-CODE */
  STUN_ATTR_REALM             = 0x0014,/**REALM attribute */
  STUN_ATTR_NONCE             = 0x0015,/**NONCE attribute */
  STUN_ATTR_XOR_RELAYED_ADDR  = 0x0016,/**TURN XOR-RELAYED-ADDRESS */
  STUN_ATTR_XOR_MAPPED_ADDR   = 0x0020,/**XOR-MAPPED-ADDRESS */

通过wireshark抓取STUN Binding Request报文，报文各字段如下所示：

2、服务端Lite-ICE工作原理：

1）服务端只处理Binding Request请求，并返回Binding Response响应报文。

2）把session状态设置为DOING_DTLS_HANDSHAKE状态，再调用SrsSecurityTransport::start_active_handshake()启动DTLS握手。

srs_error_t SrsRtcConnection::on_stun(SrsUdpMuxSocket* skt, SrsStunPacket* r) {
    
    if (!r->is_binding_request()) {
        return err;
    }
    update_sendonly_socket(skt);
    on_binding_request(r);
}

srs_error_t SrsRtcConnection::on_binding_request(SrsStunPacket* r)
{
    SrsStunPacket stun_binding_response;
    stun_binding_response.set_message_type(BindingResponse); // 构造Binding响应报文
    ......    
    sendonly_skt->sendto(stream->data(), stream->pos(), 0);  // 发送Binding响应报文
    
    if (state_ == WAITING_STUN) {
        state_ = DOING_DTLS_HANDSHAKE;
        transport_->start_active_handshake();  // 启动DTLS协议握手,实际passive端啥也没做
    }
}

通过wireshark抓取STUN Binding Response报文，报文各字段如下所示：

STUN协议涉及内容较多，但Lite-ICE模式下，主要只涉及上面两种报文，要全面了解STUN协议可参考以下链接：
https://blog.csdn.net/qq_32523587/article/details/103621017 STUN协议简要介绍
https://developer.aliyun.com/article/781948 WebRTC STUN | Short-term 消息认证

2、DTLS原理与协商过程

借用一份WebRTC协议栈的网图可知，当底层UDP通道连接建立后，接下来需要完成DTLS握手。DTLS本身很复杂，对于初学者我们大概需要明白几个关键点：

1、为了安全，公网上传输的数据必须加密。TLS是针对TCP协议的安全加密协议，而DTLS则是针对UDP协议的安全加密协议。

2、从性能的角度看，对于大批量数据传输，只能使用对称加密方式。但是，对称加密的密钥本身需要先通过非对称加密后，再经过网络传输到对端。DTLS和TLS内部已经包含了这两种加密方式。

3、WebRTC中，DTLS只是为DataChannel / SCTP提供加密服务，而音视频数据实际上是通过SRTP协议加密的，DTLS此时的工作只是为SRTP生成并导出密钥。

所以，启动DTLS握手协商之前，必须为DTLS生成密钥和证书，这部分代码在SrsDtlsCertificate::initialize()函数中，相关函数的详细说明可以参考OpenSSL接口文档。

srs_error_t SrsDtlsCertificate::initialize()
{
    srtp_init();   // 初始化SRTP加密协议库
    ......
    // 使用RSA或ECDSA算法为DTLS生成公钥和私钥(缺省使用ECDSA算法)
    ......
    dtls_cert = X509_new(); // 创建509格式的证书
    X509_set_pubkey(dtls_cert, dtls_pkey);  // 将公钥放入证书用于对外发布
    X509_sign(dtls_cert, dtls_pkey, EVP_sha1();  // 使用私钥对证书签名，防止证书被篡改
    
    // 生成证书的摘要信息,并作为SDP的fingerprint属性字段，随SDP一起与对端进行交换
    // 参与DTLS握手的双方，根据这个对端证书签名验证对端证书的有效性，最终完成DTLS握手
    X509_digest(dtls_cert, EVP_sha256(), md, &n); 
}

为每条session(SrsRtcConnection对象)创建SSL_CTX数据结构和SSL数据结构，并将证书和密钥导入SSL_CTX数据结构。

srs_error_t SrsRtcConnection::initialize(SrsRequest* r, bool dtls, bool srtp, string username) {
    ......
    transport_->initialize(cfg);  // session对象传输层初始化
}

srs_error_t SrsSecurityTransport::initialize(SrsSessionConfig* cfg){
    return dtls_->initialize(cfg->dtls_role, cfg->dtls_version);
}

srs_error_t SrsDtls::initialize(std::string role, std::string version)
{
    if (role == "active") {  // 这个role在SDP报文中也有体现，主动发起DTLS协商的一般是客户端
        impl = new SrsDtlsClientImpl(callback_);
    } else {
        impl = new SrsDtlsServerImpl(callback_); // 服务端缺省创建这个对象
    }

    return impl->initialize(version, role);
}

srs_error_t SrsDtlsServerImpl::initialize(std::string version, std::string role)
{
    SrsDtlsImpl::initialize(version, role);
    SSL_set_accept_state(dtls); // 设置Dtls工作在服务端模式
    return err;
}

srs_error_t SrsDtlsImpl::initialize(std::string version, std::string role)
{
    dtls_ctx = srs_build_dtls_ctx(version_, role); // 此函数用于创建SSL_CTX数据结构
    dtls = SSL_new(dtls_ctx); // 创建SSL数据结构，作为DTLS算法模块
    ......
    DTLS_set_timer_cb(dtls, dtls_timer_cb); // 定时器回调函数用于UDP报文重发
    
    bio_in = BIO_new(BIO_s_mem();
    bio_out = BIO_new(BIO_s_mem();
    SSL_set_bio(dtls, bio_in, bio_out); // 创建BIO对象，协助dtls算法模块读写数据
}

另外，如上最后一个函数可知，这里还引入了BIO数据结构，用于协助DTLS算法模块收发数据，具体如下：

WebRTC模块SrsUdpMuxListener::cycle()协程接收到dtls报文，并通过SrsRtcConnection::on_dtls()函数将报文转给DTLS算法模块。

srs_error_t SrsRtcConnection::on_dtls(char* data, int nb_data){
    return transport_->on_dtls(data, nb_data);
}
srs_error_t SrsSecurityTransport::on_dtls(char* data, int nb_data){
    return dtls_->on_dtls(data, nb_data);
}
srs_error_t SrsDtls::on_dtls(char* data, int nb_data){
    return impl->on_dtls(data, nb_data);
}
srs_error_t SrsDtlsImpl::on_dtls(char* data, int nb_data){
    do_on_dtls(data, nb_data)) != srs_success) 
}

srs_error_t SrsDtlsImpl::do_on_dtls(char* data, int nb_data)
{
    BIO_reset(bio_in);
    BIO_reset(bio_out);  
    BIO_write(bio_in, data, nb_data); // 将从网络接收的DTLS数据通过BIO写入SSL
    
    do_handshake(); // 处理DTLS握手

    // do_handshake()函数用于完成握手，下面的处理大概是读取BIO中生成的应答数据，再通过网络发送给对端
    for (int i = 0; i < 1024 && BIO_ctrl_pending(bio_in) > 0; i++) {
        int r0 = SSL_read(dtls, buf, sizeof(buf));
        int r1 = SSL_get_error(dtls, r0);
        if (r0 <= 0) {
            if (r1 != SSL_ERROR_WANT_READ && r1 != SSL_ERROR_WANT_WRITE) {
                break;
            }
            
            int size = BIO_get_mem_data(bio_out, (char**)&data);
            callback_->write_dtls_data(data, size);
        }
    }
}

srs_error_t SrsDtlsImpl::do_handshake()
{
    int r0 = SSL_do_handshake(dtls);
    int r1 = SSL_get_error(dtls, r0);
    
    if (r1 == SSL_ERROR_NONE) { // 如果返回值是SSL_ERROR_NONE，则表示DTLS握手成功
        handshake_done_for_us = true;
    }
    ......
    if (handshake_done_for_us) { 
        err = on_handshake_done(); // 向外通知DTLS握手成功
    }
}

srs_error_t SrsDtlsServerImpl::on_handshake_done() {    
    callback_->on_dtls_handshake_done();
}
srs_error_t SrsSecurityTransport::on_dtls_handshake_done(){
    handshake_done = true; // 设置DTLS握手成功标志
    ......
    srtp_initialize(); // 接下来完成SRTP初始化
    ......
    return session_->on_connection_established(); // 通知session对象，WebRTC连接建立
}

DTLS协商完成后，会自动生成SRTP发送和接收报文时需要的密钥，使用此密钥完成SRTP初始化：

srs_error_t SrsSecurityTransport::srtp_initialize()
{
    dtls_->get_srtp_key(recv_key, send_key); // 从DTLS中得到SRTP发送和接收报文时需要的密钥
    srtp_->initialize(recv_key, send_key);  // 将收发报文所需的密钥写入SRTP协议模块
                                            // 此函数内部通过调用srtp_create创建收发加密报文的上下文句柄
    return err;
}

Session对象(SrsRtcConnection)接收到WebRTC连接建立的通知，启动属于此连接的推流端接收应答协程SrsRtcPLIWorker::cycle()或拉流端发送协程SrsRtcPlayStream::cycle()

srs_error_t SrsRtcConnection::on_connection_established()
{
    {
        SrsRtcPublishStream* publisher = it->second;
        publisher->start();
    }
    或
    {
        SrsRtcPlayStream* player = it->second;
        player->start();
    }
}

3、连接建立，服务端处理RTP报文

DTLS协商成功后，客户端与服务器之间的WebRTC连接真正建立，服务端接下来开始处理RTP报文：
1）根据接收的RTP报文的IP找到SrsRtcConnection对象
2）根据接收的RTP报文的SSRC找到SrsRtcPublishStream对象
3）根据配置，对RTP报文做相应的特殊处理
4）调用srtp_unprotect()函数对SRTP报文进行解密
5）最终调用SrsRtcPublishStream::do_on_rtp_plaintext()函数处理RTP报文

srs_error_t SrsRtcConnection::on_rtp() {
    ......
    find_publisher(data, nb_data, &publisher); // 根据报文头部的SSRC找到对应的流对象
    return publisher->on_rtp(data, nb_data);
}

srs_error_t SrsRtcPublishStream::on_rtp(char* data, int nb_data)
{
    if (nn_simulate_nack_drop) { return err; }  // 如果设置了模拟丢包触发NACK，则直接返回
    
    if (twcc_id_) { // 如果开启了TWCC，RTP包会带扩展信息，增加预处理防止后续SRTP解密报文时出错
        srs_rtp_fast_parse_twcc(data, nb_data, twcc_id_, twcc_sn);
    }
    
    if (pt_to_drop_) {  // 如果某些类型的报文在配置文件中设置为丢弃，这里需要识别并丢弃
        uint8_t pt = srs_rtp_fast_parse_pt(data, nb_data);
        if (pt_to_drop_ == pt) {  return err; }
    }
    
    // 最终调用srtp_unprotect()函数对SRTP报文进行解密
    session_->transport_->unprotect_rtp(plaintext, &nb_plaintext);
    
    on_rtp_plaintext();  // 处理SRTP解密后的RTP明文
}

srs_error_t SrsRtcPublishStream::on_rtp_plaintext(char* plaintext, int nb_plaintext) {
    do_on_rtp_plaintext(pkt, &buf);
}

此函数内部一边将报文放入消费者队列，一边根据接收报文的序列号调整NACK队列

srs_error_t SrsRtcPublishStream::do_on_rtp_plaintext(SrsRtpPacket*& pkt, SrsBuffer* buf) {
    ......
    pkt->decode(buf);
    
    // SrsRtpPacket数据包在这里被直接放入SrsRtcConsumer对象的RTP报文缓存队列
    SrsRtcAudioRecvTrack* audio_track = get_audio_track(ssrc);
    SrsRtcVideoRecvTrack* video_track = get_video_track(ssrc);
    audio_track->on_rtp(source, pkt);
    video_track->on_rtp(source, pkt); 

    // 如果circuit-breaker使能，当判断到网络发生拥塞时，则停止发送RTP数据包
    if (_srs_circuit_breaker->hybrid_critical_water_level()) { return err;}

    // 如果NACK使能，根据接收报文的序列号调整接收对象SrsRtcRecvTrack内部的NACK队列
    if (nack_enabled_) {
        audio_track->on_nack(&pkt);
        video_track->on_nack(&pkt); // 这里两个函数其实都是SrsRtcRecvTrack::on_nack()函数
    }
}

根据接收报文的序列号调整NACK队列，用于后续指导NACK报文的发送

srs_error_t SrsRtcRecvTrack::on_nack(SrsRtpPacket** ppkt) {
    uint16_t seq = pkt->header.get_sequence();
    if (nack_receiver_->find(seq)) {
        nack_receiver_->remove(seq); // 从NACK队列中删除已接收到的序列号
        return err;
    }
    // 将接收到的序列号加入NACK队列,并重新计算NACK队列最新的起始序列号和结束序列号
    rtp_queue_->update(seq, nack_first, nack_last); 
    
    // 将最新的起始序列号和结束序列号写入NACK队列，并通过check_queue_size()检测如果队列满时清空队列
    if (srs_rtp_seq_distance(nack_first, nack_last) > 0) {
        nack_receiver_->insert(nack_first, nack_last);
        nack_receiver_->check_queue_size();
    }
    
    rtp_queue_->set(seq, pkt->copy()); // 将收到的RTP包放入一个环形缓冲区
}

4、服务端RTCP报文处理

RTP报文封装音视频数据，底层采用UDP协议发送，本身属于不可靠传输。通过引入RTCP协议，周期性的发送RTCP报文描述本端的接收和发送状态，可以提升音视频数据的可靠传输、流量控制和拥塞控制等服务质量保证。

WebRTC服务模块对于RTCP报文主要有两种处理逻辑：
1）根据服务端状态主动发送RTCP报文
2）接收并处理客户端发送的RTCP报文

NACK报文收发原理：

1）首先创建一个NACK定时器对象，它的内部工作原理就是以观察者模式订阅了一个20毫秒周期的系统定时器

SrsRtcConnectionNackTimer::SrsRtcConnectionNackTimer(SrsRtcConnection* p) : p_(p)
{
    _srs_hybrid->timer20ms()->subscribe(this);
}

2）20毫秒定时器超时，周期性调用SrsRtcPublishStream::check_send_nacks()检测推流接收端是否需要发送NACK

srs_error_t SrsRtcConnectionNackTimer::on_timer(srs_utime_t interval){
    ......
    std::map<std::string, SrsRtcPublishStream*>::iterator it;
    for (it = p_->publishers_.begin(); it != p_->publishers_.end(); it++) {
        SrsRtcPublishStream* publisher = it->second;
        publisher->check_send_nacks();  // 检测推流接收端是否需要发送NACK
    }

    return err;
}

srs_error_t SrsRtcPublishStream::check_send_nacks(){
    for (int i = 0; i < (int)video_tracks_.size(); ++i) {
        track->check_send_nacks();
    }

    for (int i = 0; i < (int)audio_tracks_.size(); ++i) {
        track->check_send_nacks()) != srs_success)
    }

    return err;
}

srs_error_t SrsRtcRecvTrack::do_check_send_nacks(uint32_t& timeout_nacks)
{
    session_->check_send_nacks(nack_receiver_, track_desc_->ssrc_, sent_nacks, timeout_nacks);
    return err;
}

3）构造RTCP类型的NACK报文并发送

void SrsRtcConnection::check_send_nacks(SrsRtpNackForReceiver* nack, uint32_t ssrc, uint32_t& sent_nacks, uint32_t& timeout_nacks)
{
    SrsRtcpNack rtcpNack(ssrc);
    rtcpNack.set_media_ssrc(ssrc);
    nack->get_nack_seqs(rtcpNack, timeout_nacks);

    if(rtcpNack.empty()){ return; }

    char buf[kRtcpPacketSize];
    SrsBuffer stream(buf, sizeof(buf));

    rtcpNack.encode(&stream);

    int nb_protected_buf = stream.pos();
    transport_->protect_rtcp(stream.data(), &nb_protected_buf);

    sendonly_skt->sendto(stream.data(), nb_protected_buf, 0);
}

PLI(Picture Loss Indication)报文收发原理：

当视频接收端向发送端反馈一个PLI报文时，发送方的编码器会重新生成关键帧并发送给接收端。

SRS4.0 WebRTC模块中推流端接收对象SrsRtcPublishStream和拉流端发送对象SrsRtcPlayStream在建立连接的时候，都会创建各自的SrsRtcPLIWorker::cycle()协程。

1）推流端接收对象SrsRtcPublishStream创建的SrsRtcPLIWorker::cycle()协程，此协程等待条件变量唤醒后，遍历plis_队列，发送PLI请求报文，

srs_error_t SrsRtcPLIWorker::cycle()
{

    while (true) {

        while (!plis_.empty()) {
            ......
            plis.swap(plis_);

            for (map<uint32_t, SrsContextId>::iterator it = plis.begin(); 
                 it != plis.end(); ++it) {
                ......
                handler_->do_request_keyframe(ssrc, cid);
            }
        }
        srs_cond_wait(wait_);
    }

    return err;
}

其中plis_队列和条件变量唤醒的操作，都是在SrsRtcSource对象的定时器超时函数中处理的。

SrsRtcSource继承ISrsFastTimer类，再通过SrsRtcSource::on_publish()函数，以观察者模式订阅了一个100毫秒周期的系统定时器

class SrsRtcSource : public ISrsFastTimer 

srs_error_t SrsRtcSource::on_publish()
{
    if (bridger_) {
        // The PLI interval for RTC2RTMP.
        pli_for_rtmp_ = _srs_config->get_rtc_pli_for_rtmp(req->vhost);

        // @see SrsRtcSource::on_timer()
        _srs_hybrid->timer100ms()->subscribe(this);
    }
}

最终SrsRtcSource::on_timer()超时函数以100毫秒的周期被执行，以默认6秒为周期调用一次publish_stream_->request_keyframe()函数，插入plis_队列，并触发条件变量。

srs_error_t SrsRtcSource::on_timer(srs_utime_t interval)
{
    ......
    if (!publish_stream_) { return err; }

    // Request PLI and reset the timer.
    if (true) {
        pli_elapsed_ += interval;
        if (pli_elapsed_ < pli_for_rtmp_) { return err; }
        
        pli_elapsed_ = 0;
    }

    for (int i = 0; i < (int)stream_desc_->video_track_descs_.size(); i++) {
        SrsRtcTrackDescription* desc = stream_desc_->video_track_descs_.at(i);
        publish_stream_->request_keyframe(desc->ssrc_);
    }

    return err;
}

void SrsRtcPublishStream::request_keyframe(uint32_t ssrc) {
    SrsContextId sub_cid = _srs_context->get_id();
    pli_worker_->request_keyframe(ssrc, sub_cid);
}

void SrsRtcPLIWorker::request_keyframe(uint32_t ssrc, SrsContextId cid) {
    plis_.insert(make_pair(ssrc, cid));
    srs_cond_signal(wait_);
}

接收并处理对端发送的RTCP报文

srs_error_t SrsRtcConnection::on_rtcp() {
    SrsSecurityTransport::unprotect_rtp(); // 使用srtp_unprotect_rtcp解密RTCP报文
    
    SrsRtcpCompound rtcp_compound;
    rtcp_compound.decode(buffer); // 解析得到具体的RTCP报文  
                                  // FIR/SR/RR/SDES/BYE/APP/RTPFB/PSFB/XR
    
    while(NULL != (rtcp = rtcp_compound.get_next_rtcp())) {
        dispatch_rtcp(rtcp); // 此函数为各类RTCP报文提供处理路由
    }
}

srs_error_t SrsRtcConnection::dispatch_rtcp(SrsRtcpCommon* rtcp)
{
   // 分类处理各类RTCP报文
}

总结：

本章以WebRTC推流端连接建立的过程为线索，分析了过程中各种协议报文的基本处理流程：
1、Lite-ICE机制、STUN Binding Request和STUN Binding Response报文格式
2、DTLS工作原理、为SRTP获取密钥并完成初始化
3、音视频RTP报文接收并放入对应的拉流端消费者队列，以及更新NACK队列
4、定时发送RTCP报文与接收处理对端发送的RTCP报文