GB28181学习（十七）——基于jrtplib实现tcp被动和主动发流

2023-11-23 04:20
文章标签 实现 学习 tcp gb28181 主动 十七 被动 jrtplib 发流

本文主要是介绍GB28181学习（十七）——基于jrtplib实现tcp被动和主动发流，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

前言

GB/T28181-2022实时流的传输方式介绍：https://blog.csdn.net/www_dong/article/details/134255185

基于jrtplib实现tcp被动和主动收流介绍：https://blog.csdn.net/www_dong/article/details/134451387

本文主要介绍下级平台或设备发流功能，用于对接特定的SIP服务器或上级平台。

UDP发流

流程图

在这里插入图片描述

发送端流程：

  • 初始化rtp参数；
  • 裸流数据做PS复用；
  • 组RTP包发送；

设计

  1. 初始化rtp参数
int CUdp::InitRtp_()
{RTPSessionParams sessionParams;sessionParams.SetMinimumRTCPTransmissionInterval(10);sessionParams.SetOwnTimestampUnit(1.0 / 90000.0);sessionParams.SetAcceptOwnPackets(true);sessionParams.SetMaximumPacketSize(1450);RTPUDPv4TransmissionParams transParams;transParams.SetRTPSendBuffer(2*1024*1024);transParams.SetBindIP(m_ip);transParams.SetPortbase((uint16_t)m_port);if (0 != Create(sessionParams, &transParams)){return -1;}SetDefaultPayloadType((uint8_t)m_payload);SetDefaultTimestampIncrement(3600);SetDefaultMark(true);RTPIPv4Address addr(ntohl(inet_addr(m_ip), (uint16_t)m_port);if(0 != AddDestination(addr)){return -1;}return 0;
}
  1. 流数据复用为PS
// 使用ireader开源库进行ps复用
// 初始化
CData2PS::CData2PS()
{struct ps_muxer_func_t func;func.alloc = Alloc;func.free = Free;func.write = Packet;m_ps = ps_muxer_create(&func, this);// TODO codecid待补充m_ps_stream = ps_muxer_add_stream(m_ps, PSI_STREAM_H264, nullptr, 0);
}// 塞数据
int CData2PS::InputData(void* data, int len)
{if (!m_ps)return -1;uint64_t clock = time64_now();if (0 == m_ps_clock)m_ps_clock = clock;return ps_muxer_input(m_ps, m_ps_stream, 0, (clock - m_ps_clock) * 90, (clock - m_ps_clock) * 90, data, len);
}
  1. 发送rtp包
// 调用jrtplib中SendPacket(data, len);接口发送数据// 以下为SendPacket部分源码
// 主要流程:
// 1. 构建packet
// 2. 发送rtp数据
int RTPSession::SendPacket(const void *data,size_t len,uint8_t pt,bool mark,uint32_t timestampinc)
{int status;if (!created)return ERR_RTP_SESSION_NOTCREATED;BUILDER_LOCKif ((status = packetbuilder.BuildPacket(data,len,pt,mark,timestampinc)) < 0){BUILDER_UNLOCKreturn status;}if ((status = SendRTPData(packetbuilder.GetPacket(),packetbuilder.GetPacketLength())) < 0){BUILDER_UNLOCKreturn status;}BUILDER_UNLOCKSOURCES_LOCKsources.SentRTPPacket();SOURCES_UNLOCKPACKSENT_LOCKsentpackets = true;PACKSENT_UNLOCKreturn 0;
}// 构建包
int RTPPacketBuilder::PrivateBuildPacket(const void *data,size_t len,uint8_t pt,bool mark,uint32_t timestampinc,bool gotextension,uint16_t hdrextID,const void *hdrextdata,size_t numhdrextwords)
{RTPPacket p(pt,data,len,seqnr,timestamp,ssrc,mark,numcsrcs,csrcs,gotextension,hdrextID,(uint16_t)numhdrextwords,hdrextdata,buffer,maxpacksize,GetMemoryManager());int status = p.GetCreationError();if (status < 0)return status;packetlength = p.GetPacketLength();if (numpackets == 0) // first packet{lastwallclocktime = RTPTime::CurrentTime();lastrtptimestamp = timestamp;prevrtptimestamp = timestamp;}else if (timestamp != prevrtptimestamp){lastwallclocktime = RTPTime::CurrentTime();lastrtptimestamp = timestamp;prevrtptimestamp = timestamp;}numpayloadbytes += (uint32_t)p.GetPayloadLength();numpackets++;timestamp += timestampinc;seqnr++;return 0;
}// 发送包
int RTPSession::SendRTPData(const void *data, size_t len)
{if (!m_changeOutgoingData)return rtptrans->SendRTPData(data, len);void *pSendData = 0;size_t sendLen = 0;int status = 0;status = OnChangeRTPOrRTCPData(data, len, true, &pSendData, &sendLen);if (status < 0)return status;if (pSendData){status = rtptrans->SendRTPData(pSendData, sendLen);OnSentRTPOrRTCPData(pSendData, sendLen, true);}return status;
}// 底层实现
int RTPUDPv4Transmitter::SendRTPData(const void *data,size_t len)	
{if (!init)return ERR_RTP_UDPV4TRANS_NOTINIT;MAINMUTEX_LOCKif (!created){MAINMUTEX_UNLOCKreturn ERR_RTP_UDPV4TRANS_NOTCREATED;}if (len > maxpacksize){MAINMUTEX_UNLOCKreturn ERR_RTP_UDPV4TRANS_SPECIFIEDSIZETOOBIG;}destinations.GotoFirstElement();while (destinations.HasCurrentElement()){// 调用sendto函数实现udp包的发送sendto(rtpsock,(const char *)data,len,0,(const struct sockaddr *)destinations.GetCurrentElement().GetRTPSockAddr(),sizeof(struct sockaddr_in));destinations.GotoNextElement();}MAINMUTEX_UNLOCKreturn 0;
}

tcp passive发流

流程图

在这里插入图片描述

发送端流程：

  • 上级平台或sip服务器以主动方式连接，对于下级平台或者设备（数据发送端）为被动方式；
  • 下级平台或者设备（数据发送端）启动端口监听；
  • 接收上级平台或sip服务器tcp连接请求；
  • 向上级平台或sip服务器发送流数据；

设计

  1. 创建socket、bind、listen，启动数据接收线程；
// TcpServer为封装的socket类int CGBTcpServer::Start()
{if (0 != m_localPort || m_tcpServer.get())return 0;int ret = -1;do {m_tcpServer = std::make_shared<TcpServer>(nullptr, this);if (!m_tcpServer.get())break;ret = m_tcpServer->TcpCreate();if (0 != ret)break;ret = m_tcpServer->TcpBind(m_localPort);if (0 != ret)break;ret = m_tcpServer->TcpListen(5);if (0 != ret)break;m_thread = std::thread(TCPData2PSThread, this);return 0;} while (0);Stop();return ret;
}
  1. 在线程内等待连接，连接成功后接收数据并回调至应用层处理
void CGBTcpServer::TCPData2PSWorker()
{if (!m_pspacker)m_pspacker = new(std::nothrow) CData2PS(PSTCPDataCB, this);bool bAccept = false;while (m_running){if (!bAccept){if (0 == m_tcpServer->TcpAccept()){bAccept = true;if (0 != InitRtp_()){break;}}continue;}std::this_thread::sleep_for(std::chrono::seconds(1));}
}
  1. 初始化rtp参数
int CGBTcpServer::InitRtp_()
{const int packetSize = 45678;RTPSessionParams sessionparams;sessionparams.SetProbationType(RTPSources::NoProbation);sessionparams.SetOwnTimestampUnit(1.0 / packetSize);sessionparams.SetMaximumPacketSize(packetSize + 64);m_rtpTcpTransmitter = new RTPTCPTransmitter(nullptr);m_rtpTcpTransmitter->Init(true);m_rtpTcpTransmitter->Create(65535, 0);int status = Create(sessionparams, m_rtpTcpTransmitter);if (status < 0){return status;}status = AddDestination(RTPTCPAddress(m_tcpServer->GetClientSocket()));if (0 != status)return status;SetDefaultPayloadType(96);SetDefaultMark(false);SetDefaultTimestampIncrement(160);return 0;
}
  1. 将数据复用为PS；
  2. tcp方式发包
// 调用jrtplib中SendPacket(data, len);接口发送数据// 以下为tcp方式SendPacket部分源码
int RTPTCPTransmitter::SendRTPData(const void *data,size_t len)	
{return SendRTPRTCPData(data, len);
}int RTPTCPTransmitter::SendRTPRTCPData(const void *data, size_t len)
{if (!m_init)return ERR_RTP_TCPTRANS_NOTINIT;MAINMUTEX_LOCKif (!m_created){MAINMUTEX_UNLOCKreturn ERR_RTP_TCPTRANS_NOTCREATED;}// #define RTPTCPTRANS_MAXPACKSIZE							65535if (len > RTPTCPTRANS_MAXPACKSIZE){MAINMUTEX_UNLOCKreturn ERR_RTP_TCPTRANS_SPECIFIEDSIZETOOBIG;}std::map<SocketType, SocketData>::iterator it = m_destSockets.begin();std::map<SocketType, SocketData>::iterator end = m_destSockets.end();vector<SocketType> errSockets;int flags = 0;
#ifdef RTP_HAVE_MSG_NOSIGNALflags = MSG_NOSIGNAL;
#endif // RTP_HAVE_MSG_NOSIGNALwhile (it != end){uint8_t lengthBytes[2] = { (uint8_t)((len >> 8)&0xff), (uint8_t)(len&0xff) };SocketType sock = it->first;// 调用send接口发送数据// 1. 先发送2字节头(固定格式)// 2. 再发送数据if (send(sock,(const char *)lengthBytes,2,flags) < 0 ||send(sock,(const char *)data,len,flags) < 0)errSockets.push_back(sock);++it;}MAINMUTEX_UNLOCKif (errSockets.size() != 0){for (size_t i = 0 ; i < errSockets.size() ; i++)OnSendError(errSockets[i]);}// Don't return an error code to avoid the poll thread exiting// due to one closed connection for examplereturn 0;
}

tcp active发流

流程图

在这里插入图片描述

发送端流程：

  • 上级平台或sip服务器启动tcp监听连接，对于下级平台或者设备（数据发送端）为主动方式；
  • 下级平台或者设备（数据发送端）发起tcp连接；
  • 接收上级平台或sip服务器tcp响应；
  • 向上级平台或sip服务器发送流数据；

设计

  1. 创建socket、connect、初始化rtp，启动数据接收线程
// TcpClient为封装的客户端socket类int CGBTcpClient::Start()
{if (0 != m_localPort || m_tcpClient.get())return 0;int ret = -1;do{m_tcpClient = std::make_shared<TcpClient>(nullptr, this);if (!m_tcpClient.get() || 0 != m_tcpClient->TcpCreate())break;ret = m_tcpClient->TcpConnectByTime(m_localIP.c_str(), m_localPort, 5);if (0 != ret)break;ret = InitRtp_();if (0 != ret)break;m_thread = std::thread(RTPPackerThread, this);return 0;} while (0);Stop();return ret;
}
  1. 初始化rtp参数
int CGBTcpClient::InitRtp_()
{const int packSize = 45678;RTPSessionParams sessionParams;sessionParams.SetProbationType(RTPSources::NoProbation);sessionParams.SetOwnTimestampUnit(90000.0 / 25.0);sessionParams.SetMaximumPacketSize(packSize + 64);m_rtpTcpTransmitter = new RTPTCPTransmitter(nullptr);m_rtpTcpTransmitter->Init(true);m_rtpTcpTransmitter->Create(65535, 0);if (0 != Create(sessionParams, m_rtpTcpTransmitter))return -1;if (0 != AddDestination(RTPTCPAddress(m_tcpClient->GetClientSocket())))return -1;return 0;
}
  1. 视音频数据复用为PS
  2. 发送数据，同tcp passive发流

这篇关于GB28181学习（十七）——基于jrtplib实现tcp被动和主动发流的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/415304。

相关文章

SpringBoot实现微信小程序支付功能

SpringBoot实现微信小程序支付功能

《SpringBoot实现微信小程序支付功能》小程序支付功能已成为众多应用的核心需求之一,本文主要介绍了SpringBoot实现微信小程序支付功能,文中通过示例代码介绍的非常详细,对大家的学习或者工作... 目录一、引言二、准备工作（一）微信支付商户平台配置（二）Spring Boot项目搭建（三）配置文件
阅读更多...
基于Python实现高效PPT转图片工具

基于Python实现高效PPT转图片工具

《基于Python实现高效PPT转图片工具》在日常工作中,PPT是我们常用的演示工具,但有时候我们需要将PPT的内容提取为图片格式以便于展示或保存,所以本文将用Python实现PPT转PNG工具,希望... 目录1. 概述2. 功能使用2.1 安装依赖2.2 使用步骤2.3 代码实现2.4 GUI界面3.效
阅读更多...
MySQL更新某个字段拼接固定字符串的实现

MySQL更新某个字段拼接固定字符串的实现

《MySQL更新某个字段拼接固定字符串的实现》在MySQL中,我们经常需要对数据库中的某个字段进行更新操作,本文就来介绍一下MySQL更新某个字段拼接固定字符串的实现,感兴趣的可以了解一下... 目录1. 查看字段当前值2. 更新字段拼接固定字符串3. 验证更新结果mysql更新某个字段拼接固定字符串 -
阅读更多...
java实现延迟/超时/定时问题

java实现延迟/超时/定时问题

《java实现延迟/超时/定时问题》：本文主要介绍java实现延迟/超时/定时问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Java实现延迟/超时/定时java 每间隔5秒执行一次,一共执行5次然后结束scheduleAtFixedRate 和 schedu
阅读更多...
Java Optional避免空指针异常的实现

Java Optional避免空指针异常的实现

《JavaOptional避免空指针异常的实现》空指针异常一直是困扰开发者的常见问题之一,本文主要介绍了JavaOptional避免空指针异常的实现,帮助开发者编写更健壮、可读性更高的代码,减少因... 目录一、Optional 概述二、Optional 的创建三、Optional 的常用方法四、Optio
阅读更多...
在Android平台上实现消息推送功能

在Android平台上实现消息推送功能

《在Android平台上实现消息推送功能》随着移动互联网应用的飞速发展,消息推送已成为移动应用中不可或缺的功能,在Android平台上,实现消息推送涉及到服务端的消息发送、客户端的消息接收、通知渠道（... 目录一、项目概述二、相关知识介绍2.1 消息推送的基本原理2.2 Firebase Cloud Me
阅读更多...
Spring Boot项目中结合MyBatis实现MySQL的自动主从切换功能

Spring Boot项目中结合MyBatis实现MySQL的自动主从切换功能

《SpringBoot项目中结合MyBatis实现MySQL的自动主从切换功能》：本文主要介绍SpringBoot项目中结合MyBatis实现MySQL的自动主从切换功能,本文分步骤给大家介绍的... 目录原理解析1. mysql主从复制（Master-Slave Replication）2. 读写分离3.
阅读更多...
Redis实现延迟任务的三种方法详解

Redis实现延迟任务的三种方法详解

《Redis实现延迟任务的三种方法详解》延迟任务（DelayedTask）是指在未来的某个时间点,执行相应的任务,本文为大家整理了三种常见的实现方法,感兴趣的小伙伴可以参考一下... 目录1.前言2.Redis如何实现延迟任务3.代码实现3.1. 过期键通知事件实现3.2. 使用ZSet实现延迟任务3.3
阅读更多...
基于Python和MoviePy实现照片管理和视频合成工具

基于Python和MoviePy实现照片管理和视频合成工具

《基于Python和MoviePy实现照片管理和视频合成工具》在这篇博客中,我们将详细剖析一个基于Python的图形界面应用程序,该程序使用wxPython构建用户界面,并结合MoviePy、Pill... 目录引言项目概述代码结构分析1. 导入和依赖2. 主类：PhotoManager初始化方法：__in
阅读更多...
springboot filter实现请求响应全链路拦截

springboot filter实现请求响应全链路拦截

《springbootfilter实现请求响应全链路拦截》这篇文章主要为大家详细介绍了SpringBoot如何结合Filter同时拦截请求和响应,从而实现​​日志采集自动化,感兴趣的小伙伴可以跟随小... 目录一、为什么你需要这个过滤器？​​​二、核心实现：一个Filter搞定双向数据流​​​​三、完整代码
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2