代码暂存，未完成

Hu Chunming
1 parent 8a3c2932
Showing 11 changed files with 872 additions and 23 deletions
.vscode/settings.json
src/decoder/gb28181/rtp/FFRtpParser.cpp
src/decoder/gb28181/rtp/FFRtpParser.h
src/decoder/gb28181/rtp/RTPReceiver.cpp
src/decoder/gb28181/rtp/RTPReceiver2.cpp
src/decoder/gb28181/rtp/RTPReceiver2.h
src/decoder/gb28181/rtp/RTPTcpReceiver.cpp
src/decoder/gb28181/rtp/RTPTcpReceiver.h
src/decoder/gb28181/rtp/Rtp.cpp
src/decoder/gb28181/rtp/Rtp.h
src/demo/demo.cpp
@@ -79,7 +79,9 @@
     "forward_list": "cpp",
     "source_location": "cpp",
     "slist": "cpp",
-    "valarray": "cpp"
+    "valarray": "cpp",
+    "hash_map": "cpp",
+    "hash_set": "cpp"
   },
   "C_Cpp_Runner.cCompilerPath": "gcc",
   "C_Cpp_Runner.cppCompilerPath": "g++",
+//
+// Created by bxc on 2023/4/18.
+// 作者：北小菜
+// 邮箱：bilibili_bxc@126.com
+// 西瓜视频主页：https://www.ixigua.com/home/4171970536803763
+// 哔哩哔哩主页：https://space.bilibili.com/487906612/
+//
+
+#include "FFRtpParser.h"
+#include "Utils.h"
+#include <string.h>
+
+int avio_read_packet(void* opaque, uint8_t* buf, int buffsize){
+	FFRtpParser* player = (FFRtpParser*)opaque;
+
+	int ret = 0;
+	if (player->bufferSize >= buffsize)
+	{
+		memcpy(buf, player->buffer, buffsize);
+		player->bufferSize = player->bufferSize - buffsize;
+		memmove(player->buffer, player->buffer + buffsize, player->bufferSize);
+		ret = buffsize;
+
+		LOG_INFO("avio_read_packet=%d", buffsize);
+	}
+	return ret;
+}
+
+FFRtpParser::FFRtpParser()
+{
+}
+
+FFRtpParser::~FFRtpParser()
+{
+	if (mVideoCodecPar) {
+		avcodec_parameters_free(&mVideoCodecPar);
+	}
+	if (mVideoCodecCtx) {
+		avcodec_close(mVideoCodecCtx);
+		mVideoCodecCtx = nullptr;
+	}
+
+	if (mFmtCtx) {
+		avformat_close_input(&mFmtCtx);
+		mFmtCtx = nullptr;
+	}
+}
+
+bool FFRtpParser::probe()
+{
+	mFmtCtx = avformat_alloc_context();
+
+	unsigned char* avioBuff = (unsigned char*)av_malloc(1920 * 1080);
+	mAvioCtx = avio_alloc_context(avioBuff, sizeof(avioBuff), 0, this, avio_read_packet, NULL, NULL);
+	//探测流（获取码流格式）
+	if (av_probe_input_buffer2(mAvioCtx, (const AVInputFormat**)&mInputFmt, "", NULL, 0, 0) < 0){
+		LOG_ERROR("av_probe_input_buffer2 error");
+		return false;
+	}
+	mFmtCtx->pb = mAvioCtx;
+
+	//配置流参数
+	//av_dict_set(&options, "fflags", "nobuffer", 0); //不缓存直接解码
+
+	//打开流
+	if (avformat_open_input(&mFmtCtx, "", mInputFmt, &net_options) != 0)
+	{
+		LOG_ERROR("avformat_open_input error");
+		return false;
+	}
+	//获取流信息
+	if (avformat_find_stream_info(mFmtCtx, NULL) < 0)//?
+	{
+		LOG_ERROR("avformat_find_stream_info error");
+		return false;
+	}
+	//获取视频流
+	mVideoStream = av_find_best_stream(mFmtCtx, AVMEDIA_TYPE_VIDEO, -1, -1, NULL, 0);
+	if (mVideoStream < 0)
+	{
+		LOG_ERROR("av_find_best_stream error");
+		return false;
+	}
+	//获取解码信息
+	mVideoCodecPar = mFmtCtx->streams[mVideoStream]->codecpar;
+	const AVCodec* videoCodec = avcodec_find_decoder(mVideoCodecPar->codec_id);
+	if (!videoCodec){
+		LOG_ERROR("avcodec_find_decoder error");
+		return false;
+	}
+	mVideoCodecCtx = avcodec_alloc_context3(videoCodec);
+
+	//codecpar为解码器上下文赋值
+	if (avcodec_parameters_to_context(mVideoCodecCtx, mVideoCodecPar) != 0)
+	{
+		LOG_ERROR("avcodec_parameters_to_context error");
+		return false;
+	}
+
+	//设置解码器参数
+	//av_dict_set(&codec_options, "tune", "zero-latency", 0);//设置零延迟
+	//av_dict_set(&codec_options, "preset", "ultrafast", 0);//设置最模糊但是最快的解码方式
+	//av_dict_set(&codec_options, "x265-params", "qp=20", 0);//设置265量化参数
+	//量化参数：控制了视频帧中每一个宏区块（Macroblock）的压缩量。较大的数值，量化值更高，意味着更多的压缩，更低的质量，较小的数值代表相反的含义。
+
+	//打开解码器
+	if (avcodec_open2(mVideoCodecCtx, videoCodec, &codec_options) < 0)
+	{
+		LOG_ERROR("avcodec_open2 error");
+		return false;
+	}
+	LOG_INFO("mVideoCodecCtx->width=%d,mVideoCodecCtx->height=%d", mVideoCodecCtx->width, mVideoCodecCtx->height);
+	return true;
+}
+
+void FFRtpParser::play(){
+	LOG_INFO("start");
+
+	AVPacket pkt;
+	while (av_read_frame(mFmtCtx, &pkt) >= 0) {
+		if (pkt.stream_index == mVideoStream){
+
+		}
+		av_packet_unref(&pkt);
+	}
+	LOG_INFO("end");
+}
 \ No newline at end of file
