diff --git a/src/Http/HttpRequestSplitter.h b/src/Http/HttpRequestSplitter.h index efe1ec90..6e77a7c9 100644 --- a/src/Http/HttpRequestSplitter.h +++ b/src/Http/HttpRequestSplitter.h @@ -27,8 +27,9 @@ public: * 添加数据 * @param data 需要添加的数据 * @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: /** diff --git a/src/Rtp/Decoder.cpp b/src/Rtp/Decoder.cpp index 5c011ae8..68243552 100644 --- a/src/Rtp/Decoder.cpp +++ b/src/Rtp/Decoder.cpp @@ -203,7 +203,8 @@ void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t d } default: - if (codecid != 0) { + // 海康的 PS 流中会有 codecid 为 0xBD 的包 + if (codecid != 0 && codecid != 0xBD) { if (_last_unsported_print.elapsedTime() / 1000 > 5) { _last_unsported_print.resetTime(); WarnL << "unsupported codec type:" << getCodecName(codecid) << " " << (int) codecid; diff --git a/src/Rtp/GB28181Process.cpp b/src/Rtp/GB28181Process.cpp index 42f004d2..c3fb9942 100644 --- a/src/Rtp/GB28181Process.cpp +++ b/src/Rtp/GB28181Process.cpp @@ -150,27 +150,6 @@ bool GB28181Process::inputRtp(bool, const char *data, size_t 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) { if (frame->getCodecId() != CodecInvalid) { //这里不是ps或ts @@ -197,7 +176,7 @@ void GB28181Process::onRtpDecode(const Frame::Ptr &frame) { } if (_decoder) { - HttpRequestSplitter::input(frame->data(), frame->size()); + _decoder->input(reinterpret_cast(frame->data()), frame->size()); } } diff --git a/src/Rtp/GB28181Process.h b/src/Rtp/GB28181Process.h index 4d5d06d4..e1a60d5c 100644 --- a/src/Rtp/GB28181Process.h +++ b/src/Rtp/GB28181Process.h @@ -22,7 +22,7 @@ namespace mediakit{ class RtpReceiverImp; -class GB28181Process : public HttpRequestSplitter, public ProcessInterface{ +class GB28181Process : public ProcessInterface { public: typedef std::shared_ptr Ptr; GB28181Process(const MediaInfo &media_info, MediaSinkInterface *interface); @@ -38,8 +38,6 @@ public: protected: 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: void onRtpDecode(const Frame::Ptr &frame); diff --git a/src/Rtp/PSDecoder.cpp b/src/Rtp/PSDecoder.cpp index 7854e1af..448f1809 100644 --- a/src/Rtp/PSDecoder.cpp +++ b/src/Rtp/PSDecoder.cpp @@ -45,7 +45,8 @@ PSDecoder::~PSDecoder() { } 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(data), bytes); + return bytes; } void PSDecoder::setOnDecode(Decoder::onDecode cb) { @@ -56,5 +57,28 @@ void PSDecoder::setOnStream(Decoder::onStream cb) { _on_stream = std::move(cb); } +const char *PSDecoder::onSearchPacketTail(const char *data, size_t len) { + try { + auto ret = ps_demuxer_input(static_cast(_ps_demuxer), reinterpret_cast(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 #endif//#if defined(ENABLE_RTPPROXY) \ No newline at end of file diff --git a/src/Rtp/PSDecoder.h b/src/Rtp/PSDecoder.h index 270f929a..6cb5b40e 100644 --- a/src/Rtp/PSDecoder.h +++ b/src/Rtp/PSDecoder.h @@ -14,17 +14,26 @@ #if defined(ENABLE_RTPPROXY) #include #include "Decoder.h" +#include "Http/HttpRequestSplitter.h" + namespace mediakit{ //ps解析器 -class PSDecoder : public Decoder { +class PSDecoder : public Decoder, private HttpRequestSplitter { public: PSDecoder(); ~PSDecoder(); + ssize_t input(const uint8_t* data, size_t bytes) override; void setOnDecode(onDecode 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: void *_ps_demuxer = nullptr; onDecode _on_decode; diff --git a/src/Rtsp/RtpReceiver.h b/src/Rtsp/RtpReceiver.h index fda96de6..64794376 100644 --- a/src/Rtsp/RtpReceiver.h +++ b/src/Rtsp/RtpReceiver.h @@ -37,7 +37,7 @@ public: */ void clear() { _seq_cycle_count = 0; - _rtp_sort_cache_map.clear(); + _pkt_sort_cache_map.clear(); _next_seq_out = 0; _max_sort_size = kMin; } @@ -46,7 +46,7 @@ public: * 获取排序缓存长度 */ size_t getJitterSize() const{ - return _rtp_sort_cache_map.size(); + return _pkt_sort_cache_map.size(); } /** @@ -67,27 +67,27 @@ public: //过滤seq回退包(回环包除外) return; } - } else if (_next_seq_out && seq - _next_seq_out > (0xFFFF >> 1)) { + } else if (_next_seq_out && seq - _next_seq_out > ((std::numeric_limits::max)() >> 1)) { //过滤seq跳变非常大的包(防止回环时乱序时收到非常大的seq) return; } //放入排序缓存 - _rtp_sort_cache_map.emplace(seq, std::move(packet)); + _pkt_sort_cache_map.emplace(seq, std::move(packet)); //尝试输出排序后的包 tryPopPacket(); } void flush(){ //清空缓存 - while (!_rtp_sort_cache_map.empty()) { - popIterator(_rtp_sort_cache_map.begin()); + while (!_pkt_sort_cache_map.empty()) { + popIterator(_pkt_sort_cache_map.begin()); } } private: void popPacket() { - auto it = _rtp_sort_cache_map.begin(); + auto it = _pkt_sort_cache_map.begin(); if (it->first >= _next_seq_out) { //过滤回跳包 popIterator(it); @@ -96,37 +96,37 @@ private: 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 return; } ++_seq_cycle_count; //找到大的SEQ并清空掉,然后从小的SEQ重新开始排序 - auto hit = _rtp_sort_cache_map.upper_bound((SEQ) (_next_seq_out - _rtp_sort_cache_map.size())); - while (hit != _rtp_sort_cache_map.end()) { + auto hit = _pkt_sort_cache_map.upper_bound((SEQ) (_next_seq_out - _pkt_sort_cache_map.size())); + while (hit != _pkt_sort_cache_map.end()) { //回环前,清空剩余的大的SEQ的数据 _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; } //删除回跳的数据包 - _rtp_sort_cache_map.erase(it); + _pkt_sort_cache_map.erase(it); } void popIterator(typename map::iterator it) { auto seq = it->first; auto data = std::move(it->second); - _rtp_sort_cache_map.erase(it); + _pkt_sort_cache_map.erase(it); _next_seq_out = seq + 1; _cb(seq, data); } void tryPopPacket() { 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(); ++count; @@ -134,7 +134,7 @@ private: if (count) { setSortSize(); - } else if (_rtp_sort_cache_map.size() > _max_sort_size) { + } else if (_pkt_sort_cache_map.size() > _max_sort_size) { //排序缓存溢出,不再继续排序 popPacket(); setSortSize(); @@ -142,7 +142,7 @@ private: } 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) { _max_sort_size = kMax; } @@ -155,8 +155,8 @@ private: size_t _seq_cycle_count = 0; //排序缓存长度 size_t _max_sort_size = kMin; - //rtp排序缓存,根据seq排序 - map _rtp_sort_cache_map; + //pkt排序缓存,根据seq排序 + map _pkt_sort_cache_map; //回调 function _cb; };