PSDecoder: 缓存未处理完的数据

src/Rtp/PSDecoder.cpp

@@ -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)
\ No newline at end of file

src/Rtp/PSDecoder.h

@@ -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;