403 lines
13 KiB
C++
403 lines
13 KiB
C++
/*
|
||
* Copyright (c) 2016-present 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-like 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_RTPRECEIVER_H
|
||
#define ZLMEDIAKIT_RTPRECEIVER_H
|
||
|
||
#include <map>
|
||
#include <string>
|
||
#include <memory>
|
||
#include "Rtsp/Rtsp.h"
|
||
#include "Extension/Frame.h"
|
||
// for NtpStamp
|
||
#include "Common/Stamp.h"
|
||
#include "Util/TimeTicker.h"
|
||
|
||
namespace mediakit {
|
||
|
||
template<typename T, typename SEQ = uint16_t>
|
||
class PacketSortor {
|
||
public:
|
||
static constexpr SEQ SEQ_MAX = (std::numeric_limits<SEQ>::max)();
|
||
using iterator = typename std::map<SEQ, T>::iterator;
|
||
|
||
virtual ~PacketSortor() = default;
|
||
|
||
void setOnSort(std::function<void(SEQ seq, T packet)> cb) { _cb = std::move(cb); }
|
||
|
||
/**
|
||
* 清空状态
|
||
* Clear the state
|
||
|
||
* [AUTO-TRANSLATED:6aadbd77]
|
||
*/
|
||
void clear() {
|
||
_started = false;
|
||
_ticker.resetTime();
|
||
_pkt_sort_cache_map.clear();
|
||
}
|
||
|
||
/**
|
||
* 获取排序缓存长度
|
||
* Get the length of the sorting cache
|
||
|
||
* [AUTO-TRANSLATED:8e05a703]
|
||
*/
|
||
size_t getJitterSize() const { return _pkt_sort_cache_map.size(); }
|
||
|
||
/**
|
||
* 输入并排序
|
||
* @param seq 序列号
|
||
* @param packet 包负载
|
||
* Input and sort
|
||
* @param seq Sequence number
|
||
* @param packet Packet payload
|
||
|
||
* [AUTO-TRANSLATED:0fbf096e]
|
||
*/
|
||
void sortPacket(SEQ seq, T packet) {
|
||
_latest_seq = seq;
|
||
if (!_started) {
|
||
// 记录第一个seq [AUTO-TRANSLATED:410c831f]
|
||
// Record the first seq
|
||
_started = true;
|
||
_last_seq_out = seq - 1;
|
||
}
|
||
auto next_seq = static_cast<SEQ>(_last_seq_out + 1);
|
||
if (seq == next_seq) {
|
||
// 收到下一个seq [AUTO-TRANSLATED:44960fea]
|
||
// Receive the next seq
|
||
output(seq, std::move(packet));
|
||
// 清空连续包列表 [AUTO-TRANSLATED:fdaafd3b]
|
||
// Clear the continuous packet list
|
||
flushPacket();
|
||
_pkt_drop_cache_map.clear();
|
||
return;
|
||
}
|
||
|
||
if (seq < next_seq && !mayLooped(next_seq, seq)) {
|
||
// 无回环风险, 缓存seq回退包 [AUTO-TRANSLATED:4200dd1b]
|
||
// No loop risk, cache seq rollback packets
|
||
_pkt_drop_cache_map.emplace(seq, std::move(packet));
|
||
if (_pkt_drop_cache_map.size() > _max_distance || _ticker.elapsedTime() > _max_buffer_ms) {
|
||
// seq回退包太多，可能源端重置seq计数器，这部分数据需要输出 [AUTO-TRANSLATED:d31aead7]
|
||
// Too many seq rollback packets, the source may reset the seq counter, this part of data needs to be output
|
||
forceFlush(next_seq);
|
||
// 旧的seq计数器的数据清空后把新seq计数器的数据赋值给排序列队 [AUTO-TRANSLATED:f69f864c]
|
||
// After clearing the data of the old seq counter, assign the data of the new seq counter to the sorting queue
|
||
_pkt_sort_cache_map = std::move(_pkt_drop_cache_map);
|
||
popIterator(_pkt_sort_cache_map.begin());
|
||
}
|
||
return;
|
||
}
|
||
_pkt_sort_cache_map.emplace(seq, std::move(packet));
|
||
|
||
if (needForceFlush(seq)) {
|
||
forceFlush(next_seq);
|
||
}
|
||
}
|
||
|
||
void flush() {
|
||
if (!_pkt_sort_cache_map.empty()) {
|
||
forceFlush(static_cast<SEQ>(_last_seq_out + 1));
|
||
_pkt_sort_cache_map.clear();
|
||
}
|
||
}
|
||
|
||
void setParams(size_t max_buffer_size, size_t max_buffer_ms, size_t max_distance) {
|
||
_max_buffer_size = max_buffer_size;
|
||
_max_buffer_ms = max_buffer_ms;
|
||
_max_distance = max_distance;
|
||
}
|
||
|
||
private:
|
||
SEQ distance(SEQ seq) {
|
||
SEQ ret;
|
||
auto next_seq = static_cast<SEQ>(_last_seq_out + 1);
|
||
if (seq > next_seq) {
|
||
ret = seq - next_seq;
|
||
} else {
|
||
ret = next_seq - seq;
|
||
}
|
||
if (ret > SEQ_MAX >> 1) {
|
||
return SEQ_MAX - ret;
|
||
}
|
||
return ret;
|
||
}
|
||
|
||
bool needForceFlush(SEQ seq) {
|
||
return _pkt_sort_cache_map.size() > _max_buffer_size || distance(seq) > _max_distance || _ticker.elapsedTime() > _max_buffer_ms;
|
||
}
|
||
|
||
void forceFlush(SEQ next_seq) {
|
||
if (_pkt_sort_cache_map.empty()) {
|
||
return;
|
||
}
|
||
// 寻找距离比next_seq大的最近的seq [AUTO-TRANSLATED:d2de6f5b]
|
||
// Find the nearest seq that is greater than next_seq
|
||
auto it = _pkt_sort_cache_map.lower_bound(next_seq);
|
||
if (it == _pkt_sort_cache_map.end()) {
|
||
// 没有比next_seq更大的seq，应该是回环时丢包导致 [AUTO-TRANSLATED:d0d6970b]
|
||
// There is no seq greater than next_seq, it should be caused by packet loss during loopback
|
||
it = _pkt_sort_cache_map.begin();
|
||
}
|
||
// 丢包无法恢复，把这个包当做next_seq [AUTO-TRANSLATED:2d8c0b9e]
|
||
// Packet loss cannot be recovered, treat this packet as next_seq
|
||
popIterator(it);
|
||
// 清空连续包列表 [AUTO-TRANSLATED:fdaafd3b]
|
||
// Clear the continuous packet list
|
||
flushPacket();
|
||
// 删除距离next_seq太大的包 [AUTO-TRANSLATED:9e774c5e]
|
||
// Delete packets that are too far away from next_seq
|
||
for (auto it = _pkt_sort_cache_map.begin(); it != _pkt_sort_cache_map.end();) {
|
||
if (distance(it->first) > _max_distance) {
|
||
it = _pkt_sort_cache_map.erase(it);
|
||
} else {
|
||
++it;
|
||
}
|
||
}
|
||
}
|
||
|
||
bool mayLooped(SEQ last_seq, SEQ now_seq) { return last_seq > SEQ_MAX - _max_distance || now_seq < _max_distance; }
|
||
|
||
void flushPacket() {
|
||
if (_pkt_sort_cache_map.empty()) {
|
||
return;
|
||
}
|
||
auto next_seq = static_cast<SEQ>(_last_seq_out + 1);
|
||
auto it = _pkt_sort_cache_map.lower_bound(next_seq);
|
||
if (!mayLooped(next_seq, next_seq)) {
|
||
// 无回环风险, 清空 < next_seq的值 [AUTO-TRANSLATED:10c77bf9]
|
||
// No loop risk, clear values less than next_seq
|
||
it = _pkt_sort_cache_map.erase(_pkt_sort_cache_map.begin(), it);
|
||
}
|
||
|
||
while (it != _pkt_sort_cache_map.end()) {
|
||
// 找到下一个包 [AUTO-TRANSLATED:8e20ab9f]
|
||
// Find the next packet
|
||
if (it->first == static_cast<SEQ>(_last_seq_out + 1)) {
|
||
it = popIterator(it);
|
||
continue;
|
||
}
|
||
break;
|
||
}
|
||
}
|
||
|
||
iterator popIterator(iterator it) {
|
||
output(it->first, std::move(it->second));
|
||
return _pkt_sort_cache_map.erase(it);
|
||
}
|
||
|
||
void output(SEQ seq, T packet) {
|
||
auto next_seq = static_cast<SEQ>(_last_seq_out + 1);
|
||
if (seq != next_seq) {
|
||
WarnL << "packet dropped: " << next_seq << " -> " << static_cast<SEQ>(seq - 1)
|
||
<< ", latest seq: " << _latest_seq
|
||
<< ", jitter buffer size: " << _pkt_sort_cache_map.size()
|
||
<< ", jitter buffer ms: " << _ticker.elapsedTime();
|
||
}
|
||
_last_seq_out = seq;
|
||
_cb(seq, std::move(packet));
|
||
_ticker.resetTime();
|
||
}
|
||
|
||
private:
|
||
bool _started = false;
|
||
// 排序缓存最大保存数据长度，单位毫秒 [AUTO-TRANSLATED:ed217b1c]
|
||
// Maximum data length of sorting cache, unit: milliseconds
|
||
size_t _max_buffer_ms = 1000;
|
||
// 排序缓存最大保存数据个数 [AUTO-TRANSLATED:9cfa91b4]
|
||
// Maximum number of data in sorting cache
|
||
size_t _max_buffer_size = 1024;
|
||
// seq最大跳跃距离 [AUTO-TRANSLATED:bb663e41]
|
||
// Maximum seq jump distance
|
||
size_t _max_distance = 256;
|
||
// 记录上次output至今的时间 [AUTO-TRANSLATED:83e53e42]
|
||
// Record the time since the last output
|
||
toolkit::Ticker _ticker;
|
||
// 最近输入的seq [AUTO-TRANSLATED:24ca96ee]
|
||
// The most recently input seq
|
||
SEQ _latest_seq = 0;
|
||
// 下次应该输出的SEQ [AUTO-TRANSLATED:e757a4fa]
|
||
// The next SEQ to be output
|
||
SEQ _last_seq_out = 0;
|
||
// pkt排序缓存，根据seq排序 [AUTO-TRANSLATED:3787f9a6]
|
||
// pkt sorting cache, sorted by seq
|
||
std::map<SEQ, T> _pkt_sort_cache_map;
|
||
// 预丢弃包列表 [AUTO-TRANSLATED:67e57ebc]
|
||
// Pre-discard packet list
|
||
std::map<SEQ, T> _pkt_drop_cache_map;
|
||
// 回调 [AUTO-TRANSLATED:03bad27d]
|
||
// Callback
|
||
std::function<void(SEQ seq, T packet)> _cb;
|
||
};
|
||
|
||
class RtpTrack : public PacketSortor<RtpPacket::Ptr> {
|
||
public:
|
||
class BadRtpException : public std::invalid_argument {
|
||
public:
|
||
template<typename Type>
|
||
BadRtpException(Type &&type) : invalid_argument(std::forward<Type>(type)) {}
|
||
};
|
||
|
||
RtpTrack();
|
||
|
||
void clear();
|
||
uint32_t getSSRC() const;
|
||
RtpPacket::Ptr inputRtp(TrackType type, int sample_rate, uint8_t *ptr, size_t len);
|
||
void setNtpStamp(uint32_t rtp_stamp, uint64_t ntp_stamp_ms);
|
||
void setPayloadType(uint8_t pt);
|
||
|
||
protected:
|
||
virtual void onRtpSorted(RtpPacket::Ptr rtp) {}
|
||
virtual void onBeforeRtpSorted(const RtpPacket::Ptr &rtp) {}
|
||
|
||
private:
|
||
bool _disable_ntp = false;
|
||
uint8_t _pt = 0xFF;
|
||
uint32_t _ssrc = 0;
|
||
toolkit::Ticker _ssrc_alive;
|
||
NtpStamp _ntp_stamp;
|
||
};
|
||
|
||
class RtpTrackImp : public RtpTrack{
|
||
public:
|
||
using OnSorted = std::function<void(RtpPacket::Ptr)>;
|
||
using BeforeSorted = std::function<void(const RtpPacket::Ptr &)>;
|
||
|
||
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下标索引
|
||
* @param type track类型
|
||
* @param samplerate rtp时间戳基准时钟，视频为90000，音频为采样率
|
||
* @param ptr rtp数据指针
|
||
* @param len rtp数据指针长度
|
||
* @return 解析成功返回true
|
||
* Generate and sort rtp packets from input data pointer
|
||
* @param index Track index
|
||
* @param type Track type
|
||
* @param samplerate RTP timestamp base clock, 90000 for video, sample rate for audio
|
||
* @param ptr RTP data pointer
|
||
* @param len RTP data pointer length
|
||
* @return Return true if parsing is successful
|
||
|
||
* [AUTO-TRANSLATED:4ec12e4a]
|
||
*/
|
||
bool handleOneRtp(int index, TrackType type, int sample_rate, uint8_t *ptr, size_t len) {
|
||
assert(index < kCount && index >= 0);
|
||
return _track[index].inputRtp(type, sample_rate, ptr, len).operator bool();
|
||
}
|
||
|
||
/**
|
||
* 设置ntp时间戳，在收到rtcp sender report时设置
|
||
* 如果rtp_stamp/sample_rate/ntp_stamp_ms都为0，那么采用rtp时间戳为ntp时间戳
|
||
* @param index track下标索引
|
||
* @param rtp_stamp rtp时间戳
|
||
* @param ntp_stamp_ms ntp时间戳
|
||
* Set ntp timestamp, set when receiving rtcp sender report
|
||
* If rtp_stamp/sample_rate/ntp_stamp_ms are all 0, then use rtp timestamp as ntp timestamp
|
||
* @param index Track index
|
||
* @param rtp_stamp RTP timestamp
|
||
* @param ntp_stamp_ms NTP timestamp
|
||
|
||
* [AUTO-TRANSLATED:1e50904e]
|
||
*/
|
||
void setNtpStamp(int index, uint32_t rtp_stamp, uint64_t ntp_stamp_ms) {
|
||
assert(index < kCount && index >= 0);
|
||
_track[index].setNtpStamp(rtp_stamp, ntp_stamp_ms);
|
||
}
|
||
|
||
void setPayloadType(int index, uint8_t pt){
|
||
assert(index < kCount && index >= 0);
|
||
_track[index].setPayloadType(pt);
|
||
}
|
||
|
||
void clear() {
|
||
for (auto &track : _track) {
|
||
track.clear();
|
||
}
|
||
}
|
||
|
||
size_t getJitterSize(int index) const {
|
||
assert(index < kCount && index >= 0);
|
||
return _track[index].getJitterSize();
|
||
}
|
||
|
||
uint32_t getSSRC(int index) const {
|
||
assert(index < kCount && index >= 0);
|
||
return _track[index].getSSRC();
|
||
}
|
||
|
||
protected:
|
||
/**
|
||
* rtp数据包排序后输出
|
||
* @param rtp rtp数据包
|
||
* @param track_index track索引
|
||
* Output rtp data packets after sorting
|
||
* @param rtp RTP data packet
|
||
* @param track_index Track index
|
||
|
||
* [AUTO-TRANSLATED:55022da9]
|
||
*/
|
||
virtual void onRtpSorted(RtpPacket::Ptr rtp, int index) {}
|
||
|
||
/**
|
||
* 解析出rtp但还未排序
|
||
* @param rtp rtp数据包
|
||
* @param track_index track索引
|
||
* RTP data packet parsed but not yet sorted
|
||
* @param rtp RTP data packet
|
||
* @param track_index Track index
|
||
|
||
* [AUTO-TRANSLATED:c1636911]
|
||
*/
|
||
virtual void onBeforeRtpSorted(const RtpPacket::Ptr &rtp, int index) {}
|
||
|
||
private:
|
||
RtpTrackImp _track[kCount];
|
||
};
|
||
|
||
using RtpReceiver = RtpMultiReceiver<2>;
|
||
|
||
}//namespace mediakit
|
||
|
||
|
||
#endif //ZLMEDIAKIT_RTPRECEIVER_H
|