ps解析添加外部缓存

src/Rtp/RtpProcess.cpp

@@ -111,7 +111,7 @@ static inline bool checkTS(const uint8_t *packet, int bytes){
 }
 
 void RtpProcess::onRtpSorted(const RtpPacket::Ptr &rtp, int) {
-    if(rtp->sequence != _sequence + (uint16_t)1 && _sequence != 0){
+    if(rtp->sequence != (uint16_t)(_sequence + 1) && _sequence != 0){
         WarnP(this) << "rtp丢包:" << rtp->sequence << " != " << _sequence << "+1" << ",公网环境下请使用tcp方式推流";
     }
     _sequence = rtp->sequence;
@@ -124,6 +124,21 @@ void RtpProcess::onRtpSorted(const RtpPacket::Ptr &rtp, int) {
     decodeRtp(rtp->data() + 4 ,rtp->size() - 4);
 }
 
+const char *RtpProcess::onSearchPacketTail(const char *packet,int bytes){
+    try {
+        auto ret = _decoder->input((uint8_t *) packet, bytes);
+        if (ret > 0) {
+            return packet + ret;
+        }
+        return nullptr;
+    } catch (std::exception &ex) {
+        InfoL << "解析ps或ts异常: bytes=" << bytes
+              << " ,exception=" << ex.what()
+              << " ,hex=" << hexdump((uint8_t *) packet, bytes);
+        return nullptr;
+    }
+}
+
 void RtpProcess::onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestamp, int flags) {
     if(_save_file_ps){
         fwrite((uint8_t *)packet,bytes, 1, _save_file_ps.get());
@@ -143,10 +158,7 @@ void RtpProcess::onRtpDecode(const uint8_t *packet, int bytes, uint32_t timestam
     }
 
     if (_decoder) {
-        auto ret = _decoder->input((uint8_t *) packet, bytes);
-        if (ret != bytes) {
-            WarnP(this) << ret << " != " << bytes << " " << flags;
-        }
+        HttpRequestSplitter::input((char *) packet, bytes);
     }
 }
 

src/Rtp/RtpProcess.h

@@ -18,11 +18,12 @@
 #include "Decoder.h"
 #include "Common/Device.h"
 #include "Common/Stamp.h"
+#include "Http/HttpRequestSplitter.h"
 using namespace mediakit;
 
 namespace mediakit{
 
-class RtpProcess : public RtpReceiver , public RtpDecoder, public SockInfo, public MediaSinkInterface, public std::enable_shared_from_this<RtpProcess>{
+class RtpProcess : public HttpRequestSplitter, public RtpReceiver , public RtpDecoder, public SockInfo, public MediaSinkInterface, public std::enable_shared_from_this<RtpProcess>{
 public:
     typedef std::shared_ptr<RtpProcess> Ptr;
     RtpProcess(const string &stream_id);
@@ -71,6 +72,9 @@ protected:
     void addTrack(const Track::Ptr & track) override;
     void resetTracks() override {};
 
+    const char *onSearchPacketTail(const char *data,int len) override;
+    int64_t onRecvHeader(const char *data,uint64_t len) override { return 0; };
+
 private:
     void emitOnPublish();