SRS:webrtc_to_rtmp详解

SRS是一个开源的（MIT协议）简单高效的实时视频服务器，支持RTMP、WebRTC、HLS、HTTP-FLV、SRT、MPEG-DASH和GB28181等协议。 SRS媒体服务器和FFmpeg、OBS、VLC、 WebRTC等客户端配合使用，提供流的接收和分发的能力，是一个典型的发布 （推流）和订阅（播放）服务器模型。 SRS支持互联网广泛应用的音视频协议转换，比如可以将RTMP或SRT， 转成HLS或HTTP-FLV或WebRTC等协议

./objs/srs -c conf/rtc.conf
#记得 rtc.conf -> rtc_to_rtmp on;

./objs/signaling

openssl genrsa -out server.key 2048
subj="/C=CN/ST=Beijing/L=Beijing/O=Me/OU=Me/CN=me.org"
openssl req -new -x509 -key server.key -out server.crt -days 365 -subj $subj
sudo ./objs/httpx-static -t 80  -s 443 -r /data/srs/trunk/3rdparty/signaling/www/ -ssk ./objs/server.key -ssc ./objs/server.crt -p http://127.0.0.1:1985/rtc -p http://127.0.0.1:1989/sig

1.数据连接通道
2.数据格式转换

rtc_server->bridge->live_server(rtmp_server)

rtc_server->live_source(bridge)->live_server(rtmp_server)

接受数据-> 读取数据 -> 解析数据 -> 数据格式转换 -> 数据发送

rtc_server(UDP:8000)_监听端口->rtc_server_接收数据->rtc_server_解析数据->rtc_server_转换数据->rtc_to_rtmp_brigre_发送数据->live_server->rtmp_client

srs_main_server.cpp => int main(int argc, char** argv, char** envp) //初始化入口函数
srs_main_server.cpp => err = do_main(argc, argv, envp) //主函数启动
srs_main_server.cpp => srs_error_t do_main(int argc, char** argv, char** envp) // 
srs_main_server.cpp => _srs_config->check_config() //配置文件读取以及检测配置
srs_main_server.cpp => run_directly_or_daemon() //进程是否守护以及后台运行
srs_main_server.cpp => run_in_thread_pool() // 运行线程池
srs_main_server.cpp => _srs_thread_pool->execute("hybrid", run_hybrid_server, (void*)NULL)) //  执行 run_hybrid_server 
srs_main_server.cpp => _srs_hybrid->register_server(new RtcServerAdapter()) //注册rtc服务
srs_main_server.cpp => _srs_hybrid->initialize() / server->initialize() //server初始化
srs_main_server.cpp => _srs_hybrid->run() / SrsHybridServer::run() //server run
srs_app_rtc_server.cpp => RtcServerAdapter::run(SrsWaitGroup* wg) //
srs_app_rtc_server.cpp => SrsRtcServer::listen_api() //进行
srs_app_rtc_conn.cpp => SrsGoApiRtcPublish::serve_http / server_->create_session(ruc, local_sdp, &session) // 进行rtc publish 会话创建
srs_app_rtc_conn.cpp => SrsGoApiRtcPlay::serve_http / server_->create_session(ruc, local_sdp, &session) // 进行rtc play 会话创建
srs_app_rtc_conn.cpp => do_create_session(ruc, local_sdp, session) // 去创建rtc 会话
srs_app_rtc_conn.cpp => SrsRtcConnection::add_publisher()/ session->add_publisher(ruc, local_sdp) //增加一个publisher
srs_app_rtc_conn.cpp => SrsRtcConnection::create_publisher(SrsRequest* req, SrsRtcSourceDescription* stream_desc) //rtc创建 Publisher
srs_app_rtc_conn.cpp => publisher->initialize(req, stream_desc) //publisher 初始化
srs_app_rtc_conn.cpp => SrsRtcPublishStream::initialize(SrsRequest* r, SrsRtcSourceDescription* stream_desc) //rtc 推流初始化
srs_app_rtc_conn.cpp => source_->set_bridge(bridge) //桥接器设置

    // Bridge to rtmp
