兼容GB28181 tcp推流缓存覆盖的垃圾设备

src/Rtp/RtpSession.cpp

@@ -12,6 +12,7 @@
 #include "RtpSession.h"
 #include "RtpSelector.h"
 #include "Network/TcpServer.h"
+#include "Rtsp/RtpReceiver.h"
 namespace mediakit{
 
 const string RtpSession::kStreamID = "stream_id";
@@ -57,23 +58,40 @@ void RtpSession::onManager() {
 }
 
 void RtpSession::onRtpPacket(const char *data, size_t len) {
+    if (_search_rtp) {
+        //搜索上下文期间，数据丢弃
+        if (_search_rtp_finished) {
+            //下个包开始就是正确的rtp包了
+            _search_rtp_finished = false;
+            _search_rtp = false;
+        }
+        return;
+    }
     if (len > 1024 * 10) {
-        throw SockException(Err_shutdown, StrPrinter << "rtp包长度异常(" << len << ")，发送端可能缓存溢出并覆盖");
+        _search_rtp = true;
+        WarnL << "rtp包长度异常(" << len << ")，发送端可能缓存溢出并覆盖，开始搜索ssrc以便恢复上下文";
+        return;
     }
     if (!_process) {
-        uint32_t ssrc;
-        if (!RtpSelector::getSSRC(data, len, ssrc)) {
+        if (!RtpSelector::getSSRC(data, len, _ssrc)) {
             return;
         }
         if (_stream_id.empty()) {
             //未指定流id就使用ssrc为流id
-            _stream_id = printSSRC(ssrc);
+            _stream_id = printSSRC(_ssrc);
         }
         //tcp情况下，一个tcp链接只可能是一路流，不需要通过多个ssrc来区分，所以不需要频繁getProcess
         _process = RtpSelector::Instance().getProcess(_stream_id, true);
         _process->setListener(dynamic_pointer_cast<RtpSession>(shared_from_this()));
     }
-    _process->inputRtp(false, getSock(), data, len, &addr);
+    try {
+        _process->inputRtp(false, getSock(), data, len, &addr);
+    } catch (RtpReceiver::BadRtpException &ex) {
+        WarnL << ex.what() << "，开始搜索ssrc以便恢复上下文";
+        _search_rtp = true;
+    } catch (...) {
+        throw;
+    }
     _ticker.resetTime();
 }
 
@@ -92,5 +110,57 @@ int RtpSession::totalReaderCount(MediaSource &sender) {
     return _process ? _process->getTotalReaderCount() : sender.totalReaderCount();
 }
 
+static const char *findSSRC(const char *data, ssize_t len, uint32_t ssrc) {
+    //rtp前面必须预留两个字节的长度字段
+    for (ssize_t i = 2; i <= len - 4; ++i) {
+        auto ptr = (const uint8_t *) data + i;
+        if (ptr[0] == (ssrc >> 24) && ptr[1] == ((ssrc >> 16) & 0xFF) &&
+            ptr[2] == ((ssrc >> 8) & 0xFF) && ptr[3] == (ssrc & 0xFF)) {
+            return (const char *) ptr;
+        }
+    }
+    return nullptr;
+}
+
+//rtp长度到ssrc间的长度固定为10
+static size_t constexpr kSSRCOffset = 2 + 4 + 4;
+
+const char *RtpSession::onSearchPacketTail(const char *data, size_t len) {
+    if (!_search_rtp) {
+        //tcp上下文正常，不用搜索ssrc
+        return RtpSplitter::onSearchPacketTail(data, len);
+    }
+    if (!_process) {
+        throw SockException(Err_shutdown, "ssrc未获取到，无法通过ssrc恢复tcp上下文");
+    }
+    //搜索第一个rtp的ssrc
+    auto ssrc_ptr0 = findSSRC(data, len, _ssrc);
+    if (!ssrc_ptr0) {
+        //未搜索到任意rtp，返回数据不够
+        return nullptr;
+    }
+    //这两个字节是第一个rtp的长度字段
+    auto rtp_len_ptr = (ssrc_ptr0 - kSSRCOffset);
+    auto rtp_len = ((uint8_t *)rtp_len_ptr)[0] << 8 | ((uint8_t *)rtp_len_ptr)[1];
+
+    //搜索第二个rtp的ssrc
+    auto ssrc_ptr1 = findSSRC(ssrc_ptr0 + rtp_len, data + (ssize_t) len - ssrc_ptr0 - rtp_len, _ssrc);
+    if (!ssrc_ptr1) {
+        //未搜索到第二个rtp，返回数据不够
+        return nullptr;
+    }
+
+    //两个ssrc的间隔正好等于rtp的长度(外加rtp长度字段)，那么说明找到rtp
+    auto ssrc_offset = ssrc_ptr1 - ssrc_ptr0;
+    if (ssrc_offset == rtp_len + 2 || ssrc_offset == rtp_len + 4) {
+        InfoL << "rtp搜索成功，tcp上下文恢复成功，丢弃的rtp残余数据为：" << rtp_len_ptr - data;
+        _search_rtp_finished = true;
+        //前面的数据都需要丢弃，这个是rtp的起始
+        return rtp_len_ptr;
+    }
+    //第一个rtp长度不匹配，说明第一个找到的ssrc不是rtp，丢弃之，我们从第二个ssrc所在rtp开始搜索
+    return ssrc_ptr1 - kSSRCOffset;
+}
+
 }//namespace mediakit
 #endif//defined(ENABLE_RTPPROXY)

src/Rtp/RtpSession.h

@@ -38,7 +38,12 @@ protected:
     // 收到rtp回调
     void onRtpPacket(const char *data, size_t len) override;
 
+    const char *onSearchPacketTail(const char *data, size_t len) override;
+
 private:
+    bool _search_rtp = false;
+    bool _search_rtp_finished = false;
+    uint32_t _ssrc = 0;
     Ticker _ticker;
     string _stream_id;
     struct sockaddr addr;

src/Rtsp/RtpReceiver.cpp

@@ -42,7 +42,7 @@ bool RtpReceiver::handleOneRtp(int index, TrackType type, int sample_rate, uint8
     }
     RtpHeader *header = (RtpHeader *) ptr;
     if (header->version != RtpPacket::kRtpVersion) {
-        throw std::invalid_argument("非法的rtp，version字段非法");
+        throw BadRtpException("非法的rtp，version字段非法");
     }
     if (!header->getPayloadSize(len)) {
         //无有效负载的rtp包

src/Rtsp/RtpReceiver.h

@@ -165,6 +165,13 @@ public:
     RtpReceiver();
     virtual ~RtpReceiver();
 
+    class BadRtpException : public invalid_argument {
+    public:
+        template<typename Type>
+        BadRtpException(Type &&type) : invalid_argument(std::forward<Type>(type)) {}
+        ~BadRtpException() = default;
+    };
+
 protected:
     /**
      * 输入数据指针生成并排序rtp包