+//
+// Created by bxc on 2023/4/18.
+// 作者：北小菜
+// 邮箱：bilibili_bxc@126.com
+// 西瓜视频主页：https://www.ixigua.com/home/4171970536803763
+// 哔哩哔哩主页：https://space.bilibili.com/487906612/
+//
+
+#ifndef GB28181_RTP_FFRTPPARSER_H
+#define GB28181_RTP_FFRTPPARSER_H
+
+#include <atomic>
+
+extern "C"
+{
+	#include <libavcodec/avcodec.h>
+	#include <libavformat/avformat.h>
+	#include <libswscale/swscale.h>
+}
+
+#define RtpParser_buffer_max_size 4194304 // 4M = 4 * 1024 * 1024 = 4194304 字节
+
+class FFRtpParser
+{
+public:
+	FFRtpParser();
+	~FFRtpParser();
+public:
+	bool probe();//阻塞式探测国标流并获取解码参数
+	void play();//在探测国标流成功以后，解码并渲染国标视频流
+public:
+	std::atomic<char> buffer[RtpParser_buffer_max_size];
+	std::atomic_int   bufferSize {0};
+private:
+	AVFormatContext    * mFmtCtx;
+	AVIOContext        * mAvioCtx;
+	const AVInputFormat* mInputFmt;
+	int                  mVideoStream = -1;
+	AVCodecParameters  * mVideoCodecPar;
+	AVCodecContext     * mVideoCodecCtx;
+
+	AVDictionary* net_options;//网络连接参数
+	AVDictionary* codec_options;//编码参数
+
+};
+#endif //GB28181_RTP_FFRTPPARSER_H
 \ No newline at end of file