#if defined(SRS_RTC) && defined(SRS_FFMPEG_FIT)
	//读取配置文件看是否开启协议转换设置
    bool rtc_to_rtmp = _srs_config->get_rtc_to_rtmp(req_->vhost); 
	//如果开启协议转换
    if (rtc_to_rtmp) {
        //遍历检查是否有对应的 liveSource,如果没有则创建
        if ((err = _srs_sources->fetch_or_create(r, _srs_hybrid->srs()->instance(), live_source)) != srs_success) {
            return srs_error_wrap(err, "create source");
        }

        // Disable GOP cache for RTC2RTMP bridge, to keep the streams in sync,
        // especially for stream merging.
        live_source->set_cache(false);
		//创建新的桥
        SrsCompositeBridge* bridge = new SrsCompositeBridge();
        //添加 to_rtmp_brige
        bridge->append(new SrsFrameToRtmpBridge(live_source));
		//brige 初始化
        if ((err = bridge->initialize(r)) != srs_success) {
            srs_freep(bridge);
            return srs_error_wrap(err, "create bridge");
        }
		//liveSource与brige进行绑定
        source_->set_bridge(bridge);
    }
#endif

srs_app_listener.cpp => srs_error_t SrsUdpMuxListener::cycle() //监听函数
srs_app_listener.cpp => int nread = skt.recvfrom(SRS_UTIME_NO_TIMEOUT) //读取数据
srs_app_listener.cpp => int SrsUdpMuxSocket::recvfrom(srs_utime_t timeout) //读取数据函数
srs_app_listener.cpp => nread = srs_recvfrom(lfd, buf, nb_buf, (sockaddr*)&from, &fromlen, timeout) //读取数据到 buf[SRS_UDP_MAX_PACKET_SIZE] buf[65535]

srs_app_listener.cpp => err = handler->on_udp_packet(&skt); //解析UDP数据入口函数
srs_app_listener.hpp =>  ISrsUdpMuxHandler* handler; //上述handler定义
srs_app_listener.cpp => SrsUdpMuxListener::SrsUdpMuxListener(ISrsUdpMuxHandler* h, std::string i, int p) //handler赋值
srs_app_rtc_server.cpp => SrsUdpMuxListener* listener = new SrsUdpMuxListener(this, ip, port); //SrsUdpMuxListener 实例化，handler入参，值为 this 即 SrsRtcServer
srs_app_rtc_server.hpp => class SrsRtcServer : public ISrsUdpMuxHandler, public ISrsFastTimer, public ISrsReloadHandler //在定义SrsRtcServer的时候可以看到继承了公有类 ISrsUdpMuxHandler，即on_udp_packet()为 virtual srs_error_t on_udp_packet(SrsUdpMuxSocket* skt)
srs_app_rtc_server.hpp => virtual srs_error_t on_udp_packet(SrsUdpMuxSocket* skt) 声明了虚函数 on_udp_packet()，且在CPP做了实现
srs_app_rtc_server.cpp => SrsRtcServer::on_udp_packet(SrsUdpMuxSocket* skt) 
srs_app_rtc_server.cpp => char* data = skt->data(); int size = skt->size(); //获取了 data = buf[SRS_UDP_MAX_PACKET_SIZE] size= nread
srs_app_rtc_server.cpp => bool is_rtp_or_rtcp = srs_is_rtp_or_rtcp((uint8_t*)data, size); //判断是否是 rtp_or_rtcp 数据包
srs_app_rtc_server.cpp => bool is_rtcp = srs_is_rtcp((uint8_t*)data, size); //判断是否是 rtcp 数据包
srs_app_rtc_server.cpp => err = session->udp()->on_rtp(data, size); //如果是rtp的数据包，进行对应的处理
srs_app_rtc_network.cpp => srs_error_t SrsRtcUdpNetwork::on_rtp(char* data, int nb_data) //定位到 rtc_network udp on_rtp()函数
srs_app_rtc_network.cpp => conn_->on_rtp_plaintext(unprotected_buf, nb_unprotected_buf) //进行rtp上下文处理
srs_app_rtc_conn.cpp => publisher->on_rtp_plaintext(data, nb_data) //SrsRtcPublishStream 进行上下文处理
srs_app_rtc_conn.cpp => SrsRtcPublishStream::on_rtp_plaintext(char* plaintext, int nb_plaintext)
srs_app_rtc_conn.cpp => err = do_on_rtp_plaintext(pkt, &buf); //SrsRtcPublishStream 进一步进行 rtp上下文处理
srs_app_rtc_conn.cpp => pkt->decode(buf) //进行SrsRtpPacket数据包解码
srs_kernel_rtc_rtp.cpp => srs_error_t SrsRtpHeader::decode(SrsBuffer* buf) //关键rtp解码函数
srs_kernel_rtc_rtp.cpp => ssrc = buf->read_4bytes() //获取ssrc
srs_kernel_rtc_rtp.cpp => SrsRtcAudioRecvTrack* audio_track = get_audio_track(ssrc) //通过ssrc进行音频实例化
srs_kernel_rtc_rtp.cpp => SrsRtcVideoRecvTrack* video_track = get_video_track(ssrc) //通过ssrc进行视频实例化
srs_kernel_rtc_rtp.cpp => audio_track->on_rtp(source, pkt) //进行音频包处理
srs_kernel_rtc_rtp.cpp => video_track->on_rtp(source, pkt) //进行视频包处理
srs_app_rtc_source.cpp => source->on_rtp(pkt) //SrsRtcSource 进行 SrsRtpPacket* pkt 包处理
srs_app_rtc_source.cpp => frame_builder_->on_rtp(pkt)//SrsRtcFrameBuilder 进行 pkt包处理
srs_app_rtc_source.cpp => SrsRtcFrameBuilder::on_rtp(SrsRtpPacket *pkt)
srs_app_rtc_source.cpp => transcode_audio(pkt) //如果是音频包，进行转码处理，转码为AAC
srs_app_rtc_source.cpp => packet_video(pkt) //如果是视频包，进行视频包处理

