amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-10-31 08:39:34 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

webrtc支持 rtcp xr(#1796)

Browse Source 
metaRTC 播放时，会发送rtcp xr，因此处理rtcp xr 来兼容
This commit is contained in:
夏楚 2022-07-15 11:11:30 +08:00 committed by GitHub
commit 6aa1c239b8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 1277 additions and 934 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

187
src/Rtcp/Rtcp.cpp
View File

@ -8,11 +8,11 @@
* may be found in the AUTHORS file in the root of the source tree.
*/
#include <stddef.h>
#include <assert.h>
#include "Rtcp.h"
#include "Util/logger.h"
#include "RtcpFCI.h"
#include "Util/logger.h"
#include <assert.h>
#include <stddef.h>
using namespace std;
using namespace toolkit;
@ -21,37 +21,49 @@ namespace mediakit {
const char *rtcpTypeToStr(RtcpType type) {
switch (type) {
#define SWITCH_CASE(key, value) case RtcpType::key : return #value "(" #key ")";
#define SWITCH_CASE(key, value) \
case RtcpType::key: \
return #value "(" #key ")";
RTCP_PT_MAP(SWITCH_CASE)
#undef SWITCH_CASE
default: return "unknown rtcp pt";
default:
return "unknown rtcp pt";
}
}
const char *sdesTypeToStr(SdesType type) {
switch (type) {
#define SWITCH_CASE(key, value) case SdesType::key : return #value "(" #key ")";
#define SWITCH_CASE(key, value) \
case SdesType::key: \
return #value "(" #key ")";
SDES_TYPE_MAP(SWITCH_CASE)
#undef SWITCH_CASE
default: return "unknown source description type";
default:
return "unknown source description type";
}
}
const char *psfbTypeToStr(PSFBType type) {
switch (type) {
#define SWITCH_CASE(key, value) case PSFBType::key : return #value "(" #key ")";
#define SWITCH_CASE(key, value) \
case PSFBType::key: \
return #value "(" #key ")";
PSFB_TYPE_MAP(SWITCH_CASE)
#undef SWITCH_CASE
default: return "unknown payload-specific fb message fmt type";
default:
return "unknown payload-specific fb message fmt type";
}
}
const char *rtpfbTypeToStr(RTPFBType type) {
switch (type) {
#define SWITCH_CASE(key, value) case RTPFBType::key : return #value "(" #key ")";
#define SWITCH_CASE(key, value) \
case RTPFBType::key: \
return #value "(" #key ")";
RTPFB_TYPE_MAP(SWITCH_CASE)
#undef SWITCH_CASE
default: return "unknown transport layer feedback messages fmt type";
default:
return "unknown transport layer feedback messages fmt type";
}
}
@ -140,7 +152,8 @@ string RtcpHeader::dumpString() const {
return rtcp->dumpString();
}
default: return StrPrinter << dumpHeader() << hexdump((char *)this + sizeof(*this), getSize() - sizeof(*this));
default:
return StrPrinter << dumpHeader() << hexdump((char *)this + sizeof(*this), getSize() - sizeof(*this));
}
}
@ -193,8 +206,22 @@ void RtcpHeader::net2Host(size_t len) {
bye->net2Host(len);
break;
}
default: throw std::runtime_error(StrPrinter << "未处理的rtcp包:" << rtcpTypeToStr((RtcpType) this->pt));
case RtcpType::RTCP_XR: {
RtcpXRRRTR *xr = (RtcpXRRRTR *)this;
if (xr->bt == 4) {
xr->net2Host(len);
// TraceL<<xr->dumpString();
} else if (xr->bt == 5) {
RtcpXRDLRR *dlrr = (RtcpXRDLRR *)this;
dlrr->net2Host(len);
TraceL << dlrr->dumpString();
} else {
throw std::runtime_error(StrPrinter << "rtcp xr bt " << xr->bt << " not support");
}
break;
}
default:
throw std::runtime_error(StrPrinter << "未处理的rtcp包:" << rtcpTypeToStr((RtcpType)this->pt));
}
}
@ -224,19 +251,13 @@ vector<RtcpHeader *> RtcpHeader::loadFromBytes(char *data, size_t len) {
class BufferRtcp : public Buffer {
public:
BufferRtcp(std::shared_ptr<RtcpHeader> rtcp) {
_rtcp = std::move(rtcp);
}
BufferRtcp(std::shared_ptr<RtcpHeader> rtcp) { _rtcp = std::move(rtcp); }
~BufferRtcp() override {}
char *data() const override {
return (char *) _rtcp.get();
}
char *data() const override { return (char *)_rtcp.get(); }
size_t size() const override {
return _rtcp->getSize();
}
size_t size() const override { return _rtcp->getSize(); }
private:
std::shared_ptr<RtcpHeader> _rtcp;
@ -254,9 +275,7 @@ std::shared_ptr<RtcpSR> RtcpSR::create(size_t item_count) {
auto ptr = (RtcpSR *)new char[bytes];
setupHeader(ptr, RtcpType::RTCP_SR, item_count, bytes);
setupPadding(ptr, bytes - real_size);
return std::shared_ptr<RtcpSR>(ptr, [](RtcpSR *ptr) {
delete[] (char *) ptr;
});
return std::shared_ptr<RtcpSR>(ptr, [](RtcpSR *ptr) { delete[] (char *)ptr; });
}
string RtcpSR::getNtpStamp() const {
@ -311,13 +330,15 @@ string RtcpSR::dumpString() const {
#define CHECK_MIN_SIZE(size, kMinSize) \
if (size < kMinSize) { \
throw std::out_of_range(StrPrinter << rtcpTypeToStr((RtcpType)pt) << " 长度不足:" << size << " < " << kMinSize); \
throw std::out_of_range( \
StrPrinter << rtcpTypeToStr((RtcpType)pt) << " 长度不足:" << size << " < " << kMinSize); \
}
#define CHECK_REPORT_COUNT(item_count) \
/*修正个数，防止getItemList时内存越界*/ \
if (report_count != item_count) { \
WarnL << rtcpTypeToStr((RtcpType)pt) << " report_count 字段不正确,已修正为:" << (int)report_count << " -> " << item_count; \
WarnL << rtcpTypeToStr((RtcpType)pt) << " report_count 字段不正确,已修正为:" << (int)report_count << " -> " \
<< item_count; \
report_count = item_count; \
}
@ -385,9 +406,7 @@ std::shared_ptr<RtcpRR> RtcpRR::create(size_t item_count) {
auto ptr = (RtcpRR *)new char[bytes];
setupHeader(ptr, RtcpType::RTCP_RR, item_count, bytes);
setupPadding(ptr, bytes - real_size);
return std::shared_ptr<RtcpRR>(ptr, [](RtcpRR *ptr) {
delete[] (char *) ptr;
});
return std::shared_ptr<RtcpRR>(ptr, [](RtcpRR *ptr) { delete[] (char *)ptr; });
}
string RtcpRR::dumpString() const {
@ -473,9 +492,7 @@ std::shared_ptr<RtcpSdes> RtcpSdes::create(const std::vector<string> &item_text)
setupHeader(ptr, RtcpType::RTCP_SDES, item_text.size(), bytes);
setupPadding(ptr, bytes - real_size);
return std::shared_ptr<RtcpSdes>(ptr, [](RtcpSdes *ptr) {
delete[] (char *) ptr;
});
return std::shared_ptr<RtcpSdes>(ptr, [](RtcpSdes *ptr) { delete[] (char *)ptr; });
}
string RtcpSdes::dumpString() const {
@ -527,9 +544,7 @@ std::shared_ptr<RtcpFB> RtcpFB::create_l(RtcpType type, int fmt, const void *fci
}
setupHeader(ptr, type, fmt, bytes);
setupPadding(ptr, bytes - real_size);
return std::shared_ptr<RtcpFB>((RtcpFB *) ptr, [](RtcpFB *ptr) {
delete[] (char *) ptr;
});
return std::shared_ptr<RtcpFB>((RtcpFB *)ptr, [](RtcpFB *ptr) { delete[] (char *)ptr; });
}
std::shared_ptr<RtcpFB> RtcpFB::create(PSFBType fmt, const void *fci, size_t fci_len) {
@ -606,7 +621,9 @@ string RtcpFB::dumpString() const {
}
break;
}
default: /*不可达*/ assert(0); break;
default: /*不可达*/
assert(0);
break;
}
return std::move(printer);
}
@ -639,9 +656,7 @@ std::shared_ptr<RtcpBye> RtcpBye::create(const std::vector<uint32_t> &ssrcs, con
memcpy(reason_len_ptr + 1, reason.data(), *reason_len_ptr);
}
return std::shared_ptr<RtcpBye>(ptr, [](RtcpBye *ptr) {
delete[] (char *) ptr;
});
return std::shared_ptr<RtcpBye>(ptr, [](RtcpBye *ptr) { delete[] (char *)ptr; });
}
vector<uint32_t *> RtcpBye::getSSRC() {
@ -691,6 +706,96 @@ void RtcpBye::net2Host(size_t size) {
}
}
}
////////////////////////////////////////////
string RtcpXRRRTR::dumpString() const {
_StrPrinter printer;
printer << RtcpHeader::dumpHeader();
printer << "ssrc :" << ssrc << "\r\n";
printer << "bt :" << (int)bt << "\r\n";
printer << "block_length : " << block_length << "\r\n";
printer << "ntp msw : " << ntpmsw << "\r\n";
printer << "ntp lsw : " << ntplsw << "\r\n";
return std::move(printer);
}
void RtcpXRRRTR::net2Host(size_t size) {
static const size_t kMinSize = sizeof(RtcpHeader);
CHECK_MIN_SIZE(size, kMinSize);
if (size != sizeof(RtcpXRRRTR)) {
throw std::invalid_argument(
StrPrinter << "rtcp xr Receiver Reference Time Report Block must is " << sizeof(RtcpXRRRTR)
<< " actual size " << size);
}
ssrc = ntohl(ssrc);
block_length = ntohs(block_length);
ntpmsw = ntohl(ntpmsw);
ntplsw = ntohl(ntplsw);
}
string RtcpXRDLRRReportItem::dumpString() const {
_StrPrinter printer;
printer << "ssrc :" << ssrc << "\r\n";
printer << "last RR (lrr) :" << lrr << "\r\n";
printer << "delay since last RR (dlrr): " << dlrr << "\r\n";
return std::move(printer);
}
void RtcpXRDLRRReportItem::net2Host() {
ssrc = ntohl(ssrc);
lrr = ntohl(lrr);
dlrr = ntohl(dlrr);
}
std::vector<RtcpXRDLRRReportItem *> RtcpXRDLRR::getItemList() {
auto count = block_length / 3;
RtcpXRDLRRReportItem *ptr = &items;
vector<RtcpXRDLRRReportItem *> ret;
for (int i = 0; i < (int)count; ++i) {
ret.emplace_back(ptr);
++ptr;
}
return ret;
}
string RtcpXRDLRR::dumpString() const {
_StrPrinter printer;
printer << RtcpHeader::dumpHeader();
printer << "ssrc :" << ssrc << "\r\n";
printer << "bt :" << (int)bt << "\r\n";
printer << "block_length : " << block_length << "\r\n";
auto items_list = ((RtcpXRDLRR *)this)->getItemList();
auto i = 0;
for (auto &item : items_list) {
printer << "---- item:" << i++ << " ----\r\n";
printer << item->dumpString();
}
return std::move(printer);
}
void RtcpXRDLRR::net2Host(size_t size) {
static const size_t kMinSize = sizeof(RtcpHeader);
CHECK_MIN_SIZE(size, kMinSize);
ssrc = ntohl(ssrc);
block_length = ntohs(block_length);
auto count = block_length / 3;
for (int i = 0; i < (int)count; ++i) {
RtcpXRDLRRReportItem *ptr = &items;
ptr->net2Host();
ptr++;
}
}
std::shared_ptr<RtcpXRDLRR> RtcpXRDLRR::create(size_t item_count) {
auto real_size = sizeof(RtcpXRDLRR) - sizeof(RtcpXRDLRRReportItem) + item_count * sizeof(RtcpXRDLRRReportItem);
auto bytes = alignSize(real_size);
auto ptr = (RtcpXRDLRR *)new char[bytes];
setupHeader(ptr, RtcpType::RTCP_XR, 0, bytes);
setupPadding(ptr, bytes - real_size);
return std::shared_ptr<RtcpXRDLRR>(ptr, [](RtcpXRDLRR *ptr) { delete[] (char *)ptr; });
}
#if 0
#include "Util/onceToken.h"

