Merge remote-tracking branch 'upstream/master' into master

src/Record/TsMuxer.cpp

@@ -172,6 +172,7 @@ void TsMuxer::init() {
                 TsMuxer *muxer = (TsMuxer *) param;
                 muxer->onTs(packet, bytes, muxer->_timestamp, muxer->_is_idr_fast_packet);
                 muxer->_is_idr_fast_packet = false;
+                return 0;
             }
     };
     if (_context == nullptr) {

src/Rtp/PSDecoder.cpp

@@ -26,6 +26,7 @@ PSDecoder::PSDecoder() {
         if(thiz->_on_decode){
             thiz->_on_decode(stream, codecid, flags, pts, dts, data, bytes);
         }
+        return 0;
     },this);
 }
 

src/Rtp/PSEncoder.cpp

@@ -38,6 +38,7 @@ void PSEncoder::init() {
             [](void *param, int stream, void *packet, size_t bytes) {
                 PSEncoder *thiz = (PSEncoder *) param;
                 thiz->onPS(thiz->_timestamp, packet, bytes);
+                return 0;
             }
     };
 

src/Rtp/RtpSession.cpp

@@ -57,9 +57,12 @@ void RtpSession::onManager() {
 }
 
 void RtpSession::onRtpPacket(const char *data, uint64_t len) {
+    if (len > 1024 * 2) {
+        throw SockException(Err_shutdown, "rtp包长度异常，发送端可能缓存溢出并覆盖");
+    }
     if (!_process) {
         uint32_t ssrc;
-        if (!RtpSelector::getSSRC(data + 2, len - 2, ssrc)) {
+        if (!RtpSelector::getSSRC(data, len, ssrc)) {
             return;
         }
         if (_stream_id.empty()) {
@@ -70,7 +73,7 @@ void RtpSession::onRtpPacket(const char *data, uint64_t len) {
         _process = RtpSelector::Instance().getProcess(_stream_id, true);
         _process->setListener(dynamic_pointer_cast<RtpSession>(shared_from_this()));
     }
-    _process->inputRtp(getSock(), data + 2, len - 2, &addr);
+    _process->inputRtp(getSock(), data, len, &addr);
     _ticker.resetTime();
 }
 

src/Rtp/RtpSplitter.cpp

@@ -17,20 +17,22 @@ RtpSplitter::RtpSplitter() {}
 RtpSplitter::~RtpSplitter() {}
 
 const char *RtpSplitter::onSearchPacketTail(const char *data, int len) {
+    if (len < 4) {
+        //数据不够
+        return nullptr;
+    }
     if (data[0] == '$') {
         //可能是4个字节的rtp头
+        _offset = 4;
         return onSearchPacketTail_l(data + 2, len - 2);
     }
     //两个字节的rtp头
+    _offset = 2;
     return onSearchPacketTail_l(data, len);
 }
 
 const char *RtpSplitter::onSearchPacketTail_l(const char *data, int len) {
     //这是rtp包
-    if (len < 2) {
-        //数据不够
-        return nullptr;
-    }
     uint16_t length = (((uint8_t *) data)[0] << 8) | ((uint8_t *) data)[1];
     if (len < length + 2) {
         //数据不够
@@ -41,7 +43,7 @@ const char *RtpSplitter::onSearchPacketTail_l(const char *data, int len) {
 }
 
 int64_t RtpSplitter::onRecvHeader(const char *data, uint64_t len) {
-    onRtpPacket(data,len);
+    onRtpPacket(data + _offset, len - _offset);
     return 0;
 }
 

src/Rtp/RtpSplitter.h

@@ -31,6 +31,9 @@ protected:
     const char *onSearchPacketTail(const char *data,int len) override ;
     const char *onSearchPacketTail_l(const char *data,int len);
     int64_t onRecvHeader(const char *data,uint64_t len) override;
+
+private:
+    int _offset = 0;
 };
 
 }//namespace mediakit

src/Rtsp/RtpReceiver.h

@@ -61,33 +61,75 @@ public:
      * @param packet 包负载
      */
     void sortPacket(SEQ seq, T packet) {
-        if (seq < _next_seq_out && _next_seq_out - seq > kMax) {
-            //回环
-            ++_seq_cycle_count;
+        if (seq < _next_seq_out) {
+            if (_next_seq_out - seq < kMax) {
+                //过滤seq回退包(回环包除外)
+                return;
+            }
+        } else if (_next_seq_out && seq - _next_seq_out > (0xFFFF >> 1)) {
+            //过滤seq跳变非常大的包(防止回环时乱序时收到非常大的seq)
+            return;
         }
+
         //放入排序缓存
         _rtp_sort_cache_map.emplace(seq, std::move(packet));
         //尝试输出排序后的包
         tryPopPacket();
     }
 
+    void flush(){
+        //清空缓存
+        while (!_rtp_sort_cache_map.empty()) {
+            popIterator(_rtp_sort_cache_map.begin());
+        }
+    }
+
 private:
     void popPacket() {
         auto it = _rtp_sort_cache_map.begin();
+        if (it->first >= _next_seq_out) {
+            //过滤回跳包
+            popIterator(it);
+            return;
+        }
+
+        if (_next_seq_out - it->first > (0xFFFF >> 1)) {
+            //产生回环了
+            if (_rtp_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()) {
+                //回环前，清空剩余的大的SEQ的数据
+                _cb(hit->first, hit->second);
+                hit = _rtp_sort_cache_map.erase(hit);
+            }
+            //下一个回环的数据
+            popIterator(_rtp_sort_cache_map.begin());
+            return;
+        }
+        //删除回跳的数据包
+        _rtp_sort_cache_map.erase(it);
+    }
+
+    void popIterator(typename map<SEQ, T>::iterator it) {
         _cb(it->first, it->second);
         _next_seq_out = it->first + 1;
         _rtp_sort_cache_map.erase(it);
     }
 
     void tryPopPacket() {
-        bool flag = false;
+        int count = 0;
         while ((!_rtp_sort_cache_map.empty() && _rtp_sort_cache_map.begin()->first == _next_seq_out)) {
             //找到下个包，直接输出
             popPacket();
-            flag = true;
+            ++count;
         }
 
-        if (flag) {
+        if (count) {
             setSortSize();
         } else if (_rtp_sort_cache_map.size() > _max_sort_size) {
             //排序缓存溢出，不再继续排序
@@ -97,11 +139,9 @@ private:
     }
 
     void setSortSize() {
-        _max_sort_size = 2 * _rtp_sort_cache_map.size();
+        _max_sort_size = kMin + _rtp_sort_cache_map.size();
         if (_max_sort_size > kMax) {
             _max_sort_size = kMax;
-        } else if (_max_sort_size < kMin) {
-            _max_sort_size = kMin;
         }
     }