Skip to content

Z
ZLMediaKit
Commit 6d8d64ca by xiongguangjie
Options

for mergen

parents e7e7906e 8c670c65
正在显示 包含 644 行增加 和 488 行删除
src/Extension/H265.cpp
......@@ -68,7 +68,6 @@ bool H265Frame::isKeyFrame(int type, const char *ptr) {
return (((*((uint8_t *) ptr + 2)) >> 7) & 0x01) == 1 && (type == NAL_IDR_N_LP || type == NAL_IDR_W_RADL);
}
return false;
}
H265Frame::H265Frame(){
......
src/Rtcp/RtcpContext.cpp
......@@ -18,8 +18,7 @@ void RtcpContext::clear() {
memset(this, 0, sizeof(RtcpContext));
}
RtcpContext::RtcpContext(uint32_t sample_rate, bool is_receiver) {
_sample_rate = sample_rate;
RtcpContext::RtcpContext(bool is_receiver) {
_is_receiver = is_receiver;
}
......@@ -35,7 +34,6 @@ void RtcpContext::onRtp(uint16_t seq, uint32_t stamp, size_t bytes) {
diff = -diff;
}
//抖动单位为采样次数
diff *= (_sample_rate / 1000.0);
_jitter += (diff - _jitter) / 16.0;
} else {
_jitter = 0;
......@@ -129,7 +127,7 @@ Buffer::Ptr RtcpContext::createRtcpSR(uint32_t rtcp_ssrc) {
rtcp->setNtpStamp(tv);
//转换成rtp时间戳
rtcp->rtpts = htonl(uint32_t(_last_rtp_stamp * (_sample_rate / 1000.0)));
rtcp->rtpts = htonl(_last_rtp_stamp);
rtcp->packet_count = htonl((uint32_t) _packets);
rtcp->octet_count = htonl((uint32_t) _bytes);
return RtcpHeader::toBuffer(std::move(rtcp));
......
src/Rtcp/RtcpContext.h
......@@ -22,15 +22,14 @@ public:
using Ptr = std::shared_ptr<RtcpContext>;
/**
* 创建rtcp上下文
* @param sample_rate 音频采用率，视频一般为90000
* @param is_receiver 是否为rtp接收者，接收者更消耗性能
*/
RtcpContext(uint32_t sample_rate, bool is_receiver);
RtcpContext(bool is_receiver);
/**
* 输出或输入rtp时调用
* @param seq rtp的seq
* @param stamp rtp的时间戳，单位毫秒
* @param stamp rtp的时间戳，单位采样数(非毫秒)
* @param bytes rtp数据长度
*/
void onRtp(uint16_t seq, uint32_t stamp, size_t bytes);
......@@ -87,8 +86,6 @@ private:
bool _is_receiver;
//时间戳抖动值
double _jitter = 0;
//视频默认90000,音频为采样率
uint32_t _sample_rate;
//收到或发送的rtp的字节数
size_t _bytes = 0;
//收到或发送的rtp的个数
......
src/Rtp/GB28181Process.cpp
......@@ -24,36 +24,23 @@ static inline bool checkTS(const uint8_t *packet, size_t bytes){
return bytes % TS_PACKET_SIZE == 0 && packet[0] == TS_SYNC_BYTE;
}
class RtpReceiverImp : public RtpReceiver {
class RtpReceiverImp : public RtpTrackImp {
public:
using Ptr = std::shared_ptr<RtpReceiverImp>;
RtpReceiverImp(int sample_rate, function<void(RtpPacket::Ptr rtp)> cb, function<void(const RtpPacket::Ptr &rtp)> cb_before = nullptr){
RtpReceiverImp(int sample_rate, RtpTrackImp::OnSorted cb, RtpTrackImp::BeforeSorted cb_before = nullptr){
_sample_rate = sample_rate;
_on_sort = std::move(cb);
_on_before_sort = std::move(cb_before);
setOnSorted(std::move(cb));
setBeforeSorted(std::move(cb_before));
}
~RtpReceiverImp() override = default;
bool inputRtp(TrackType type, uint8_t *ptr, size_t len){
return handleOneRtp((int) type, type, _sample_rate, ptr, len);
}
protected:
void onRtpSorted(RtpPacket::Ptr rtp, int track_index) override {
_on_sort(std::move(rtp));
}
void onBeforeRtpSorted(const RtpPacket::Ptr &rtp, int track_index) override {
if (_on_before_sort) {
_on_before_sort(rtp);
}
return RtpTrack::inputRtp(type, _sample_rate, ptr, len);
}
private:
int _sample_rate;
function<void(RtpPacket::Ptr rtp)> _on_sort;
function<void(const RtpPacket::Ptr &rtp)> _on_before_sort;
};
///////////////////////////////////////////////////////////////////////////////////////////
......
src/Rtp/RtpServer.cpp
......@@ -27,7 +27,7 @@ class RtcpHelper : public RtcpContext, public std::enable_shared_from_this<RtcpH
public:
using Ptr = std::shared_ptr<RtcpHelper>;
RtcpHelper(Socket::Ptr rtcp_sock, uint32_t sample_rate) : RtcpContext(sample_rate, true){
RtcpHelper(Socket::Ptr rtcp_sock, uint32_t sample_rate) : RtcpContext(true){
_rtcp_sock = std::move(rtcp_sock);
_sample_rate = sample_rate;
}
......@@ -35,7 +35,7 @@ public:
void onRecvRtp(const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len){
//统计rtp接受情况，用于发送rr包
auto header = (RtpHeader *) buf->data();
onRtp(ntohs(header->seq), ntohl(header->stamp) * uint64_t(1000) / _sample_rate, buf->size());
onRtp(ntohs(header->seq), ntohl(header->stamp), buf->size());
sendRtcp(ntohl(header->ssrc), addr, addr_len);
}
......
src/Rtp/RtpSession.cpp
......@@ -103,7 +103,7 @@ void RtpSession::onRtpPacket(const char *data, size_t len) {
}
try {
_process->inputRtp(false, getSock(), data, len, &_addr);
} catch (RtpReceiver::BadRtpException &ex) {
} catch (RtpTrack::BadRtpException &ex) {
if (!_is_udp) {
WarnL << ex.what() << "，开始搜索ssrc以便恢复上下文";
_search_rtp = true;
......
src/Rtsp/RtpReceiver.cpp
......@@ -15,19 +15,23 @@
namespace mediakit {
RtpReceiver::RtpReceiver() {
int index = 0;
for (auto &sortor : _rtp_sortor) {
sortor.setOnSort([this, index](uint16_t seq, RtpPacket::Ptr &packet) {
onRtpSorted(std::move(packet), index);
});
++index;
}
RtpTrack::RtpTrack() {
setOnSort([this](uint16_t seq, RtpPacket::Ptr &packet) {
onRtpSorted(std::move(packet));
});
}
uint32_t RtpTrack::getSSRC() const {
return _ssrc;
}
RtpReceiver::~RtpReceiver() {}
void RtpTrack::clear() {
_ssrc = 0;
_ssrc_alive.resetTime();
PacketSortor<RtpPacket::Ptr>::clear();
}
bool RtpReceiver::handleOneRtp(int index, TrackType type, int sample_rate, uint8_t *ptr, size_t len) {
bool RtpTrack::inputRtp(TrackType type, int sample_rate, uint8_t *ptr, size_t len) {
if (len < RtpPacket::kRtpHeaderSize) {
WarnL << "rtp包太小:" << len;
return false;
......@@ -52,23 +56,23 @@ bool RtpReceiver::handleOneRtp(int index, TrackType type, int sample_rate, uint8
//比对缓存ssrc
auto ssrc = ntohl(header->ssrc);
if (!_ssrc[index]) {
if (!_ssrc) {
//记录并锁定ssrc
_ssrc[index] = ssrc;
_ssrc_alive[index].resetTime();
} else if (_ssrc[index] == ssrc) {
_ssrc = ssrc;
_ssrc_alive.resetTime();
} else if (_ssrc == ssrc) {
//ssrc匹配正确,刷新计时器
_ssrc_alive[index].resetTime();
_ssrc_alive.resetTime();
} else {
//ssrc错误
if (_ssrc_alive[index].elapsedTime() < 3 * 1000) {
if (_ssrc_alive.elapsedTime() < 3 * 1000) {
//接受正确ssrc的rtp在10秒内，那么我们认为存在多路rtp,忽略掉ssrc不匹配的rtp
WarnL << "ssrc不匹配,rtp已丢弃:" << ssrc << " != " << _ssrc[index];
WarnL << "ssrc不匹配,rtp已丢弃:" << ssrc << " != " << _ssrc;
return false;
}
InfoL << "rtp流ssrc切换:" << _ssrc[index] << " -> " << ssrc;
_ssrc[index] = ssrc;
_ssrc_alive[index].resetTime();
InfoL << "rtp流ssrc切换:" << _ssrc << " -> " << ssrc;
_ssrc = ssrc;
_ssrc_alive.resetTime();
}
auto rtp = RtpPacket::create();
......@@ -87,29 +91,32 @@ bool RtpReceiver::handleOneRtp(int index, TrackType type, int sample_rate, uint8
//拷贝rtp
memcpy(&data[4], ptr, len);
onBeforeRtpSorted(rtp, index);
onBeforeRtpSorted(rtp);
auto seq = rtp->getSeq();
_rtp_sortor[index].sortPacket(seq, std::move(rtp));
sortPacket(seq, std::move(rtp));
return true;
}
void RtpReceiver::clear() {
CLEAR_ARR(_ssrc);
for (auto &sortor : _rtp_sortor) {
sortor.clear();
}
////////////////////////////////////////////////////////////////////////////////////
void RtpTrackImp::setOnSorted(OnSorted cb) {
_on_sorted = std::move(cb);
}
size_t RtpReceiver::getJitterSize(int index) const{
return _rtp_sortor[index].getJitterSize();
void RtpTrackImp::setBeforeSorted(BeforeSorted cb) {
_on_before_sorted = std::move(cb);
}
size_t RtpReceiver::getCycleCount(int index) const{
return _rtp_sortor[index].getCycleCount();
void RtpTrackImp::onRtpSorted(RtpPacket::Ptr rtp) {
if (_on_sorted) {
_on_sorted(std::move(rtp));
}
}
uint32_t RtpReceiver::getSSRC(int index) const{
return _ssrc[index];
void RtpTrackImp::onBeforeRtpSorted(const RtpPacket::Ptr &rtp) {
if (_on_before_sorted) {
_on_before_sorted(rtp);
}
}
}//namespace mediakit
src/Rtsp/RtpReceiver.h
......@@ -160,11 +160,8 @@ private:
function<void(SEQ seq, T &packet)> _cb;
};
class RtpReceiver {
class RtpTrack : private PacketSortor<RtpPacket::Ptr>{
public:
RtpReceiver();
virtual ~RtpReceiver();
class BadRtpException : public invalid_argument {
public:
template<typename Type>
......@@ -172,7 +169,60 @@ public:
~BadRtpException() = default;
};
RtpTrack();
virtual ~RtpTrack() = default;
void clear();
uint32_t getSSRC() const;
bool inputRtp(TrackType type, int sample_rate, uint8_t *ptr, size_t len);
protected:
virtual void onRtpSorted(RtpPacket::Ptr rtp) {}
virtual void onBeforeRtpSorted(const RtpPacket::Ptr &rtp) {}
private:
uint32_t _ssrc = 0;
Ticker _ssrc_alive;
};
class RtpTrackImp : public RtpTrack{
public:
using OnSorted = function<void(RtpPacket::Ptr)>;
using BeforeSorted = function<void(const RtpPacket::Ptr &)>;
RtpTrackImp() = default;
~RtpTrackImp() override = default;
void setOnSorted(OnSorted cb);
void setBeforeSorted(BeforeSorted cb);
protected:
void onRtpSorted(RtpPacket::Ptr rtp) override;
void onBeforeRtpSorted(const RtpPacket::Ptr &rtp) override;
private:
OnSorted _on_sorted;
BeforeSorted _on_before_sorted;
};
template<int kCount = 2>
class RtpMultiReceiver {
public:
RtpMultiReceiver() {
int index = 0;
for (auto &track : _track) {
track.setOnSorted([this, index](RtpPacket::Ptr rtp) {
onRtpSorted(std::move(rtp), index);
});
track.setBeforeSorted([this, index](const RtpPacket::Ptr &rtp) {
onBeforeRtpSorted(rtp, index);
});
++index;
}
}
virtual ~RtpMultiReceiver() = default;
/**
* 输入数据指针生成并排序rtp包
* @param index track下标索引
......@@ -182,34 +232,49 @@ protected:
* @param len rtp数据指针长度
* @return 解析成功返回true
*/
bool handleOneRtp(int index, TrackType type, int samplerate, uint8_t *ptr, size_t len);
bool handleOneRtp(int index, TrackType type, int sample_rate, uint8_t *ptr, size_t len){
return _track[index].inputRtp(type, sample_rate, ptr, len);
}
void clear() {
for (auto &track : _track) {
track.clear();
}
}
size_t getJitterSize(int index) const {
return _track[index].getJitterSize();
}
size_t getCycleCount(int index) const {
return _track[index].getCycleCount();
}
uint32_t getSSRC(int index) const {
return _track[index].getSSRC();
}
protected:
/**
* rtp数据包排序后输出
* @param rtp rtp数据包
* @param track_index track索引
*/
virtual void onRtpSorted(RtpPacket::Ptr rtp, int track_index) {}
virtual void onRtpSorted(RtpPacket::Ptr rtp, int index) {}
/**
* 解析出rtp但还未排序
* @param rtp rtp数据包
* @param track_index track索引
*/
virtual void onBeforeRtpSorted(const RtpPacket::Ptr &rtp, int track_index) {}
void clear();
size_t getJitterSize(int track_index) const;
size_t getCycleCount(int track_index) const;
uint32_t getSSRC(int track_index) const;
virtual void onBeforeRtpSorted(const RtpPacket::Ptr &rtp, int index) {}
private:
uint32_t _ssrc[2] = {0, 0};
Ticker _ssrc_alive[2];
//rtp排序缓存，根据seq排序
PacketSortor<RtpPacket::Ptr> _rtp_sortor[2];
RtpTrackImp _track[kCount];
};
using RtpReceiver = RtpMultiReceiver<2>;
}//namespace mediakit
......
src/Rtsp/RtspPlayer.cpp
......@@ -205,7 +205,7 @@ void RtspPlayer::handleResDESCRIBE(const Parser& parser) {
}
_rtcp_context.clear();
for (auto &track : _sdp_track) {
_rtcp_context.emplace_back(std::make_shared<RtcpContext>(track->_samplerate, true));
_rtcp_context.emplace_back(std::make_shared<RtcpContext>(true));
}
sendSetup(0);
}
......@@ -591,7 +591,7 @@ void RtspPlayer::sendRtspRequest(const string &cmd, const string &url,const StrC
void RtspPlayer::onBeforeRtpSorted(const RtpPacket::Ptr &rtp, int track_idx){
auto &rtcp_ctx = _rtcp_context[track_idx];
rtcp_ctx->onRtp(rtp->getSeq(), rtp->getStampMS(), rtp->size() - RtpPacket::kRtpTcpHeaderSize);
rtcp_ctx->onRtp(rtp->getSeq(), ntohl(rtp->getHeader()->stamp), rtp->size() - RtpPacket::kRtpTcpHeaderSize);
auto &ticker = _rtcp_send_ticker[track_idx];
if (ticker.elapsedTime() < 3 * 1000) {
......
src/Rtsp/RtspPusher.cpp
......@@ -179,7 +179,7 @@ void RtspPusher::sendAnnounce() {
}
_rtcp_context.clear();
for (auto &track : _track_vec) {
_rtcp_context.emplace_back(std::make_shared<RtcpContext>(track->_samplerate, false));
_rtcp_context.emplace_back(std::make_shared<RtcpContext>(false));
}
_on_res_func = std::bind(&RtspPusher::handleResAnnounce, this, placeholders::_1);
sendRtspRequest("ANNOUNCE", _url, {}, src->getSdp());
......@@ -360,7 +360,7 @@ void RtspPusher::updateRtcpContext(const RtpPacket::Ptr &rtp){
int track_index = getTrackIndexByTrackType(rtp->type);
auto &ticker = _rtcp_send_ticker[track_index];
auto &rtcp_ctx = _rtcp_context[track_index];
rtcp_ctx->onRtp(rtp->getSeq(), rtp->getStampMS(), rtp->size() - RtpPacket::kRtpTcpHeaderSize);
rtcp_ctx->onRtp(rtp->getSeq(), ntohl(rtp->getHeader()->stamp), rtp->size() - RtpPacket::kRtpTcpHeaderSize);
//send rtcp every 5 second
if (ticker.elapsedTime() > 5 * 1000) {
......
src/Rtsp/RtspSession.cpp
......@@ -252,7 +252,7 @@ void RtspSession::handleReq_ANNOUNCE(const Parser &parser) {
}
_rtcp_context.clear();
for (auto &track : _sdp_track) {
_rtcp_context.emplace_back(std::make_shared<RtcpContext>(track->_samplerate, true));
_rtcp_context.emplace_back(std::make_shared<RtcpContext>(true));
}
_push_src = std::make_shared<RtspMediaSourceImp>(_media_info._vhost, _media_info._app, _media_info._streamid);
_push_src->setListener(dynamic_pointer_cast<MediaSourceEvent>(shared_from_this()));
......@@ -413,7 +413,7 @@ void RtspSession::onAuthSuccess() {
}
strongSelf->_rtcp_context.clear();
for (auto &track : strongSelf->_sdp_track) {
strongSelf->_rtcp_context.emplace_back(std::make_shared<RtcpContext>(track->_samplerate, false));
strongSelf->_rtcp_context.emplace_back(std::make_shared<RtcpContext>(false));
}
strongSelf->_sessionid = makeRandStr(12);
strongSelf->_play_src = rtsp_src;
......@@ -1126,7 +1126,7 @@ void RtspSession::onBeforeRtpSorted(const RtpPacket::Ptr &rtp, int track_index){
void RtspSession::updateRtcpContext(const RtpPacket::Ptr &rtp){
int track_index = getTrackIndexByTrackType(rtp->type);
auto &rtcp_ctx = _rtcp_context[track_index];
rtcp_ctx->onRtp(rtp->getSeq(), rtp->getStampMS(), rtp->size() - RtpPacket::kRtpTcpHeaderSize);
rtcp_ctx->onRtp(rtp->getSeq(), ntohl(rtp->getHeader()->stamp), rtp->size() - RtpPacket::kRtpTcpHeaderSize);
auto &ticker = _rtcp_send_tickers[track_index];
//send rtcp every 5 second
......
webrtc/Nack.cpp 0 → 100644
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/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.
*/
#include "Nack.h"
static constexpr uint32_t kMaxNackMS = 10 * 1000;
void NackList::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();
}
}
void NackList::for_each_nack(const FCI_NACK &nack, const function<void(const RtpPacket::Ptr &rtp)> &func) {
auto seq = nack.getPid();
for (auto bit : nack.getBitArray()) {
if (bit) {
//丢包
RtpPacket::Ptr *ptr = get_rtp(seq);
if (ptr) {
func(*ptr);
}
}
++seq;
}
}
void NackList::pop_front() {
if (_nack_cache_seq.empty()) {
return;
}
_nack_cache_pkt.erase(_nack_cache_seq.front());
_nack_cache_seq.pop_front();
}
RtpPacket::Ptr *NackList::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 NackList::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);
}
////////////////////////////////////////////////////////////////////////////////////////////////
void NackContext::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 NackContext::setOnNack(onNack cb) {
_cb = std::move(cb);
}
void NackContext::doNack(const FCI_NACK &nack) {
if (_cb) {
_cb(nack);
}
}
void NackContext::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);
}
}
\ No newline at end of file
webrtc/Nack.h 0 → 100644
/*
* Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
*
* This file is part of ZLMediaKit(https://github.com/ZLMediaKit/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.
*/
#ifndef ZLMEDIAKIT_NACK_H
#define ZLMEDIAKIT_NACK_H
#include "Rtsp/Rtsp.h"
#include "Rtcp/RtcpFCI.h"
using namespace mediakit;
class NackList {
public:
NackList() = default;
~NackList() = default;
void push_back(RtpPacket::Ptr rtp);
void for_each_nack(const FCI_NACK &nack, const function<void(const RtpPacket::Ptr &rtp)> &cb);
private:
void pop_front();
uint32_t get_cache_ms();
RtpPacket::Ptr *get_rtp(uint16_t seq);
private:
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)>;
NackContext() = default;
~NackContext() = default;
void received(uint16_t seq);
void setOnNack(onNack cb);
private:
void eraseFrontSeq();
void doNack(const FCI_NACK &nack);
private:
onNack _cb;
set<uint16_t> _seq;
uint16_t _last_max_seq = 0;
};
#endif //ZLMEDIAKIT_NACK_H
webrtc/RtpExt.cpp
......@@ -561,4 +561,90 @@ void RtpExt::setType(RtpExtType type) {
RtpExtType RtpExt::getType() const {
return _type;
}
\ No newline at end of file
}
RtpExtContext::RtpExtContext(const RtcMedia &m){
for (auto &ext : m.extmap) {
auto ext_type = RtpExt::getExtType(ext.ext);
_rtp_ext_id_to_type.emplace(ext.id, ext_type);
_rtp_ext_type_to_id.emplace(ext_type, ext.id);
}
}
string RtpExtContext::getRid(uint32_t ssrc) const{
auto it = _ssrc_to_rid.find(ssrc);
if (it == _ssrc_to_rid.end()) {
return "";
}
return it->second;
}
void RtpExtContext::setRid(uint32_t ssrc, const string &rid) {
_ssrc_to_rid[ssrc] = rid;
}
void RtpExtContext::changeRtpExtId(const RtpHeader *header, bool is_recv, string *rid_ptr) {
string rid, repaired_rid;
auto ext_map = RtpExt::getExtValue(header);
for (auto &pr : ext_map) {
if (is_recv) {
auto it = _rtp_ext_id_to_type.find(pr.first);
if (it == _rtp_ext_id_to_type.end()) {
WarnL << "接收rtp时,忽略不识别的rtp ext, id=" << (int) pr.first;
pr.second.clearExt();
continue;
}
pr.second.setType(it->second);
//重新赋值ext id为 ext type，作为后面处理ext的统一中间类型
pr.second.setExtId((uint8_t) it->second);
switch (it->second) {
case RtpExtType::sdes_rtp_stream_id : rid = pr.second.getRtpStreamId(); break;
case RtpExtType::sdes_repaired_rtp_stream_id : repaired_rid = pr.second.getRepairedRtpStreamId(); break;
default : break;
}
} else {
pr.second.setType((RtpExtType) pr.first);
auto it = _rtp_ext_type_to_id.find((RtpExtType) pr.first);
if (it == _rtp_ext_type_to_id.end()) {
WarnL << "发送rtp时, 忽略不被客户端支持rtp ext:" << pr.second.dumpString();
pr.second.clearExt();
continue;
}
//重新赋值ext id为客户端sdp声明的类型
pr.second.setExtId(it->second);
}
}
if (!is_recv) {
return;
}
if (rid.empty()) {
rid = repaired_rid;
}
auto ssrc = ntohl(header->ssrc);
if (rid.empty()) {
//获取rid
rid = _ssrc_to_rid[ssrc];
} else {
//设置rid
auto it = _ssrc_to_rid.find(ssrc);
if (it == _ssrc_to_rid.end() || it->second != rid) {
_ssrc_to_rid[ssrc] = rid;
onGetRtp(header->pt, ssrc, rid);
}
}
if (rid_ptr) {
*rid_ptr = rid;
}
}
void RtpExtContext::setOnGetRtp(OnGetRtp cb) {
_cb = std::move(cb);
}
void RtpExtContext::onGetRtp(uint8_t pt, uint32_t ssrc, const string &rid){
if (_cb) {
_cb(pt, ssrc, rid);
}
}
webrtc/RtpExt.h
......@@ -108,5 +108,31 @@ private:
RtpExtType _type = RtpExtType::padding;
};
class RtcMedia;
class RtpExtContext {
public:
using Ptr = std::shared_ptr<RtpExtContext>;
using OnGetRtp = function<void(uint8_t pt, uint32_t ssrc, const string &rid)>;
RtpExtContext(const RtcMedia &media);
~RtpExtContext() = default;
void setOnGetRtp(OnGetRtp cb);
string getRid(uint32_t ssrc) const;
void setRid(uint32_t ssrc, const string &rid);
void changeRtpExtId(const RtpHeader *header, bool is_recv, string *rid_ptr = nullptr);
private:
void onGetRtp(uint8_t pt, uint32_t ssrc, const string &rid);
private:
OnGetRtp _cb;
//发送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;
//ssrc --> rid
unordered_map<uint32_t/*simulcast ssrc*/, string/*rid*/> _ssrc_to_rid;
};
#endif //ZLMEDIAKIT_RTPEXT_H
webrtc/Sdp.cpp
......@@ -1636,7 +1636,8 @@ RETRY:
if (configure.direction != RtpDirection::recvonly &&
configure.direction != RtpDirection::sendrecv) {
//我们不支持接收
continue;
answer_media.direction = RtpDirection::inactive;
break;
}
answer_media.direction = RtpDirection::recvonly;
break;
......@@ -1645,7 +1646,8 @@ RETRY:
if (configure.direction != RtpDirection::sendonly &&
configure.direction != RtpDirection::sendrecv) {
//我们不支持发送
continue;
answer_media.direction = RtpDirection::inactive;
break;
}
answer_media.direction = RtpDirection::sendonly;
break;
......
webrtc/WebRtcTransport.cpp
......@@ -148,7 +148,6 @@ void WebRtcTransport::sendRtcpRemb(uint32_t ssrc, size_t bit_rate) {
fb->ssrc = htonl(0);
fb->ssrc_media = htonl(ssrc);
sendRtcpPacket((char *) fb.get(), fb->getSize(), true);
TraceL << ssrc << " " << bit_rate;
}
void WebRtcTransport::sendRtcpPli(uint32_t ssrc) {
......@@ -255,7 +254,8 @@ void WebRtcTransport::inputSockData(char *buf, size_t len, RTC::TransportTuple *
if (_srtp_session_recv->DecryptSrtp((uint8_t *) buf, &len)) {
onRtp(buf, len);
} else {
WarnL;
RtpHeader *rtp = (RtpHeader *) buf;
WarnL << "srtp_unprotect rtp failed, pt:" << (int)rtp->pt;
}
return;
}
......@@ -399,37 +399,49 @@ void WebRtcTransportImp::onStartWebRTC() {
//获取ssrc和pt相关信息,届时收到rtp和rtcp时分别可以根据pt和ssrc找到相关的信息
for (auto &m_answer : getSdp(SdpType::answer).media) {
auto m_offer = getSdp(SdpType::offer).getMedia(m_answer.type);
auto info = std::make_shared<RtpPayloadInfo>();
info->media = &m_answer;
info->answer_ssrc_rtp = m_answer.getRtpSSRC();
info->answer_ssrc_rtx = m_answer.getRtxSSRC();
info->offer_ssrc_rtp = m_offer->getRtpSSRC();
info->offer_ssrc_rtx = m_offer->getRtxSSRC();
info->plan_rtp = &m_answer.plan[0];;
info->plan_rtx = m_answer.getRelatedRtxPlan(info->plan_rtp->pt);
info->rtcp_context_send = std::make_shared<RtcpContext>(info->plan_rtp->sample_rate, false);
//send ssrc --> RtpPayloadInfo
_rtp_info_ssrc[info->answer_ssrc_rtp] = std::make_pair(false, info);
_rtp_info_ssrc[info->answer_ssrc_rtx] = std::make_pair(true, info);
//recv ssrc --> RtpPayloadInfo
_rtp_info_ssrc[info->offer_ssrc_rtp] = std::make_pair(false, info);;
_rtp_info_ssrc[info->offer_ssrc_rtx] = std::make_pair(true, info);;
//rtp pt --> RtpPayloadInfo
_rtp_info_pt.emplace(info->plan_rtp->pt, std::make_pair(false, info));
if (info->plan_rtx) {
//rtx pt --> RtpPayloadInfo
_rtp_info_pt.emplace(info->plan_rtx->pt, std::make_pair(true, info));
auto track = std::make_shared<MediaTrack>();
track->media = &m_answer;
track->answer_ssrc_rtp = m_answer.getRtpSSRC();
track->answer_ssrc_rtx = m_answer.getRtxSSRC();
track->offer_ssrc_rtp = m_offer->getRtpSSRC();
track->offer_ssrc_rtx = m_offer->getRtxSSRC();
track->plan_rtp = &m_answer.plan[0];;
track->plan_rtx = m_answer.getRelatedRtxPlan(track->plan_rtp->pt);
track->rtcp_context_send = std::make_shared<RtcpContext>(false);
//send ssrc --> MediaTrack
_ssrc_to_track[track->answer_ssrc_rtp] = track;
_ssrc_to_track[track->answer_ssrc_rtx] = track;
//recv ssrc --> MediaTrack
_ssrc_to_track[track->offer_ssrc_rtp] = track;
_ssrc_to_track[track->offer_ssrc_rtx] = track;
//rtp pt --> MediaTrack
_pt_to_track.emplace(track->plan_rtp->pt, std::make_pair(false, track));
if (track->plan_rtx) {
//rtx pt --> MediaTrack
_pt_to_track.emplace(track->plan_rtx->pt, std::make_pair(true, track));
}
if (m_offer->type != TrackApplication) {
//记录rtp ext类型与id的关系，方便接收或发送rtp时修改rtp ext id
for (auto &ext : m_offer->extmap) {
auto ext_type = RtpExt::getExtType(ext.ext);
_rtp_ext_id_to_type.emplace(ext.id, ext_type);
_rtp_ext_type_to_id.emplace(ext_type, ext.id);
track->rtp_ext_ctx = std::make_shared<RtpExtContext>(*m_offer);
track->rtp_ext_ctx->setOnGetRtp([this, track](uint8_t pt, uint32_t ssrc, const string &rid) {
//ssrc --> MediaTrack
_ssrc_to_track[ssrc] = track;
InfoL << "get rtp, pt:" << (int) pt << ", ssrc:" << ssrc << ", rid:" << rid;
});
int index = 0;
for (auto &ssrc : m_offer->rtp_ssrc_sim) {
//记录ssrc对应的MediaTrack
_ssrc_to_track[ssrc.ssrc] = track;
if (m_offer->rtp_rids.size() > index) {
//支持firefox的simulcast, 提前映射好ssrc和rid的关系
track->rtp_ext_ctx->setRid(ssrc.ssrc, m_offer->rtp_rids[index]);
}
++index;
}
}
}
......@@ -466,10 +478,10 @@ void WebRtcTransportImp::onStartWebRTC() {
}
auto rtsp_media = rtsp_send_sdp.getMedia(m.type);
if (rtsp_media && getCodecId(rtsp_media->plan[0].codec) == getCodecId(m.plan[0].codec)) {
auto it = _rtp_info_pt.find(m.plan[0].pt);
CHECK(it != _rtp_info_pt.end());
auto it = _pt_to_track.find(m.plan[0].pt);
CHECK(it != _pt_to_track.end());
//记录发送rtp时约定的信息，届时发送rtp时需要修改pt和ssrc
_send_rtp_info[m.type] = it->second.second;
_type_to_track[m.type] = it->second.second;
}
}
}
......@@ -558,27 +570,45 @@ SdpAttrCandidate::Ptr WebRtcTransportImp::getIceCandidate() const{
///////////////////////////////////////////////////////////////////
class RtpReceiverImp : public RtpReceiver {
class RtpChannel : public RtpTrackImp {
public:
RtpReceiverImp( function<void(RtpPacket::Ptr rtp)> cb){
_on_sort = std::move(cb);
RtpChannel(RtpTrackImp::OnSorted cb, function<void(const FCI_NACK &nack)> on_nack) {
setOnSorted(std::move(cb));
_nack_ctx.setOnNack(std::move(on_nack));
}
~RtpReceiverImp() override = default;
~RtpChannel() override = default;
bool inputRtp(TrackType type, int samplerate, uint8_t *ptr, size_t len){
return handleOneRtp((int) type, type, samplerate, ptr, len);
bool inputRtp(TrackType type, int sample_rate, uint8_t *ptr, size_t len, bool is_rtx){
if (!is_rtx) {
RtpHeader *rtp = (RtpHeader *) ptr;
auto seq = ntohs(rtp->seq);
//统计rtp接受情况，便于生成nack rtcp包
_nack_ctx.received(seq);
//统计rtp收到的情况，好做rr汇报
_rtcp_context.onRtp(seq, ntohl(rtp->stamp), len);
}
return RtpTrack::inputRtp(type, sample_rate, ptr, len);
}
protected:
void onRtpSorted(RtpPacket::Ptr rtp, int track_index) override {
_on_sort(std::move(rtp));
Buffer::Ptr createRtcpRR(RtcpHeader *sr, uint32_t ssrc) {
_rtcp_context.onRtcp(sr);
return _rtcp_context.createRtcpRR(ssrc, getSSRC());
}
private:
function<void(RtpPacket::Ptr rtp)> _on_sort;
NackContext _nack_ctx;
RtcpContext _rtcp_context{true};
};
std::shared_ptr<RtpChannel> MediaTrack::getRtpChannel(uint32_t ssrc) const{
auto it_chn = rtp_channel.find(rtp_ext_ctx->getRid(ssrc));
if (it_chn == rtp_channel.end()) {
return nullptr;
}
return it_chn->second;
}
void WebRtcTransportImp::onRtcp(const char *buf, size_t len) {
_bytes_usage += len;
auto rtcps = RtcpHeader::loadFromBytes((char *) buf, len);
......@@ -587,19 +617,15 @@ void WebRtcTransportImp::onRtcp(const char *buf, size_t len) {
case RtcpType::RTCP_SR : {
//对方汇报rtp发送情况
RtcpSR *sr = (RtcpSR *) rtcp;
auto it = _rtp_info_ssrc.find(sr->ssrc);
if (it != _rtp_info_ssrc.end()) {
auto rtx = it->second.first;
if (!rtx) {
auto &info = it->second.second;
auto it = info->rtcp_context_recv.find(sr->ssrc);
if (it != info->rtcp_context_recv.end()) {
it->second->onRtcp(sr);
auto rr = it->second->createRtcpRR(info->answer_ssrc_rtp, sr->ssrc);
sendRtcpPacket(rr->data(), rr->size(), true);
} else {
WarnL << "未识别的sr rtcp包:" << rtcp->dumpString();
}
auto it = _ssrc_to_track.find(sr->ssrc);
if (it != _ssrc_to_track.end()) {
auto &track = it->second;
auto rtp_chn = track->getRtpChannel(sr->ssrc);
if(!rtp_chn){
WarnL << "未识别的sr rtcp包:" << rtcp->dumpString();
} else {
auto rr = rtp_chn->createRtcpRR(sr, track->answer_ssrc_rtp);
sendRtcpPacket(rr->data(), rr->size(), true);
}
} else {
WarnL << "未识别的sr rtcp包:" << rtcp->dumpString();
......@@ -611,14 +637,11 @@ void WebRtcTransportImp::onRtcp(const char *buf, size_t len) {
//对方汇报rtp接收情况
RtcpRR *rr = (RtcpRR *) rtcp;
for (auto item : rr->getItemList()) {
auto it = _rtp_info_ssrc.find(item->ssrc);
if (it != _rtp_info_ssrc.end()) {
auto rtx = it->second.first;
if (!rtx) {
auto &info = it->second.second;
auto sr = info->rtcp_context_send->createRtcpSR(info->answer_ssrc_rtp);
sendRtcpPacket(sr->data(), sr->size(), true);
}
auto it = _ssrc_to_track.find(item->ssrc);
if (it != _ssrc_to_track.end()) {
auto &track = it->second;
auto sr = track->rtcp_context_send->createRtcpSR(track->answer_ssrc_rtp);
sendRtcpPacket(sr->data(), sr->size(), true);
} else {
WarnL << "未识别的rr rtcp包:" << rtcp->dumpString();
}
......@@ -629,12 +652,12 @@ void WebRtcTransportImp::onRtcp(const char *buf, size_t len) {
//对方汇报停止发送rtp
RtcpBye *bye = (RtcpBye *) rtcp;
for (auto ssrc : bye->getSSRC()) {
auto it = _rtp_info_ssrc.find(*ssrc);
if (it == _rtp_info_ssrc.end()) {
auto it = _ssrc_to_track.find(*ssrc);
if (it == _ssrc_to_track.end()) {
WarnL << "未识别的bye rtcp包:" << rtcp->dumpString();
continue;
}
_rtp_info_ssrc.erase(it);
_ssrc_to_track.erase(it);
}
onShutdown(SockException(Err_eof, "rtcp bye message received"));
break;
......@@ -648,20 +671,17 @@ void WebRtcTransportImp::onRtcp(const char *buf, size_t len) {
switch ((RTPFBType) rtcp->report_count) {
case RTPFBType::RTCP_RTPFB_NACK : {
RtcpFB *fb = (RtcpFB *) rtcp;
auto it = _rtp_info_ssrc.find(fb->ssrc_media);
if (it == _rtp_info_ssrc.end()) {
auto it = _ssrc_to_track.find(fb->ssrc_media);
if (it == _ssrc_to_track.end()) {
WarnL << "未识别的 rtcp包:" << rtcp->dumpString();
return;
}
auto rtx = it->second.first;
if (!rtx) {
auto &info = it->second.second;
auto &fci = fb->getFci<FCI_NACK>();
info->nack_list.for_each_nack(fci, [&](const RtpPacket::Ptr &rtp) {
//rtp重传
onSendRtp(rtp, true, true);
});
}
auto &track = it->second;
auto &fci = fb->getFci<FCI_NACK>();
track->nack_list.for_each_nack(fci, [&](const RtpPacket::Ptr &rtp) {
//rtp重传
onSendRtp(rtp, true, true);
});
break;
}
default: break;
......@@ -675,122 +695,55 @@ void WebRtcTransportImp::onRtcp(const char *buf, size_t len) {
///////////////////////////////////////////////////////////////////
void WebRtcTransportImp::changeRtpExtId(RtpPayloadInfo &info, const RtpHeader *header, bool is_recv, bool is_rtx, string *rid_ptr) const{
auto ext_map = RtpExt::getExtValue(header);
for (auto &pr : ext_map) {
if (is_recv) {
auto it = _rtp_ext_id_to_type.find(pr.first);
if (it == _rtp_ext_id_to_type.end()) {
WarnL << "接收rtp时,忽略不识别的rtp ext, id=" << (int) pr.first;
pr.second.clearExt();
continue;
}
pr.second.setType(it->second);
//重新赋值ext id为 ext type，作为后面处理ext的统一中间类型
pr.second.setExtId((uint8_t) it->second);
switch(it->second){
case RtpExtType::sdes_repaired_rtp_stream_id :
case RtpExtType::sdes_rtp_stream_id : {
auto ssrc = ntohl(header->ssrc);
auto rid = it->second == RtpExtType::sdes_rtp_stream_id ? pr.second.getRtpStreamId() : pr.second.getRepairedRtpStreamId();
//根据rid获取rtp或rtx的ssrc
auto &ssrc_ref = is_rtx ? info.rid_ssrc[rid].second : info.rid_ssrc[rid].first;
if (!ssrc_ref) {
//ssrc未赋值，赋值
ssrc_ref = ssrc;
DebugL << (is_rtx ? "got rid of rtx:" : "got rid:") << rid << ", ssrc:" << ssrc;
}
if (is_rtx) {
//rtx ssrc --> rtp ssrc
auto &rtp_ssrc_ref = info.rtx_ssrc_to_rtp_ssrc[ssrc];
if (!rtp_ssrc_ref && info.rid_ssrc[rid].first) {
//未找到rtx到rtp ssrc的映射关系，且已经获取rtp的ssrc，那么设置映射关系
rtp_ssrc_ref = info.rid_ssrc[rid].first;
DebugL << "got ssrc of rid:" << rid << ", [rtx-rtp]:" << ssrc << "-" << rtp_ssrc_ref;
}
}
if (rid_ptr) {
*rid_ptr = rid;
}
break;
}
default : break;
}
} else {
pr.second.setType((RtpExtType) pr.first);
auto it = _rtp_ext_type_to_id.find((RtpExtType) pr.first);
if (it == _rtp_ext_type_to_id.end()) {
WarnL << "发送rtp时, 忽略不被客户端支持rtp ext:" << pr.second.dumpString();
pr.second.clearExt();
continue;
}
//重新赋值ext id为客户端sdp声明的类型
pr.second.setExtId(it->second);
}
}
void WebRtcTransportImp::createRtpChannel(const string &rid, uint32_t ssrc, const MediaTrack::Ptr &track) {
//rid --> RtpReceiverImp
auto &ref = track->rtp_channel[rid];
ref = std::make_shared<RtpChannel>([track, this, rid](RtpPacket::Ptr rtp) mutable {
onSortedRtp(*track, rid, std::move(rtp));
}, [track, this, ssrc](const FCI_NACK &nack) mutable {
onSendNack(*track, nack, ssrc);
});
InfoL << "create rtp receiver of ssrc:" << ssrc << ", rid:" << rid << ", codec:" << track->plan_rtp->codec;
}
void WebRtcTransportImp::onRtp(const char *buf, size_t len) {
onRtp_l(buf, len, false);
}
void WebRtcTransportImp::onRtp_l(const char *buf, size_t len, bool rtx) {
if (!rtx) {
_bytes_usage += len;
_alive_ticker.resetTime();
}
_bytes_usage += len;
_alive_ticker.resetTime();
RtpHeader *rtp = (RtpHeader *) buf;
auto ssrc = ntohl(rtp->ssrc);
//根据接收到的rtp的pt信息，找到该流的信息
auto it = _rtp_info_pt.find(rtp->pt);
if (it == _rtp_info_pt.end()) {
WarnL;
auto it = _pt_to_track.find(rtp->pt);
if (it == _pt_to_track.end()) {
WarnL << "unknown rtp pt:" << (int)rtp->pt;
return;
}
auto &info = it->second.second;
if (!it->second.first) {
bool is_rtx = it->second.first;
auto ssrc = ntohl(rtp->ssrc);
auto &track = it->second.second;
//修改ext id至统一
string rid;
track->rtp_ext_ctx->changeRtpExtId(rtp, true, &rid);
auto &ref = track->rtp_channel[rid];
if (!ref) {
if (is_rtx) {
//再接收到对应的rtp前，丢弃rtx包
WarnL << "unknown rtx rtp, rid:" << rid << ", ssrc:" << ssrc << ", codec:" << track->plan_rtp->codec << ", seq:" << ntohs(rtp->seq);
return;
}
createRtpChannel(rid, ssrc, track);
}
if (!is_rtx) {
//这是普通的rtp数据
auto seq = ntohs(rtp->seq);
#if 0
if (!rtx && info->media->type == TrackVideo && seq % 100 == 0) {
auto seq = ntohs(rtp->seq);
if (track->media->type == TrackVideo && seq % 100 == 0) {
//此处模拟接受丢包
DebugL << "recv dropped:" << seq;
return;
}
#endif
auto &ref = info->receiver[ssrc];
if (!rtx) {
//统计rtp接受情况，便于生成nack rtcp包
info->nack_ctx[ssrc].received(seq);
//时间戳转换成毫秒
auto stamp_ms = ntohl(rtp->stamp) * uint64_t(1000) / info->plan_rtp->sample_rate;
//统计rtp收到的情况，好做rr汇报
auto &cxt_ref = info->rtcp_context_recv[ssrc];
if (!cxt_ref) {
cxt_ref = std::make_shared<RtcpContext>(info->plan_rtp->sample_rate, true);
}
cxt_ref->onRtp(seq, stamp_ms, len);
//修改ext id至统一
string rid;
changeRtpExtId(*info, rtp, true, false, &rid);
if (!ref) {
ref = std::make_shared<RtpReceiverImp>([info, this, rid](RtpPacket::Ptr rtp) mutable {
onSortedRtp(*info, rid, std::move(rtp));
});
info->nack_ctx[ssrc].setOnNack([info, this, ssrc](const FCI_NACK &nack) mutable {
onSendNack(*info, nack, ssrc);
});
//recv simulcast ssrc --> RtpPayloadInfo
_rtp_info_ssrc[ssrc] = std::make_pair(false, info);
InfoL << "receive rtp of ssrc:" << ssrc;
}
}
//解析并排序rtp
if(!ref){
InfoL << "ignore no rtp receiver of ssrc:" << ssrc<<" is rtx:"<<rtx;
......@@ -809,31 +762,22 @@ void WebRtcTransportImp::onRtp_l(const char *buf, size_t len, bool rtx) {
return;
}
//修改ext id至统一
changeRtpExtId(*info, rtp, true, true);
//前两个字节是原始的rtp的seq
auto origin_seq = payload[0] << 8 | payload[1];
//rtx 转换为 rtp
rtp->pt = track->plan_rtp->pt;
rtp->seq = htons(origin_seq);
if (info->offer_ssrc_rtp) {
//非simulcast或音频
rtp->ssrc = htonl(info->offer_ssrc_rtp);
TraceL << "received rtx rtp,ssrc: " << ssrc << ", seq:" << origin_seq << ", pt:" << (int)rtp->pt;
} else {
//todo simulcast下，辅码流通过rtx传输？
//simulcast情况下，根据rtx的ssrc查找rtp的ssrc
rtp->ssrc = htonl(info->rtx_ssrc_to_rtp_ssrc[ntohl(rtp->ssrc)]);
}
rtp->pt = info->plan_rtp->pt;
rtp->ssrc = htonl(ref->getSSRC());
memmove((uint8_t *) buf + 2, buf, payload - (uint8_t *) buf);
buf += 2;
len -= 2;
onRtp_l(buf, len, true);
ref->inputRtp(track->media->type, track->plan_rtp->sample_rate, (uint8_t *) buf, len, true);
}
void WebRtcTransportImp::onSendNack(RtpPayloadInfo &info, const FCI_NACK &nack, uint32_t ssrc) {
void WebRtcTransportImp::onSendNack(MediaTrack &track, const FCI_NACK &nack, uint32_t ssrc) {
auto rtcp = RtcpFB::create(RTPFBType::RTCP_RTPFB_NACK, &nack, FCI_NACK::kSize);
rtcp->ssrc = htons(info.answer_ssrc_rtp);
rtcp->ssrc = htons(track.answer_ssrc_rtp);
rtcp->ssrc_media = htonl(ssrc);
DebugL << htonl(ssrc) << " " << nack.getPid();
sendRtcpPacket((char *) rtcp.get(), rtcp->getSize(), true);
......@@ -841,8 +785,8 @@ void WebRtcTransportImp::onSendNack(RtpPayloadInfo &info, const FCI_NACK &nack,
///////////////////////////////////////////////////////////////////
void WebRtcTransportImp::onSortedRtp(RtpPayloadInfo &info, const string &rid, RtpPacket::Ptr rtp) {
if (info.media->type == TrackVideo && _pli_ticker.elapsedTime() > 2000) {
void WebRtcTransportImp::onSortedRtp(MediaTrack &track, const string &rid, RtpPacket::Ptr rtp) {
if (track.media->type == TrackVideo && _pli_ticker.elapsedTime() > 2000) {
//定期发送pli请求关键帧，方便非rtc等协议
_pli_ticker.resetTime();
sendRtcpPli(rtp->getSSRC());
......@@ -879,42 +823,41 @@ void WebRtcTransportImp::onSortedRtp(RtpPayloadInfo &info, const string &rid, Rt
///////////////////////////////////////////////////////////////////
void WebRtcTransportImp::onSendRtp(const RtpPacket::Ptr &rtp, bool flush, bool rtx){
auto &info = _send_rtp_info[rtp->type];
if (!info) {
auto &track = _type_to_track[rtp->type];
if (!track) {
//忽略，对方不支持该编码类型
return;
}
if (!rtx) {
//统计rtp发送情况，好做sr汇报
info->rtcp_context_send->onRtp(rtp->getSeq(), rtp->getStampMS(), rtp->size() - RtpPacket::kRtpTcpHeaderSize);
info->nack_list.push_back(rtp);
track->rtcp_context_send->onRtp(rtp->getSeq(), ntohl(rtp->getHeader()->stamp), rtp->size() - RtpPacket::kRtpTcpHeaderSize);
track->nack_list.push_back(rtp);
#if 0
//此处模拟发送丢包
if (rtp->type == TrackVideo && rtp->getSeq() % 100 == 0) {
DebugL << "send dropped:" << rtp->getSeq();
return;
}
#endif
} else {
WarnL << "send rtx rtp:" << rtp->getSeq();
}
pair<bool/*rtx*/, RtpPayloadInfo *> ctx{rtx, info.get()};
pair<bool/*rtx*/, MediaTrack *> ctx{rtx, track.get()};
sendRtpPacket(rtp->data() + RtpPacket::kRtpTcpHeaderSize, rtp->size() - RtpPacket::kRtpTcpHeaderSize, flush, &ctx);
_bytes_usage += rtp->size() - RtpPacket::kRtpTcpHeaderSize;
}
void WebRtcTransportImp::onBeforeEncryptRtp(const char *buf, size_t &len, void *ctx) {
auto pr = (pair<bool/*rtx*/, RtpPayloadInfo *> *) ctx;
auto pr = (pair<bool/*rtx*/, MediaTrack *> *) ctx;
auto header = (RtpHeader *) buf;
if (!pr->first || !pr->second->plan_rtx) {
//普通的rtp,或者不支持rtx, 修改目标pt和ssrc
changeRtpExtId(*pr->second, header, false, false);
pr->second->rtp_ext_ctx->changeRtpExtId(header, false);
header->pt = pr->second->plan_rtp->pt;
header->ssrc = htonl(pr->second->answer_ssrc_rtp);
} else {
//重传的rtp, rtx
changeRtpExtId(*pr->second, header, false, true);
pr->second->rtp_ext_ctx->changeRtpExtId(header, false);
header->pt = pr->second->plan_rtx->pt;
if (pr->second->answer_ssrc_rtx) {
//有rtx单独的ssrc,有些情况下，浏览器支持rtx，但是未指定rtx单独的ssrc
......@@ -949,7 +892,7 @@ void WebRtcTransportImp::onShutdown(const SockException &ex){
bool WebRtcTransportImp::close(MediaSource &sender, bool force) {
//此回调在其他线程触发
if(!_push_src || (!force && _push_src->totalReaderCount())){
if (!force && totalReaderCount(sender)) {
return false;
}
string err = StrPrinter << "close media:" << sender.getSchema() << "/" << sender.getVhost() << "/" << sender.getApp() << "/" << sender.getId() << " " << force;
......@@ -958,7 +901,11 @@ bool WebRtcTransportImp::close(MediaSource &sender, bool force) {
}
int WebRtcTransportImp::totalReaderCount(MediaSource &sender) {
return _push_src ? _push_src->totalReaderCount() : sender.readerCount();
auto total_count = 0;
for (auto &src : _push_src_simulcast) {
total_count += src.second->totalReaderCount();
}
return total_count;
}
MediaOriginType WebRtcTransportImp::getOriginType(MediaSource &sender) const {
......
webrtc/WebRtcTransport.h
......@@ -22,6 +22,8 @@
#include "Rtsp/RtspMediaSourceImp.h"
#include "Rtcp/RtcpContext.h"
#include "Rtcp/RtcpFCI.h"
#include "Nack.h"
using namespace toolkit;
using namespace mediakit;
......@@ -125,151 +127,26 @@ private:
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 {
class RtpChannel;
class MediaTrack {
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;
using Ptr = std::shared_ptr<MediaTrack>;
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;
RtpExtContext::Ptr rtp_ext_ctx;
//for send rtp
NackList nack_list;
RtcpContext::Ptr rtcp_context_send;
//for recv rtp
unordered_map<string/*rid*/, std::shared_ptr<RtpChannel> > rtp_channel;
std::shared_ptr<RtpChannel> getRtpChannel(uint32_t ssrc) const;
};
class WebRtcTransportImp : public WebRtcTransport, public MediaSourceEvent, public SockInfo, public std::enable_shared_from_this<WebRtcTransportImp>{
......@@ -298,8 +175,6 @@ protected:
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 {};
......@@ -339,28 +214,9 @@ private:
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;
void onSortedRtp(MediaTrack &track, const string &rid, RtpPacket::Ptr rtp);
void onSendNack(MediaTrack &track, const FCI_NACK &nack, uint32_t ssrc);
void createRtpChannel(const string &rid, uint32_t ssrc, const MediaTrack::Ptr &track);
private:
uint16_t _rtx_seq[2] = {0, 0};
......@@ -386,13 +242,9 @@ private:
//播放rtsp源的reader对象
RtspMediaSource::RingType::RingReader::Ptr _reader;
//根据发送rtp的track类型获取相关信息
RtpPayloadInfo::Ptr _send_rtp_info[2];
MediaTrack::Ptr _type_to_track[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;
unordered_map<uint8_t/*pt*/, std::pair<bool/*is rtx*/,MediaTrack::Ptr> > _pt_to_track;
//根据rtcp的ssrc获取相关信息，收发rtp和rtx的ssrc都会记录
unordered_map<uint32_t/*ssrc*/, MediaTrack::Ptr> _ssrc_to_track;
};
\ No newline at end of file
www/webrtc/ZLMRTCClient.js
......@@ -7399,18 +7399,11 @@ var ZLMRTCClient = (function (exports) {
};
if (this.options.simulcast && stream.getVideoTracks().length > 0) {
VideoTransceiverInit.sendEncodings = [{
rid: 'q',
active: true,
scaleResolutionDownBy: 4.0
}, {
rid: 'h',
active: true,
scaleResolutionDownBy: 2.0
}, {
rid: 'f',
active: true
}];
VideoTransceiverInit.sendEncodings = [
{ rid: "h", active: true, maxBitrate: 1000000 },
{ rid: "m", active: true, maxBitrate: 500000, scaleResolutionDownBy: 2 },
{ rid: "l", active: true, maxBitrate: 200000, scaleResolutionDownBy: 4 }
];
}
if (stream.getAudioTracks().length > 0) {
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论