feat parse ps

.gitignore

@@ -1,4 +1,6 @@
-
+docs/
+build*
+.gdb_history
 .vscode/
 
 build/

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required(VERSION 3.9)
 project(
   SecMedia
-  VERSION 0.0.1
+  VERSION 0.0.1 LANGUAGES C CXX
   DESCRIPTION "Security Media Package")
 set(CMAKE_CXX_STANDARD 11)
 
@@ -83,7 +83,8 @@ file(
   ${SecMedia_Root}/*/*.c
   ${SecMedia_Root}/*/*/*.cpp
   ${SecMedia_Root}/*/*/*.h
-  ${SecMedia_Root}/*/*/*.c)
+  ${SecMedia_Root}/*/*/*.c
+)
 file(GLOB SecMedia_api_list ${CMAKE_CURRENT_SOURCE_DIR}/include/common.h)
 
 # # target_compile_options(${PROJECT_NAME} PRIVATE -fvisibility=hidden)

PcapSender/main.cpp

@@ -1,14 +1,14 @@
-#include <iostream>
+#include "HuaWei/HWsec.h"
+#include "HuaWei/HWsign.h"
 #include <Packet.h>
 #include <PcapFileDevice.h>
 #include <UdpLayer.h>
-#include <string>
 #include <arpa/inet.h>
+#include <iostream>
 #include <stdio.h>
-#include <sys/socket.h> 
+#include <string>
 #include <sys/select.h>
-#include "HuaWei/HWsign.h"
+#include <sys/socket.h>
-#include "HuaWei/HWsec.h"
 
 using namespace std;
 using namespace pcpp;
@@ -18,27 +18,28 @@ sec_set_info sign_info={
     "2021-07-06T17:27:19.000",
     32,
     64,
-            {
+    {0x24, 0x88, 0xc8, 0xdc, 0x7f, 0xd7, 0xe0, 0x91, 0x30, 0x1b, 0x5c, 0x58, 0x2f, 0xe7, 0x44, 0x7d,
-                0x24,0x88,0xc8,0xdc,0x7f,0xd7,0xe0,0x91,0x30,0x1b,0x5c,0x58,0x2f,0xe7,0x44,0x7d,
+     0x2f, 0x43, 0xe4, 0xee, 0xc8, 0x7d, 0xc0, 0xfb, 0xa4, 0xb8, 0x7d, 0x4b, 0x8a, 0x69, 0x7c, 0x4e},
-                0x2f,0x43,0xe4,0xee,0xc8,0x7d,0xc0,0xfb,0xa4,0xb8,0x7d,0x4b,0x8a,0x69,0x7c,0x4e
-            },
     {
         0xaa, 0xb1, 0x3f, 0xd7, 0x66, 0xe2, 0x75, 0x97, 0xc0, 0x03, 0xe6, 0xe4, 0x1d, 0x77, 0x54, 0x78,
         0xc8, 0x29, 0xb2, 0x0b, 0x9e, 0xd1, 0xff, 0xa3, 0x6a, 0x6f, 0xd2, 0x7f, 0xd6, 0x2d, 0xaa, 0x3f,
         0xc9, 0x24, 0xec, 0x6c, 0x96, 0x0a, 0x7b, 0x73, 0xf6, 0xe6, 0xfc, 0xda, 0x3a, 0x08, 0xfd, 0x92,
         0xfc, 0x00, 0x08, 0x97, 0x78, 0x2c, 0x71, 0x6b, 0xe1, 0x26, 0xf5, 0x1e, 0xba, 0x31, 0xf5, 0xb2,
-            }
+    }};
-        };
+
-void * EncrypInit(){
+void *
+EncrypInit()
+{
     // auto Verify_handle=HWVerify_init();
 
     auto sign_handle = GB28181_stream_init(&sign_info);//HWSign_init(&sign_info);
     return sign_handle;
 }
-std::string getProtocolTypeAsString(pcpp::ProtocolType protocolType)
+
-{
+std::string
-    switch (protocolType)
+getProtocolTypeAsString(pcpp::ProtocolType protocolType)
 {
+    switch (protocolType) {
     case pcpp::Ethernet:
         return "Ethernet";
     case pcpp::IPv4:
@@ -53,15 +54,19 @@ std::string getProtocolTypeAsString(pcpp::ProtocolType protocolType)
     default:
         return "Unknown";
     }
-
 }
 
-inline void sleep(timeval & delta){
+inline void
+sleep(timeval &delta)
+{
     // delta.tv_sec
     select(0, NULL, NULL, NULL, &delta);
 }
+
 template<typename T, typename U>
-timespec TimeDiff(T && minu,U && sub){
+timespec
+TimeDiff(T &&minu, U &&sub)
+{
     timespec deltatime;
     deltatime.tv_nsec = minu.tv_nsec - sub.tv_nsec;
     deltatime.tv_sec = minu.tv_sec - sub.tv_sec;
@@ -72,24 +77,23 @@ timespec TimeDiff(T && minu,U && sub){
     return move(deltatime);
 }
 
-int ReadPcapAndSend(int socket,sockaddr_in & addr,const string & filename,const string & filter,void * sign_handle){
+int
+ReadPcapAndSend(int socket, sockaddr_in &addr, const string &filename, const string &filter, void *sign_handle)
+{
     // auto sign_h2=EncrypInit();
     auto reader = pcpp::IFileReaderDevice::getReader(filename);
     // verify that a reader interface was indeed created
-    if (reader == NULL)
+    if (reader == NULL) {
-    {
         std::cerr << "Cannot determine reader for file type" << std::endl;
         return 1;
     }
     // open the reader for reading
-    if (!reader->open())
+    if (!reader->open()) {
-    {
         std::cerr << "Cannot open input.pcap for reading" << std::endl;
         return 1;
     }
 
-    if (!reader->setFilter(filter))
+    if (!reader->setFilter(filter)) {
-    {
         std::cerr << "Cannot set filter for file reader" << std::endl;
         return 1;
     }
@@ -105,8 +109,7 @@ int ReadPcapAndSend(int socket,sockaddr_in & addr,const string & filename,const
     unsigned int sign_out_len;
     void *param = nullptr;
     uint16_t offset_len, append_len;
-    while (reader->getNextPacket(rawPacket))
+    while (reader->getNextPacket(rawPacket)) {
-    {
         pcpp::Packet parsedPacket(&rawPacket, OsiModelTransportLayer);
         auto layer = parsedPacket.getLayerOfType<UdpLayer>(false);
 
@@ -123,8 +126,7 @@ int ReadPcapAndSend(int socket,sockaddr_in & addr,const string & filename,const
             current_time.tv_sec = inital_time.tv_sec + real_now.tv_sec;
             current_time.tv_nsec = inital_time.tv_nsec + real_now.tv_nsec;
             gaptime = TimeDiff(nowtime, current_time);
-            if (gaptime.tv_nsec>=0 && gaptime.tv_sec>=0)
+            if (gaptime.tv_nsec >= 0 && gaptime.tv_sec >= 0) {
-            {
                 nanosleep(&gaptime, NULL);
             } else {
                 // cout<<" s:" << gaptime.tv_sec<<" ns:" << gaptime.tv_nsec <<endl;
@@ -134,33 +136,26 @@ int ReadPcapAndSend(int socket,sockaddr_in & addr,const string & filename,const
 
                 GB28181_stream_in(sign_handle, (char *) payload, payload_len, nullptr);
                 GB28181_stream_out(sign_handle, sign_out_buf, &sign_out_len, &offset_len, &append_len, &param);
-                if (append_len==0)
+                if (append_len == 0) {
-                {
                     sendto(socket, sign_out_buf, sign_out_len, 0, (const sockaddr *) &addr, sizeof(addr));
-                }else
+                } else {
-                {
                     sendto(socket, sign_out_buf, offset_len, 0, (const sockaddr *) &addr, sizeof(addr));
-                    sendto(socket,sign_out_buf+offset_len,sign_out_len-offset_len,0,(const sockaddr*)&addr,sizeof(addr));
+                    sendto(socket, sign_out_buf + offset_len, sign_out_len - offset_len, 0, (const sockaddr *) &addr,
+                           sizeof(addr));
                 }
                 // GB28181_stream_in(sign_h2,(char*)payload,payload_len,nullptr);
                 // GB28181_stream_out(sign_h2,sign_out_buf,&sign_out_len,&offset_len,&append_len, &param);
 
-                
-                
             } else {
                 if (sendto(socket, payload, payload_len, 0, (const sockaddr *) &addr, sizeof(addr)) == -1) {
                     printf("send failed : %s\n", strerror(errno));
                 }
             }
-            
-
         }
 
-
         // cout<<" stamp:" << rawPacket.getPacketTimeStamp().tv_sec<<" len:"<< rawPacket.getRawDataLen()<<endl;
     }
 
-
     // reader->getNextPacket(rawPacket);
 
     // for (pcpp::Layer* curLayer = parsedPacket.getFirstLayer(); curLayer != NULL; curLayer = curLayer->getNextLayer())
@@ -175,20 +170,49 @@ int ReadPcapAndSend(int socket,sockaddr_in & addr,const string & filename,const
     // create the stats object
     pcpp::IPcapDevice::PcapStats stats;
     reader->getStatistics(stats);
-    std::cout << "Read " << stats.packetsRecv << " packets successfully and " << stats.packetsDrop << " packets could not be read" << std::flush;
+    std::cout << "Read " << stats.packetsRecv << " packets successfully and " << stats.packetsDrop
+              << " packets could not be read" << std::flush;
     return 0;
 }
-int main(int argc, char *argv[]){
+
+void
+ShowAuthHelper()
+{
+    struct Config {
+        uint8_t prikey_size;
+        uint8_t pubkey_size;
+        uint8_t prikey[64];
+        uint8_t pubkey[128];
+    } cfg;
+
+    cfg.prikey_size = sign_info.prikey_size;
+    cfg.pubkey_size = sign_info.pubkey_size;
+    memcpy(cfg.prikey, sign_info.prikey, 64);
+    memcpy(cfg.pubkey, sign_info.pubkey, 128);
+    const char *tpl =
+        "http://127.0.0.1/index/api/SetAuthHashCig"
+        "?deviceId=%s"
+        "&KeyVersion=%s"
+        "DecryptType=2"
+        "ConfigData=%s";
+    char out_buf[1024];
+    sprintf(out_buf, tpl, sign_info.camera_id, sign_info.vkek_version, "base64(config)");
+    printf("%s\n", out_buf);
+}
+
+int
+main(int argc, char *argv[])
+{
 
     char ip[16] = {0};
     int sockfd, port;
-    if (argc>4)
+    if (argc > 4) {
-	{
         string filename(argv[1]);
         strcpy(ip, argv[2]);
         port = atoi(argv[3]);
         string filter(argv[4]);
         printf("filename:%s\nip:%s\nport:%d\nfilter:%s\n", filename.data(), ip, port, filter.data());
+        ShowAuthHelper();
 
         sockfd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);//SOCK_STREAM
         if (sockfd == -1) {
@@ -198,13 +222,9 @@ int main(int argc, char *argv[]){
         int rcvBufSize = 2097152;
         int rcvlen = sizeof(rcvBufSize);
         setsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, (char *) &rcvBufSize, sizeof(rcvBufSize));
-        if (getsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, (char*)&rcvBufSize, (socklen_t *)&rcvlen) >= 0)
+        if (getsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, (char *) &rcvBufSize, (socklen_t *) &rcvlen) >= 0) {
-        {
             printf("set udp socket send buff size to : %d\n", rcvBufSize);
-            if (rcvBufSize < 4194304)
+            if (rcvBufSize < 4194304) { printf("socket send buff too small, please set up to 2097152(2M)\n"); }
-            {
-                printf("socket send buff too small, please set up to 2097152(2M)");
-            }
         } else {
             printf("socket failed : %s\n", strerror(errno));
             return -1;
@@ -226,13 +246,10 @@ int main(int argc, char *argv[]){
         ReadPcapAndSend(sockfd, addr, filename, filter, sign_h);
         // }
 
-	}
+    } else {
-	else
-	{
         printf("CMD as: filename.pcap ip port\n");
         system("pause");
         return 0;
     }
     return 0;
-    
 }

doxide.yaml

@@ -0,0 +1,10 @@
+title: AW-Security-Media-Lib
+description: 
+files:
+  - "src/HuaWei/*.h"
+  - "src/GB28181/*.h"
+  - "src/DecEnc/*.h"
+
+defines:
+  API_EXPORT: ""
+

mkdocs.yaml

@@ -0,0 +1,44 @@
+site_name: ASM
+site_description:  
+theme:
+  name: material
+  # custom_dir: docs/overrides
+  features:
+    - navigation.indexes
+  palette:
+    # Palette toggle for light mode
+    - scheme: default
+      primary: red
+      accent: red
+      toggle:
+        icon: material/brightness-7 
+        name: Switch to dark mode
+
+    # Palette toggle for dark mode
+    - scheme: slate
+      primary: red
+      accent: red
+      toggle:
+        icon: material/brightness-4
+        name: Switch to light mode
+
+markdown_extensions:
+  - def_list
+  - attr_list
+  - admonition
+  - pymdownx.details
+  - pymdownx.superfences
+  - pymdownx.arithmatex:
+      generic: true
+  - pymdownx.emoji:
+      emoji_index: !!python/name:material.extensions.emoji.twemoji
+      emoji_generator: !!python/name:material.extensions.emoji.to_svg
+plugins:
+  - search
+extra_css:
+  - stylesheets/doxide.css
+extra_javascript:
+  - javascripts/mathjax.js
+  - https://polyfill.io/v3/polyfill.min.js?features=es6
+  - https://cdn.jsdelivr.net/npm/mathjax@3/es5/tex-mml-chtml.js
+

src/GB28181/L2PSstream.cpp

@@ -1,9 +1,10 @@
 #include "L2PSstream.h"
-#include "L2PSstream.h"
+#include "StreamCodec.h"
 #include "mpeg-ps.h"
 #include "mpeg-ts-proto.h"
-#include "StreamCodec.h"
+
-static inline void nbo_w16(uint8_t* ptr, uint16_t val)
+static inline void
+nbo_w16(uint8_t *ptr, uint16_t val)
 {
     ptr[0] = (uint8_t) ((val >> 8) & 0xFF);
     ptr[1] = (uint8_t) (val & 0xFF);
@@ -11,100 +12,104 @@ static inline void nbo_w16(uint8_t* ptr, uint16_t val)
 
 L2PSstream::L2PSstream(sec_set_info *set_info) : L2UdpStream(set_info)
 {
-    _ps_demuxer = ps_demuxer_create([](void* param,
+    _ps_demuxer = ps_demuxer_create(
-                                       int stream,
+        [](void *param, int stream, int codecid, int flags, int64_t pts, int64_t dts, const void *data, size_t bytes) {
+            L2PSstream *thiz = (L2PSstream *) param;
+            thiz->on_ps_decode(stream, codecid, thiz->_ps_demuxer, flags, pts, dts, data, bytes);
+            return 0;
+        },
+        this);
+}
+
+L2PSstream::~L2PSstream() { ps_demuxer_destroy((struct ps_demuxer_t *) _ps_demuxer); }
+
+void
+L2PSstream::on_rtp_out(const uint8_t *packet, int bytes, uint32_t timestamp, int flags)
+{
+    TRACEL("timestamp:%u, bytes:%u, flag:%d\n", timestamp, bytes, flags);
+    ps_demuxer_input((struct ps_demuxer_t *) _ps_demuxer, packet, bytes);
+};
+
+void
+L2PSstream::on_ps_decode(int stream,
                          int codecid,
+                         void *ps,
                          int flags,
                          int64_t pts,
                          int64_t dts,
                          const void *data,
-                                       size_t bytes){
+                         int bytes)
-        L2PSstream *thiz = (L2PSstream *)param;
-        thiz->on_ps_decode(stream, codecid,thiz->_ps_demuxer, flags, pts, dts, data, bytes);
-        return 0;
-    },this);
-}
-
-L2PSstream::~L2PSstream()
 {
-    ps_demuxer_destroy((struct ps_demuxer_t*)_ps_demuxer);
-}
-
-void L2PSstream::on_rtp_out(const uint8_t *packet, int bytes, uint32_t timestamp, int flags){
-    // TRACEL("timestamp:%u, bytes:%u, flag:%d\n",timestamp,bytes,flags);
-    ps_demuxer_input((struct ps_demuxer_t*)_ps_demuxer,packet,bytes);
-};
-
-void L2PSstream::on_ps_decode(int stream, int codecid, void *ps, int flags, int64_t pts, int64_t dts, const void *data, int bytes){
     // TRACEL("dts:%lu, bytes:%d, codecid:%d\n",dts,bytes,codecid);
     auto prefix = prefixSize((char *) data, bytes);
     pts /= 90;
     dts /= 90;
     pts %= ((int64_t) UINT32_MAX + 1);
     dts %= ((int64_t) UINT32_MAX + 1);
-    switch (codecid)
+    switch (codecid) {
-    {
     case PSI_STREAM_H264:
         _code_id = CodecId::CodecH264;
-        stream_split((char*)data,bytes,prefix,[&,dts](const char *ptr, int len, int prefix){
+        stream_split((char *) data, bytes, prefix,
-            decodeH264(ptr,len,dts,prefix);
+                     [&, dts](const char *ptr, int len, int prefix) { decodeH264(ptr, len, dts, prefix); });
-        });
         break;
 
     case PSI_STREAM_H265:
         _code_id = CodecId::CodecH265;
-        stream_split((char*)data,bytes,prefix,[&](const char *ptr, int len, int prefix){
+        stream_split((char *) data, bytes, prefix,
-            decodeH265(ptr,len,dts,prefix);
+                     [&](const char *ptr, int len, int prefix) { decodeH265(ptr, len, dts, prefix); });
-        });
         break;
     default:
         break;
     }
 }
-bool L2PSstream::splitPES(const uint8_t * data, uint16_t len,const std::function<bool(const char *,const char *, int, int)> &cb){
+
+bool
+L2PSstream::splitPES(const uint8_t *data,
+                     uint16_t len,
+                     const std::function<bool(const char *, const char *, int, int)> &cb)
+{
     auto prefix = prefixSize((char *) data, len);
     uint8_t ned[] = {0x00, 0x00, 0x01, 0xe0};
     auto start = (uint8_t *) memmem(data, len, ned, sizeof(ned));
     char *pes_head = nullptr;
     bool ret = false;
-    ret= stream_split_nd((char*)start,len-(start-data),prefix,ned,sizeof(ned),[&](const char *ptr, int len, int prefix){
+    ret = stream_split_nd(
+        (char *) start, len - (start - data), prefix, ned, sizeof(ned), [&](const char *ptr, int len, int prefix) {
             pes_head = (char *) ptr;
             stream_split((char *) ptr, len, prefix, [&](const char *ptr, int len, int prefix) {
                 //INFOL("%02X %02X %02X %02X %02X \n",(uint8_t)ptr[0],(uint8_t)ptr[1],(uint8_t)ptr[2],(uint8_t)ptr[3],(uint8_t)ptr[4]);
                 if (cb) {
                     ret = cb(ptr, pes_head, len, prefix);
-            }else
+                } else {
-            {
                     throw runtime_error("empty callback function");
                 }
-            
-            
             });
             return ret;
         });
-    if(!ret && start){
+    if (!ret && start) { record_PES_head(start, len - (start - data)); }
-        record_PES_head(start,len-(start-data));
-    }
     return ret;
 }
 
-void L2PSstream::rtp_post_process(const RTPcell::Ptr &rtp_cell){
+void
+L2PSstream::rtp_post_process(const RTPcell::Ptr &rtp_cell)
+{
     const uint8_t h264key[] = {0x00, 0x00, 0x00, 0x01, 0x65};
     const uint8_t PEShead[] = {0x00, 0x00, 0x01, 0xe0};
     const uint8_t h264Sei[] = {0x00, 0x00, 0x00, 0x01, 0x06};
     uint8_t *keyframe_pos, *PES_pos;
     uint16_t payload_len = 0;
-    switch (_code_id)
+    switch (_code_id) {
-    {
     case CodecId::CodecH264:
 
         keyframe_pos = (uint8_t *) memmem(rtp_cell->_PT_data, rtp_cell->_PT_len, h264key, sizeof(h264key));
         if (keyframe_pos) {//the rtp contain key frame
-            auto ret=splitPES(rtp_cell->_PT_data, keyframe_pos-(rtp_cell->_PT_data),[&](const char *ptr,const char *pes, int len, int prefix){
+            INFOL("Find Key frame\n");
+            auto ret = splitPES(
+                rtp_cell->_PT_data, keyframe_pos - (rtp_cell->_PT_data),
+                [&](const char *ptr, const char *pes, int len, int prefix) {
                     if (prefix == 4) {
                         int type = H264_TYPE(((uint8_t *) ptr + prefix)[0]);
-                        if (type==H264Nal::NAL_SEI)
+                        if (type == H264Nal::NAL_SEI) {
-                        {
                             uint8_t *sei_ptr = _sei_race_buf;
                             // if(len>sizeof(_sei_race_buf)) return false;
                             // memcpy(sei_ptr,ptr,len);sei_ptr+=len;
@@ -113,14 +118,15 @@ void L2PSstream::rtp_post_process(const RTPcell::Ptr &rtp_cell){
                             uint16_t pes_pt_len = (((uint16_t) pes[4]) << 8) | (uint8_t) pes[5];
                             pes_pt_len += gen_ptr - sei_ptr;
                             uint16_t pes_pos = ((uint8_t *) pes - rtp_cell->_buffer) + rtp_cell->_RTPpkg_head_pos;
-                            if(pes_pos+2+4>rtp_cell->_RTPpkg_ptr->_buffer.size()) 
+                            if (pes_pos + 2 + 4 > rtp_cell->_RTPpkg_ptr->_buffer.size()) return false;
-                                return false;
                             nbo_w16((uint8_t *) (rtp_cell->_RTPpkg_ptr->_buffer.data() + pes_pos + 4), pes_pt_len);
                             ///////////add rear rtp header/////////////////
-                            sei_ptr=addRTPheader(rtp_cell,gen_ptr,sizeof(_sei_race_buf)-(gen_ptr-_sei_race_buf));
+                            sei_ptr =
+                                addRTPheader(rtp_cell, gen_ptr, sizeof(_sei_race_buf) - (gen_ptr - _sei_race_buf));
                             if (!sei_ptr) return false;
                             _sei_race_len = sei_ptr - _sei_race_buf;
-                            rtp_cell->_RTPpkg_ptr->insertData(_sei_race_buf,_sei_race_len,rtp_cell->_RTPpkg_head_pos,
+                            rtp_cell->_RTPpkg_ptr->insertData(
+                                _sei_race_buf, _sei_race_len, rtp_cell->_RTPpkg_head_pos,
                                 ((uint8_t *) ptr - rtp_cell->_buffer) + len, sei_ptr - gen_ptr);
                             // SUCSL("SUCESS SIGN!!!!!!!!!!!!!!\n");
                             return true;
@@ -148,6 +154,8 @@ void L2PSstream::rtp_post_process(const RTPcell::Ptr &rtp_cell){
             //     }
 
             // }
+        } else {
+            INFOL("NO key frame\n");
         }
         break;
 
@@ -162,11 +170,11 @@ void L2PSstream::rtp_post_process(const RTPcell::Ptr &rtp_cell){
     }
 }
 
-
+void
-void L2PSstream::decodeH264(const char *data, int bytes, int64_t dts, int prefix){
+L2PSstream::decodeH264(const char *data, int bytes, int64_t dts, int prefix)
-    int type=H264_TYPE(((uint8_t *)data+prefix)[0]);
-    switch (type)
 {
+    int type = H264_TYPE(((uint8_t *) data + prefix)[0]);
+    switch (type) {
     case H264Nal::NAL_IDR:
         TRACEL("NAL_IDR dts:%lu, bytes:%d\n", dts, bytes);
         addSignData(data + prefix, bytes - prefix, dts, type);
@@ -177,12 +185,13 @@ void L2PSstream::decodeH264(const char *data, int bytes, int64_t dts, int prefix
     default:
         break;
     }
-
 };
-void L2PSstream::decodeH265(const char *data, int bytes, int64_t dts, int prefix){
+
-    int type=H265_TYPE(((uint8_t *)data+prefix)[0]);
+void
-    switch (type)
+L2PSstream::decodeH265(const char *data, int bytes, int64_t dts, int prefix)
 {
+    int type = H265_TYPE(((uint8_t *) data + prefix)[0]);
+    switch (type) {
     case H265Nal::NAL_BLA_W_LP... H265Nal::NAL_RSV_IRAP_VCL23://keyframe
         TRACEL("NAL_IDR dts:%lu, bytes:%d\n", dts, bytes);
         addSignData(data + prefix, bytes - prefix, dts, type);
@@ -198,12 +207,4 @@ void L2PSstream::decodeH265(const char *data, int bytes, int64_t dts, int prefix
     }
 };
 
-
-
-
-
-
-
-
 ///////////////////////////////////////////////////////////////////
-

src/GB28181/L2SecurityStream.cpp

@@ -1,40 +1,34 @@
 #include <mutex>
 
 #include "L2SecurityStream.h"
+#include "layer2/frame.h"
+#include "layer2/l2_manager.h"
+#include "layer4/rtp_manager.h"
 
-	void printdata(uint8_t* out_buff,int frameLen,int type,int dts){
+void
+printdata(uint8_t *out_buff, int frameLen, int type, int dts)
+{
     printf("\n");
     printf("type: %d %d,%d\n", type, dts, frameLen);
-		for (int num = 0; num < 10; num++)
+    for (int num = 0; num < 10; num++) { printf("%02X ", (uint8_t) out_buff[num]); }
-		{
-			printf("%02X ", (uint8_t)out_buff[num]);
-		}
     printf("---- ");
-		for (int num = frameLen/2; num < frameLen / 2+10; num++)
+    for (int num = frameLen / 2; num < frameLen / 2 + 10; num++) { printf("%02X ", (uint8_t) out_buff[num]); }
-		{
-			printf("%02X ", (uint8_t)out_buff[num]);
-		}
     printf("---- ");
-		for (int num = 10; num >0; num--)
+    for (int num = 10; num > 0; num--) { printf("%02X ", (uint8_t) out_buff[frameLen - num]); }
-		{
-			printf("%02X ", (uint8_t)out_buff[frameLen-num]);
-		}
     printf("\n");
 }
 
-L2SecurityStream::L2SecurityStream(sec_set_info* set_info){
+L2SecurityStream::L2SecurityStream(sec_set_info *set_info)
+{
     _keyframe_seq = -1;
     _code_id = CodecInvalid;
     sm3_init(&_sm3);
     if (set_info) {
         _set_info = *set_info;
-        }else
+    } else {
-        {
         throw runtime_error("L2SecurityStream initial error: empty sec_set_info");
     }
 
-        
-
     uint8_t security_set_version[19] = "Ver 0.0.2";
     memset(&sec_set, 0, sizeof(sec_set));
     memset(&(sec_set.evek), 0xff, sizeof(sec_set.evek));
@@ -62,28 +56,22 @@ L2SecurityStream::L2SecurityStream(sec_set_info* set_info){
 
     _thread_exit = 0;
     for (auto &th : _sign_thread_list) {
-            th=new std::thread([&](){
+        th = new std::thread([&]() { ThreadSign(); });
-                ThreadSign();
-            });
-        }
-        for(auto &th:_sign_thread_list){
-            th->detach();
     }
+    for (auto &th : _sign_thread_list) { th->detach(); }
     usleep(1000);
-        
 }
 
-void L2SecurityStream::ThreadSign(){
+void
+L2SecurityStream::ThreadSign()
+{
     uint8_t sha[128];
     cpu_set_t mask;
     CPU_ZERO(&mask);
     CPU_SET(_set_info.cpu_core, &mask);
-        if(pthread_setaffinity_np(pthread_self(),sizeof(mask),&mask)<0){
+    if (pthread_setaffinity_np(pthread_self(), sizeof(mask), &mask) < 0) { ERROL("Sign Thread start fail"); }
-            ERROL("Sign Thread start fail");
-        }
     SUCSL("sign thread created\n");
-        while (1)
+    while (1) {
-        {
         {
             std::unique_lock<std::mutex> lock(_sign_mtx);
             _sign_start.wait(lock, [&] { return _start_sign; });
@@ -114,11 +102,19 @@ void L2SecurityStream::ThreadSign(){
             _keyframe_seq = -5;
             NOTICEL("End sign\n");
         }
-            
     }
 }
 
-    int L2SecurityStream::RtpOutput(char * buf,  uint32_t * len, uint16_t* sei_tail_pos, uint16_t* extra_len, void ** param){
+int
+L2SecurityStream::RtpOutput(char *buf, uint32_t *len, uint16_t *sei_tail_pos, uint16_t *extra_len, void **param)
+{
+    if (_l2_manager.empty()) { return -1; }
+    auto frame = _l2_manager.FrameOutput();
+    std::copy(frame->data().begin(), frame->data().end(), buf);
+    *len = frame->data().size();
+    *param = frame->user_param();
+
+    /*
     auto element = _package_queue.pop();
     // *param=nullptr;
     if (!element) return -1;
@@ -132,25 +128,28 @@ void L2SecurityStream::ThreadSign(){
     // DEBUGL("in param address: %p \n",element->_usr_param);
     // DEBUGL("%p out>> in %u out %u que: %lu frame_len %u tail_pos %u ins_len %u\n",this,_all_in,_all_out, _package_queue.size(),*len,*sei_tail_pos,*extra_len);
     // DEBUGL("OUT len %u\n", *len);
+    */
     return 1;
 }
 