@@ -93,7 +93,7 @@ void RTPReceiver::ClosePsThread(){
     LOG_INFO("[{}] 3.", m_SipChannelId);
     m_bPsExit = true;
     // PS解包线程退出
-    if (m_psThreadPtr->joinable())
+    if (m_psThreadPtr && m_psThreadPtr->joinable())
     {
         m_psThreadPtr->join();
         delete m_psThreadPtr;
+#include "RTPReceiver2.h"
+#include "rtppacket.h"
+#include <thread>
+
+#include "../common_header.h"
+#include "../websocket/WebsocketClient.h"
+
+#ifdef __linux__
+#include "arpa/inet.h"
+#endif
+
+#include "Rtp.h"
+
+const int MAX_RTP_BUFFER_SIZE = 1024*1024*10;
+
+#define Server_cache_max_size 4194304 // 1M = 1 * 1024 * 1024 = 1048576 字节
+#define Server_rtp_max_size 1800
+
+
+RTPReceiver2::RTPReceiver2()
+{
+    mRecvCache = (uint8_t*)malloc(Server_cache_max_size);
+	mRecvRtpBuffer = (uint8_t*)malloc(Server_rtp_max_size);
+}
+
+RTPReceiver2::~RTPReceiver2(){
+    if (mRecvCache) {
+		free(mRecvCache);
+		mRecvCache = nullptr;
+	}
+
+	if (mRecvRtpBuffer) {
+		free(mRecvRtpBuffer);
+		mRecvRtpBuffer = nullptr;
+	}
+}
+
+int RTPReceiver2::init(const char* ip, uint16_t port, bool isUdp) {
+	if (!isUdp) {
+		LOG_INFO("tcp://%s:%d", ip, port);
+		startTcpServer(ip, port);
+	}
+	else {
+		LOG_INFO("udp://%s:%d", ip, port);
+		startUdpServer(ip, port);
+	}
+}
+
+int RTPReceiver2::startUdpServer(const char* ip, uint16_t port) {
+
+	int server_fd, ret;
+    struct sockaddr_in ser_addr;
+
+    server_fd = socket(AF_INET, SOCK_DGRAM, 0); //AF_INET:IPV4;SOCK_DGRAM:UDP
+    if(server_fd < 0)
+    {
+        printf("create socket fail!\n");
+        return -1;
+    }
+
+    memset(&ser_addr, 0, sizeof(ser_addr));
+    ser_addr.sin_family = AF_INET;
+    ser_addr.sin_addr.s_addr = htonl(INADDR_ANY); //IP地址，需要进行网络序转换，INADDR_ANY：本地地址
+    ser_addr.sin_port = htons(port);  //端口号，需要网络序转换
+
+    ret = bind(server_fd, (struct sockaddr*)&ser_addr, sizeof(ser_addr));
+    if(ret < 0)
+    {
+        printf("socket bind fail!\n");
+        return -1;
+    }
+
+
+	char recvBuf[10000];
+	int  recvBufSize;
+
+    socklen_t len;
+    struct sockaddr_in clent_addr;  //clent_addr用于记录发送方的地址信息
+    while(!m_bRtpExit)
+    {
+        memset(recvBuf, 0, sizeof(recvBuf));
+        len = sizeof(clent_addr);
+        recvBufSize = recvfrom(server_fd, recvBuf, sizeof(recvBuf), 0, (struct sockaddr*)&clent_addr, &len);  //recvfrom是拥塞函数，没有数据就一直拥塞
+        if(recvBufSize <= 0) {
+            printf("recieve data fail!\n");
+            break;
+        }
+
+		if ((mPlayer->bufferSize + recvBufSize - RTP_HEADER_SIZE) < MAX_RTP_BUFFER_SIZE) {
+			memcpy(mPlayer->buffer + mPlayer->bufferSize, recvBuf + RTP_HEADER_SIZE, recvBufSize - RTP_HEADER_SIZE);
+			mPlayer->bufferSize += recvBufSize - RTP_HEADER_SIZE;
+		} else {
+			LOG_ERROR("recvBufSize = {} over GB28181Player_buffer_max_size ", recvBufSize);
+		}
+    }
+
+	close(server_fd);
+
+	return 0;
+}
+
+int RTPReceiver2::startTcpServer(const char* ip, uint16_t port) {
+
+	int  listenfd, connfd;
+    struct sockaddr_in  servaddr;
+    char  buff[4096];
+    int  n;
+
+    if( (listenfd = socket(AF_INET, SOCK_STREAM, 0)) == -1 ){
+        printf("create socket error: %s(errno: %d)\n",strerror(errno),errno);
+        return 0;
+    }
+
+    memset(&servaddr, 0, sizeof(servaddr));
+    servaddr.sin_family = AF_INET;
+    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
+    servaddr.sin_port = htons(port);
+
+    if( bind(listenfd, (struct sockaddr*)&servaddr, sizeof(servaddr)) == -1){
+        printf("bind socket error: %s(errno: %d)\n",strerror(errno),errno);
+        return 0;
+    }
+
+    if( listen(listenfd, 10) == -1){
+        printf("listen socket error: %s(errno: %d)\n",strerror(errno),errno);
+        return 0;
+    }
+
+
+	char recvBuf[10000];
+	int  recvBufSize = 0;
+
+	while (!m_bRtpExit)
+	{
+		LOG_INFO("阻塞监听新连接...");
+		// 阻塞接收请求 start
+
+		int clientFd = accept(listenfd, (struct sockaddr*)NULL, NULL);
+		if (clientFd < 0) {
+			LOG_ERROR("accept connection error");
+			continue;
+		}
+		// 阻塞接收请求 end
+		LOG_INFO("发现新连接：clientFd=%d", clientFd);
+		
+		while (!m_bRtpExit) {
+			recvBufSize = recv(clientFd, recvBuf, sizeof(recvBuf), 0);
+			if (recvBufSize <= 0) {
+				LOG_ERROR("::recv error: clientFd={},recvBufSize={}", clientFd, recvBufSize);
+				break;
+			}
+
+			parseTcpData(recvBuf, recvBufSize);
+		}
+
+		close(clientFd);
+		LOG_INFO("关闭连接 clientFd={}", clientFd);
+
+	}
+
+	close(listenfd);
+	return 0;
+}
+
+void RTPReceiver2::parseTcpData(char* recvBuf, int recvBufSize) {
+
+	if ((mRecvCacheSize + recvBufSize) > Server_cache_max_size) {
+		LOG_ERROR("超过缓冲容量上限，忽略本次读取的数据。mRecvCacheSize=%d,recvBufSize=%d",mRecvCacheSize, recvBufSize);
+	
+	}
+	else {
+		memcpy(mRecvCache + mRecvCacheSize, recvBuf, recvBufSize);
+		mRecvCacheSize += recvBufSize;
+	}
+	//LOGI("cacheSize=%d，开始进入解析 ... ...", cacheSize);
+
+	while (true) {
+
+		if (mRecvCacheSize > 2) {
+
+			bool success = false;
+
+			if (mRecvCacheSize > 2) {
+				mRecvRtpBufferSize = ntohs(*(int16_t*)(mRecvCache));
+				if ((mRecvCacheSize - 2) >= mRecvRtpBufferSize) {
+					success = true;
+				}
+			}
+
+			if (success) {
+				mRecvCacheSize -= 2;
+				mRecvCacheSize -= mRecvRtpBufferSize;
+
+				// 提取RTP
+				memcpy(mRecvRtpBuffer, mRecvCache + 2, mRecvRtpBufferSize);
+				memmove(mRecvCache, mRecvCache + 2 + mRecvRtpBufferSize, mRecvCacheSize);
+
+				// RTP
+				RtpHeader rtpHeader;
+				parseRtpHeader(mRecvRtpBuffer, &rtpHeader);
+				printf("get a rtp seq=%d,RtpBufferSize=%d,mRecvCacheSize=%d,marker=%d,timestamp=%d\n", 
+					rtpHeader.seq, 
+					mRecvRtpBufferSize, 
+					mRecvCacheSize,rtpHeader.marker, rtpHeader.timestamp);
+
+
+				// 将从mRecvCache提取出来的rtp字节流 mRecvRtpBuffer去掉RTP_HEADER_SIZE，存储到播放器缓存中
+				if ((mPlayer->bufferSize + mRecvRtpBufferSize - RTP_HEADER_SIZE) < MAX_RTP_BUFFER_SIZE) {
+					memcpy(mPlayer->buffer + mPlayer->bufferSize, mRecvRtpBuffer + RTP_HEADER_SIZE, mRecvRtpBufferSize - RTP_HEADER_SIZE);
+					mPlayer->bufferSize += mRecvRtpBufferSize - RTP_HEADER_SIZE;
+				}
+				else {
+					LOG_ERROR("recvBufSize = %d over MAX_RTP_BUFFER_SIZE ", recvBufSize);
+				}
+
+			}
+			else {
+				//LOGI("跳出解析:cacheSize=%d,pktSize=%d", cacheSize, pktSize);
+				break;
+			}
+		}
+		else {
+			//LOGI("跳出解析:缓冲数据未发现完整数据包");
+			break;
+		}
+	}
+}
+
+int RTPReceiver2::allocRtpPort() {
+
+	WebsocketClient* pServer = WebsocketClient::getInstance();
+	int MIN_RTP_PORT = pServer->GetMinRtpPort() ;
+	int MAX_RTP_PORT = pServer->GetMaxRtpPort();
+
+	int s_rtpPort = MIN_RTP_PORT;
+
+	srand((unsigned int)time(NULL));
+	s_rtpPort = MIN_RTP_PORT + (rand() % MIN_RTP_PORT);
+
+	if (s_rtpPort % 2)
+		++s_rtpPort;
+
+	int count = 0;
+
+	while (true)
+	{
+		if (s_rtpPort >= MAX_RTP_PORT) {
+			s_rtpPort = MIN_RTP_PORT;
+			count ++;
+			if (count > 1) {
+				LOG_ERROR("[{}] - 范围内没有可用的port", m_SipChannelId);
+			}
+		}
+
+		int i = 0;
+		for (; i < 2; i++) {
+			sockaddr_in sRecvAddr;
+			int s = socket(AF_INET, SOCK_DGRAM, 0);
+
+			sRecvAddr.sin_family = AF_INET;        
+			sRecvAddr.sin_addr.s_addr = htonl(INADDR_ANY);    
+			sRecvAddr.sin_port = htons(s_rtpPort + i); 
+
+			int nResult = bind(s, (sockaddr *)&sRecvAddr, sizeof(sRecvAddr));
+			if (nResult != 0) {
+				break;
+			}
+
+			nResult = close(s);
+			if (nResult != 0) {
+				LOG_ERROR("[{}] - closesocket failed : {}", m_SipChannelId, nResult);
+				break;
+			}
+		}
+
+		if (i == 2)
+			break;
+
+		s_rtpPort += 2;
+	}
+
+	return s_rtpPort;
+}
+
+void RTPReceiver2::RequestStreamFailed() {
+	m_bRtpExit = true;
+}
 \ No newline at end of file
+#ifndef _RTP_RECEIVER_H_
+#define _RTP_RECEIVER_H_
+
+#include <stdint.h>
+#include <atomic>
+#include <thread>
+
+using namespace std;
+
+
+class RTPReceiver2{
+
+public:
+    RTPReceiver2();
+    virtual ~RTPReceiver2();
+
+	int init(const char* ip, uint16_t port, bool isUdp);
+
+	void RequestStreamFailed();
+
+	int allocRtpPort();
+
+private:
+	int startUdpServer(const char* ip, uint16_t port);
+	int startTcpServer(const char* ip, uint16_t port);
+
+	void parseTcpData(char* recvBuf, int recvBufSize);
+
+public:
+    uint8_t* mRecvCache {nullptr};
+	uint64_t mRecvCacheSize {0};
+
+	uint8_t* mRecvRtpBuffer {nullptr};    // 从mRecvCache提取出来的rtp字节流
+	int16_t  mRecvRtpBufferSize {0};// 从mRecvCache提取出来的rtp字节流总长度 （rtpHeader+rtpBody）
+
+	bool m_bRtpExit {false};
+};
+
+#endif  // _RTP_RECEIVER_H_
 \ No newline at end of file
@@ -110,8 +110,8 @@ RTPTcpReceiver::RTPTcpReceiver()
 }
 RTPTcpReceiver::~RTPTcpReceiver(){
-    if (IsOpened())
-		Close();
+
+	Close();
     if(m_rtpSessionPtr != nullptr){
         delete m_rtpSessionPtr;
@@ -149,7 +149,7 @@ bool RTPTcpReceiver::Open(string channel_id){
 }
 bool RTPTcpReceiver::IsOpened(){
-    LOG_INFO("[{}] isopng:{} ", m_SipChannelId, m_bOpened);
+    LOG_INFO("[{}] isopen:{} ", m_SipChannelId, m_bOpened);
     return m_bOpened;
 }
@@ -176,8 +176,7 @@ void RTPTcpReceiver::close_task(){
     LOG_DEBUG("[{}] 1.", m_SipChannelId);
     // rtp接收线程退出
-    if (m_rtpThread.joinable())
-    {
+    if (m_rtpThread.joinable()) {
         m_rtpThread.join();
     }
@@ -239,7 +238,9 @@ int RTPTcpReceiver::initSession(int localPort){
     m_rtpThread = std::thread(rtp_revc_thread_, this);
     m_listenFinishThread = std::thread(listen_finish_thread_, this);
-    InitPS();
+    if (InitPS() != 0) {
+		return false;
+	}
     bool bRet = RequestStream();
     if (!bRet)
@@ -253,6 +254,7 @@ int RTPTcpReceiver::initSession(int localPort){
     return 0;
 }
+// 图灵版本的请求
 int RTPTcpReceiver::OnRtpRecv()
 {
     if(nullptr == m_rtpSessionPtr){
@@ -360,6 +362,114 @@ end_flag:
     return 0;
 }
+
+int RTPTcpReceiver::OnRtpRecv2()
+{
+    if(nullptr == m_rtpSessionPtr){
+        return -1;
+    }
+
+    LOG_INFO("[{}] OnRtpRecv started, m_nListener : {}", m_SipChannelId, m_nListener);
+
+    sockaddr_in   clientAddr;
+	int  nLen = sizeof(sockaddr_in);
+	SocketType nServer = -1;
+    
+    LOG_INFO("[{}] Poll started.", m_SipChannelId);
+    int reconn_times = 0;
+    int reaccept_times = 0;
+    bool bReconn = false;
+    while(!m_bRtpExit){
+        while(!m_bAccepted){
+            if(m_bRtpExit){
+                goto end_flag;
+            }
+            
+            while (!bReconn){
+                if(m_bRtpExit){
+                    goto end_flag;
+                }
+
+                reconn_times++;
+                if(reconn_times > 10){
+                    // 10次请求都失败，结束任务
+                    m_bRtpExit = true;
+                    goto end_flag;
+                }
+                LOG_DEBUG("[{}] RequestStream...", m_SipChannelId);
+                bReconn = RequestStream();
+                if (bReconn){
+                    LOG_DEBUG("[{}] RequestStream, True", m_SipChannelId);
+                   continue;
+                }
+                LOG_DEBUG("[{}] RequestStream, False", m_SipChannelId);
+                
+                std::this_thread::sleep_for(std::chrono::seconds(5));
+            }
+            
+            LOG_DEBUG("[{}] accepting...", m_SipChannelId);
+            nServer = accept(m_nListener, (sockaddr*)&clientAddr, (socklen_t * ) &nLen);
+            if (-1 == nServer){
+                reaccept_times++;
+                LOG_DEBUG("[{}] reaccept_times =  {}", m_SipChannelId, reaccept_times);
+                if(reaccept_times > 600){
+                    LOG_DEBUG("[{}] reaccept_times > 600", m_SipChannelId);
+                    bReconn = false;
+                    reaccept_times = 0;
+                }
+                std::this_thread::sleep_for(std::chrono::milliseconds(100));
+                continue;
+            }
+            LOG_DEBUG("[{}] accept success", m_SipChannelId);
+            m_rtpSessionPtr->AddDestination(RTPTCPAddress(nServer));
+            m_bAccepted = true;
+            bReconn = false;
+            reconn_times = 0;
+            reaccept_times = 0;
+
+            LOG_INFO("[{}] nServer={}", m_SipChannelId, nServer);
+            break;
+        }
+
+        m_rtpSessionPtr->BeginDataAccess();
+        if (m_rtpSessionPtr->GotoFirstSourceWithData())
+        {
+            do
+            {
+                RTPPacket *pack;
+                
+                while ((pack = m_rtpSessionPtr->GetNextPacket()) != NULL)
+                {
+                    // LOG_DEBUG("[{}] time: {} ", m_SipChannelId, UtilTools::get_cur_time_ms());
+                    ParsePacket(pack);
+                    
+                    m_rtpSessionPtr->DeletePacket(pack);
+                }
+            } while (m_rtpSessionPtr->GotoNextSourceWithData());
+        }
+        
+        m_rtpSessionPtr->EndDataAccess();
+
+        m_rtpSessionPtr->Poll();
+		std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+
+end_flag:
+
+    m_rtpSessionPtr->Destroy();
+
+    if(nServer > 0){
+        close(nServer);
+    }
+    if(m_nListener > 0){
+        close(m_nListener);
+    }
+
+    LOG_INFO("[{}] OnRtpRecv exited.", m_SipChannelId);
+
+    return 0;
+}
+
 int RTPTcpReceiver::ListenFinish(){
     while(!m_bRtpExit){
         std::this_thread::sleep_for(std::chrono::seconds(3));
@@ -47,6 +47,7 @@ public:
 public:
 	int OnRtpRecv();
+    int OnRtpRecv2();
     bool ReConnect();
     int ListenFinish();
     bool isClosing();
@@ -72,6 +73,7 @@ private:
     std::thread      	m_listenFinishThread; 					// RTP接收线程
+    std::atomic_bool 	m_bNoData{false};
 };
 #endif  // _RTP_TCP_RECEIVER_H_
+//
+// Created by bxc on 2023/4/18.
+// 作者：北小菜
+// 邮箱：bilibili_bxc@126.com
+// 西瓜视频主页：https://www.ixigua.com/home/4171970536803763
+// 哔哩哔哩主页：https://space.bilibili.com/487906612/
+//
+
+#include "Rtp.h"
+#include <stdio.h>
+#include <string.h>
+
+void rtpHeaderInit(struct RtpPacket* rtpPacket, uint8_t csrcLen, uint8_t extension,
+                   uint8_t padding, uint8_t version, uint8_t payloadType, uint8_t marker,
+                   uint16_t seq, uint32_t timestamp, uint32_t ssrc){
+    rtpPacket->rtpHeader.csrcLen = csrcLen;
+    rtpPacket->rtpHeader.extension = extension;
+    rtpPacket->rtpHeader.padding = padding;
+    rtpPacket->rtpHeader.version = version;
+    rtpPacket->rtpHeader.payloadType =  payloadType;
+    rtpPacket->rtpHeader.marker = marker;
+    rtpPacket->rtpHeader.seq = seq;
+    rtpPacket->rtpHeader.timestamp = timestamp;
+    rtpPacket->rtpHeader.ssrc = ssrc;
+}
+int parseRtpHeader(uint8_t* headerBuf, struct RtpHeader* rtpHeader){
+    memset(rtpHeader,0,sizeof(*rtpHeader));
+    /* byte 0 */
+    rtpHeader->version = (headerBuf[0] & 0xC0) >> 6;
+    rtpHeader->padding = (headerBuf[0] & 0x20) >> 5;
+    rtpHeader->extension = (headerBuf[0] & 0x10) >> 4;
+    rtpHeader->csrcLen = (headerBuf[0] & 0x0F);
+    /* byte 1 */
+    rtpHeader->marker = (headerBuf[1] & 0x80) >> 7;
+    rtpHeader->payloadType = (headerBuf[1] & 0x7F);
+    /* bytes 2,3 */
+    rtpHeader->seq = ((headerBuf[2] & 0xFF) << 8) | (headerBuf[3] & 0xFF);
+    /* bytes 4-7 */
+    rtpHeader->timestamp = ((headerBuf[4] & 0xFF) << 24) | ((headerBuf[5] & 0xFF) << 16)
+            | ((headerBuf[6] & 0xFF) << 8)
+            | ((headerBuf[7] & 0xFF));
+    /* bytes 8-11 */
+    rtpHeader->ssrc = ((headerBuf[8] & 0xFF) << 24) | ((headerBuf[9] & 0xFF) << 16)
+                           | ((headerBuf[10] & 0xFF) << 8)
+                           | ((headerBuf[11] & 0xFF));
+
+    return 0;
+}
+
+//
+// Created by bxc on 2023/4/18.
+// 作者：北小菜
+// 邮箱：bilibili_bxc@126.com
+// 西瓜视频主页：https://www.ixigua.com/home/4171970536803763
+// 哔哩哔哩主页：https://space.bilibili.com/487906612/
+//
+
+#ifndef GB28181PLAYER_RTP_H
+#define GB28181PLAYER_RTP_H
+
+#include <stdint.h>
+
+#define RTP_VESION              2
+#define RTP_PAYLOAD_TYPE_H264   96
+#define RTP_PAYLOAD_TYPE_AAC    97
+
+#define RTP_HEADER_SIZE 12
+#define RTP_MAX_SIZE    1400
+
+/*
+ *
+ *    0                   1                   2                   3
+ *    7 6 5 4 3 2 1 0|7 6 5 4 3 2 1 0|7 6 5 4 3 2 1 0|7 6 5 4 3 2 1 0
+ *   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *   |V=2|P|X|  CC   |M|     PT      |       sequence number         |
+ *   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *   |                           timestamp                           |
+ *   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *   |           synchronization source (SSRC) identifier            |
+ *   +=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
+ *   |            contributing source (CSRC) identifiers             |
+ *   :                             ....                              :
+ *   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ *
+ */
+struct RtpHeader{
+    /* byte 0 */
+    uint8_t csrcLen:4;
+    uint8_t extension:1;
+    uint8_t padding:1;
+    uint8_t version:2; // 最高2位
+
+    /* byte 1 */
+    uint8_t payloadType:7;
+    uint8_t marker:1;
+
+    /* bytes 2,3 */
+    uint16_t seq;
+
+    /* bytes 4-7 */
+    uint32_t timestamp;
+
+    /* bytes 8-11 */
+    uint32_t ssrc;
+};
+struct RtpPacket{
+    struct RtpHeader rtpHeader;
+    uint8_t payload[0];
+};
+
+void rtpHeaderInit(struct RtpPacket* rtpPacket, uint8_t csrcLen, uint8_t extension,
+                   uint8_t padding, uint8_t version, uint8_t payloadType, uint8_t marker,
+                   uint16_t seq, uint32_t timestamp, uint32_t ssrc);
+
+int parseRtpHeader(uint8_t* headerBuf, struct RtpHeader* rtpHeader);
+
+#endif //GB28181PLAYER_RTP_H
+
@@ -917,7 +917,7 @@ string createTask(void *handle, std::vector&lt;algorithm_type_t&gt; algor_vec, int gi,
       tparam.ipc_url = "rtsp://admin:admin@123456@192.168.60.176:554/cam/realmonitor?channel=1&subtype=0";
       break;
     case 1:
-      tparam.ipc_url = "rtsp://122.97.218.170:8604/openUrl/V5nXRHa?params=eyJwcm90b2NhbCI6InJ0c3AiLCJjbGllbnRUeXBlIjoib3Blbl9hcGkiLCJleHByaWVUaW1lIjotMSwicHJvdG9jb2wiOiJydHNwIiwiZXhwaXJlVGltZSI6MzAwLCJlbmFibGVNR0MiOnRydWUsImV4cGFuZCI6InN0YW5kYXJkPXJ0c3Amc3RyZWFtZm9ybT1ydHAiLCJhIjoiMTBjZjM4N2JjY2Y5NDg3YzhjNWYzNjE2M2ViMWUyNTJ8MXwwfDEiLCJ0IjoxfQ==";
+      tparam.ipc_url = "rtsp://admin:ad123456@192.168.60.165:554/cam/realmonitor?channel=1&subtype=0";
       break;
     case 2:
       tparam.ipc_url = "rtsp://admin:ad123456@192.168.10.166:554/cam/realmonitor?channel=1&subtype=0";
@@ -1001,6 +1001,12 @@ string createTask(void *handle, std::vector&lt;algorithm_type_t&gt; algor_vec, int gi,
     case 27:
       tparam.ipc_url = "/data/share/data/Street_4k_265.mp4";
       break;
+    case 28:
+      tparam.ipc_url = "http://111.200.42.56:8889/gajlqt.mp4";
+      break;
+    case 29:
+      tparam.ipc_url = "http://192.168.60.179:10016/110.mp4";
+      break;
     default:
       tparam.ipc_url = "/opt/share/data/Street.uvf";
       break;
@@ -1071,7 +1077,7 @@ string createTask_dvpp28181(void *handle, std::vector&lt;algorithm_type_t&gt; algor_ve
       tparam.ipc_url = "34020000001310004065";
       break;
     case 1:
-      tparam.ipc_url = "34020000001310000001";
+      tparam.ipc_url = "34020000001320000109";
       break;
     case 2:
       tparam.ipc_url = "34020000001320000166";
@@ -1147,6 +1153,90 @@ string createTask_dvpp28181(void *handle, std::vector&lt;algorithm_type_t&gt; algor_ve
   return task_id_str;
 }
+string createTask_dvpp28181_tcp(void *handle, std::vector<algorithm_type_t> algor_vec, int gi, bool bFlag = true){
+  task_param tparam;
+
+  switch(gi){
+    case 8:
+      tparam.ipc_url = "34020000001310004065";
+      break;
+    case 9:
+      tparam.ipc_url = "34020000001320000109";
+      break;
+    case 10:
+      tparam.ipc_url = "34020000001320000166";
+      break;
+    case 11:
+      tparam.ipc_url = "32120200002160000077";
+      break;
+    case 12:
+      tparam.ipc_url = "34020000001320000207";
+      break;
+    case 13:
+      tparam.ipc_url = "34020000001310000176";
+      break;
+    default:
+      tparam.ipc_url = "34020000001310004065";
+      break;
+  }
+  
+  tparam.algor_counts = algor_vec.size();
+  tparam.dec_type = 3;
+  tparam.protocal = 1;
+
+  if (bFlag){
+    nTaskId = gi;
+  }
+
+  std::string task_id_str = "test_task_id_" + std::to_string(nTaskId);
+  tparam.task_id = task_id_str.c_str();
+
+  nTaskId++;
+
+  tparam.algor_config_params = new algor_config_param[tparam.algor_counts];
+
+  for (size_t idx = 0; idx < algor_vec.size(); ++idx)
+    set_task_params(tparam, idx, algor_vec.at(idx));
+
+  const int result_code = add_task(handle, tparam);
+  if (result_code != 0)
+    printf("[Error]: ");
+  printf("--- task_id: %s result code: %d\n", tparam.task_id, result_code);
+
+
+  // 释放参数
+  for (size_t idx = 0; idx < algor_vec.size(); ++idx) {
+    if(tparam.algor_config_params[idx].algor_type == algorithm_type_t::VIDEO_TIMING_SNAPSHOT) {
+      algor_config_param_road_work* algor_param = (algor_config_param_road_work*)tparam.algor_config_params[idx].algor_init_config_param->algor_param;
+      delete algor_param;
+      algor_basic_config_param_t* basic_param = (algor_basic_config_param_t*)tparam.algor_config_params[idx].algor_init_config_param->basic_param;
+      delete basic_param;
+
+      algor_init_config_param_t* config_param = tparam.algor_config_params[idx].algor_init_config_param;
+      delete config_param;
+    } else if(tparam.algor_config_params[idx].algor_type == algorithm_type_t::VEHICLE_SOLIDLINETURNAROUND) {
+      algor_config_param_manned_incident* algor_param = (algor_config_param_manned_incident*)tparam.algor_config_params[idx].algor_init_config_param->algor_param;
+      delete algor_param;
+      algor_basic_config_param_t* basic_param = (algor_basic_config_param_t*)tparam.algor_config_params[idx].algor_init_config_param->basic_param;
+      delete basic_param;
+
+      algor_init_config_param_t* config_param = tparam.algor_config_params[idx].algor_init_config_param;
+      delete config_param;
+    } else if(tparam.algor_config_params[idx].algor_type == algorithm_type_t::VEHICLE_SOLIDLINETURNAROUND) {
+      algor_config_param_manned_incident* algor_param = (algor_config_param_manned_incident*)tparam.algor_config_params[idx].algor_init_config_param->algor_param;
+      delete algor_param;
+      algor_basic_config_param_t* basic_param = (algor_basic_config_param_t*)tparam.algor_config_params[idx].algor_init_config_param->basic_param;
+      delete basic_param;
+
+      algor_init_config_param_t* config_param = tparam.algor_config_params[idx].algor_init_config_param;
+      delete config_param;
+    }
+  }
+  delete[] tparam.algor_config_params;
+
+  return task_id_str;
+}
+
 void test_snapshot(void *handle){
    task_param tparam;
    tparam.ipc_url = "rtsp://admin:ad123456@192.168.60.165:554/cam/realmonitor?channel=1&subtype=0";
@@ -1328,36 +1418,64 @@ init_mq_conn(handle);
-char ch = 'a';
-while (ch != 'q') {
-  ch = getchar();
-  switch (ch)
+char ch[10] = "a";
+while (strcmp(ch,"q") != 0) {
+  fgets(ch,10,stdin);
+  int num = atoi(ch);
+  switch (num)
   {
-  case '0':
+  case 0:
     createTask_dvpp28181(handle, algor_vec, 0, false);  
     break;
-  case '1':
+  case 1:
     createTask_dvpp28181(handle, algor_vec, 1, false);  
     break;
-  case '2':
+  case 2:
     createTask_dvpp28181(handle, algor_vec, 2, false);  
     break;
-  case '3':
+  case 3:
     createTask_dvpp28181(handle, algor_vec, 3, false);  
     break;
-  case '4':
+  case 4:
     createTask_dvpp28181(handle, algor_vec, 4, false);  
     break;
-  case '5':
+  case 5:
     createTask_dvpp28181(handle, algor_vec, 5, false);  
     break;
-  case '6':
+  case 6:
     createTask(handle, algor_vec2, 2, false);
     break;
-  case '7':
+  case 7:
     createTask(handle, algor_vec2, 0, false);
     break;
-  case 'c':
+  case 8:
+    createTask_dvpp28181_tcp(handle, algor_vec, 8, false);  
+    break;
+  case 9:
+    createTask_dvpp28181_tcp(handle, algor_vec, 9, false);  
+    break;
+  case 10:
+    createTask_dvpp28181_tcp(handle, algor_vec, 10, false);  
+    break;
+  case 11:
+    createTask_dvpp28181_tcp(handle, algor_vec, 11, false);  
+    break;
+  case 12:
+    createTask_dvpp28181_tcp(handle, algor_vec, 12, false);  
+    break;
+  case 13:
+    createTask_dvpp28181_tcp(handle, algor_vec, 13, false);  
+    break;
+  case 14:
+    createTask(handle, algor_vec2, 28, false);
+    break;
+  case 15:
+    createTask(handle, algor_vec2, 1, false);
+    break;
+  case 16:
+    createTask(handle, algor_vec2, 29, false);
+    break;
+  case 100:
     close_all_task(handle);
     break;
   default: