srs4.0 webrtc分析（五）网络模块

网络模块介绍

webrtc 中音视频数据是通过udp传输，所以srs4.0要支持webrtc得开启一个udp server进行收发流。
具体代码见文件：srs_app_rtc_server.cpp
监听逻辑

srs_error_t SrsRtcServer::listen_udp()
{//首先获得监听的IP、PORTint port = _srs_config->get_rtc_server_listen();string ip = srs_any_address_for_listener();srs_assert(listeners.empty());//基于配置文件中的reuseport值，决定在同一个端口进行监听的次数。//默认创建一个socket监听一次。如果reuseport大于一就会创建多个socket监听多次// 每个SrsUdpMuxListener内部会启动一个协程进行网络收发int nn_listeners = _srs_config->get_rtc_server_reuseport();for (int i = 0; i < nn_listeners; i++) { //把该类的this指针传进去，接收到网络数据后回传到SrsRtcServerSrsUdpMuxListener* listener = new SrsUdpMuxListener(this, ip, port);if ((err = listener->listen()) != srs_success) {srs_freep(listener);return srs_error_wrap(err, "listen %s:%d", ip.c_str(), port);}listeners.push_back(listener);}return err;
}

SrsUdpMuxListener 网络处理类

启动监听

srs_error_t SrsUdpMuxListener::listen()
{srs_error_t err = srs_success;if ((err = srs_udp_listen(ip, port, &lfd)) != srs_success) {return srs_error_wrap(err, "listen %s:%d", ip.c_str(), port);}//启动一个协程接收网络数据srs_freep(trd);trd = new SrsSTCoroutine("udp", this, cid);if ((err = trd->start()) != srs_success) {return srs_error_wrap(err, "start thread");}return err;
}

修改udp 接收、发送buffer 大小，发送接收buffer 都设置为10M

void SrsUdpMuxListener::set_socket_buffer()
{int default_sndbuf = 0;// TODO: FIXME: Config it.int expect_sndbuf = 1024*1024*10; // 10Mint actual_sndbuf = expect_sndbuf;int r0_sndbuf = 0;if (true) {if ((r0_sndbuf = setsockopt(fd(), SOL_SOCKET, SO_SNDBUF, (void*)&actual_sndbuf, sizeof(actual_sndbuf))) < 0) {srs_warn("set SO_SNDBUF failed, expect=%d, r0=%d", expect_sndbuf, r0_sndbuf);}}int default_rcvbuf = 0;// TODO: FIXME: Config it.int expect_rcvbuf = 1024*1024*10; // 10Mint actual_rcvbuf = expect_rcvbuf;int r0_rcvbuf = 0;if (true) {if ((r0_rcvbuf = setsockopt(fd(), SOL_SOCKET, SO_RCVBUF, (void*)&actual_rcvbuf, sizeof(actual_rcvbuf))) < 0) {srs_warn("set SO_RCVBUF failed, expect=%d, r0=%d", expect_rcvbuf, r0_rcvbuf);}}
}

接收网络数据

