Skip to content

Z
ZLMediaKit
Commit 84fde961 by xiongziliang
Options

提取RTP解析代码

parent dd2192fd
正在显示 包含 268 行增加 和 127 行删除
src/Rtsp/RtpReceiver.cpp 0 → 100644
/*
* MIT License
*
* Copyright (c) 2016 xiongziliang <771730766@qq.com>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#include <set>
#include <cmath>
#include <stdarg.h>
#include <algorithm>
#include "Common/config.h"
#include "RtspPlayer.h"
#include "H264/SPSParser.h"
#include "Util/MD5.h"
#include "Util/mini.h"
#include "Util/util.h"
#include "Util/base64.h"
#include "Network/sockutil.h"
#include "RtpReceiver.h"
#define POP_HEAD(trackidx) \
auto it = _amapRtpSort[trackidx].begin(); \
onRtpSorted(it->second, trackidx); \
_amapRtpSort[trackidx].erase(it);
# define AV_RB16(x) \
((((const uint8_t*)(x))[0] << 8) | \
((const uint8_t*)(x))[1])
namespace mediakit {
RtpReceiver::RtpReceiver() {}
RtpReceiver::~RtpReceiver() {}
bool RtpReceiver::handleOneRtp(int iTrackidx,SdpTrack::Ptr &track, unsigned char *pucData, unsigned int uiLen) {
auto pt_ptr=_pktPool.obtain();
auto &rtppt=*pt_ptr;
rtppt.interleaved = track->_interleaved;
rtppt.length = uiLen + 4;
rtppt.mark = pucData[1] >> 7;
rtppt.PT = pucData[1] & 0x7F;
//序列号
memcpy(&rtppt.sequence,pucData+2,2);//内存对齐
rtppt.sequence = ntohs(rtppt.sequence);
//时间戳
memcpy(&rtppt.timeStamp, pucData+4, 4);//内存对齐
//时间戳转换成毫秒
rtppt.timeStamp = ntohl(rtppt.timeStamp) * 1000L / track->_samplerate;
//ssrc
memcpy(&rtppt.ssrc,pucData+8,4);//内存对齐
rtppt.ssrc = ntohl(rtppt.ssrc);
rtppt.type = track->_type;
if (track->_ssrc == 0) {
track->_ssrc = rtppt.ssrc;
//保存SSRC
} else if (track->_ssrc != rtppt.ssrc) {
//ssrc错误
WarnL << "ssrc错误";
if (_aui32SsrcErrorCnt[iTrackidx]++ > 10) {
track->_ssrc = rtppt.ssrc;
WarnL << "ssrc更换!";
}
return false;
}
_aui32SsrcErrorCnt[iTrackidx] = 0;
rtppt.payload[0] = '$';
rtppt.payload[1] = rtppt.interleaved;
rtppt.payload[2] = (uiLen & 0xFF00) >> 8;
rtppt.payload[3] = (uiLen & 0x00FF);
rtppt.offset = 16;
int csrc = pucData[0] & 0x0f;
int ext = pucData[0] & 0x10;
rtppt.offset += 4 * csrc;
if (ext) {
if(uiLen < rtppt.offset){
return false;
}
/* calculate the header extension length (stored as number of 32-bit words) */
ext = (AV_RB16(pucData + rtppt.offset - 2) + 1) << 2;
rtppt.offset += ext;
}
memcpy(rtppt.payload + 4, pucData, uiLen);
/////////////////////////////////RTP排序逻辑///////////////////////////////////
if(rtppt.sequence != (uint16_t)(_aui16LastSeq[iTrackidx] + 1) && _aui16LastSeq[iTrackidx] != 0){
//包乱序或丢包
_aui64SeqOkCnt[iTrackidx] = 0;
_abSortStarted[iTrackidx] = true;
//WarnL << "包乱序或丢包:" << trackidx <<" " << rtppt.sequence << " " << _aui16LastSeq[trackidx];
}else{
//正确序列的包
_aui64SeqOkCnt[iTrackidx]++;
}
_aui16LastSeq[iTrackidx] = rtppt.sequence;
//开始排序缓存
if (_abSortStarted[iTrackidx]) {
_amapRtpSort[iTrackidx].emplace(rtppt.sequence, pt_ptr);
GET_CONFIG_AND_REGISTER(uint32_t,clearCount,Rtp::kClearCount);
GET_CONFIG_AND_REGISTER(uint32_t,maxRtpCount,Rtp::kMaxRtpCount);
if (_aui64SeqOkCnt[iTrackidx] >= clearCount) {
//网络环境改善，需要清空排序缓存
_aui64SeqOkCnt[iTrackidx] = 0;
_abSortStarted[iTrackidx] = false;
while (_amapRtpSort[iTrackidx].size()) {
POP_HEAD(iTrackidx)
}
} else if (_amapRtpSort[iTrackidx].size() >= maxRtpCount) {
//排序缓存溢出
POP_HEAD(iTrackidx)
}
}else{
//正确序列
onRtpSorted(pt_ptr, iTrackidx);
}
//////////////////////////////////////////////////////////////////////////////////
return true;
}
void RtpReceiver::clear() {
CLEAR_ARR(_aui16LastSeq)
CLEAR_ARR(_aui32SsrcErrorCnt)
CLEAR_ARR(_aui64SeqOkCnt)
CLEAR_ARR(_abSortStarted)
_amapRtpSort[0].clear();
_amapRtpSort[1].clear();
}
void RtpReceiver::setPoolSize(int size) {
_pktPool.setSize(size);
}
}//namespace mediakit
src/Rtsp/RtpReceiver.h 0 → 100644
/*
* MIT License
*
* Copyright (c) 2016 xiongziliang <771730766@qq.com>
*
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*/
#ifndef ZLMEDIAKIT_RTPRECEIVER_H
#define ZLMEDIAKIT_RTPRECEIVER_H
#include <string>
#include <memory>
#include "Rtsp.h"
#include "RtspSession.h"
#include "RtspMediaSource.h"
#include "Player/PlayerBase.h"
#include "Util/util.h"
#include "Util/logger.h"
#include "Util/TimeTicker.h"
#include "Poller/Timer.h"
#include "Network/Socket.h"
#include "Network/TcpClient.h"
#include "RtspSplitter.h"
using namespace std;
using namespace toolkit;
namespace mediakit {
class RtpReceiver {
public:
RtpReceiver();
virtual ~RtpReceiver();
protected:
/**
* 输入数据指针生成并排序rtp包
* @param iTrackidx track下标索引
* @param track sdp track相关信息
* @param pucData rtp数据指针
* @param uiLen rtp数据指针长度
* @return 解析成功返回true
*/
bool handleOneRtp(int iTrackidx,SdpTrack::Ptr &track, unsigned char *pucData, unsigned int uiLen);
/**
* rtp数据包排序后输出
* @param rtppt rtp数据包
* @param trackidx track索引
*/
virtual void onRtpSorted(const RtpPacket::Ptr &rtppt, int trackidx){}
void clear();
void setPoolSize(int size);
private:
uint32_t _aui32SsrcErrorCnt[2] = { 0, 0 };
/* RTP包排序所用参数 */
uint16_t _aui16LastSeq[2] = { 0 , 0 };
uint64_t _aui64SeqOkCnt[2] = { 0 , 0};
bool _abSortStarted[2] = { 0 , 0};
map<uint32_t , RtpPacket::Ptr> _amapRtpSort[2];
RtspMediaSource::PoolType _pktPool;
};
}//namespace mediakit
#endif //ZLMEDIAKIT_RTPRECEIVER_H
src/Rtsp/RtspPlayer.cpp
...@@ -41,18 +41,11 @@ using namespace toolkit; ...@@ -41,18 +41,11 @@ using namespace toolkit;
namespace mediakit { namespace mediakit {
#define POP_HEAD(trackidx) \
auto it = _amapRtpSort[trackidx].begin(); \
onRecvRTP_l(it->second, trackidx); \
_amapRtpSort[trackidx].erase(it);
#define RTP_BUF_SIZE (4 * 1024)
const char kRtspMd5Nonce[] = "rtsp_md5_nonce"; const char kRtspMd5Nonce[] = "rtsp_md5_nonce";
const char kRtspRealm[] = "rtsp_realm"; const char kRtspRealm[] = "rtsp_realm";
RtspPlayer::RtspPlayer(void){ RtspPlayer::RtspPlayer(void){
_pktPool.setSize(64); RtpReceiver::setPoolSize(64);
} }
RtspPlayer::~RtspPlayer(void) { RtspPlayer::~RtspPlayer(void) {
DebugL<<endl; DebugL<<endl;
...@@ -65,29 +58,26 @@ void RtspPlayer::teardown(){ ...@@ -65,29 +58,26 @@ void RtspPlayer::teardown(){
erase(kRtspMd5Nonce); erase(kRtspMd5Nonce);
erase(kRtspRealm); erase(kRtspRealm);
_aTrackInfo.clear(); _aTrackInfo.clear();
_onHandshake = nullptr;
_strSession.clear(); _strSession.clear();
_uiCseq = 1;
_strContentBase.clear(); _strContentBase.clear();
RtpReceiver::clear();
CLEAR_ARR(_apUdpSock); CLEAR_ARR(_apUdpSock);
CLEAR_ARR(_aui16LastSeq)
CLEAR_ARR(_aui16FirstSeq) CLEAR_ARR(_aui16FirstSeq)
CLEAR_ARR(_aui32SsrcErrorCnt)
CLEAR_ARR(_aui64RtpRecv) CLEAR_ARR(_aui64RtpRecv)
CLEAR_ARR(_aui64SeqOkCnt)
CLEAR_ARR(_abSortStarted)
CLEAR_ARR(_aui64RtpRecv) CLEAR_ARR(_aui64RtpRecv)
CLEAR_ARR(_aui16NowSeq) CLEAR_ARR(_aui16NowSeq)
_amapRtpSort[0].clear(); CLEAR_ARR(_aiFistStamp);
_amapRtpSort[1].clear(); CLEAR_ARR(_aiNowStamp);
_pBeatTimer.reset(); _pBeatTimer.reset();
_pPlayTimer.reset(); _pPlayTimer.reset();
_pRtpTimer.reset(); _pRtpTimer.reset();
_iSeekTo = 0; _iSeekTo = 0;
CLEAR_ARR(_aiFistStamp); _uiCseq = 1;
CLEAR_ARR(_aiNowStamp); _onHandshake = nullptr;
} }
void RtspPlayer::play(const char* strUrl){ void RtspPlayer::play(const char* strUrl){
...@@ -234,7 +224,6 @@ void RtspPlayer::handleResDESCRIBE(const Parser& parser) { ...@@ -234,7 +224,6 @@ void RtspPlayer::handleResDESCRIBE(const Parser& parser) {
throw std::runtime_error("onCheckSDP faied"); throw std::runtime_error("onCheckSDP faied");
} }
CLEAR_ARR(_aui32SsrcErrorCnt)
sendSetup(0); sendSetup(0);
} }
//发送SETUP命令 //发送SETUP命令
...@@ -346,10 +335,10 @@ void RtspPlayer::handleResSETUP(const Parser &parser, unsigned int uiTrackIndex) ...@@ -346,10 +335,10 @@ void RtspPlayer::handleResSETUP(const Parser &parser, unsigned int uiTrackIndex)
return; return;
} }
if(((struct sockaddr_in *)addr)->sin_addr.s_addr != srcIP) { if(((struct sockaddr_in *)addr)->sin_addr.s_addr != srcIP) {
WarnL << "收到请他地址的UDP数据:" << inet_ntoa(((struct sockaddr_in *) addr)->sin_addr); WarnL << "收到其他地址的UDP数据:" << inet_ntoa(((struct sockaddr_in *) addr)->sin_addr);
return; return;
} }
strongSelf->handleOneRtp(i,(unsigned char *)buf->data(),buf->size()); strongSelf->handleOneRtp(i,strongSelf->_aTrackInfo[i],(unsigned char *)buf->data(),buf->size());
}); });
} }
/////////////////////////心跳///////////////////////////////// /////////////////////////心跳/////////////////////////////////
...@@ -470,105 +459,12 @@ void RtspPlayer::onRtpPacket(const char *data, uint64_t len) { ...@@ -470,105 +459,12 @@ void RtspPlayer::onRtpPacket(const char *data, uint64_t len) {
trackIdx = getTrackIndexByInterleaved(interleaved); trackIdx = getTrackIndexByInterleaved(interleaved);
} }
if (trackIdx != -1) { if (trackIdx != -1) {
handleOneRtp(trackIdx, (unsigned char *)data + 4, len - 4); handleOneRtp(trackIdx,_aTrackInfo[trackIdx],(unsigned char *)data + 4, len - 4);
} }
} }
# define AV_RB16(x) \
((((const uint8_t*)(x))[0] << 8) | \
((const uint8_t*)(x))[1])
bool RtspPlayer::handleOneRtp(int iTrackidx, unsigned char *pucData, unsigned int uiLen) {
auto &track = _aTrackInfo[iTrackidx];
auto pt_ptr=_pktPool.obtain();
auto &rtppt=*pt_ptr;
rtppt.interleaved = track->_interleaved;
rtppt.length = uiLen + 4;
rtppt.mark = pucData[1] >> 7;
rtppt.PT = pucData[1] & 0x7F;
//序列号
memcpy(&rtppt.sequence,pucData+2,2);//内存对齐
rtppt.sequence = ntohs(rtppt.sequence);
//时间戳
memcpy(&rtppt.timeStamp, pucData+4, 4);//内存对齐
//时间戳转换成毫秒
rtppt.timeStamp = ntohl(rtppt.timeStamp) * 1000L / track->_samplerate;
//ssrc
memcpy(&rtppt.ssrc,pucData+8,4);//内存对齐
rtppt.ssrc = ntohl(rtppt.ssrc);
rtppt.type = track->_type;
if (track->_ssrc == 0) {
track->_ssrc = rtppt.ssrc;
//保存SSRC
} else if (track->_ssrc != rtppt.ssrc) {
//ssrc错误
WarnL << "ssrc错误";
if (_aui32SsrcErrorCnt[iTrackidx]++ > 10) {
track->_ssrc = rtppt.ssrc;
WarnL << "ssrc更换!";
}
return false;
}
_aui32SsrcErrorCnt[iTrackidx] = 0;
rtppt.payload[0] = '$';
rtppt.payload[1] = rtppt.interleaved;
rtppt.payload[2] = (uiLen & 0xFF00) >> 8;
rtppt.payload[3] = (uiLen & 0x00FF);
rtppt.offset = 16;
int csrc = pucData[0] & 0x0f;
int ext = pucData[0] & 0x10;
rtppt.offset += 4 * csrc;
if (ext) {
if(uiLen < rtppt.offset){
return false;
}
/* calculate the header extension length (stored as number of 32-bit words) */
ext = (AV_RB16(pucData + rtppt.offset - 2) + 1) << 2;
rtppt.offset += ext;
}
memcpy(rtppt.payload + 4, pucData, uiLen);
/////////////////////////////////RTP排序逻辑/////////////////////////////////// void RtspPlayer::onRtpSorted(const RtpPacket::Ptr &rtppt, int trackidx){
if(rtppt.sequence != (uint16_t)(_aui16LastSeq[iTrackidx] + 1) && _aui16LastSeq[iTrackidx] != 0){
//包乱序或丢包
_aui64SeqOkCnt[iTrackidx] = 0;
_abSortStarted[iTrackidx] = true;
//WarnL << "包乱序或丢包:" << trackidx <<" " << rtppt.sequence << " " << _aui16LastSeq[trackidx];
}else{
//正确序列的包
_aui64SeqOkCnt[iTrackidx]++;
}
_aui16LastSeq[iTrackidx] = rtppt.sequence;
//开始排序缓存
if (_abSortStarted[iTrackidx]) {
_amapRtpSort[iTrackidx].emplace(rtppt.sequence, pt_ptr);
GET_CONFIG_AND_REGISTER(uint32_t,clearCount,Rtp::kClearCount);
GET_CONFIG_AND_REGISTER(uint32_t,maxRtpCount,Rtp::kMaxRtpCount);
if (_aui64SeqOkCnt[iTrackidx] >= clearCount) {
//网络环境改善，需要清空排序缓存
_aui64SeqOkCnt[iTrackidx] = 0;
_abSortStarted[iTrackidx] = false;
while (_amapRtpSort[iTrackidx].size()) {
POP_HEAD(iTrackidx)
}
} else if (_amapRtpSort[iTrackidx].size() >= maxRtpCount) {
//排序缓存溢出
POP_HEAD(iTrackidx)
}
}else{
//正确序列
onRecvRTP_l(pt_ptr, iTrackidx);
}
//////////////////////////////////////////////////////////////////////////////////
return true;
}
void RtspPlayer::onRecvRTP_l(const RtpPacket::Ptr &rtppt, int trackidx){
//统计丢包率 //统计丢包率
if (_aui16FirstSeq[trackidx] == 0 || rtppt->sequence < _aui16FirstSeq[trackidx]) { if (_aui16FirstSeq[trackidx] == 0 || rtppt->sequence < _aui16FirstSeq[trackidx]) {
_aui16FirstSeq[trackidx] = rtppt->sequence; _aui16FirstSeq[trackidx] = rtppt->sequence;
......
src/Rtsp/RtspPlayer.h
...@@ -40,6 +40,7 @@ ...@@ -40,6 +40,7 @@
#include "Network/Socket.h" #include "Network/Socket.h"
#include "Network/TcpClient.h" #include "Network/TcpClient.h"
#include "RtspSplitter.h" #include "RtspSplitter.h"
#include "RtpReceiver.h"
using namespace std; using namespace std;
using namespace toolkit; using namespace toolkit;
...@@ -47,7 +48,7 @@ using namespace toolkit; ...@@ -47,7 +48,7 @@ using namespace toolkit;
namespace mediakit { namespace mediakit {
//实现了rtsp播放器协议部分的功能 //实现了rtsp播放器协议部分的功能
class RtspPlayer: public PlayerBase,public TcpClient, public RtspSplitter { class RtspPlayer: public PlayerBase,public TcpClient, public RtspSplitter, public RtpReceiver {
public: public:
typedef std::shared_ptr<RtspPlayer> Ptr; typedef std::shared_ptr<RtspPlayer> Ptr;
...@@ -76,9 +77,15 @@ protected: ...@@ -76,9 +77,15 @@ protected:
* @param len * @param len
*/ */
void onRtpPacket(const char *data,uint64_t len) override ; void onRtpPacket(const char *data,uint64_t len) override ;
/**
* rtp数据包排序后输出
* @param rtppt rtp数据包
* @param trackidx track索引
*/
void onRtpSorted(const RtpPacket::Ptr &rtppt, int trackidx) override;
private: private:
void onShutdown_l(const SockException &ex); void onShutdown_l(const SockException &ex);
void onRecvRTP_l(const RtpPacket::Ptr &pRtppt, int iTrackidx);
void onRecvRTP_l(const RtpPacket::Ptr &pRtppt, const SdpTrack::Ptr &track); void onRecvRTP_l(const RtpPacket::Ptr &pRtppt, const SdpTrack::Ptr &track);
void onPlayResult_l(const SockException &ex); void onPlayResult_l(const SockException &ex);
...@@ -95,8 +102,6 @@ private: ...@@ -95,8 +102,6 @@ private:
void handleResDESCRIBE(const Parser &parser); void handleResDESCRIBE(const Parser &parser);
bool handleAuthenticationFailure(const string &wwwAuthenticateParamsStr); bool handleAuthenticationFailure(const string &wwwAuthenticateParamsStr);
void handleResPAUSE(const Parser &parser, bool bPause); void handleResPAUSE(const Parser &parser, bool bPause);
//处理一个rtp包
bool handleOneRtp(int iTrackidx, unsigned char *ucData, unsigned int uiLen);
//发送SETUP命令 //发送SETUP命令
bool sendSetup(unsigned int uiTrackIndex); bool sendSetup(unsigned int uiTrackIndex);
...@@ -110,19 +115,12 @@ private: ...@@ -110,19 +115,12 @@ private:
vector<SdpTrack::Ptr> _aTrackInfo; vector<SdpTrack::Ptr> _aTrackInfo;
function<void(const Parser&)> _onHandshake; function<void(const Parser&)> _onHandshake;
RtspMediaSource::PoolType _pktPool;
Socket::Ptr _apUdpSock[2]; Socket::Ptr _apUdpSock[2];
//rtsp info //rtsp info
string _strSession; string _strSession;
unsigned int _uiCseq = 1; unsigned int _uiCseq = 1;
uint32_t _aui32SsrcErrorCnt[2] = { 0, 0 };
string _strContentBase; string _strContentBase;
eRtpType _eType = RTP_TCP; eRtpType _eType = RTP_TCP;
/* RTP包排序所用参数 */
uint16_t _aui16LastSeq[2] = { 0 , 0 };
uint64_t _aui64SeqOkCnt[2] = { 0 , 0};
bool _abSortStarted[2] = { 0 , 0};
map<uint32_t , RtpPacket::Ptr> _amapRtpSort[2];
/* 丢包率统计需要用到的参数 */ /* 丢包率统计需要用到的参数 */
uint16_t _aui16FirstSeq[2] = { 0 , 0}; uint16_t _aui16FirstSeq[2] = { 0 , 0};
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论