137
src/Rtcp/Rtcp.h
View File

@ -11,11 +11,11 @@
#ifndef ZLMEDIAKIT_RTCP_H
#define ZLMEDIAKIT_RTCP_H
#include "Common/macros.h"
#include "Network/Buffer.h"
#include "Util/util.h"
#include <stdint.h>
#include <vector>
#include "Util/util.h"
#include "Network/Buffer.h"
#include "Common/macros.h"
namespace mediakit {
@ -222,7 +222,6 @@ public:
void setSize(size_t size);
protected:
/**
* 打印字段详情
* 使用net2Host转换成主机字节序后才可使用此函数
@ -687,6 +686,136 @@ private:
void net2Host(size_t size);
} PACKED;
/*
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
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|V=2|P|reserved | PT=XR=207 | length |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| SSRC |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
: report blocks :
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
*/
/*
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
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| BT=4 | reserved | block length = 2 |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| NTP timestamp, most significant word |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| NTP timestamp, least significant word |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
*/
class RtcpXRRRTR : public RtcpHeader {
public:
friend class RtcpHeader;
uint32_t ssrc;
// 4
uint8_t bt;
uint8_t reserved;
// 2
uint16_t block_length;
// ntp timestamp MSW(in second)
uint32_t ntpmsw;
// ntp timestamp LSW(in picosecond)
uint32_t ntplsw;
private:
/**
* 打印字段详情
* 使用net2Host转换成主机字节序后才可使用此函数
*/
std::string dumpString() const;
/**
* 网络字节序转换为主机字节序
* @param size 字节长度，防止内存越界
*/
void net2Host(size_t size);
} PACKED;
/*
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
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| BT=5 | reserved | block length |
+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
| SSRC_1 (SSRC of first receiver) | sub-
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ block
| last RR (LRR) | 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| delay since last RR (DLRR) |
+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
| SSRC_2 (SSRC of second receiver) | sub-
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ block
: ... : 2
+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*/
class RtcpXRDLRRReportItem {
public:
friend class RtcpXRDLRR;
uint32_t ssrc;
uint32_t lrr;
uint32_t dlrr;
private:
/**
* 打印字段详情
* 使用net2Host转换成主机字节序后才可使用此函数
*/
std::string dumpString() const;
/**
* 网络字节序转换为主机字节序
* @param size 字节长度，防止内存越界
*/
void net2Host();
} PACKED;
class RtcpXRDLRR : public RtcpHeader {
public:
friend class RtcpHeader;
uint32_t ssrc;
uint8_t bt;
uint8_t reserved;
uint16_t block_length;
RtcpXRDLRRReportItem items;
/**
* 创建RtcpXRDLRR包，只赋值了RtcpHeader部分(网络字节序)
* @param item_count RtcpXRDLRRReportItem对象个数
* @return RtcpXRDLRR包
*/
static std::shared_ptr<RtcpXRDLRR> create(size_t item_count);
/**
* 获取RtcpXRDLRRReportItem对象指针列表
* 使用net2Host转换成主机字节序后才可使用此函数
*/
std::vector<RtcpXRDLRRReportItem *> getItemList();
private:
/**
* 打印字段详情
* 使用net2Host转换成主机字节序后才可使用此函数
*/
std::string dumpString() const;
/**
* 网络字节序转换为主机字节序
* @param size 字节长度，防止内存越界
*/
void net2Host(size_t size);
} PACKED;
#if defined(_WIN32)
#pragma pack(pop)
#endif // defined(_WIN32)

60
src/Rtcp/RtcpContext.cpp
View File

@ -14,7 +14,8 @@ using namespace toolkit;
namespace mediakit {
void RtcpContext::onRtp(uint16_t /*seq*/, uint32_t stamp, uint64_t ntp_stamp_ms, uint32_t /*sample_rate*/, size_t bytes) {
void RtcpContext::onRtp(
uint16_t /*seq*/, uint32_t stamp, uint64_t ntp_stamp_ms, uint32_t /*sample_rate*/, size_t bytes) {
++_packets;
_bytes += bytes;
_last_rtp_stamp = stamp;
@ -45,6 +46,10 @@ Buffer::Ptr RtcpContext::createRtcpRR(uint32_t rtcp_ssrc, uint32_t rtp_ssrc) {
throw std::runtime_error("没有实现, rtp发送者尝试发送rr包");
}
Buffer::Ptr RtcpContext::createRtcpXRDLRR(uint32_t rtcp_ssrc, uint32_t rtp_ssrc) {
throw std::runtime_error("没有实现, rtp发送者尝试发送xr dlrr包");
}
////////////////////////////////////////////////////////////////////////////////////
void RtcpContextForSend::onRtcp(RtcpHeader *rtcp) {
@ -72,7 +77,21 @@ void RtcpContextForSend::onRtcp(RtcpHeader *rtcp) {
}
break;
}
default: break;
case RtcpType::RTCP_XR: {
auto rtcp_xr = (RtcpXRRRTR *)rtcp;
if (rtcp_xr->bt == 4) {
_xr_xrrtr_recv_last_rr[rtcp_xr->ssrc]
= ((rtcp_xr->ntpmsw & 0xFFFF) << 16) | ((rtcp_xr->ntplsw >> 16) & 0xFFFF);
_xr_rrtr_recv_sys_stamp[rtcp_xr->ssrc] = getCurrentMillisecond();
} else if (rtcp_xr->bt == 5) {
TraceL << "for sender not recive dlrr";
} else {
TraceL << "not support xr bt " << rtcp_xr->bt;
}
break;
}
default:
break;
}
}
@ -103,15 +122,45 @@ Buffer::Ptr RtcpContextForSend::createRtcpSR(uint32_t rtcp_ssrc) {
return RtcpHeader::toBuffer(std::move(rtcp));
}
toolkit::Buffer::Ptr RtcpContextForSend::createRtcpXRDLRR(uint32_t rtcp_ssrc, uint32_t rtp_ssrc) {
auto rtcp = RtcpXRDLRR::create(1);
rtcp->bt = 5;
rtcp->reserved = 0;
rtcp->block_length = htons(3);
rtcp->ssrc = htonl(rtcp_ssrc);
rtcp->items.ssrc = htonl(rtp_ssrc);
if (_xr_xrrtr_recv_last_rr.find(rtp_ssrc) == _xr_xrrtr_recv_last_rr.end()) {
rtcp->items.lrr = 0;
WarnL;
} else {
rtcp->items.lrr = htonl(_xr_xrrtr_recv_last_rr[rtp_ssrc]);
}
if (_xr_rrtr_recv_sys_stamp.find(rtp_ssrc) == _xr_rrtr_recv_sys_stamp.end()) {
rtcp->items.dlrr = 0;
WarnL;
} else {
// now - Last SR time,单位毫秒
auto delay = getCurrentMillisecond() - _xr_rrtr_recv_sys_stamp[rtp_ssrc];
// in units of 1/65536 seconds
auto dlsr = (uint32_t)(delay / 1000.0f * 65536);
rtcp->items.dlrr = htonl(dlsr);
}
return RtcpHeader::toBuffer(std::move(rtcp));
}
////////////////////////////////////////////////////////////////////////////////////
void RtcpContextForRecv::onRtp(uint16_t seq, uint32_t stamp, uint64_t ntp_stamp_ms, uint32_t sample_rate, size_t bytes) {
void RtcpContextForRecv::onRtp(
uint16_t seq, uint32_t stamp, uint64_t ntp_stamp_ms, uint32_t sample_rate, size_t bytes) {
{
// 接收者才做复杂的统计运算
auto sys_stamp = getCurrentMillisecond();
if (_last_rtp_sys_stamp) {
// 计算时间戳抖动值
double diff = double((int64_t(sys_stamp) - int64_t(_last_rtp_sys_stamp)) * (sample_rate / double(1000.0))
double diff = double(
(int64_t(sys_stamp) - int64_t(_last_rtp_sys_stamp)) * (sample_rate / double(1000.0))
- (int64_t(stamp) - int64_t(_last_rtp_stamp)));
if (diff < 0) {
diff = -diff;
@ -162,7 +211,8 @@ void RtcpContextForRecv::onRtcp(RtcpHeader *rtcp) {
_last_sr_ntp_sys = getCurrentMillisecond();
break;
}
default: break;
default:
break;
}
}

17
src/Rtcp/RtcpContext.h
View File

@ -11,9 +11,9 @@
#ifndef ZLMEDIAKIT_RTCPCONTEXT_H
#define ZLMEDIAKIT_RTCPCONTEXT_H
#include <stdint.h>
#include <stddef.h>
#include "Rtcp.h"
#include <stddef.h>
#include <stdint.h>
namespace mediakit {
@ -55,6 +55,13 @@ public:
*/
virtual toolkit::Buffer::Ptr createRtcpSR(uint32_t rtcp_ssrc);
/**
* @brief 创建xr的dlrr包，用于接收者估算rtt
*
* @return toolkit::Buffer::Ptr
*/
virtual toolkit::Buffer::Ptr createRtcpXRDLRR(uint32_t rtcp_ssrc, uint32_t rtp_ssrc);
/**
* 创建RR rtcp包
* @param rtcp_ssrc rtcp的ssrc
@ -86,8 +93,11 @@ protected:
class RtcpContextForSend : public RtcpContext {
public:
toolkit::Buffer::Ptr createRtcpSR(uint32_t rtcp_ssrc) override;
void onRtcp(RtcpHeader *rtcp) override;
toolkit::Buffer::Ptr createRtcpXRDLRR(uint32_t rtcp_ssrc, uint32_t rtp_ssrc) override;
/**
* 获取rtt
* @param ssrc rtp ssrc
@ -98,6 +108,9 @@ public:
private:
std::map<uint32_t /*ssrc*/, uint32_t /*rtt*/> _rtt;
std::map<uint32_t /*last_sr_lsr*/, uint64_t /*ntp stamp*/> _sender_report_ntp;
std::map<uint32_t /*ssrc*/, uint64_t /*xr rrtr sys stamp*/> _xr_rrtr_recv_sys_stamp;
std::map<uint32_t /*ssrc*/, uint32_t /*last rr */> _xr_xrrtr_recv_last_rr;
};
class RtcpContextForRecv : public RtcpContext {

16
src/Rtcp/RtcpFCI.cpp
View File

@ -465,9 +465,11 @@ FCI_TWCC::TwccPacketStatus FCI_TWCC::getPacketChunkList(size_t total_size) const
string FCI_TWCC::dumpString(size_t total_size) const {
_StrPrinter printer;
auto map = getPacketChunkList(total_size);
printer << "twcc fci, base_seq:" << getBaseSeq() << ", pkt_status_count:" << getPacketCount() << ", ref time:" << getReferenceTime() << ", fb count:" << (int)fb_pkt_count << "\n";
printer << "twcc fci, base_seq:" << getBaseSeq() << ", pkt_status_count:" << getPacketCount()
<< ", ref time:" << getReferenceTime() << ", fb count:" << (int)fb_pkt_count << "\n";
for (auto &pr : map) {
printer << "rtp seq:" << pr.first <<", packet status:" << (int)(pr.second.first) << ", delta:" << pr.second.second << "\n";
printer << "rtp seq:" << pr.first << ", packet status:" << (int)(pr.second.first)
<< ", delta:" << pr.second.second << "\n";
}
return std::move(printer);
}
@ -476,10 +478,14 @@ static void appendDeltaString(string &delta_str, FCI_TWCC::TwccPacketStatus &sta
for (auto it = status.begin(); it != status.end() && count--;) {
switch (it->second.first) {
// large delta模式先写高字节，再写低字节
case SymbolStatus::large_delta: delta_str.push_back((it->second.second >> 8) & 0xFF);
case SymbolStatus::large_delta:
delta_str.push_back((it->second.second >> 8) & 0xFF);
// small delta模式只写低字节
case SymbolStatus::small_delta: delta_str.push_back(it->second.second & 0xFF); break;
default: break;
case SymbolStatus::small_delta:
delta_str.push_back(it->second.second & 0xFF);
break;
default:
break;
}
// 移除已经处理过的数据
it = status.erase(it);

5
src/Rtcp/RtcpFCI.h
View File

@ -11,8 +11,8 @@
#ifndef ZLMEDIAKIT_RTCPFCI_H
#define ZLMEDIAKIT_RTCPFCI_H
#include "Rtcp.h"
#include "Common/config.h"
#include "Rtcp.h"
namespace mediakit {
@ -354,7 +354,8 @@ enum class SymbolStatus : uint8_t{
class FCI_TWCC {
public:
static size_t constexpr kSize = 8;
using TwccPacketStatus = std::map<uint16_t/*rtp ext seq*/, std::pair<SymbolStatus, int16_t/*recv delta,单位为250us*/> >;
using TwccPacketStatus
= std::map<uint16_t /*rtp ext seq*/, std::pair<SymbolStatus, int16_t /*recv delta,单位为250us*/>>;
void check(size_t size);
std::string dumpString(size_t total_size) const;
uint16_t getBaseSeq() const;

151
webrtc/WebRtcTransport.cpp
View File

@ -9,11 +9,11 @@
*/
#include "WebRtcTransport.h"
#include <iostream>
#include "RtpExt.h"
#include "Rtcp/Rtcp.h"
#include "Rtcp/RtcpFCI.h"
#include "RtpExt.h"
#include "Rtsp/RtpReceiver.h"
#include <iostream>
#define RTP_SSRC_OFFSET 1
#define RTX_SSRC_OFFSET 2
@ -92,7 +92,8 @@ const string &WebRtcTransport::getIdentifier() const {
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void WebRtcTransport::OnIceServerSendStunPacket(const RTC::IceServer *iceServer, const RTC::StunPacket *packet, RTC::TransportTuple *tuple) {
void WebRtcTransport::OnIceServerSendStunPacket(
const RTC::IceServer *iceServer, const RTC::StunPacket *packet, RTC::TransportTuple *tuple) {
sendSockData((char *)packet->GetData(), packet->GetSize(), tuple);
}
@ -120,16 +121,13 @@ void WebRtcTransport::OnIceServerDisconnected(const RTC::IceServer *iceServer) {
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void WebRtcTransport::OnDtlsTransportConnected(
const RTC::DtlsTransport *dtlsTransport,
RTC::SrtpSession::CryptoSuite srtpCryptoSuite,
uint8_t *srtpLocalKey,
size_t srtpLocalKeyLen,
uint8_t *srtpRemoteKey,
size_t srtpRemoteKeyLen,
std::string &remoteCert) {
const RTC::DtlsTransport *dtlsTransport, RTC::SrtpSession::CryptoSuite srtpCryptoSuite, uint8_t *srtpLocalKey,
size_t srtpLocalKeyLen, uint8_t *srtpRemoteKey, size_t srtpRemoteKeyLen, std::string &remoteCert) {
InfoL;
_srtp_session_send = std::make_shared<RTC::SrtpSession>(RTC::SrtpSession::Type::OUTBOUND, srtpCryptoSuite, srtpLocalKey, srtpLocalKeyLen);
_srtp_session_recv = std::make_shared<RTC::SrtpSession>(RTC::SrtpSession::Type::INBOUND, srtpCryptoSuite, srtpRemoteKey, srtpRemoteKeyLen);
_srtp_session_send = std::make_shared<RTC::SrtpSession>(
RTC::SrtpSession::Type::OUTBOUND, srtpCryptoSuite, srtpLocalKey, srtpLocalKeyLen);
_srtp_session_recv = std::make_shared<RTC::SrtpSession>(
RTC::SrtpSession::Type::INBOUND, srtpCryptoSuite, srtpRemoteKey, srtpRemoteKeyLen);
#ifdef ENABLE_SCTP
_sctp = std::make_shared<RTC::SctpAssociationImp>(getPoller(), this, 128, 128, 262144, true);
_sctp->TransportConnected();
@ -137,7 +135,8 @@ void WebRtcTransport::OnDtlsTransportConnected(
onStartWebRTC();
}
void WebRtcTransport::OnDtlsTransportSendData(const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) {
void WebRtcTransport::OnDtlsTransportSendData(
const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) {
sendSockData((char *)data, len, nullptr);
}
@ -155,7 +154,8 @@ void WebRtcTransport::OnDtlsTransportClosed(const RTC::DtlsTransport *dtlsTransp
onShutdown(SockException(Err_shutdown, "dtls close notify received"));
}
void WebRtcTransport::OnDtlsTransportApplicationDataReceived(const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) {
void WebRtcTransport::OnDtlsTransportApplicationDataReceived(
const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) {
#ifdef ENABLE_SCTP
_sctp->ProcessSctpData(data, len);
#else
@ -181,12 +181,13 @@ void WebRtcTransport::OnSctpAssociationClosed(RTC::SctpAssociation* sctpAssociat
InfoL << getIdentifier();
}
void WebRtcTransport::OnSctpAssociationSendData(RTC::SctpAssociation* sctpAssociation, const uint8_t* data, size_t len) {
void WebRtcTransport::OnSctpAssociationSendData(
RTC::SctpAssociation *sctpAssociation, const uint8_t *data, size_t len) {
_dtls_transport->SendApplicationData(data, len);
}
void WebRtcTransport::OnSctpAssociationMessageReceived(RTC::SctpAssociation *sctpAssociation, uint16_t streamId,
uint32_t ppid, const uint8_t *msg, size_t len) {
void WebRtcTransport::OnSctpAssociationMessageReceived(
RTC::SctpAssociation *sctpAssociation, uint16_t streamId, uint32_t ppid, const uint8_t *msg, size_t len) {
InfoL << getIdentifier() << " " << streamId << " " << ppid << " " << len << " " << string((char *)msg, len);
RTC::SctpStreamParameters params;
params.streamId = streamId;
@ -234,7 +235,8 @@ string getFingerprint(const string &algorithm_str, const std::shared_ptr<RTC::Dt
void WebRtcTransport::setRemoteDtlsFingerprint(const RtcSession &remote) {
// 设置远端dtls签名
RTC::DtlsTransport::Fingerprint remote_fingerprint;
remote_fingerprint.algorithm = RTC::DtlsTransport::GetFingerprintAlgorithm(_offer_sdp->media[0].fingerprint.algorithm);
remote_fingerprint.algorithm
= RTC::DtlsTransport::GetFingerprintAlgorithm(_offer_sdp->media[0].fingerprint.algorithm);
remote_fingerprint.value = _offer_sdp->media[0].fingerprint.hash;
_dtls_transport->SetRemoteFingerprint(remote_fingerprint);
}
@ -259,8 +261,8 @@ std::string WebRtcTransport::getAnswerSdp(const string &offer){
fingerprint.algorithm = _offer_sdp->media[0].fingerprint.algorithm;
fingerprint.hash = getFingerprint(fingerprint.algorithm, _dtls_transport);
RtcConfigure configure;
configure.setDefaultSetting(_ice_server->GetUsernameFragment(), _ice_server->GetPassword(),
RtpDirection::sendrecv, fingerprint);
configure.setDefaultSetting(
_ice_server->GetUsernameFragment(), _ice_server->GetPassword(), RtpDirection::sendrecv, fingerprint);
onRtcConfigure(configure);
//// 生成answer sdp ////
@ -364,7 +366,9 @@ void WebRtcTransportImp::onCreate(){
weak_ptr<WebRtcTransportImp> weak_self = static_pointer_cast<WebRtcTransportImp>(shared_from_this());
GET_CONFIG(float, timeoutSec, RTC::kTimeOutSec);
_timer = std::make_shared<Timer>(timeoutSec / 2, [weak_self]() {
_timer = std::make_shared<Timer>(
timeoutSec / 2,
[weak_self]() {
auto strong_self = weak_self.lock();
if (!strong_self) {
return false;
@ -373,14 +377,14 @@ void WebRtcTransportImp::onCreate(){
strong_self->onShutdown(SockException(Err_timeout, "接受rtp和rtcp超时"));
}
return true;
}, getPoller());
},
getPoller());
_twcc_ctx.setOnSendTwccCB([this](uint32_t ssrc, string fci) {
onSendTwcc(ssrc, fci);
});
_twcc_ctx.setOnSendTwccCB([this](uint32_t ssrc, string fci) { onSendTwcc(ssrc, fci); });
}
WebRtcTransportImp::WebRtcTransportImp(const EventPoller::Ptr &poller) : WebRtcTransport(poller) {
WebRtcTransportImp::WebRtcTransportImp(const EventPoller::Ptr &poller)
: WebRtcTransport(poller) {
InfoL << getIdentifier();
}
@ -455,7 +459,8 @@ void WebRtcTransportImp::onStartWebRTC() {
_ssrc_to_track[track->offer_ssrc_rtx] = track;
// rtp pt --> MediaTrack
_pt_to_track.emplace(track->plan_rtp->pt, std::unique_ptr<WrappedMediaTrack>(new WrappedRtpTrack(track, _twcc_ctx, *this)));
_pt_to_track.emplace(
track->plan_rtp->pt, std::unique_ptr<WrappedMediaTrack>(new WrappedRtpTrack(track, _twcc_ctx, *this)));
if (track->plan_rtx) {
// rtx pt --> MediaTrack
_pt_to_track.emplace(track->plan_rtx->pt, std::unique_ptr<WrappedMediaTrack>(new WrappedRtxTrack(track)));
@ -546,14 +551,19 @@ void WebRtcTransportImp::onCheckAnswer(RtcSession &sdp) {
void WebRtcTransportImp::onCheckSdp(SdpType type, RtcSession &sdp) {
switch (type) {
case SdpType::answer: onCheckAnswer(sdp); break;
case SdpType::offer: break;
default: /*不可达*/ assert(0); break;
case SdpType::answer:
onCheckAnswer(sdp);
break;
case SdpType::offer:
break;
default: /*不可达*/
assert(0);
break;
}
}
SdpAttrCandidate::Ptr makeIceCandidate(std::string ip, uint16_t port,
uint32_t priority = 100, std::string proto = "udp") {
SdpAttrCandidate::Ptr
makeIceCandidate(std::string ip, uint16_t port, uint32_t priority = 100, std::string proto = "udp") {
auto candidate = std::make_shared<SdpAttrCandidate>();
// rtp端口
candidate->component = 1;
@ -593,19 +603,18 @@ void WebRtcTransportImp::onRtcConfigure(RtcConfigure &configure) const {
}
}
///////////////////////////////////////////////////////////////////
class RtpChannel : public RtpTrackImp, public std::enable_shared_from_this<RtpChannel> {
class RtpChannel
: public RtpTrackImp
, public std::enable_shared_from_this<RtpChannel> {
public:
RtpChannel(EventPoller::Ptr poller, RtpTrackImp::OnSorted cb, function<void(const FCI_NACK &nack)> on_nack) {
_poller = std::move(poller);
_on_nack = std::move(on_nack);
setOnSorted(std::move(cb));
_nack_ctx.setOnNack([this](const FCI_NACK &nack) {
onNack(nack);
});
_nack_ctx.setOnNack([this](const FCI_NACK &nack) { onNack(nack); });
}
~RtpChannel() override = default;
@ -770,11 +779,30 @@ void WebRtcTransportImp::onRtcp(const char *buf, size_t len) {
});
break;
}
default: break;
default:
break;
}
break;
}
default: break;
case RtcpType::RTCP_XR: {
RtcpXRRRTR *xr = (RtcpXRRRTR *)rtcp;
if (xr->bt != 4) {
break;
}
auto it = _ssrc_to_track.find(xr->ssrc);
if (it == _ssrc_to_track.end()) {
WarnL << "未识别的 rtcp包:" << rtcp->dumpString();
return;
}
auto &track = it->second;
track->rtcp_context_send->onRtcp(rtcp);
auto xrdlrr = track->rtcp_context_send->createRtcpXRDLRR(track->answer_ssrc_rtp, track->answer_ssrc_rtp);
sendRtcpPacket(xrdlrr->data(), xrdlrr->size(), true);
break;
}
default:
break;
}
}
}
@ -785,9 +813,9 @@ void WebRtcTransportImp::createRtpChannel(const string &rid, uint32_t ssrc, Medi
// rid --> RtpReceiverImp
auto &ref = track.rtp_channel[rid];
weak_ptr<WebRtcTransportImp> weak_self = dynamic_pointer_cast<WebRtcTransportImp>(shared_from_this());
ref = std::make_shared<RtpChannel>(getPoller(), [&track, this, rid](RtpPacket::Ptr rtp) mutable {
onSortedRtp(track, rid, std::move(rtp));
}, [&track, weak_self, ssrc](const FCI_NACK &nack) mutable {
ref = std::make_shared<RtpChannel>(
getPoller(), [&track, this, rid](RtpPacket::Ptr rtp) mutable { onSortedRtp(track, rid, std::move(rtp)); },
[&track, weak_self, ssrc](const FCI_NACK &nack) mutable {
// nack发送可能由定时器异步触发
auto strong_self = weak_self.lock();
if (strong_self) {
@ -851,7 +879,8 @@ void WrappedRtxTrack::inputRtp(const char *buf, size_t len, uint64_t stamp_ms, R
auto &ref = track->rtp_channel[rid];
if (!ref) {
// 再接收到对应的rtp前，丢弃rtx包
WarnL << "unknown rtx rtp, rid:" << rid << ", ssrc:" << ntohl(rtp->ssrc) << ", codec:" << track->plan_rtp->codec << ", seq:" << ntohs(rtp->seq);
WarnL << "unknown rtx rtp, rid:" << rid << ", ssrc:" << ntohl(rtp->ssrc) << ", codec:" << track->plan_rtp->codec
<< ", seq:" << ntohs(rtp->seq);
return;
}
@ -918,7 +947,9 @@ void WebRtcTransportImp::onSendRtp(const RtpPacket::Ptr &rtp, bool flush, bool r
}
if (!rtx) {
// 统计rtp发送情况，好做sr汇报
track->rtcp_context_send->onRtp(rtp->getSeq(), rtp->getStamp(), rtp->ntp_stamp, rtp->sample_rate, rtp->size() - RtpPacket::kRtpTcpHeaderSize);
track->rtcp_context_send->onRtp(
rtp->getSeq(), rtp->getStamp(), rtp->ntp_stamp, rtp->sample_rate,
rtp->size() - RtpPacket::kRtpTcpHeaderSize);
track->nack_list.pushBack(rtp);
#if 0
//此处模拟发送丢包
@ -988,8 +1019,9 @@ void WebRtcTransportImp::onShutdown(const SockException &ex){
void WebRtcTransportImp::setSession(Session::Ptr session) {
_history_sessions.emplace(session.get(), session);
if (_selected_session) {
InfoL << "rtc network changed: " << _selected_session->get_peer_ip() << ":" << _selected_session->get_peer_port()
<< " -> " << session->get_peer_ip() << ":" << session->get_peer_port() << ", id:" << getIdentifier();
InfoL << "rtc network changed: " << _selected_session->get_peer_ip() << ":"
<< _selected_session->get_peer_port() << " -> " << session->get_peer_ip() << ":"
<< session->get_peer_port() << ", id:" << getIdentifier();
}
_selected_session = std::move(session);
unrefSelf();
@ -1062,8 +1094,8 @@ void WebRtcPluginManager::registerPlugin(const string &type, Plugin cb) {
_map_creator[type] = std::move(cb);
}
void WebRtcPluginManager::getAnswerSdp(Session &sender, const string &type, const string &offer, const WebRtcArgs &args,
const onCreateRtc &cb) {
void WebRtcPluginManager::getAnswerSdp(
Session &sender, const string &type, const string &offer, const WebRtcArgs &args, const onCreateRtc &cb) {
lock_guard<mutex> lck(_mtx_creator);
auto it = _map_creator.find(type);
if (it == _map_creator.end()) {
@ -1073,17 +1105,20 @@ void WebRtcPluginManager::getAnswerSdp(Session &sender, const string &type, cons
it->second(sender, offer, args, cb);
}
#include "WebRtcEchoTest.h"
#include "WebRtcPlayer.h"
#include "WebRtcPusher.h"
#include "WebRtcEchoTest.h"
void echo_plugin(Session &sender, const string &offer, const WebRtcArgs &args, const WebRtcPluginManager::onCreateRtc &cb) {
void echo_plugin(
Session &sender, const string &offer, const WebRtcArgs &args, const WebRtcPluginManager::onCreateRtc &cb) {
cb(*WebRtcEchoTest::create(EventPollerPool::Instance().getPoller()));
}
void push_plugin(Session &sender, const string &offer_sdp, const WebRtcArgs &args, const WebRtcPluginManager::onCreateRtc &cb) {
void push_plugin(
Session &sender, const string &offer_sdp, const WebRtcArgs &args, const WebRtcPluginManager::onCreateRtc &cb) {
MediaInfo info(args["url"]);
Broadcast::PublishAuthInvoker invoker = [cb, offer_sdp, info](const string &err, const ProtocolOption &option) mutable {
Broadcast::PublishAuthInvoker invoker = [cb, offer_sdp,
info](const string &err, const ProtocolOption &option) mutable {
if (!err.empty()) {
cb(WebRtcException(SockException(Err_other, err)));
return;
@ -1122,20 +1157,23 @@ void push_plugin(Session &sender, const string &offer_sdp, const WebRtcArgs &arg
push_src_ownership = push_src->getOwnership();
push_src->setProtocolOption(option);
}
auto rtc = WebRtcPusher::create(EventPollerPool::Instance().getPoller(), push_src, push_src_ownership, info, option);
auto rtc
= WebRtcPusher::create(EventPollerPool::Instance().getPoller(), push_src, push_src_ownership, info, option);
push_src->setListener(rtc);
cb(*rtc);
};
// rtsp推流需要鉴权
auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPublish, MediaOriginType::rtc_push, info, invoker, static_cast<SockInfo &>(sender));
auto flag = NoticeCenter::Instance().emitEvent(
Broadcast::kBroadcastMediaPublish, MediaOriginType::rtc_push, info, invoker, static_cast<SockInfo &>(sender));
if (!flag) {
// 该事件无人监听,默认不鉴权
invoker("", ProtocolOption());
}
}
void play_plugin(Session &sender, const string &offer_sdp, const WebRtcArgs &args, const WebRtcPluginManager::onCreateRtc &cb) {
void play_plugin(
Session &sender, const string &offer_sdp, const WebRtcArgs &args, const WebRtcPluginManager::onCreateRtc &cb) {
MediaInfo info(args["url"]);
auto session_ptr = sender.shared_from_this();
Broadcast::AuthInvoker invoker = [cb, offer_sdp, info, session_ptr](const string &err) mutable {
@ -1160,7 +1198,8 @@ void play_plugin(Session &sender, const string &offer_sdp, const WebRtcArgs &arg
};
// 广播通用播放url鉴权事件
auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed, info, invoker, static_cast<SockInfo &>(sender));
auto flag = NoticeCenter::Instance().emitEvent(
Broadcast::kBroadcastMediaPlayed, info, invoker, static_cast<SockInfo &>(sender));
if (!flag) {
// 该事件无人监听,默认不鉴权
invoker("");