From e7e8969b4fef9b7f4a25329dc29e04a535825d4f Mon Sep 17 00:00:00 2001
From: xiongziliang <771730766@qq.com>
Date: Sat, 8 Aug 2020 12:13:35 +0800
Subject: [PATCH] =?UTF-8?q?ps=E8=A7=A3=E6=9E=90=E6=B7=BB=E5=8A=A0=E5=A4=96?=
 =?UTF-8?q?=E9=83=A8=E7=BC=93=E5=AD=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/Rtp/RtpProcess.cpp | 22 +++++++++++++++++-----
 src/Rtp/RtpProcess.h   |  6 +++++-
 2 files changed, 22 insertions(+), 6 deletions(-)

diff --git a/src/Rtp/RtpProcess.cpp b/src/Rtp/RtpProcess.cpp
index 3b9a689d..a7092eef 100644
--- a/src/Rtp/RtpProcess.cpp
+++ b/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);
     }
 }
 
diff --git a/src/Rtp/RtpProcess.h b/src/Rtp/RtpProcess.h
index d2d9c5ee..b819169d 100644
--- a/src/Rtp/RtpProcess.h
+++ b/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();