srs_app_rtc_source.cpp => SrsRtcFrameBuilder::transcode_audio(SrsRtpPacket *pkt)//入口函数
srs_app_rtc_source.cpp => if (is_first_audio_) //判断是否为首帧音频帧。进行首帧音频帧处理
    srs_app_rtc_source.cpp => codec_->aac_codec_header(&header, &header_len); //进行首帧header处理
    srs_app_rtc_source.cpp => SrsCommonMessage out_rtmp //定义rtmp msg消息
    srs_app_rtc_source.cpp => packet_aac(&out_rtmp, (char *)header, header_len, ts, is_first_audio_) //进行rtmp消息的格式化
    srs_app_rtc_source.cpp => SrsSharedPtrMessage msg; //定义通用发送的消息包
    srs_app_rtc_source.cpp => msg.create(&out_rtmp) //根据音频包，格式化通用消息包
    srs_app_rtc_source.cpp => bridge_->on_frame(&msg)) //桥接器进行消息消息包处理
srs_app_rtc_source.cpp => SrsRtpRawPayload *payload = dynamic_cast<SrsRtpRawPayload*>(pkt->payload()); //获取pkt内音频包消息
srs_app_rtc_source.cpp => codec_->transcode(&frame, out_pkts) //使用ffmpeg api 进行音频帧转码
srs_app_rtc_source.cpp => std::vector<SrsAudioFrame*>::iterator it = out_pkts.begin(); it != out_pkts.end(); ++it //进行转码后音频帧处理
srs_app_rtc_source.cpp => SrsCommonMessage out_rtmp //定义rtmp msg消息
srs_app_rtc_source.cpp => packet_aac(&out_rtmp, (char *)header, header_len, ts, is_first_audio_) //进行rtmp消息的格式化
srs_app_rtc_source.cpp => SrsSharedPtrMessage msg; //定义通用发送的消息包
srs_app_rtc_source.cpp => msg.create(&out_rtmp) //根据音频包，格式化通用消息包
srs_app_rtc_source.cpp => bridge_->on_frame(&msg)) //桥接器进行消息消息包处理

    std::vector<SrsAudioFrame*> out_pkts;
    //获取rtp音频包
    SrsRtpRawPayload *payload = dynamic_cast<SrsRtpRawPayload*>(pkt->payload());
	//定义一个SrsAudioFrame变量
    SrsAudioFrame frame;
	//拷贝音频帧数据
    frame.add_sample(payload->payload, payload->nn_payload);
    frame.dts = ts;
    frame.cts = 0;
	//音频帧转码
    err = codec_->transcode(&frame, out_pkts);
    if (err != srs_success) {
        return err;
    }

void SrsRtcFrameBuilder::packet_aac(SrsCommonMessage* audio, char* data, int len, uint32_t pts, bool is_header)
{
    //增加2个字节的tag header
    int rtmp_len = len + 2;
    audio->header.initialize_audio(rtmp_len, pts, 1);
    audio->create_payload(rtmp_len);
    SrsBuffer stream(audio->payload, rtmp_len);
    //AAC flag 属性设置
    uint8_t aac_flag = (SrsAudioCodecIdAAC << 4) | (SrsAudioSampleRate44100 << 2) | (SrsAudioSampleBits16bit << 1) | SrsAudioChannelsStereo;
    //写入AAC 详细属性
    stream.write_1bytes(aac_flag);
    //针对header处理
    if (is_header) {
        stream.write_1bytes(0);
    } else {
        stream.write_1bytes(1);
    }
    //写入flv数据
    stream.write_bytes(data, len);
    audio->size = rtmp_len;
}

