Merge remote-tracking branch 'origin/master'

This commit is contained in:
ziyue 2021-07-20 12:43:28 +08:00
commit 46722546a6
7 changed files with 60 additions and 48 deletions

View File

@ -27,8 +27,9 @@ public:
* *
* @param data * @param data
* @param len * @param len
* @warning len + 1, 使 strstr , , @p len + 1 '\0' .
*/ */
virtual void input(const char *data,size_t len); virtual void input(const char *data, size_t len);
protected: protected:
/** /**

View File

@ -203,7 +203,8 @@ void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t d
} }
default: default:
if (codecid != 0) { // 海康的 PS 流中会有 codecid 为 0xBD 的包
if (codecid != 0 && codecid != 0xBD) {
if (_last_unsported_print.elapsedTime() / 1000 > 5) { if (_last_unsported_print.elapsedTime() / 1000 > 5) {
_last_unsported_print.resetTime(); _last_unsported_print.resetTime();
WarnL << "unsupported codec type:" << getCodecName(codecid) << " " << (int) codecid; WarnL << "unsupported codec type:" << getCodecName(codecid) << " " << (int) codecid;

View File

@ -150,27 +150,6 @@ bool GB28181Process::inputRtp(bool, const char *data, size_t data_len) {
return ref->inputRtp(TrackVideo, (unsigned char *) data, data_len); return ref->inputRtp(TrackVideo, (unsigned char *) data, data_len);
} }
const char *GB28181Process::onSearchPacketTail(const char *packet,size_t bytes){
try {
auto ret = _decoder->input((uint8_t *) packet, bytes);
if (ret >= 0) {
//解析成功全部或部分
return packet + ret;
}
//解析失败,丢弃所有数据
return packet + bytes;
} catch (std::exception &ex) {
InfoL << "解析ps或ts异常: bytes=" << bytes
<< " ,exception=" << ex.what()
<< " ,hex=" << hexdump((uint8_t *) packet, MIN(bytes,32));
if (remainDataSize() > 256 * 1024) {
//缓存太多数据无法处理则上抛异常
throw;
}
return nullptr;
}
}
void GB28181Process::onRtpDecode(const Frame::Ptr &frame) { void GB28181Process::onRtpDecode(const Frame::Ptr &frame) {
if (frame->getCodecId() != CodecInvalid) { if (frame->getCodecId() != CodecInvalid) {
//这里不是ps或ts //这里不是ps或ts
@ -197,7 +176,7 @@ void GB28181Process::onRtpDecode(const Frame::Ptr &frame) {
} }
if (_decoder) { if (_decoder) {
HttpRequestSplitter::input(frame->data(), frame->size()); _decoder->input(reinterpret_cast<const uint8_t *>(frame->data()), frame->size());
} }
} }

View File

@ -22,7 +22,7 @@
namespace mediakit{ namespace mediakit{
class RtpReceiverImp; class RtpReceiverImp;
class GB28181Process : public HttpRequestSplitter, public ProcessInterface{ class GB28181Process : public ProcessInterface {
public: public:
typedef std::shared_ptr<GB28181Process> Ptr; typedef std::shared_ptr<GB28181Process> Ptr;
GB28181Process(const MediaInfo &media_info, MediaSinkInterface *interface); GB28181Process(const MediaInfo &media_info, MediaSinkInterface *interface);
@ -38,8 +38,6 @@ public:
protected: protected:
void onRtpSorted(RtpPacket::Ptr rtp); void onRtpSorted(RtpPacket::Ptr rtp);
const char *onSearchPacketTail(const char *data,size_t len) override;
ssize_t onRecvHeader(const char *data,size_t len) override { return 0; };
private: private:
void onRtpDecode(const Frame::Ptr &frame); void onRtpDecode(const Frame::Ptr &frame);

View File

@ -45,7 +45,8 @@ PSDecoder::~PSDecoder() {
} }
ssize_t PSDecoder::input(const uint8_t *data, size_t bytes) { ssize_t PSDecoder::input(const uint8_t *data, size_t bytes) {
return ps_demuxer_input((struct ps_demuxer_t*)_ps_demuxer,data,bytes); HttpRequestSplitter::input(reinterpret_cast<const char *>(data), bytes);
return bytes;
} }
void PSDecoder::setOnDecode(Decoder::onDecode cb) { void PSDecoder::setOnDecode(Decoder::onDecode cb) {
@ -56,5 +57,28 @@ void PSDecoder::setOnStream(Decoder::onStream cb) {
_on_stream = std::move(cb); _on_stream = std::move(cb);
} }
const char *PSDecoder::onSearchPacketTail(const char *data, size_t len) {
try {
auto ret = ps_demuxer_input(static_cast<struct ps_demuxer_t *>(_ps_demuxer), reinterpret_cast<const uint8_t *>(data), len);
if (ret >= 0) {
//解析成功全部或部分
return data + ret;
}
//解析失败,丢弃所有数据
return data + len;
} catch (std::exception &ex) {
InfoL << "解析 ps 异常: bytes=" << len
<< ", exception=" << ex.what()
<< ", hex=" << hexdump(data, MIN(len, 32));
if (remainDataSize() > 256 * 1024) {
//缓存太多数据无法处理则上抛异常
throw;
}
return nullptr;
}
}
}//namespace mediakit }//namespace mediakit
#endif//#if defined(ENABLE_RTPPROXY) #endif//#if defined(ENABLE_RTPPROXY)

View File

@ -14,17 +14,26 @@
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
#include <stdint.h> #include <stdint.h>
#include "Decoder.h" #include "Decoder.h"
#include "Http/HttpRequestSplitter.h"
namespace mediakit{ namespace mediakit{
//ps解析器 //ps解析器
class PSDecoder : public Decoder { class PSDecoder : public Decoder, private HttpRequestSplitter {
public: public:
PSDecoder(); PSDecoder();
~PSDecoder(); ~PSDecoder();
ssize_t input(const uint8_t* data, size_t bytes) override; ssize_t input(const uint8_t* data, size_t bytes) override;
void setOnDecode(onDecode cb) override; void setOnDecode(onDecode cb) override;
void setOnStream(onStream cb) override; void setOnStream(onStream cb) override;
// HttpRequestSplitter interface
private:
using HttpRequestSplitter::input;
const char *onSearchPacketTail(const char *data, size_t len) override;
ssize_t onRecvHeader(const char *, size_t) override { return 0; };
private: private:
void *_ps_demuxer = nullptr; void *_ps_demuxer = nullptr;
onDecode _on_decode; onDecode _on_decode;

View File

@ -37,7 +37,7 @@ public:
*/ */
void clear() { void clear() {
_seq_cycle_count = 0; _seq_cycle_count = 0;
_rtp_sort_cache_map.clear(); _pkt_sort_cache_map.clear();
_next_seq_out = 0; _next_seq_out = 0;
_max_sort_size = kMin; _max_sort_size = kMin;
} }
@ -46,7 +46,7 @@ public:
* *
*/ */
size_t getJitterSize() const{ size_t getJitterSize() const{
return _rtp_sort_cache_map.size(); return _pkt_sort_cache_map.size();
} }
/** /**
@ -67,27 +67,27 @@ public:
//过滤seq回退包(回环包除外) //过滤seq回退包(回环包除外)
return; return;
} }
} else if (_next_seq_out && seq - _next_seq_out > (0xFFFF >> 1)) { } else if (_next_seq_out && seq - _next_seq_out > ((std::numeric_limits<SEQ>::max)() >> 1)) {
//过滤seq跳变非常大的包(防止回环时乱序时收到非常大的seq) //过滤seq跳变非常大的包(防止回环时乱序时收到非常大的seq)
return; return;
} }
//放入排序缓存 //放入排序缓存
_rtp_sort_cache_map.emplace(seq, std::move(packet)); _pkt_sort_cache_map.emplace(seq, std::move(packet));
//尝试输出排序后的包 //尝试输出排序后的包
tryPopPacket(); tryPopPacket();
} }
void flush(){ void flush(){
//清空缓存 //清空缓存
while (!_rtp_sort_cache_map.empty()) { while (!_pkt_sort_cache_map.empty()) {
popIterator(_rtp_sort_cache_map.begin()); popIterator(_pkt_sort_cache_map.begin());
} }
} }
private: private:
void popPacket() { void popPacket() {
auto it = _rtp_sort_cache_map.begin(); auto it = _pkt_sort_cache_map.begin();
if (it->first >= _next_seq_out) { if (it->first >= _next_seq_out) {
//过滤回跳包 //过滤回跳包
popIterator(it); popIterator(it);
@ -96,37 +96,37 @@ private:
if (_next_seq_out - it->first > (0xFFFF >> 1)) { if (_next_seq_out - it->first > (0xFFFF >> 1)) {
//产生回环了 //产生回环了
if (_rtp_sort_cache_map.size() < 2 * kMin) { if (_pkt_sort_cache_map.size() < 2 * kMin) {
//等足够多的数据后才处理回环, 因为后面还可能出现大的SEQ //等足够多的数据后才处理回环, 因为后面还可能出现大的SEQ
return; return;
} }
++_seq_cycle_count; ++_seq_cycle_count;
//找到大的SEQ并清空掉然后从小的SEQ重新开始排序 //找到大的SEQ并清空掉然后从小的SEQ重新开始排序
auto hit = _rtp_sort_cache_map.upper_bound((SEQ) (_next_seq_out - _rtp_sort_cache_map.size())); auto hit = _pkt_sort_cache_map.upper_bound((SEQ) (_next_seq_out - _pkt_sort_cache_map.size()));
while (hit != _rtp_sort_cache_map.end()) { while (hit != _pkt_sort_cache_map.end()) {
//回环前清空剩余的大的SEQ的数据 //回环前清空剩余的大的SEQ的数据
_cb(hit->first, hit->second); _cb(hit->first, hit->second);
hit = _rtp_sort_cache_map.erase(hit); hit = _pkt_sort_cache_map.erase(hit);
} }
//下一个回环的数据 //下一个回环的数据
popIterator(_rtp_sort_cache_map.begin()); popIterator(_pkt_sort_cache_map.begin());
return; return;
} }
//删除回跳的数据包 //删除回跳的数据包
_rtp_sort_cache_map.erase(it); _pkt_sort_cache_map.erase(it);
} }
void popIterator(typename map<SEQ, T>::iterator it) { void popIterator(typename map<SEQ, T>::iterator it) {
auto seq = it->first; auto seq = it->first;
auto data = std::move(it->second); auto data = std::move(it->second);
_rtp_sort_cache_map.erase(it); _pkt_sort_cache_map.erase(it);
_next_seq_out = seq + 1; _next_seq_out = seq + 1;
_cb(seq, data); _cb(seq, data);
} }
void tryPopPacket() { void tryPopPacket() {
int count = 0; int count = 0;
while ((!_rtp_sort_cache_map.empty() && _rtp_sort_cache_map.begin()->first == _next_seq_out)) { while ((!_pkt_sort_cache_map.empty() && _pkt_sort_cache_map.begin()->first == _next_seq_out)) {
//找到下个包,直接输出 //找到下个包,直接输出
popPacket(); popPacket();
++count; ++count;
@ -134,7 +134,7 @@ private:
if (count) { if (count) {
setSortSize(); setSortSize();
} else if (_rtp_sort_cache_map.size() > _max_sort_size) { } else if (_pkt_sort_cache_map.size() > _max_sort_size) {
//排序缓存溢出,不再继续排序 //排序缓存溢出,不再继续排序
popPacket(); popPacket();
setSortSize(); setSortSize();
@ -142,7 +142,7 @@ private:
} }
void setSortSize() { void setSortSize() {
_max_sort_size = kMin + _rtp_sort_cache_map.size(); _max_sort_size = kMin + _pkt_sort_cache_map.size();
if (_max_sort_size > kMax) { if (_max_sort_size > kMax) {
_max_sort_size = kMax; _max_sort_size = kMax;
} }
@ -155,8 +155,8 @@ private:
size_t _seq_cycle_count = 0; size_t _seq_cycle_count = 0;
//排序缓存长度 //排序缓存长度
size_t _max_sort_size = kMin; size_t _max_sort_size = kMin;
//rtp排序缓存根据seq排序 //pkt排序缓存根据seq排序
map<SEQ, T> _rtp_sort_cache_map; map<SEQ, T> _pkt_sort_cache_map;
//回调 //回调
function<void(SEQ seq, T &packet)> _cb; function<void(SEQ seq, T &packet)> _cb;
}; };