ZLMediaKit/webrtc/WebRtcTransport.h
2021-06-24 11:39:55 +08:00

398 lines
13 KiB
C++
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit).
*
* Use of this source code is governed by MIT license that can be found in the
* LICENSE file in the root of the source tree. All contributing project authors
* may be found in the AUTHORS file in the root of the source tree.
*/
#pragma once
#include <memory>
#include <string>
#include "DtlsTransport.hpp"
#include "IceServer.hpp"
#include "SrtpSession.hpp"
#include "StunPacket.hpp"
#include "Sdp.h"
#include "Poller/EventPoller.h"
#include "Network/Socket.h"
#include "Rtsp/RtspMediaSourceImp.h"
#include "Rtcp/RtcpContext.h"
#include "Rtcp/RtcpFCI.h"
using namespace toolkit;
using namespace mediakit;
class WebRtcTransport : public RTC::DtlsTransport::Listener, public RTC::IceServer::Listener {
public:
using Ptr = std::shared_ptr<WebRtcTransport>;
WebRtcTransport(const EventPoller::Ptr &poller);
~WebRtcTransport() override = default;
/**
* 创建对象
*/
virtual void onCreate();
/**
* 销毁对象
*/
virtual void onDestory();
/**
* 创建webrtc answer sdp
* @param offer offer sdp
* @return answer sdp
*/
std::string getAnswerSdp(const string &offer);
/**
* socket收到udp数据
* @param buf 数据指针
* @param len 数据长度
* @param tuple 数据来源
*/
void inputSockData(char *buf, size_t len, RTC::TransportTuple *tuple);
/**
* 发送rtp
* @param buf rtcp内容
* @param len rtcp长度
* @param flush 是否flush socket
* @param ctx 用户指针
*/
void sendRtpPacket(const char *buf, size_t len, bool flush, void *ctx = nullptr);
void sendRtcpPacket(const char *buf, size_t len, bool flush, void *ctx = nullptr);
const EventPoller::Ptr& getPoller() const;
protected:
//// dtls相关的回调 ////
void OnDtlsTransportConnecting(const RTC::DtlsTransport *dtlsTransport) override;
void OnDtlsTransportConnected(const RTC::DtlsTransport *dtlsTransport,
RTC::SrtpSession::CryptoSuite srtpCryptoSuite,
uint8_t *srtpLocalKey,
size_t srtpLocalKeyLen,
uint8_t *srtpRemoteKey,
size_t srtpRemoteKeyLen,
std::string &remoteCert) override;
void OnDtlsTransportFailed(const RTC::DtlsTransport *dtlsTransport) override;
void OnDtlsTransportClosed(const RTC::DtlsTransport *dtlsTransport) override;
void OnDtlsTransportSendData(const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) override;
void OnDtlsTransportApplicationDataReceived(const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) override;
protected:
//// ice相关的回调 ///
void OnIceServerSendStunPacket(const RTC::IceServer *iceServer, const RTC::StunPacket *packet, RTC::TransportTuple *tuple) override;
void OnIceServerSelectedTuple(const RTC::IceServer *iceServer, RTC::TransportTuple *tuple) override;
void OnIceServerConnected(const RTC::IceServer *iceServer) override;
void OnIceServerCompleted(const RTC::IceServer *iceServer) override;
void OnIceServerDisconnected(const RTC::IceServer *iceServer) override;
protected:
virtual void onStartWebRTC() = 0;
virtual void onRtcConfigure(RtcConfigure &configure) const;
virtual void onCheckSdp(SdpType type, RtcSession &sdp);
virtual void onSendSockData(const char *buf, size_t len, struct sockaddr_in *dst, bool flush = true) = 0;
virtual void onRtp(const char *buf, size_t len) = 0;
virtual void onRtcp(const char *buf, size_t len) = 0;
virtual void onShutdown(const SockException &ex) = 0;
virtual void onBeforeEncryptRtp(const char *buf, size_t &len, void *ctx) = 0;
virtual void onBeforeEncryptRtcp(const char *buf, size_t &len, void *ctx) = 0;
protected:
const RtcSession& getSdp(SdpType type) const;
RTC::TransportTuple* getSelectedTuple() const;
void sendRtcpRemb(uint32_t ssrc, size_t bit_rate);
void sendRtcpPli(uint32_t ssrc);
private:
void onSendSockData(const char *buf, size_t len, bool flush = true);
void setRemoteDtlsFingerprint(const RtcSession &remote);
private:
uint8_t _srtp_buf[2000];
EventPoller::Ptr _poller;
std::shared_ptr<RTC::IceServer> _ice_server;
std::shared_ptr<RTC::DtlsTransport> _dtls_transport;
std::shared_ptr<RTC::SrtpSession> _srtp_session_send;
std::shared_ptr<RTC::SrtpSession> _srtp_session_recv;
RtcSession::Ptr _offer_sdp;
RtcSession::Ptr _answer_sdp;
};
class RtpReceiverImp;
class NackList {
public:
void push_back(RtpPacket::Ptr rtp) {
auto seq = rtp->getSeq();
_nack_cache_seq.emplace_back(seq);
_nack_cache_pkt.emplace(seq, std::move(rtp));
while (get_cache_ms() > kMaxNackMS) {
//需要清除部分nack缓存
pop_front();
}
}
template<typename FUNC>
void for_each_nack(const FCI_NACK &nack, const FUNC &func) {
auto seq = nack.getPid();
for (auto bit : nack.getBitArray()) {
if (bit) {
//丢包
RtpPacket::Ptr *ptr = get_rtp(seq);
if (ptr) {
func(*ptr);
}
}
++seq;
}
}
private:
void pop_front() {
if (_nack_cache_seq.empty()) {
return;
}
_nack_cache_pkt.erase(_nack_cache_seq.front());
_nack_cache_seq.pop_front();
}
RtpPacket::Ptr *get_rtp(uint16_t seq) {
auto it = _nack_cache_pkt.find(seq);
if (it == _nack_cache_pkt.end()) {
return nullptr;
}
return &it->second;
}
uint32_t get_cache_ms() {
if (_nack_cache_seq.size() < 2) {
return 0;
}
uint32_t back = _nack_cache_pkt[_nack_cache_seq.back()]->getStampMS();
uint32_t front = _nack_cache_pkt[_nack_cache_seq.front()]->getStampMS();
if (back > front) {
return back - front;
}
//很有可能回环了
return back + (UINT32_MAX - front);
}
private:
static constexpr uint32_t kMaxNackMS = 10 * 1000;
deque<uint16_t> _nack_cache_seq;
unordered_map<uint16_t, RtpPacket::Ptr > _nack_cache_pkt;
};
class NackContext {
public:
using onNack = function<void(const FCI_NACK &nack)>;
void received(uint16_t seq) {
if (!_last_max_seq && _seq.empty()) {
_last_max_seq = seq - 1;
}
_seq.emplace(seq);
auto max_seq = *_seq.rbegin();
auto min_seq = *_seq.begin();
auto diff = max_seq - min_seq;
if (!diff) {
return;
}
if (diff > UINT32_MAX / 2) {
//回环
_seq.clear();
_last_max_seq = min_seq;
return;
}
if (_seq.size() == diff + 1 && _last_max_seq + 1 == min_seq) {
//都是连续的seq未丢包
_seq.clear();
_last_max_seq = max_seq;
} else {
//seq不连续有丢包
if (min_seq == _last_max_seq + 1) {
//前面部分seq是连续的未丢包移除之
eraseFrontSeq();
}
//有丢包丢包从_last_max_seq开始
if (max_seq - _last_max_seq > FCI_NACK::kBitSize) {
vector<bool> vec;
vec.resize(FCI_NACK::kBitSize);
for (auto i = 0; i < FCI_NACK::kBitSize; ++i) {
vec[i] = _seq.find(_last_max_seq + i + 2) == _seq.end();
}
doNack(FCI_NACK(_last_max_seq + 1, vec));
_last_max_seq += FCI_NACK::kBitSize + 1;
if (_last_max_seq >= max_seq) {
_seq.clear();
} else {
auto it = _seq.emplace_hint(_seq.begin(), _last_max_seq);
_seq.erase(_seq.begin(), it);
}
}
}
}
void setOnNack(onNack cb) {
_cb = std::move(cb);
}
private:
void doNack(const FCI_NACK &nack) {
if (_cb) {
_cb(nack);
}
}
void eraseFrontSeq(){
//前面部分seq是连续的未丢包移除之
for (auto it = _seq.begin(); it != _seq.end();) {
if (*it != _last_max_seq + 1) {
//seq不连续丢包了
break;
}
_last_max_seq = *it;
it = _seq.erase(it);
}
}
private:
onNack _cb;
set<uint16_t> _seq;
uint16_t _last_max_seq = 0;
};
class WebRtcTransportImp : public WebRtcTransport, public MediaSourceEvent, public SockInfo, public std::enable_shared_from_this<WebRtcTransportImp>{
public:
using Ptr = std::shared_ptr<WebRtcTransportImp>;
~WebRtcTransportImp() override;
/**
* 创建WebRTC对象
* @param poller 改对象需要绑定的线程
* @return 对象
*/
static Ptr create(const EventPoller::Ptr &poller);
/**
* 绑定rtsp媒体源
* @param src 媒体源
* @param is_play 是播放还是推流
*/
void attach(const RtspMediaSource::Ptr &src, const MediaInfo &info, bool is_play = true);
protected:
void onStartWebRTC() override;
void onSendSockData(const char *buf, size_t len, struct sockaddr_in *dst, bool flush = true) override;
void onCheckSdp(SdpType type, RtcSession &sdp) override;
void onRtcConfigure(RtcConfigure &configure) const override;
void onRtp(const char *buf, size_t len) override;
void onRtp_l(const char *buf, size_t len, bool rtx);
void onRtcp(const char *buf, size_t len) override;
void onBeforeEncryptRtp(const char *buf, size_t &len, void *ctx) override;
void onBeforeEncryptRtcp(const char *buf, size_t &len, void *ctx) override {};
void onShutdown(const SockException &ex) override;
///////MediaSourceEvent override///////
// 关闭
bool close(MediaSource &sender, bool force) override;
// 播放总人数
int totalReaderCount(MediaSource &sender) override;
// 获取媒体源类型
MediaOriginType getOriginType(MediaSource &sender) const override;
// 获取媒体源url或者文件路径
string getOriginUrl(MediaSource &sender) const override;
// 获取媒体源客户端相关信息
std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
///////SockInfo override///////
//获取本机ip
string get_local_ip() override;
//获取本机端口号
uint16_t get_local_port() override;
//获取对方ip
string get_peer_ip() override;
//获取对方端口号
uint16_t get_peer_port() override;
//获取标识符
string getIdentifier() const override;
private:
WebRtcTransportImp(const EventPoller::Ptr &poller);
void onCreate() override;
void onDestory() override;
void onSendRtp(const RtpPacket::Ptr &rtp, bool flush, bool rtx = false);
SdpAttrCandidate::Ptr getIceCandidate() const;
bool canSendRtp() const;
bool canRecvRtp() const;
class RtpPayloadInfo {
public:
using Ptr = std::shared_ptr<RtpPayloadInfo>;
const RtcCodecPlan *plan_rtp;
const RtcCodecPlan *plan_rtx;
uint32_t offer_ssrc_rtp = 0;
uint32_t offer_ssrc_rtx = 0;
uint32_t answer_ssrc_rtp = 0;
uint32_t answer_ssrc_rtx = 0;
const RtcMedia *media;
NackList nack_list;
RtcpContext::Ptr rtcp_context_send;
unordered_map<string/*rid*/, std::pair<uint32_t/*rtp ssrc*/, uint32_t/*rtx ssrc*/> > rid_ssrc;
unordered_map<uint32_t/*rtx ssrc*/, uint32_t/*rtp ssrc*/> rtx_ssrc_to_rtp_ssrc;
unordered_map<uint32_t/*simulcast ssrc*/, NackContext> nack_ctx;
unordered_map<uint32_t/*simulcast ssrc*/, RtcpContext::Ptr> rtcp_context_recv;
unordered_map<uint32_t/*simulcast ssrc*/, std::shared_ptr<RtpReceiverImp> > receiver;
};
void onSortedRtp(RtpPayloadInfo &info, const string &rid, RtpPacket::Ptr rtp);
void onSendNack(RtpPayloadInfo &info, const FCI_NACK &nack, uint32_t ssrc);
void changeRtpExtId(RtpPayloadInfo &info, const RtpHeader *header, bool is_recv, bool is_rtx = false, string *rid_ptr = nullptr) const;
private:
uint16_t _rtx_seq[2] = {0, 0};
//用掉的总流量
uint64_t _bytes_usage = 0;
//媒体相关元数据
MediaInfo _media_info;
//保持自我强引用
Ptr _self;
//检测超时的定时器
Timer::Ptr _timer;
//刷新计时器
Ticker _alive_ticker;
//pli rtcp计时器
Ticker _pli_ticker;
//复合udp端口接收一切rtp与rtcp
Socket::Ptr _socket;
//推流的rtsp源
RtspMediaSource::Ptr _push_src;
unordered_map<string/*rid*/, RtspMediaSource::Ptr> _push_src_simulcast;
//播放的rtsp源
RtspMediaSource::Ptr _play_src;
//播放rtsp源的reader对象
RtspMediaSource::RingType::RingReader::Ptr _reader;
//根据发送rtp的track类型获取相关信息
RtpPayloadInfo::Ptr _send_rtp_info[2];
//根据接收rtp的pt获取相关信息
unordered_map<uint8_t/*pt*/, std::pair<bool/*is rtx*/,RtpPayloadInfo::Ptr> > _rtp_info_pt;
//根据rtcp的ssrc获取相关信息
unordered_map<uint32_t/*ssrc*/, std::pair<bool/*is rtx*/,RtpPayloadInfo::Ptr> > _rtp_info_ssrc;
//发送rtp时需要修改rtp ext id
map<RtpExtType, uint8_t> _rtp_ext_type_to_id;
//接收rtp时需要修改rtp ext id
unordered_map<uint8_t, RtpExtType> _rtp_ext_id_to_type;
};