srs_error_t SrsUdpMuxListener::cycle()
{while (true) {if ((err = trd->pull()) != srs_success) {return srs_error_wrap(err, "udp listener");}nn_loop++;int nread = skt.recvfrom(SRS_UTIME_NO_TIMEOUT);if (nread <= 0) {if (nread < 0) {srs_warn("udp recv error nn=%d", nread);}// remux udp never returncontinue;}nn_msgs++;nn_msgs_stage++;// Handle the UDP packet.//这里handler就是初始化SrsUdpMuxListener是传进来的SrsRtcServer//所以收到的网络数据最终传到SrsRtcServer的on_udp_packet()函数err = handler->on_udp_packet(&skt);}
}

处理UDP 数据包

有四种类型的UDP 包：stun 、dtls、srtp、srtcp，基于包头的类型分别处理这四种包
如：判断是否是rtcp

// For RTCP, PT is [128, 223] (or without marker [0, 95]).
// Literally, RTCP starts from 64 not 0, so PT is [192, 223] (or without marker [64, 95]).
// @note For RTP, the PT is [96, 127], or [224, 255] with marker.
bool srs_is_rtcp(const uint8_t* data, size_t len)
{return (len >= 12) && (data[0] & 0x80) && (data[1] >= 192 && data[1] <= 223);
}

处理收到的udp包，根据包类型分别处理上面提到的四种包

srs_error_t SrsRtcServer::on_udp_packet(SrsUdpMuxSocket* skt)
{SrsRtcConnection* session = NULL;uint64_t fast_id = skt->fast_id();// Try fast id first, if not found, search by long peer id.//基于用户的IP 、PORT信息找到该用户if (fast_id) {session = (SrsRtcConnection*)_srs_rtc_manager->find_by_fast_id(fast_id);}if (!session) {string peer_id = skt->peer_id();session = (SrsRtcConnection*)_srs_rtc_manager->find_by_id(peer_id);}// For STUN, the peer address may change.处理stun包if (!is_rtp_or_rtcp && srs_is_stun((uint8_t*)data, size)) {SrsStunPacket ping;if ((err = ping.decode(data, size)) != srs_success) {return srs_error_wrap(err, "decode stun packet failed");}return session->on_stun(skt, &ping);}// Note that we don't(except error) switch to the context of session, for performance issue.//处理rtp包if (is_rtp_or_rtcp && !is_rtcp) {++_srs_pps_rrtps->sugar;err = session->on_rtp(data, size);if (err != srs_success) {session->switch_to_context();}return err;}session->switch_to_context();//处理rtcp包if (is_rtp_or_rtcp && is_rtcp) {++_srs_pps_rrtcps->sugar;return session->on_rtcp(data, size);}//处理dtls包if (srs_is_dtls((uint8_t*)data, size)) {++_srs_pps_rstuns->sugar;return session->on_dtls(data, size);}
}

处理rtp

首先确定该包是谁发的，即找到该包的发送者

srs_error_t SrsRtcConnection::on_rtp(char* data, int nb_data)
{srs_error_t err = srs_success;SrsRtcPublishStream* publisher = NULL;if ((err = find_publisher(data, nb_data, &publisher)) != srs_success) {return srs_error_wrap(err, "find");}srs_assert(publisher);return publisher->on_rtp(data, nb_data);
}

基于包中的ssrc找到发布者

srs_error_t SrsRtcConnection::find_publisher(char* buf, int size, SrsRtcPublishStream** ppublisher)
{srs_error_t err = srs_success;if (publishers_.size() == 0) {return srs_error_new(ERROR_RTC_RTCP, "no publisher");}//从包中解析出ssrcuint32_t ssrc = srs_rtp_fast_parse_ssrc(buf, size);if (ssrc == 0) {return srs_error_new(ERROR_RTC_NO_PUBLISHER, "invalid ssrc");}//基于ssrc找到该包的发布者map<uint32_t, SrsRtcPublishStream*>::iterator it = publishers_ssrc_map_.find(ssrc);if(it == publishers_ssrc_map_.end()) {return srs_error_new(ERROR_RTC_NO_PUBLISHER, "no publisher for ssrc:%u", ssrc);}*ppublisher = it->second;return err;
}

发布者处理rtp 包，主要进行解密，然后基于ssrc分别把数据传给videotrack 、audioTrack进行处理

srs_error_t SrsRtcPublishStream::on_rtp(char* data, int nb_data)
{//解密 srtp ，得到 rtpif ((err = session_->transport_->unprotect_rtp(plaintext, &nb_plaintext)) != srs_success) {}//处理rtpif ((err = on_rtp_plaintext(plaintext, nb_plaintext)) != srs_success) { err = do_on_rtp_plaintext(pkt, &buf);}
}

do_on_rtp_plaintext()函数把数据传给SrsRtcVideoRecvTrack、SrsRtcAudioRecvTrack。
Track中的source是 SrsRtcStream* source;

 srs_error_t SrsRtcPublishStream::do_on_rtp_plaintext(SrsRtpPacket2*& pkt, SrsBuffer* buf){// For source to consume packet.uint32_t ssrc = pkt->header.get_ssrc();SrsRtcAudioRecvTrack* audio_track = get_audio_track(ssrc);SrsRtcVideoRecvTrack* video_track = get_video_track(ssrc);if (audio_track) {pkt->frame_type = SrsFrameTypeAudio;if ((err = audio_track->on_rtp(source, pkt)) != srs_success) {return srs_error_wrap(err, "on audio");}} else if (video_track) {pkt->frame_type = SrsFrameTypeVideo;if ((err = video_track->on_rtp(source, pkt)) != srs_success) {return srs_error_wrap(err, "on video");}}}

把流转发给该发布者的所有消费者（订阅者），流没有立即传发给消费者的网络模块，而是进入每个消费者的队列，消费者会启动一个新的协程从队列中读取数据，再发到网络层。

srs_error_t SrsRtcStream::on_rtp(SrsRtpPacket2* pkt)
{srs_error_t err = srs_success;for (int i = 0; i < (int)consumers.size(); i++) {SrsRtcConsumer* consumer = consumers.at(i);if ((err = consumer->enqueue(pkt->copy())) != srs_success) {return srs_error_wrap(err, "consume message");}}return err;
}

下篇文章再分析消费者逻辑

如果有帮助欢迎关注我哈