-    uint8_t * L2SecurityStream::gen_sec_sei(uint8_t *ptr){
+uint8_t *
+L2SecurityStream::gen_sec_sei(uint8_t *ptr)
+{
     std::lock_guard<std::mutex> lock(_sign_process_mtx);
     if (_keyframe_complete && !_start_sign && !_keyframe_DATA.empty()) {
         {
             // std::unique_lock<std::mutex> lock(_sign_process_mtx);
             if (_keyframe_seq == -5) {
-                    ptr=SecMedia::appendSEIframe(ptr,SEC_SET_PT,(uint8_t*)SecMedia::Sec_set_UUID,(uint8_t*)&sec_set,sizeof(sec_set));
+                ptr = SecMedia::appendSEIframe(
-                    ptr=SecMedia::appendSEIframe(ptr,VER_SET_PT,(uint8_t*)SecMedia::Ver_set_UUID,(uint8_t*)&ver_set,sizeof(ver_set));
+                    ptr, SEC_SET_PT, (uint8_t *) SecMedia::Sec_set_UUID, (uint8_t *) &sec_set, sizeof(sec_set));
+                ptr = SecMedia::appendSEIframe(
+                    ptr, VER_SET_PT, (uint8_t *) SecMedia::Ver_set_UUID, (uint8_t *) &ver_set, sizeof(ver_set));
                 INFOL("gen_sec_sei dts:%u\n", ver_set.head_frame_dts);
-                }
+            } else {
-                else{
                 if (_keyframe_seq == -2) {
                     WRNGL("Sign not complete\n");
                     err_ver_set.head_frame_type = SecMedia::ERR_incomplete_Sign;
-                    }
+                } else {
-                    else{
                     WRNGL("Sign loss\n");
                     err_ver_set.head_frame_type = SecMedia::ERR_Without_Sign_Data;
                 }
@@ -160,21 +159,29 @@ void L2SecurityStream::ThreadSign(){
             _keyframe_complete = false;
         }
         return ptr;
-        }else
+    } else {
-        {
         WRNGL("Without Sign data\n");
         err_ver_set.head_frame_type = SecMedia::ERR_Without_Frame_Data;
     }
-        ptr=SecMedia::appendSEIframe(ptr,SEC_SET_PT,(uint8_t*)SecMedia::Sec_set_UUID,(uint8_t*)&sec_set,sizeof(sec_set));
+    ptr = SecMedia::appendSEIframe(
-        ptr=SecMedia::appendSEIframe(ptr,VER_SET_PT,(uint8_t*)SecMedia::Ver_set_UUID,(uint8_t*)&err_ver_set,sizeof(err_ver_set));
+        ptr, SEC_SET_PT, (uint8_t *) SecMedia::Sec_set_UUID, (uint8_t *) &sec_set, sizeof(sec_set));
+    ptr = SecMedia::appendSEIframe(
+        ptr, VER_SET_PT, (uint8_t *) SecMedia::Ver_set_UUID, (uint8_t *) &err_ver_set, sizeof(err_ver_set));
 
     return ptr;
 }
 
 //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
-    int L2UdpStream::RtpInput(const uint8_t* data,uint16_t len, void *param){
+int
+L2UdpStream::RtpInput(const uint8_t *data, uint16_t len, void *param)
+{
+    Frame::Ptr frame = Frame::CreateUDPFrame(data, len);
+    _l2_manager.FrameInput(frame);
+    return 0;
 
     RTPpackage::rtp_cb_f rtp_cell_cb = [&](const RTPcell::Ptr &rtp_cell) {
+        // 完成的一个RTP包到达
+        // 解析头部信息
         rtp_cell->precessRTP();
 
         if ((uint32_t) (_prv_seq + 1) % (uint32_t) (UINT16_MAX + 1) == rtp_cell->seq || _prv_seq == 0) {
@@ -184,12 +191,10 @@ void L2SecurityStream::ThreadSign(){
             // rtp_cell->_prefix
             incrementRTPseq(rtp_cell);
             rtp_post_process(rtp_cell);
-            }else
+        } else {
-            {
             WRNGL("Disorder rtp Seq: %u+1 != %u \n", _prv_seq, rtp_cell->seq);
         }
         _prv_seq = rtp_cell->seq;
-            
     };
     RTPpackage::Ptr RTPpkg = make_shared<RTPpackage>(rtp_cell_cb, true);
     RTPpkg->assignRTPpackage(data, len, param);
@@ -198,8 +203,3 @@ void L2SecurityStream::ThreadSign(){
     _package_queue.push(RTPpkg);
     return 0;
 }
-
-    
-
-
-

src/GB28181/L2SecurityStream.h

@@ -6,6 +6,8 @@
 #include "HuaWei/HWsign.h"
 #include "HuaWei/RTP.h"
 #include "RtpDecoder.h"
+#include "layer2/l2_manager.h"
+#include "layer4/rtp_manager.h"
 #include <array>
 #include <atomic>
 #include <fstream>
@@ -23,7 +25,7 @@ public:
         _keyframe_seq = -2;
         _sign_start.notify_all();
         usleep(10000);
-    };
+    }
 
     void ThreadSign();
 
@@ -55,7 +57,7 @@ protected:
         // }
         usleep(1000);
         // sleep(1);
-    };
+    }
 
 private:
     const char _class_name[17] = "L2SecurityStream";
@@ -67,7 +69,8 @@ protected:
     uint32_t _all_out = 0;
     bool _keyframe_complete = false;
     uint16_t _sei_race_len;
-    HWsec_queue _package_queue;
+    // HWsec_queue _package_queue;
+    L2Manager _l2_manager;
 
 public:
     uint8_t _sei_race_buf[4096];
@@ -88,10 +91,13 @@ public:
     bool _start_sign = false;
 };
 
-class L2UdpStream : public L2SecurityStream {
+class L2UdpStream : public L2SecurityStream, public sigslot::has_slots<> {
 protected:
     uint16_t _seq_increment = 0;
     RtpDecoder _rtp_decoder;
+    RTPManager::Ptr _rtp_manager;
+    L2Manager _l2_manager;
+    ThreadsafeQueue<RTPpackage::Ptr> _package_queue;
 
 private:
     uint16_t _prv_seq = 0;
@@ -102,7 +108,11 @@ public:
         _rtp_decoder.setOnDecode([&](const uint8_t *packet, int bytes, uint32_t timestamp, int flags) {
             on_rtp_out(packet, bytes, timestamp, flags);
         });
-    };
+
+        _rtp_manager = std::shared_ptr<RTPManager>(new RTPManager());
+        _l2_manager.OnFrameInput.connect(_rtp_manager.get(), &RTPManager::OnFrame);
+        _rtp_manager->OnPSFrame.connect(this, &L2UdpStream::on_rtp_out);
+    }
 
     int RtpInput(const uint8_t *data, uint16_t len, void *param = nullptr) override;
     virtual void on_rtp_out(const uint8_t *packet, int bytes, uint32_t timestamp, int flags) = 0;

src/GB28181/RtpDecoder.cpp

@@ -1,23 +1,26 @@
 #include "RtpDecoder.h"
 #include "rtp-payload.h"
 
-RtpDecoder::~RtpDecoder() {
+RtpDecoder::~RtpDecoder()
+{
     if (_rtp_decoder) {
         rtp_payload_decode_destroy(_rtp_decoder);
         _rtp_decoder = nullptr;
     }
 }
 
-void RtpDecoder::decodeRtp(const void *data, int bytes) {
+void
+RtpDecoder::decodeRtp(const void *data, int bytes)
+{
+    if (bytes < 2) { return; }
+    const uint8_t cur_rtp_type = 0x7F & ((uint8_t *) data)[1];
+
     if (!_rtp_decoder) {
         static rtp_payload_t s_func = {
             [](void *param, int bytes) {
                 RtpDecoder *obj = (RtpDecoder *) param;
                 obj->_buffer.resize(bytes);
-                    if (obj->_buffer.capacity()>2*bytes && bytes>2048)
+                if (obj->_buffer.capacity() > 2 * bytes && bytes > 2048) { obj->_buffer.shrink_to_fit(); }
-                    {
-                        obj->_buffer.shrink_to_fit();
-                    }
                 // obj->_buffer->setCapacity(bytes);
                 // return (void *)obj->_buffer->data();
                 return (void *) obj->_buffer.data();
@@ -28,20 +31,17 @@ void RtpDecoder::decodeRtp(const void *data, int bytes) {
             [](void *param, const void *packet, int bytes, uint32_t timestamp, int flags) {
                 RtpDecoder *obj = (RtpDecoder *) param;
                 obj->onDecode((uint8_t *) packet, bytes, timestamp, flags);
-                }
+            }};
-        };
 
-        _rtp_type = 0x7F & ((uint8_t *) data)[1];
+        _rtp_type = cur_rtp_type;
         INFOL("rtp type:%d\n", (int) _rtp_type);
 
         _rtp_decoder = rtp_payload_decode_create(_rtp_type, _codec.data(), &s_func, this);
-        if (!_rtp_decoder) {
+        if (!_rtp_decoder) { WRNGL("unsupported rtp type:%d ,size:%u\n", (int) _rtp_type, bytes); }
-            WRNGL("unsupported rtp type:%d ,size:%u\n",(int) _rtp_type,bytes);
-        }
     }
 
     if (_rtp_decoder) {
-        if(( 0x7F & ((uint8_t *) data)[1])==_rtp_type){ // 直接收一种类型了
+        if (cur_rtp_type == _rtp_type) {// 直接收一种类型了
             rtp_payload_decode_input(_rtp_decoder, data, bytes);
         }
     }

src/GB28181/RtpDecoder.h

@@ -10,13 +10,12 @@ public:
     typedef function<void(const uint8_t *packet, int bytes, uint32_t timestamp, int flags)> on_decode_t;
 
 public:
-    RtpDecoder() {
+    RtpDecoder(const std::string &codec_name = "MP2P") : _codec(codec_name) {}
 
-    };
     ~RtpDecoder();
     void decodeRtp(const void *data, int bytes);
 
-    void setOnDecode(on_decode_t cb) { _on_decoded = cb; };
+    void setOnDecode(on_decode_t cb) { _on_decoded = cb; }
 
 private:
     void onDecode(const uint8_t *packet, int bytes, uint32_t timestamp, int flags)
@@ -27,7 +26,7 @@ private:
 private:
     void *_rtp_decoder = nullptr;
     std::vector<uint8_t> _buffer;
-    std::string _codec = "MP2P";
+    std::string _codec;
     uint8_t _rtp_type;
     on_decode_t _on_decoded;
 };

src/HuaWei/HWsec.h

@@ -1,25 +1,26 @@
 #ifndef _HWsec_H
 #define _HWsec_H
-#include <thread>
+#include "HWsign.h"
-#include <chrono>
-#include <fstream>
 #include "RTP.h"
 #include "queue.h"
-#include "HWsign.h"
+#include <chrono>
-    class HWsec_queue: public ThreadsafeQueue<RTPpackage::Ptr>
+#include <fstream>
-    {
+#include <thread>
+
+class HWsec_queue : public ThreadsafeQueue<RTPpackage::Ptr> {
 public:
-        HWsec_queue(){};
+    HWsec_queue() {}
-        ~HWsec_queue(){};
+
-        RTPpackage::Ptr pop(){
+    ~HWsec_queue() {}
-            std::lock_guard<decltype(mutex_)> lock(mutex_);
+
-            if (queue_.empty()) {
+    RTPpackage::Ptr pop()
-            return nullptr;
-            }
-            RTPpackage::Ptr tmp = queue_.front();
-            if (tmp->isPause())
     {
-                WRNGL("pause@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@pause@@@@@@@@pause@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@pause@@@@@@@@pause@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@pause@@@@@@@@ \n");
+        std::lock_guard<decltype(mutex_)> lock(mutex_);
+        if (queue_.empty()) { return nullptr; }
+        RTPpackage::Ptr tmp = queue_.front();
+        if (tmp->isPause()) {
+            WRNGL("pause@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@pause@@@@@@@@pause@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@pause@@@@"
+                  "@@@@pause@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@pause@@@@@@@@ \n");
             return nullptr;
         }
 
@@ -28,9 +29,9 @@
     }
 };
 
-    class HWsec
+class HWsec {
-    {
     typedef std::chrono::high_resolution_clock Clock;
+
 private:
     CodecId _code_id;
     RTPcell::Ptr _SEI_rtp;
@@ -46,6 +47,7 @@
     shared_ptr<ofstream> _len_file;
 #endif
     std::thread *_sign_thread;
+
 public:
     uint8_t _sei_race_buf[4096];
     uint8_t _thread_exit;
@@ -58,30 +60,29 @@
     std::mutex _sign_mtx;
     std::condition_variable _sign_start;
     bool _start_sign = false;
+
 public:
     HWsec(sec_set_info *set_info);
     ~HWsec();
-        bool check_class(){
+
-            return strcmp(_class_name,"HWsec")==0;
+    bool check_class() { return strcmp(_class_name, "HWsec") == 0; }
-        }
+
-        void HWsec_input(const uint8_t* data,uint16_t len, void *param=nullptr){
+    void HWsec_input(const uint8_t *data, uint16_t len, void *param = nullptr)
+    {
         RTPpackage::rtp_cb_f rtp_cell_cb = [&](const RTPcell::Ptr &rtp_cell) {
             if (rtp_cell) {
                 rtp_cell->precessRTP();
 
-                switch (rtp_cell->PT)
+                switch (rtp_cell->PT) {
-                {
                 case 99://h264
-                    if (_code_id!=CodecId::CodecH264)
+                    if (_code_id != CodecId::CodecH264) {
-                    {
                         _code_id = CodecId::CodecH264;
                         INFOL("Track type is changed to H.264\n");
                     }
                     // decode_264( rtp_cell);
                     break;
                 case 103 ... 108://h265
-                    if (_code_id!=CodecId::CodecH265)
+                    if (_code_id != CodecId::CodecH265) {
-                    {
                         _code_id = CodecId::CodecH265;
                         INFOL("Track type is changed to H.265\n");
                     }
@@ -94,11 +95,9 @@
                     break;
                 }
 
-            }else
+            } else {
-            {
                 WRNGL("error");
             }
-            
         };
 #ifdef DUMP_FILE
         _data_file->write((char *) data, len);
@@ -119,7 +118,8 @@
         // TRACEL("time:%ld size: %lu \n",delta.count(),RTPpkg->_buffer.size());
     }
 
-        int HWsec_output(char * buf,  uint32_t * len, uint16_t* sei_tail_pos, uint16_t* extra_len, void ** param){
+    int HWsec_output(char *buf, uint32_t *len, uint16_t *sei_tail_pos, uint16_t *extra_len, void **param)
+    {
 
         auto element = _RTPpackage_queue.pop();
         // *param=nullptr;
@@ -132,10 +132,12 @@
         element->clear_all();
         _all_out++;
         // DEBUGL("in param address: %p \n",element->_usr_param);
-            DEBUGL("%p out>> in %u out %u que: %lu frame_len %u tail_pos %u ins_len %u\n",this,_all_in,_all_out, _RTPpackage_queue.size(),*len,*sei_tail_pos,*extra_len);
+        DEBUGL("%p out>> in %u out %u que: %lu frame_len %u tail_pos %u ins_len %u\n", this, _all_in, _all_out,
+               _RTPpackage_queue.size(), *len, *sei_tail_pos, *extra_len);
         // DEBUGL("OUT len %u\n", *len);
         return 1;
     }
+
 private:
     void gen_sec_data();
     void sign_data(const char *buf, const uint32_t len);
@@ -144,5 +146,4 @@
     void setfile();
 };
 
-
 #endif//_HWSIGN_TCP_H

src/HuaWei/HWsign.h

@@ -2,8 +2,8 @@
 #ifndef _HWSIGN_H
 #define _HWSIGN_H
 
+#include <stdint.h>
 #include <stdlib.h>
-#include<stdio.h>
 //#include <functional>
 #define API_EXPORT __attribute__((visibility("default")))
 
@@ -16,6 +16,7 @@ struct sec_set_info{
     uint8_t prikey[64];
     uint8_t pubkey[128];
 };
+
 // void * HWSign_init(const std::function<void(const char * rtp_ptr, const uint32_t rtp_len)> rtp_callback);
 API_EXPORT int SDF_Device_open();
 API_EXPORT int SDF_Device_close();
@@ -23,7 +24,12 @@ API_EXPORT int SDF_Device_close();
 API_EXPORT void *GB28181_stream_init(struct sec_set_info *sign_info);
 API_EXPORT int GB28181_stream_in(void *Handle, const char *buf, const uint32_t len, void *param);
 API_EXPORT void GB28181_stream_release(void *Handle);
-API_EXPORT int GB28181_stream_out(void* Handle, char * buf,  uint32_t * len, uint16_t* sei_end_offset,uint16_t* append_length,void ** param);
+API_EXPORT int GB28181_stream_out(void *Handle,
+                                  char *buf,
+                                  uint32_t *len,
+                                  uint16_t *sei_end_offset,
+                                  uint16_t *append_length,
+                                  void **param);
 
 API_EXPORT void *HWSign_init(struct sec_set_info *sign_info);
 API_EXPORT void HWSign_release(void *Handle);
@@ -33,8 +39,12 @@ API_EXPORT int HWSign_rtp_out(void* Handle, char * buf,  uint32_t * len, void **
 API_EXPORT void *HWSign_tcp_init(struct sec_set_info *sign_info);
 API_EXPORT void HWSign_tcp_release(void *Handle);
 API_EXPORT int HWSign_tcp_rtp_input(void *Handle, const char *buf, const uint32_t len, void *param);
-API_EXPORT int HWSign_tcp_rtp_out(void* Handle, char * buf,  uint32_t * len, uint16_t* sei_end_offset,uint16_t* append_length,void ** param);
+API_EXPORT int HWSign_tcp_rtp_out(void *Handle,
-
+                                  char *buf,
+                                  uint32_t *len,
+                                  uint16_t *sei_end_offset,
+                                  uint16_t *append_length,
+                                  void **param);
 
 API_EXPORT void *HWVerify_init();
 API_EXPORT void HWVerify_release(void *Handle);

src/HuaWei/RTP.h

@@ -1,22 +1,23 @@
 #ifndef _RTP_H
 #define _RTP_H
-#include <string>
+#include "HWcommon.h"
-#include <functional>
-#include <memory>
 #include "string.h"
 #include <arpa/inet.h>
-#include "HWcommon.h"
+#include <functional>
+#include <memory>
+#include <string>
 using namespace std;
 // extern uint32_t RTPcell_total_count;
 // extern uint32_t RTPpackage_total_count;
 class RTPpackage;
-    class RTPcell
+
-    {
+class RTPcell {
 public:
     typedef shared_ptr<RTPcell> Ptr;
 
 public:
-        RTPcell(uint16_t len, uint8_t prefix, uint16_t buff_pos,shared_ptr<RTPpackage> RTPpkg_ptr){
+    RTPcell(uint16_t len, uint8_t prefix, uint16_t buff_pos, shared_ptr<RTPpackage> RTPpkg_ptr)
+    {
 
         _assign_len = 0;
         _RTPpkg_head_pos = buff_pos;
@@ -24,14 +25,18 @@ using namespace std;
         _missing_head = false;
         _buffer = new uint8_t[(len / 2048 + 1) * 2048];
         _RTPpkg_ptr = RTPpkg_ptr;
-            if(_prefix==16) _tcp_prefix=4;
+        if (_prefix == 16)
-            else _tcp_prefix=0;
+            _tcp_prefix = 4;
+        else
+            _tcp_prefix = 0;
         _RTP_len = len + _tcp_prefix;
 
         // RTPcell_total_count++;
         // SUCSL("create RTPcell %u\n",RTPcell_total_count);
     }
-        RTPcell(const uint8_t *head_d,uint16_t len, uint8_t prefix, uint16_t buff_pos,shared_ptr<RTPpackage> RTPpkg_ptr){
+
+    RTPcell(const uint8_t *head_d, uint16_t len, uint8_t prefix, uint16_t buff_pos, shared_ptr<RTPpackage> RTPpkg_ptr)
+    {
         _missing_head = true;
         _RTPpkg_ptr = RTPpkg_ptr;
         _assign_len = len;
@@ -40,12 +45,15 @@ using namespace std;
         _RTP_len = prefix;
         _buffer = new uint8_t[(len / 2048 + 1) * 2048];
         memcpy(_buffer, head_d, len);
-            if(_prefix==16) _tcp_prefix=4;
+        if (_prefix == 16)
-            else _tcp_prefix=0;
+            _tcp_prefix = 4;
+        else
+            _tcp_prefix = 0;
 
         // RTPcell_total_count++;
         // SUCSL("create RTPcell %u\n",RTPcell_total_count);
     }
+
     // RTPcell(const uint8_t *data,uint16_t len, uint8_t prefix){
     //     _buffer=new uint8_t[(len/2048+1)*2048];
     //     _RTP_len=len;
@@ -53,20 +61,25 @@ using namespace std;
     //     if(_prefix==16) _tcp_prefix=4;
     //     else _tcp_prefix=0;
     // }
-        ~RTPcell(){
+    ~RTPcell()
+    {
         // RTPcell_total_count--;
         // WRNGL("~RTPcell %u\n",RTPcell_total_count);
 
         delete _buffer;
     }
-        void setRTPlen(){
+
+    void setRTPlen()
+    {
         if (_tcp_prefix >= 4) {
             _RTP_len = (_buffer[2] << 8) & 0xff00 | (_buffer[3]) + _tcp_prefix;
             _buffer = (uint8_t *) realloc(_buffer, (_RTP_len / 2048 + 1) * 2048);
         }
         _missing_head = false;
     }
-        int append_data(const uint8_t * data, uint16_t len){
+
+    int append_data(const uint8_t *data, uint16_t len)
+    {
         if (_RTP_len >= _assign_len + len) {
             memcpy(_assign_len + _buffer, data, len);
             _assign_len += len;
@@ -74,10 +87,11 @@ using namespace std;
         }
         return -1;
     }
-        int getLackLen(){
+
-            return (int)_RTP_len-(int)_assign_len;
+    int getLackLen() { return (int) _RTP_len - (int) _assign_len; }
-        }
+
-        void precessRTP(){
+    void precessRTP()
+    {
 
         uint8_t *ptr = _buffer + _tcp_prefix;
         mark = ptr[1] >> 7;
@@ -93,12 +107,14 @@ using namespace std;
         _PT_len = _RTP_len - 12 - _tcp_prefix;
         _PT_data = ptr + 12 + _tcp_prefix;
     }
+
 public:
     uint8_t _tcp_prefix = 0;
     uint8_t _prefix = 0;
     uint16_t _RTP_len = 0;
     uint16_t _assign_len = 0;
     uint16_t _RTPpkg_head_pos = 0;
+    // 数据长度不足以判断是否构成RTP头部
     bool _missing_head = false;
     uint8_t *_buffer = NULL;
     shared_ptr<RTPpackage> _RTPpkg_ptr;
@@ -111,39 +127,51 @@ using namespace std;
     uint16_t seq;
     uint32_t timestamp;
     uint32_t ssrc;
-
 };
 
-
+class RTPpackage : public enable_shared_from_this<RTPpackage> {
-    class RTPpackage: public enable_shared_from_this<RTPpackage>
-    {
 public:
     typedef shared_ptr<RTPpackage> Ptr;
     typedef function<void(const RTPcell::Ptr &rtp_cell)> rtp_cb_f;
+
 private:
-        int isRTP(const uint8_t * data, uint16_t len,bool tail=false){
+    /**
-            if(len==0) return tail?0:-1;
+    * 判断数据是否为RTP包
-            if(len>=_prefix)
+    *
-                if((data[0]==0x24 && data[4]==0x80) || (_is_UDP && data[0]==0x80))
+    * @param data 数据指针
-                    return 1; // strong true
+    * @param len  数据长度
-                else
+    * @param tail 是否为最后一包，默认false
-                    return -1; // strong false
+    *
-            else
+    * @return int, 1: 是RTP包，-1: 不是RTP包，0: 不确定
-                return 0; // I don't no
+    */
-        }
+    enum RTPHeaderState { kNotFounded = -1, kUnknown = 0, kFounded = 1 };
-        uint16_t to_uint16(uint8_t H, uint8_t L){
+
-            return (H<<8) &0xff00 | (L);
+    int isRTPHeader(const uint8_t *data, uint16_t len, bool tail = false)
-        }
-        RTPcell::Ptr makeRTPcell(const uint8_t * data, uint16_t len, uint16_t * buff_pos){
-            auto ret=isRTP(data,len);
-            RTPcell::Ptr rtpcell;
-            if(ret==1){
-                if (!_is_UDP)
     {
+        const auto FirstByteIsRTPHeader = [](uint8_t c) { return c == 0x80 || c == 0xA0; };
+        if (len == 0) return tail ? kUnknown : kNotFounded;
+
+        if (len >= _prefix)
+            if ((data[0] == 0x24 && FirstByteIsRTPHeader(data[4])) || (_is_UDP && FirstByteIsRTPHeader(data[0])))
+                return kFounded;// strong true
+            else
+                return kNotFounded;// strong false
+        else
+            return kUnknown;// I don't no
+    }
+
+    uint16_t to_uint16(uint8_t H, uint8_t L) { return (H << 8) & 0xff00 | (L); }
+
+    RTPcell::Ptr makeRTPcell(const uint8_t *data, uint16_t len, uint16_t *buff_pos)
+    {
+        auto ret = isRTPHeader(data, len);
+        RTPcell::Ptr rtpcell;
+        if (ret == kFounded) {
+            if (!_is_UDP) {
+                // rtp over tcp
                 uint16_t length = to_uint16(data[2], data[3]);
                 rtpcell = make_shared<RTPcell>(length, _prefix, *buff_pos, shared_from_this());
-                }else
+            } else {
-                {
                 rtpcell = make_shared<RTPcell>(len, _prefix, *buff_pos, shared_from_this());
             }
 
@@ -151,21 +179,19 @@ using namespace std;
             rtpcell->append_data(data, apd_len);
             *buff_pos += apd_len;
             return rtpcell;
-            }else if (ret==0)
+        } else if (ret == 0) {
-            {
+            // 长度不足以不确定是不是RTP包，需要等下一个包
             auto rtpcell = make_shared<RTPcell>(data, len, _prefix, *buff_pos, shared_from_this());
             *buff_pos += len;
             return rtpcell;
-            }else
+        } else {
-            {
             return nullptr;
         }
-            
-            
-            
     }
+
 public:
-        RTPpackage(const rtp_cb_f & cb,const RTPcell::Ptr & prev_rtp_cell){
+    RTPpackage(const rtp_cb_f &cb, const RTPcell::Ptr &prev_rtp_cell)
+    {
         _rtp_cb = cb;
         _first_start = true;
         _Rtp_cell = prev_rtp_cell;
@@ -175,35 +201,45 @@ using namespace std;
         _insert_tail_pos = 0;
         // RTPpackage_total_count++;
         // WRNGL("create RTPpackage %u\n", RTPpackage_total_count);
-        };
+    }
-        RTPpackage(const rtp_cb_f & cb,bool is_udp){
+
+    RTPpackage(const rtp_cb_f &cb, bool is_udp)
+    {
         _rtp_cb = cb;
         _first_start = true;
         _Rtp_cell = nullptr;
-            if (is_udp) _prefix=12;
+        if (is_udp)
-            else _prefix=16;
+            _prefix = 12;
+        else
+            _prefix = 16;
         _is_UDP = is_udp;
         _insert_len = 0;
         _inserted_len = 0;
         _insert_tail_pos = 0;
         // RTPpackage_total_count++;
         // WRNGL("create RTPpackage %u\n", RTPpackage_total_count);
-        };
+    }
-        ~RTPpackage(){
+
+    ~RTPpackage()
+    {
 
         // RTPpackage_total_count--;
         // WRNGL("~RTPpackage %u\n", RTPpackage_total_count);
-        };
+    }
-        void clear_all(){
+
+    void clear_all()
+    {
         _Rtp_cell.reset();
         // _buffer.clear();
     }
-        void insertData(const uint8_t *data, uint16_t len, uint16_t head_pos, uint16_t insert_pos,uint16_t tail_remain=0){
+
+    void insertData(const uint8_t *data, uint16_t len, uint16_t head_pos, uint16_t insert_pos, uint16_t tail_remain = 0)
+    {
         uint8_t *head_ptr = (uint8_t *) _buffer.data() + head_pos;
         size_t tail_pos = _buffer.size();
         if (insert_pos < head_pos) return;
         if (tail_pos < insert_pos) return;
-            if(isRTP(head_ptr,_prefix)<1) return;
+        if (isRTPHeader(head_ptr, _prefix) < 1) return;
         if (len > sizeof(_insert_buff)) return;
         memcpy(_insert_buff, data, len);
         _insert_len = len;
@@ -212,7 +248,8 @@ using namespace std;
         _tail_remain = tail_remain;
     }
 
-        void insertData(){
+    void insertData()
+    {
         if (_insert_len <= 0) return;
         _buffer.insert(_insert_pos, (char *) _insert_buff, _insert_len);
         uint8_t *head_ptr = (uint8_t *) _buffer.data() + _head_pos;
@@ -223,78 +260,83 @@ using namespace std;
             head_ptr[3] = (length) & 0x00FF;
         }
 
-            SUCSL("!!!!!!!!!!  sucess insert sign data len:%u ,buf size %lu, head_pos:%d, insert_pos:%d\n",_insert_len,_buffer.size(),_head_pos,_insert_pos);
+        SUCSL("!!!!!!!!!!  sucess insert sign data len:%u ,buf size %lu, head_pos:%d, insert_pos:%d\n", _insert_len,
+              _buffer.size(), _head_pos, _insert_pos);
         _inserted_len += _insert_len;
         _insert_tail_pos = _insert_pos + _insert_len - _tail_remain;
         _insert_len = 0;
     }
 
-        int modifyData(uint16_t pos,uint8_t *data, size_t len){
+    int modifyData(uint16_t pos, uint8_t *data, size_t len)
+    {
         if (pos + len > _buffer.size()) return -1;
         auto ptr = _buffer.data() + pos;
         memcpy((char *) ptr, data, len);
         return 1;
     }
 
+    void setPause(bool enable) { _pop_pause = enable; }
 
-        void setPause(bool enable){
+    bool isPause() { return _pop_pause; }
-            _pop_pause=enable;
-        }
-        bool isPause(){
-            return _pop_pause;
-        }
 
-        int findStartHead(const uint8_t * package, uint16_t len){
+    /**
+     * @brief 查找RTP over TCP头部
+     * 
+     */
+    int findRTPOverTCPHeader(const uint8_t *package, uint16_t len)
+    {
         const uint8_t needle[] = {0x24, 0x00};
         uint8_t *sus_head = (uint8_t *) package;
         uint16_t rest_len = len;
         do {
             sus_head = (uint8_t *) memmem(sus_head, rest_len - 5, needle, 2);
             if (sus_head) {
-                    if( sus_head[4]==0x80){
+                if (sus_head[4] == 0x80) { return sus_head - package; }
-                        return sus_head-package;
-                    }
                 rest_len -= sus_head - package;
             }
             return -1;
         } while (rest_len < 5);
+
         return -1;
     }
 
-        RTPcell::Ptr assignRTPpackage(const uint8_t * package, uint16_t len,void* usr_param){ 
+    RTPcell::Ptr assignRTPpackage(const uint8_t *package, uint16_t len, void *usr_param)
-            int ret=0;
+    {
+        int rtp_header_start_pos = 0;
         _buffer.assign((char *) package, len);
         _usr_param = usr_param;
-            if(!_Rtp_cell && isRTP(package,len)<1){
+        if (!_Rtp_cell && isRTPHeader(package, len) != kFounded) {
-                if(_is_UDP) ret=0;
+            if (_is_UDP)
-                else ret=findStartHead(package,len);
+                rtp_header_start_pos = 0;
-                if(ret<0) return nullptr;
+            else
+                rtp_header_start_pos = findRTPOverTCPHeader(package, len);
+
+            if (rtp_header_start_pos < 0) return nullptr;
         }
         auto buf_ptr = (uint8_t *) _buffer.data();
 
         if (_Rtp_cell && _Rtp_cell->_missing_head) {
-                auto lacklen=_Rtp_cell->getLackLen();
+            const auto lacklen = _Rtp_cell->getLackLen();
+            // 如果长度已经凑齐Header Length
             if (lacklen <= len) {
                 _Rtp_cell->append_data(package, lacklen);
-                    if(isRTP(_Rtp_cell->_buffer,_Rtp_cell->_assign_len)!=1) {
+                if (isRTPHeader(_Rtp_cell->_buffer, _Rtp_cell->_assign_len) != kFounded) {
                     _Rtp_cell = nullptr;
                     assignRTPpackage(package, len, usr_param);
-                    }else
+                } else {
-                    {
+                    rtp_header_start_pos += lacklen;
-                        ret+=lacklen;
                     _Rtp_cell->setRTPlen();
                     // lacklen=_Rtp_cell->getLackLen();
                     // _Rtp_cell->append_data(package,lacklen);
                     // offeset+=lacklen;
                 }
 
-                }
+            } else {
-                else{
                 _Rtp_cell = nullptr;
                 assignRTPpackage(package, len, usr_param);
             }
         }
-            return splitRTP(buf_ptr,len,ret);
+        return splitRTP(buf_ptr, len, rtp_header_start_pos);
     }
 
 private:
@@ -308,43 +350,40 @@ using namespace std;
     //         /* code */
     //     }
 
-
-
     // }
-        RTPcell::Ptr splitRTP(const uint8_t * package, uint16_t len,uint16_t ofst=0){
+    RTPcell::Ptr splitRTP(const uint8_t *package, uint16_t len, uint16_t ofst = 0)
+    {
         int lacklen = 0;
         uint16_t offeset = ofst;
 
-
+        do {
-            do
-            {
             if (!_Rtp_cell) {
                 _Rtp_cell = makeRTPcell(package + offeset, len - offeset, &offeset);
                 if (!_Rtp_cell) {
                     WRNGL("NOT RTP PACKAGE \n");
                     return nullptr;
                 }
+                // 尚未确定是否构成一个RTP包，先返回等待数据
                 if (_Rtp_cell->_missing_head) return _Rtp_cell;
             }
 
             lacklen = _Rtp_cell->getLackLen();
+            // 如果长度已经凑齐，调用回调函数解析
             if (lacklen <= 0) {
                 _rtp_cb(_Rtp_cell);
                 _Rtp_cell.reset();
                 if ((int) len - (int) offeset == 0) return nullptr;
-                }
+            } else {
-                else{
                 if ((int) len - (int) offeset - lacklen < 0) return _Rtp_cell;
 
-                    auto ret=isRTP(package+offeset+lacklen,len-offeset-lacklen,true);
+                auto ret = isRTPHeader(package + offeset + lacklen, len - offeset - lacklen, true);
                 if (ret >= 0) {
                     _Rtp_cell->append_data(package + offeset, lacklen);
                     offeset += lacklen;
 
-                    }
+                } else {
-                    else{
 
-                        ret=findStartHead(package,len);
+                    ret = findRTPOverTCPHeader(package, len);
                     if (ret < 0) {
                         WRNGL("NOT CONTINUE RTP PACKAGE \n");
                         return nullptr;
@@ -352,12 +391,10 @@ using namespace std;
                     NOTICEL("NOT CONTINUE RTP PACKAGE BUT START NEW \n");
                     _Rtp_cell = nullptr;
                     offeset += ret;
-                        
                 }
             }
 
         } while (1);
-            
     }
 
 private:
@@ -372,16 +409,15 @@ using namespace std;
     uint16_t _insert_pos;
     uint16_t _tail_remain;
     bool _is_UDP = false;
+
 public:
     string _buffer;
     uint16_t _insert_tail_pos;
     uint16_t _inserted_len;
     void *_usr_param = nullptr;
-
 };
 
-    class RTPHeader
+class RTPHeader {
-    {
 private:
     bool _mark;
     uint8_t _PT;
@@ -389,13 +425,13 @@ using namespace std;
     uint32_t _timestamp;
     uint32_t _ssrc;
     uint8_t _header_buff[16];
+
 public:
-        RTPHeader(){};
+    RTPHeader() {}
-        ~RTPHeader(){};
+
-        void setRawHeader(uint8_t* data,uint8_t len){
+    ~RTPHeader() {}
-            memcpy(_header_buff,data,len);
+
-        }
+    void setRawHeader(uint8_t *data, uint8_t len) { memcpy(_header_buff, data, len); }
 };
 
-
 #endif

src/layer2/frame.cpp

@@ -0,0 +1,39 @@
+#include "layer2/frame.h"
+
+Frame::Ptr
+Frame::CreateUDPFrame(const uint8_t *const data, std::size_t len)
+{
+    return std::shared_ptr<Frame>(new Frame(kUDP, data, len));
+}
+
+Frame::Ptr
+Frame::CreateTCPFrame(const uint8_t *const data, std::size_t len)
+{
+    return std::shared_ptr<Frame>(new Frame(kTCP, data, len));
+}
+
+Frame::Type
+Frame::type() const
+{
+    return _type;
+}
+
+void *
+Frame::user_param() const
+{
+    return _user_param;
+}
+
+void
+Frame::set_user_param(void *param)
+{
+    _user_param = param;
+}
+
+const std::vector<uint8_t> &
+Frame::data() const
+{
+    return _data;
+}
+
+Frame::Frame(Type type, const uint8_t *const data, std::size_t len) : _type(type), _data(data, data + len) {}

src/layer2/frame.h

@@ -0,0 +1,37 @@
+#ifndef AW_SECURITY_MEDIA_LIB_LAYER2_FRAME_H
+#define AW_SECURITY_MEDIA_LIB_LAYER2_FRAME_H
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+class Frame {
+public:
+    using Ptr = std::shared_ptr<Frame>;
+
+    enum Type {
+        kTCP,
+        kUDP,
+    };
+
+    static Frame::Ptr CreateUDPFrame(const uint8_t *const data, std::size_t len);
+    static Frame::Ptr CreateTCPFrame(const uint8_t *const data, std::size_t len);
+
+    Type type() const;
+    const std::vector<uint8_t> &data() const;
+
+    void *user_param() const;
+    void set_user_param(void *param);
+
+private:
+    Frame(Type type, const uint8_t *const data, std::size_t len);
+
+private:
+    Type _type;
+    std::vector<uint8_t> _data;
+    void *_user_param = nullptr;
+};
+
+#endif// AW_SECURITY_MEDIA_LIB_LAYER2_FRAME_H

src/layer2/l2_manager.cpp

@@ -0,0 +1,37 @@
+#include "l2_manager.h"
+#include "HuaWei/HWcommon.h"
+
+void
+L2Manager::FrameInput(Frame::Ptr frame)
+{
+    std::lock_guard<std::mutex> _(_mutex);
+    OnFrameInput(frame);
+
+    _frame_queue.push(frame);
+}
+
+Frame::Ptr
+L2Manager::FrameOutput()
+{
+    std::lock_guard<std::mutex> _(_mutex);
+    auto frame = _frame_queue.front();
+    _frame_queue.pop();
+
+    OnFrameOutput(_frame_queue.front());
+
+    return frame;
+}
+
+std::size_t
+L2Manager::size() const
+{
+    std::lock_guard<std::mutex> _(_mutex);
+    return _frame_queue.size();
+}
+
+bool
+L2Manager::empty() const
+{
+    std::lock_guard<std::mutex> _(_mutex);
+    return _frame_queue.empty();
+}

src/layer2/l2_manager.h

@@ -0,0 +1,25 @@
+#ifndef AW_SECURITY_MEDIA_LIB_LAYER2_L2_MANAGER_H
+#define AW_SECURITY_MEDIA_LIB_LAYER2_L2_MANAGER_H
+
+#pragma once
+
+#include "HuaWei/queue.h"
+#include "layer2/frame.h"
+#include "sigslot.h"
+#include <queue>
+
+class L2Manager {
+public:
+    sigslot::signal1<Frame::Ptr> OnFrameInput;
+    sigslot::signal1<Frame::Ptr> OnFrameOutput;
+
+    void FrameInput(Frame::Ptr frame);
+    Frame::Ptr FrameOutput();
+    std::size_t size() const;
+    bool empty() const;
+
+private:
+    std::queue<Frame::Ptr> _frame_queue;
+    mutable std::mutex _mutex;
+};
+#endif// AW_SECURITY_MEDIA_LIB_LAYER2_L2_MANAGER_H

src/layer4/rtp_manager.cpp

@@ -0,0 +1,77 @@
+#include "layer4/rtp_manager.h"
+#include "HuaWei/HWcommon.h"
+
+RTPManager::RTPManager() : _stopped(false), _cv(), _mutex(), _packet_queue(), _rtp_decoder()
+{
+    _worker = new std::thread([this] { WorkProc(); });
+    _rtp_decoder.setOnDecode([this](const uint8_t *packet, int bytes, uint32_t timestamp, int flags) {
+        OnDecodeCB(packet, bytes, timestamp, flags);
+    });
+}
+
+RTPManager::~RTPManager()
+{
+    _stopped.store(true);
+    _worker->join();
+    delete _worker;
+}
+
+void
+RTPManager::OnFrame(Frame::Ptr frame)
+{
+    RTPPacket::Ptr packet = RTPPacket::CreateUDP(frame->data().data(), frame->data().size());
+    if (packet->packet_type() == RTPPacket::kTCP) {
+        WRNGL("Unsupport TCP\n");
+    } else if (packet->packet_type() == RTPPacket::kUDP) {
+        if (!packet->PacketIsComplete()) {
+            INFOL("Incomplete RTP Packet\n");
+            return;
+        }
+        Enqueue(packet);
+    }
+}
+
+void
+RTPManager::Enqueue(RTPPacket::Ptr packet)
+{
+    std::lock_guard<std::mutex> _(this->_mutex);
+    _packet_queue.push(packet);
+    _cv.notify_all();
+}
+
+RTPPacket::Ptr
+RTPManager::Dequeue(bool wait)
+{
+    std::unique_lock<std::mutex> _(this->_mutex);
+    if (wait) {
+        _cv.wait(_, [this] { return !_packet_queue.empty(); });
+    } else if (_packet_queue.empty()) {
+        return nullptr;
+    }
+
+    auto packet = _packet_queue.front();
+    _packet_queue.pop();
+    return packet;
+}
+
+void
+RTPManager::OnDecodeCB(const uint8_t *packet, int bytes, uint32_t timestamp, int flags)
+{
+    TRACEL("RTP Packet: %p, %d, %u, %d\n", packet, bytes, timestamp, flags);
+    OnPSFrame(packet, bytes, timestamp, flags);
+}
+
+void
+RTPManager::WorkProc()
+{
+    while (!_stopped.load(std::memory_order_relaxed)) {
+        RTPPacket::Ptr packet = Dequeue(true);
+        if (!packet) {
+            ERROL("RTP Packet is nullptr\n");
+            continue;
+        }
+        TRACEL("RTP Pcket seq: %u , ssrc=%u, len=%lu\n", packet->sequence_number(), packet->ssrc(),
+               packet->rtp_packet_size());
+        _rtp_decoder.decodeRtp(packet->rtp_packet(), packet->rtp_packet_size());
+    }
+}

src/layer4/rtp_manager.h

@@ -0,0 +1,45 @@
+#ifndef AW_SECURITY_MEDIA_LIB_LAYER3_RTP_MANAGER_H
+#define AW_SECURITY_MEDIA_LIB_LAYER3_RTP_MANAGER_H
+
+#pragma once
+
+#include "GB28181/RtpDecoder.h"
+#include "HuaWei/queue.h"
+#include "layer2/frame.h"
+#include "layer4/rtp_packet.h"
+#include "sigslot.h"
+#include <atomic>
+
+class RTPManager : public sigslot::has_slots<> {
+public:
+    // packet, bytes, timestamp, flags
+    sigslot::signal4<const uint8_t *, int, uint32_t, int> OnPSFrame;
+    using Ptr = std::shared_ptr<RTPManager>;
+
+public:
+    RTPManager();
+    ~RTPManager();
+
+    void OnFrame(Frame::Ptr frame);
+
+private:
+    void Enqueue(RTPPacket::Ptr packet);
+    RTPPacket::Ptr Dequeue(bool wait = true);
+
+    void WorkProc();
+
+    // typedef function<void(const uint8_t *packet, int bytes, uint32_t timestamp, int flags)> on_decode_t;
+    void OnDecodeCB(const uint8_t *packet, int bytes, uint32_t timestamp, int flags);
+
+private:
+    std::atomic<bool> _stopped;
+    std::thread *_worker;
+
+    std::mutex _mutex;
+    std::condition_variable _cv;
+    std::queue<RTPPacket::Ptr> _packet_queue;
+
+    RtpDecoder _rtp_decoder;
+};
+
+#endif// AW_SECURITY_MEDIA_LIB_LAYER3_RTP_MANAGER_H

src/layer4/rtp_packet.cpp

@@ -0,0 +1,216 @@
+#include "layer4/rtp_packet.h"
+#include "HuaWei/HWcommon.h"
+#include <arpa/inet.h>
+#include <assert.h>
+#include <limits>
+
+RTPPacket::Ptr
+RTPPacket::CreateUDP(const uint8_t *data, std::size_t len)
+{
+    auto p = std::shared_ptr<RTPPacket>(new RTPPacket(kUDP));
+    p->Append(data, len);
+    return p;
+}
+
+RTPPacket::Ptr
+RTPPacket::CreateTCP(const uint8_t *data, std::size_t len)
+{
+    auto p = std::shared_ptr<RTPPacket>(new RTPPacket(kUDP));
+    p->Append(data, len);
+    return p;
+}
+
+std::size_t
+RTPPacket::GetHeaderSize(PacketType type)
+{
+    switch (type) {
+    case RTPPacket::kTCP:
+        return 16;
+    case RTPPacket::kUDP:
+        return 12;
+    default:
+        ERROL("Unknown packet type.\n");
+        return std::numeric_limits<std::size_t>::max();
+    }
+}
+
+RTPPacket::~RTPPacket() {}
+
+bool
+RTPPacket::CanBeParsed() const
+{
+    if (_data.empty()) { return true; }
+    if (packet_type() == kTCP) {
+        if (_data[0] != '$') { return false; }
+        if (_data.size() >= 5 && version() != 2) { return false; }
+    } else if (packet_type() == kUDP) {
+        if (version() != 2) { return false; }
+    } else {
+        return false;
+    }
+    return true;
+}
+
+bool
+RTPPacket::FixedHeaderIsComplete() const
+{
+    if (!CanBeParsed()) { return false; }
+
+    if (GetHeaderSize(packet_type()) > _data.size()) { return false; }
+    return true;
+}
+
+bool
+RTPPacket::HeaderIsComplete() const
+{
+    if (!FixedHeaderIsComplete()) {
+        TRACEL("FixedHeader not is complete.\n");
+        return false;
+    }
+    const std::size_t base_header_len = GetHeaderSize(packet_type());
+    // 首先检查固定头长度
+    const std::size_t fixed_header_len = base_header_len + csrc_count() * 4;
+    if (_data.size() < fixed_header_len) { return false; }
+
+    // 检查扩展头长度
+    if (has_extension()) {
+        if (_data.size() < fixed_header_len + 4) { return false; }
+        const std::size_t header_len = fixed_header_len + 4 + extension_length();
+        if (_data.size() < header_len) { return false; }
+    }
+
+    return true;
+}
+
+bool
+RTPPacket::PacketIsComplete() const
+{
+    if (!HeaderIsComplete()) { return false; }
+
+    // tcp 需要检查后续长度是否满足 tcp_payload_size
+    if (packet_type() == kTCP && tcp_payload_size() + 4 > _data.size()) { return false; }
+
+    // 如果已经是整个rtp packet，检查 padding
+    // if (has_padding()) {
+    //     const std::size_t padding_len = _data.back();
+    //     if (padding_len > _data.size() - rtp_fixed_header_size()) { return false; }
+    // }
+
+    return true;
+}
+
+void
+RTPPacket::Append(const uint8_t *const data, std::size_t len)
+{
+    std::copy(data, data + len, std::back_inserter(_data));
+}
+
+std::size_t
+RTPPacket::lack_len() const
+{
+    if (packet_type() == kTCP) {
+        if (FixedHeaderIsComplete()) {}
+    }
+    return 0;
+}
+
+std::uint8_t
+RTPPacket::version() const
+{
+    return *rtp_header() >> 6;
+}
+
+bool
+RTPPacket::has_padding() const
+{
+    return *rtp_header() & 0x20;
+}
+
+bool
+RTPPacket::has_extension() const
+{
+    return *rtp_header() & 0x10;
+}
+
+std::uint8_t
+RTPPacket::csrc_count() const
+{
+    uint8_t cnt = *rtp_header() & 0x0f;
+    return cnt;
+}
+
+bool
+RTPPacket::has_marker() const
+{
+    return *(rtp_header() + 1) & 0x80;
+}
+
+std::uint8_t
+RTPPacket::payload_type() const
+{
+    return *(rtp_header() + 1) & 0x7f;
+}
+
+std::uint16_t
+RTPPacket::sequence_number() const
+{
+    uint16_t *p = (uint16_t *) (rtp_header() + 2);
+    return ntohs(*p);
+}
+
+std::uint32_t
+RTPPacket::timestamp() const
+{
+    uint32_t *p = (uint32_t *) (rtp_header() + 4);
+    return ntohl(*p);
+}
+
+std::uint32_t
+RTPPacket::ssrc() const
+{
+    uint32_t *p = (uint32_t *) (rtp_header() + 8);
+    return ntohl(*p);
+}
+
+std::size_t
+RTPPacket::tcp_payload_size() const
+{
+    assert(_type == kTCP && "Must be TCP packet.\n");
+    if (CanBeParsed() && _data.size() >= 4) { return ntohs(*(uint16_t *) (_data.data() + 2)); }
+    return 0;
+}
+
+std::uint8_t *
+RTPPacket::rtp_packet() const
+{
+    return rtp_header();
+}
+
+std::size_t
+RTPPacket::rtp_packet_size() const
+{
+    return _data.size() - _rtp_header_offset;
+}
+
+std::uint8_t *
+RTPPacket::rtp_header() const
+{
+    return const_cast<std::uint8_t *>(_data.data()) + _rtp_header_offset;
+}
+
+std::size_t
+RTPPacket::rtp_fixed_header_size() const
+{
+    return 12 + csrc_count() * 4;
+}
+
+std::size_t
+RTPPacket::extension_length() const
+{
+    if (!has_extension()) { return 0; }
+    auto extension_header = rtp_header() + rtp_fixed_header_size();
+    uint16_t *p = (uint16_t *) (extension_header + 2);
+    return ntohs(*p);
+}
+
+RTPPacket::RTPPacket(PacketType type) : _type(type), _rtp_header_offset(type == kTCP ? 4 : 0) {}

src/layer4/rtp_packet.h

@@ -0,0 +1,75 @@
+#ifndef AW_SECURITY_MEDIA_LIB_LAYER3_RTP_PACKET_H
+#define AW_SECURITY_MEDIA_LIB_LAYER3_RTP_PACKET_H
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+class RTPPacket {
+public:
+    using Ptr = std::shared_ptr<RTPPacket>;
+
+    enum PacketType {
+        kTCP,
+        kUDP,
+    };
+
+    class RTPHeaderExtension {
+    public:
+        RTPHeaderExtension(RTPPacket::Ptr rtp_packet);
+        std::uint16_t profile();
+        std::uint16_t length();
+        std::uint16_t data();
+    };
+
+public:
+    static RTPPacket::Ptr CreateUDP(const uint8_t *data, std::size_t len);
+    static RTPPacket::Ptr CreateTCP(const uint8_t *data, std::size_t len);
+    static std::size_t GetHeaderSize(PacketType type);
+
+    ~RTPPacket();
+    bool CanBeParsed() const;
+    bool FixedHeaderIsComplete() const;
+    bool HeaderIsComplete() const;
+    bool PacketIsComplete() const;
+
+    void Append(const uint8_t *const data, std::size_t len);
+    std::size_t lack_len() const;
+
+    PacketType packet_type() const { return _type; }
+
+    // Header Info
+    std::uint8_t version() const;
+    bool has_padding() const;
+    bool has_extension() const;
+    std::uint8_t csrc_count() const;
+    bool has_marker() const;
+    std::uint8_t payload_type() const;
+    std::uint16_t sequence_number() const;
+    std::uint32_t timestamp() const;
+    std::uint32_t ssrc() const;
+
+    std::size_t tcp_payload_size() const;
+
+    std::uint8_t *rtp_packet() const;
+    std::size_t rtp_packet_size() const;
+
+    std::size_t rtp_fixed_header_size() const;
+    std::size_t extension_length() const;
+    std::size_t padding_length() const;
+
+private:
+    RTPPacket(PacketType type);
+    std::uint8_t *rtp_header() const;
+
+private:
+    PacketType _type;
+    std::vector<std::uint8_t> _data;
+    // for tcp: offset = 4
+    // for udp: offset = 0
+    std::size_t _rtp_header_offset;
+};
+
+#endif// AW_SECURITY_MEDIA_LIB_LAYER3_RTP_PACKET_H