srs_app_rtc_source.cpp => SrsRtcFrameBuilder::packet_video(SrsRtpPacket* src)//入口函数
srs_app_rtc_source.cpp => packet_video_key_frame(pkt) //如果为关键帧 - 进行关键帧处理
    srs_app_rtc_source.cpp => SrsSample* sps = stap_payload ? stap_payload->get_sps() : NULL; //获取关键帧 sps 信息
    srs_app_rtc_source.cpp => SrsSample* pps = stap_payload ? stap_payload->get_pps() : NULL; //获取关键帧 pps 信息
    srs_app_rtc_source.cpp => avc->mux_sequence_header(string(sps->bytes, sps->size), string(pps->bytes, pps->size), sh)) //生成rtmp:flv 关键帧 sps/pps 信息头
    srs_app_rtc_source.cpp => avc->mux_avc2flv(sh, SrsVideoAvcFrameTypeKeyFrame, SrsVideoAvcFrameTraitSequenceHeader, pkt->get_avsync_time(), pkt->get_avsync_time(), &flv, &nb_flv)) != srs_success) // 帧内容转 h264 raw换为flv - h264格式
    srs_app_rtc_source.cpp => SrsSharedPtrMessage msg; \  msg.create(&rtmp) // 生成通用消息包
    srs_app_rtc_source.cpp => bridge_->on_frame(&msg)  //桥接器进行消息处理
srs_app_rtc_source.cpp => packet_video_rtmp(const uint16_t start, const uint16_t end)//普通视频帧以及关键帧共同适用的帧内容处理
srs_app_rtc_source.cpp => nb_payload //传输数据包的大小，要记得是按照flv数据格式来进行计算的 rtp_body_payload + flv_header + flv_tag_header 来处理的

    int nb_payload = 0;
    int16_t cnt = srs_rtp_seq_distance(start, end) + 1;
    srs_assert(cnt >= 1);

    for (uint16_t i = 0; i < (uint16_t)cnt; ++i) {
        uint16_t sn = start + i;
        uint16_t index = cache_index(sn);
        SrsRtpPacket* pkt = cache_video_pkts_[index].pkt;

        // fix crash when pkt->payload() if pkt is nullptr;
        if (!pkt) continue;

        // calculate nalu len
        SrsRtpFUAPayload2* fua_payload = dynamic_cast<SrsRtpFUAPayload2*>(pkt->payload());
        if (fua_payload && fua_payload->size > 0) {
            if (fua_payload->start) {
                // 在FUA分片传输中，第一片的大小及为包头怎么处理， 1个字节的格式为， 4个字节的整包大小
                nb_payload += 1 + 4;
            }
            nb_payload += fua_payload->size;
            continue;
        }

        SrsRtpSTAPPayload* stap_payload = dynamic_cast<SrsRtpSTAPPayload*>(pkt->payload());
        if (stap_payload) {
            for (int j = 0; j < (int)stap_payload->nalus.size(); ++j) {
                SrsSample* sample = stap_payload->nalus.at(j);
                if (sample->size > 0) {
                    // 单包的处理相对简单， 大小为 包的大小 + 包的数据大小
                    nb_payload += 4 + sample->size;
                }
            }
            continue;
        }

        SrsRtpRawPayload* raw_payload = dynamic_cast<SrsRtpRawPayload*>(pkt->payload());
        if (raw_payload && raw_payload->nn_payload > 0) {
            //多包，在FLV中，每一个子包都为单独的发送的，即每个子包都要 + 子包的大小
            nb_payload += 4 + raw_payload->nn_payload;
            continue;
        }
    }

    if (0 == nb_payload) {
        srs_warn("empty nalu");
        return err;
    }
   //type_codec1 + avc_type + composition time + nalu size + nalu
	//还需要加上TAG header 的大小
    nb_payload += 1 + 1 + 3;

   SrsCommonMessage rtmp;
    SrsRtpPacket* pkt = cache_video_pkts_[cache_index(start)].pkt;
    rtmp.header.initialize_video(nb_payload, pkt->get_avsync_time(), 1);
    rtmp.create_payload(nb_payload);
    rtmp.size = nb_payload;
    SrsBuffer payload(rtmp.payload, rtmp.size);
 //写入包头
    if (pkt->is_keyframe()) {
        payload.write_1bytes(0x17); // type(4 bits): key frame; code(4bits): avc
        rtp_key_frame_ts_ = -1;
    } else {
        payload.write_1bytes(0x27); // type(4 bits): inter frame; code(4bits): avc
    }
    payload.write_1bytes(0x01); // avc_type: nalu
    payload.write_1bytes(0x0);  // composition time
    payload.write_1bytes(0x0);
    payload.write_1bytes(0x0);
	//一个包的大小
    int nalu_len = 0;
    for (uint16_t i = 0; i < (uint16_t)cnt; ++i) {
        uint16_t index = cache_index((start + i));
        SrsRtpPacket* pkt = cache_video_pkts_[index].pkt;

        // fix crash when pkt->payload() if pkt is nullptr;
        if (!pkt) continue;

        cache_video_pkts_[index].in_use = false;
        cache_video_pkts_[index].pkt = NULL;
        cache_video_pkts_[index].ts = 0;
        cache_video_pkts_[index].rtp_ts = 0;
        cache_video_pkts_[index].sn = 0;

        SrsRtpFUAPayload2* fua_payload = dynamic_cast<SrsRtpFUAPayload2*>(pkt->payload());
        if (fua_payload && fua_payload->size > 0) {
            //处理第一个分片包
            if (fua_payload->start) {
                //计算分片包第一个包的大小
                nalu_len = fua_payload->size + 1;
                //skip 4 bytes to write nalu_len future
                //跳过开始的4个字节，在整个包写入完成后，计算出整包大小，然后在写入
                payload.skip(4);
                //写入一个字节的格式位
                payload.write_1bytes(fua_payload->nri | fua_payload->nalu_type);
                //拷贝分片包
                payload.write_bytes(fua_payload->payload, fua_payload->size);
            } else {
                //中间包累加
                nalu_len += fua_payload->size;
                //拷贝中间包
                payload.write_bytes(fua_payload->payload, fua_payload->size);
                //处理最后一个包
                if (fua_payload->end) {
                    //write nalu_len back
                    //回写整包大小 - 指针跳到分片包开始位置
                    payload.skip(-(4 + nalu_len));
                    //写入整包大小
                    payload.write_4bytes(nalu_len);
                    //指针跳到包尾
                    payload.skip(nalu_len);
                }
            }
            srs_freep(pkt);
            continue;
        }

        SrsRtpSTAPPayload* stap_payload = dynamic_cast<SrsRtpSTAPPayload*>(pkt->payload());
        if (stap_payload) {
            for (int j = 0; j < (int)stap_payload->nalus.size(); ++j) {
                SrsSample* sample = stap_payload->nalus.at(j);
                if (sample->size > 0) {
                    //在每一个子包，包头写入子包大小
                    payload.write_4bytes(sample->size);
                    //写入子包数据
                    payload.write_bytes(sample->bytes, sample->size);
                }
            }
            srs_freep(pkt);
            continue;
        }

        SrsRtpRawPayload* raw_payload = dynamic_cast<SrsRtpRawPayload*>(pkt->payload());
        if (raw_payload && raw_payload->nn_payload > 0) {
            //写入单包大小
            payload.write_4bytes(raw_payload->nn_payload);
            //写入单包数据
            payload.write_bytes(raw_payload->payload, raw_payload->nn_payload);
            srs_freep(pkt);
            continue;
        }

        srs_freep(pkt);
    }
	//创建发送包
    SrsSharedPtrMessage msg;
    if ((err = msg.create(&rtmp)) != srs_success) {
        return srs_error_wrap(err, "create message");
    }
	//桥接触处理转换后的包
    if ((err = bridge_->on_frame(&msg)) != srs_success) {
        srs_warn("fail to pack video frame");
    }

srs_app_stream_brige.cpp => srs_error_t SrsFrameToRtmpBridge::on_frame(SrsSharedPtrMessage* frame)//入口函数
srs_app_stream_brige.cpp => source_->on_frame(frame) // SrsLiveSource 进行消息包处理
srs_app_source.app => SrsLiveSource::on_frame(SrsSharedPtrMessage* msg) // liveSource 消息包处理入口函数
srs_app_source.app => mix_queue->push(msg->copy()) //将消息push到mix_queque队列
srs_app_source.app => SrsSharedPtrMessage* m = mix_queue->pop() //从队列中取出一个消息
srs_app_source.app => on_audio_imp(m) //如果消息是音频，进行音频消息处理
srs_app_source.app => on_video_imp(m) //如果消息是视频，进行视频消息处理
srs_app_source.app => consumer->enqueue(msg, atc, jitter_algorithm) //消费者消费一条对应的音视频包