amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-11-26 20:47:08 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

修复海康rtsp点播拉流seek相关bug(#2501 #2511)

Browse Source
This commit is contained in:
xiongguangjie 2023-05-31 09:59:41 +08:00 committed by GitHub
parent 0232caf068
commit 11a39c68c4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 170 additions and 177 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

339
src/Rtsp/RtspPlayer.cpp
View File

@ -8,40 +8,35 @@
* may be found in the AUTHORS file in the root of the source tree. * may be found in the AUTHORS file in the root of the source tree.
*/ */
#include <set>
#include <cmath>
#include <algorithm>
#include <iomanip>
#include "Common/config.h"
#include "RtspPlayer.h" #include "RtspPlayer.h"
#include "Util/MD5.h" #include "Common/config.h"
#include "Util/base64.h"
#include "Rtcp/Rtcp.h" #include "Rtcp/Rtcp.h"
#include "Rtcp/RtcpContext.h" #include "Rtcp/RtcpContext.h"
#include "RtspMediaSource.h"
#include "RtspDemuxer.h" #include "RtspDemuxer.h"
#include "RtspMediaSource.h"
#include "RtspPlayerImp.h" #include "RtspPlayerImp.h"
#include "Util/MD5.h"
#include "Util/base64.h"
#include <algorithm>
#include <cmath>
#include <iomanip>
#include <set>
using namespace toolkit; using namespace toolkit;
using namespace std; using namespace std;
namespace mediakit { namespace mediakit {
enum PlayType { enum PlayType { type_play = 0, type_pause, type_seek, type_speed };
type_play = 0,
type_pause,
type_seek,
type_speed
};
RtspPlayer::RtspPlayer(const EventPoller::Ptr &poller) : TcpClient(poller){ RtspPlayer::RtspPlayer(const EventPoller::Ptr &poller)
} : TcpClient(poller) {}
RtspPlayer::~RtspPlayer(void) { RtspPlayer::~RtspPlayer(void) {
DebugL << endl; DebugL << endl;
} }
void RtspPlayer::sendTeardown(){ void RtspPlayer::sendTeardown() {
if (alive()) { if (alive()) {
if (!_content_base.empty()) { if (!_content_base.empty()) {
sendRtspRequest("TEARDOWN", _content_base); sendRtspRequest("TEARDOWN", _content_base);
@ -50,7 +45,7 @@ void RtspPlayer::sendTeardown(){
} }
} }
void RtspPlayer::teardown(){ void RtspPlayer::teardown() {
sendTeardown(); sendTeardown();
_md5_nonce.clear(); _md5_nonce.clear();
_realm.clear(); _realm.clear();
@ -69,7 +64,7 @@ void RtspPlayer::teardown(){
_on_response = nullptr; _on_response = nullptr;
} }
void RtspPlayer::play(const string &strUrl){ void RtspPlayer::play(const string &strUrl) {
RtspUrl url; RtspUrl url;
try { try {
url.parse(strUrl); url.parse(strUrl);
@ -94,32 +89,35 @@ void RtspPlayer::play(const string &strUrl){
weak_ptr<RtspPlayer> weakSelf = static_pointer_cast<RtspPlayer>(shared_from_this()); weak_ptr<RtspPlayer> weakSelf = static_pointer_cast<RtspPlayer>(shared_from_this());
float playTimeOutSec = (*this)[Client::kTimeoutMS].as<int>() / 1000.0f; float playTimeOutSec = (*this)[Client::kTimeoutMS].as<int>() / 1000.0f;
_play_check_timer.reset(new Timer(playTimeOutSec, [weakSelf]() { _play_check_timer.reset(new Timer(
auto strongSelf=weakSelf.lock(); playTimeOutSec,
if(!strongSelf) { [weakSelf]() {
auto strongSelf = weakSelf.lock();
if (!strongSelf) {
return false; return false;
} }
strongSelf->onPlayResult_l(SockException(Err_timeout,"play rtsp timeout"),false); strongSelf->onPlayResult_l(SockException(Err_timeout, "play rtsp timeout"), false);
return false; return false;
}, getPoller())); },
getPoller()));
if(!(*this)[Client::kNetAdapter].empty()){ if (!(*this)[Client::kNetAdapter].empty()) {
setNetAdapter((*this)[Client::kNetAdapter]); setNetAdapter((*this)[Client::kNetAdapter]);
} }
startConnect(url._host, url._port, playTimeOutSec); startConnect(url._host, url._port, playTimeOutSec);
} }
void RtspPlayer::onConnect(const SockException &err){ void RtspPlayer::onConnect(const SockException &err) {
if(err.getErrCode() != Err_success) { if (err.getErrCode() != Err_success) {
onPlayResult_l(err,false); onPlayResult_l(err, false);
return; return;
} }
sendOptions(); sendOptions();
} }
void RtspPlayer::onRecv(const Buffer::Ptr& buf) { void RtspPlayer::onRecv(const Buffer::Ptr &buf) {
if(_benchmark_mode && !_play_check_timer){ if (_benchmark_mode && !_play_check_timer) {
//在性能测试模式下，如果rtsp握手完毕后，不再解析rtp包 // 在性能测试模式下，如果rtsp握手完毕后，不再解析rtp包
_rtp_recv_ticker.resetTime(); _rtp_recv_ticker.resetTime();
return; return;
} }
@ -132,21 +130,21 @@ void RtspPlayer::onRecv(const Buffer::Ptr& buf) {
} }
void RtspPlayer::onError(const SockException &ex) { void RtspPlayer::onError(const SockException &ex) {
//定时器_pPlayTimer为空后表明握手结束了 // 定时器_pPlayTimer为空后表明握手结束了
onPlayResult_l(ex,!_play_check_timer); onPlayResult_l(ex, !_play_check_timer);
} }
// from live555 // from live555
bool RtspPlayer::handleAuthenticationFailure(const string &paramsStr) { bool RtspPlayer::handleAuthenticationFailure(const string &paramsStr) {
if(!_realm.empty()){ if (!_realm.empty()) {
//已经认证过了 // 已经认证过了
return false; return false;
} }
char *realm = new char[paramsStr.size()]; char *realm = new char[paramsStr.size()];
char *nonce = new char[paramsStr.size()]; char *nonce = new char[paramsStr.size()];
char *stale = new char[paramsStr.size()]; char *stale = new char[paramsStr.size()];
onceToken token(nullptr,[&](){ onceToken token(nullptr, [&]() {
delete[] realm; delete[] realm;
delete[] nonce; delete[] nonce;
delete[] stale; delete[] stale;
@ -169,16 +167,16 @@ bool RtspPlayer::handleAuthenticationFailure(const string &paramsStr) {
return false; return false;
} }
bool RtspPlayer::handleResponse(const string &cmd, const Parser &parser){ bool RtspPlayer::handleResponse(const string &cmd, const Parser &parser) {
string authInfo = parser["WWW-Authenticate"]; string authInfo = parser["WWW-Authenticate"];
//发送DESCRIBE命令后的回复 // 发送DESCRIBE命令后的回复
if ((parser.Url() == "401") && handleAuthenticationFailure(authInfo)) { if ((parser.Url() == "401") && handleAuthenticationFailure(authInfo)) {
sendOptions(); sendOptions();
return false; return false;
} }
if(parser.Url() == "302" || parser.Url() == "301"){ if (parser.Url() == "302" || parser.Url() == "301") {
auto newUrl = parser["Location"]; auto newUrl = parser["Location"];
if(newUrl.empty()){ if (newUrl.empty()) {
throw std::runtime_error("未找到Location字段(跳转url)"); throw std::runtime_error("未找到Location字段(跳转url)");
} }
play(newUrl); play(newUrl);
@ -190,19 +188,19 @@ bool RtspPlayer::handleResponse(const string &cmd, const Parser &parser){
return true; return true;
} }
void RtspPlayer::handleResDESCRIBE(const Parser& parser) { void RtspPlayer::handleResDESCRIBE(const Parser &parser) {
if (!handleResponse("DESCRIBE", parser)) { if (!handleResponse("DESCRIBE", parser)) {
return; return;
} }
_content_base = parser["Content-Base"]; _content_base = parser["Content-Base"];
if(_content_base.empty()){ if (_content_base.empty()) {
_content_base = _play_url; _content_base = _play_url;
} }
if (_content_base.back() == '/') { if (_content_base.back() == '/') {
_content_base.pop_back(); _content_base.pop_back();
} }
//解析sdp // 解析sdp
SdpParser sdpParser(parser.Content()); SdpParser sdpParser(parser.Content());
string sdp; string sdp;
@ -224,16 +222,16 @@ void RtspPlayer::handleResDESCRIBE(const Parser& parser) {
} }
_rtcp_context.clear(); _rtcp_context.clear();
for (auto &track : _sdp_track) { for (auto &track : _sdp_track) {
if(track->_pt != 0xff){ if (track->_pt != 0xff) {
setPayloadType(_rtcp_context.size(),track->_pt); setPayloadType(_rtcp_context.size(), track->_pt);
} }
_rtcp_context.emplace_back(std::make_shared<RtcpContextForRecv>()); _rtcp_context.emplace_back(std::make_shared<RtcpContextForRecv>());
} }
sendSetup(0); sendSetup(0);
} }
//有必要的情况下创建udp端口 // 有必要的情况下创建udp端口
void RtspPlayer::createUdpSockIfNecessary(int track_idx){ void RtspPlayer::createUdpSockIfNecessary(int track_idx) {
auto &rtpSockRef = _rtp_sock[track_idx]; auto &rtpSockRef = _rtp_sock[track_idx];
auto &rtcpSockRef = _rtcp_sock[track_idx]; auto &rtcpSockRef = _rtcp_sock[track_idx];
if (!rtpSockRef || !rtcpSockRef) { if (!rtpSockRef || !rtcpSockRef) {
@ -244,30 +242,27 @@ void RtspPlayer::createUdpSockIfNecessary(int track_idx){
} }
} }
//发送SETUP命令 // 发送SETUP命令
void RtspPlayer::sendSetup(unsigned int track_idx) { void RtspPlayer::sendSetup(unsigned int track_idx) {
_on_response = std::bind(&RtspPlayer::handleResSETUP, this, placeholders::_1, track_idx); _on_response = std::bind(&RtspPlayer::handleResSETUP, this, placeholders::_1, track_idx);
auto &track = _sdp_track[track_idx]; auto &track = _sdp_track[track_idx];
auto control_url = track->getControlUrl(_content_base); auto control_url = track->getControlUrl(_content_base);
switch (_rtp_type) { switch (_rtp_type) {
case Rtsp::RTP_TCP: { case Rtsp::RTP_TCP: {
sendRtspRequest("SETUP", control_url, {"Transport", StrPrinter << "RTP/AVP/TCP;unicast;interleaved=" << track->_type * 2 << "-" << track->_type * 2 + 1}); sendRtspRequest(
} "SETUP", control_url, { "Transport", StrPrinter << "RTP/AVP/TCP;unicast;interleaved=" << track->_type * 2 << "-" << track->_type * 2 + 1 });
break; } break;
case Rtsp::RTP_MULTICAST: { case Rtsp::RTP_MULTICAST: {
sendRtspRequest("SETUP", control_url, {"Transport", "RTP/AVP;multicast"}); sendRtspRequest("SETUP", control_url, { "Transport", "RTP/AVP;multicast" });
} } break;
break;
case Rtsp::RTP_UDP: { case Rtsp::RTP_UDP: {
createUdpSockIfNecessary(track_idx); createUdpSockIfNecessary(track_idx);
sendRtspRequest("SETUP", control_url, {"Transport", sendRtspRequest(
StrPrinter << "RTP/AVP;unicast;client_port=" "SETUP", control_url,
<< _rtp_sock[track_idx]->get_local_port() << "-" { "Transport",
<< _rtcp_sock[track_idx]->get_local_port()}); StrPrinter << "RTP/AVP;unicast;client_port=" << _rtp_sock[track_idx]->get_local_port() << "-" << _rtcp_sock[track_idx]->get_local_port() });
} } break;
break; default: break;
default:
break;
} }
} }
@ -292,9 +287,9 @@ void RtspPlayer::handleResSETUP(const Parser &parser, unsigned int track_idx) {
auto transport_map = Parser::parseArgs(strTransport, ";", "="); auto transport_map = Parser::parseArgs(strTransport, ";", "=");
RtspSplitter::enableRecvRtp(_rtp_type == Rtsp::RTP_TCP); RtspSplitter::enableRecvRtp(_rtp_type == Rtsp::RTP_TCP);
string ssrc = transport_map["ssrc"]; string ssrc = transport_map["ssrc"];
if(!ssrc.empty()){ if (!ssrc.empty()) {
sscanf(ssrc.data(), "%x", &_sdp_track[track_idx]->_ssrc); sscanf(ssrc.data(), "%x", &_sdp_track[track_idx]->_ssrc);
} else{ } else {
_sdp_track[track_idx]->_ssrc = 0; _sdp_track[track_idx]->_ssrc = 0;
} }
@ -310,47 +305,47 @@ void RtspPlayer::handleResSETUP(const Parser &parser, unsigned int track_idx) {
auto &pRtcpSockRef = _rtcp_sock[track_idx]; auto &pRtcpSockRef = _rtcp_sock[track_idx];
if (_rtp_type == Rtsp::RTP_MULTICAST) { if (_rtp_type == Rtsp::RTP_MULTICAST) {
//udp组播 // udp组播
auto multiAddr = transport_map["destination"]; auto multiAddr = transport_map["destination"];
pRtpSockRef = createSocket(); pRtpSockRef = createSocket();
//目前组播仅支持ipv4 // 目前组播仅支持ipv4
if (!pRtpSockRef->bindUdpSock(rtp_port, "0.0.0.0")) { if (!pRtpSockRef->bindUdpSock(rtp_port, "0.0.0.0")) {
pRtpSockRef.reset(); pRtpSockRef.reset();
throw std::runtime_error("open udp sock err"); throw std::runtime_error("open udp sock err");
} }
auto fd = pRtpSockRef->rawFD(); auto fd = pRtpSockRef->rawFD();
if (-1 == SockUtil::joinMultiAddrFilter(fd, multiAddr.data(), get_peer_ip().data(),get_local_ip().data())) { if (-1 == SockUtil::joinMultiAddrFilter(fd, multiAddr.data(), get_peer_ip().data(), get_local_ip().data())) {
SockUtil::joinMultiAddr(fd, multiAddr.data(),get_local_ip().data()); SockUtil::joinMultiAddr(fd, multiAddr.data(), get_local_ip().data());
} }
//设置rtcp发送端口 // 设置rtcp发送端口
pRtcpSockRef = createSocket(); pRtcpSockRef = createSocket();
//目前组播仅支持ipv4 // 目前组播仅支持ipv4
if (!pRtcpSockRef->bindUdpSock(0, "0.0.0.0")) { if (!pRtcpSockRef->bindUdpSock(0, "0.0.0.0")) {
//分配端口失败 // 分配端口失败
throw runtime_error("open udp socket failed"); throw runtime_error("open udp socket failed");
} }
//设置发送地址和发送端口 // 设置发送地址和发送端口
auto dst = SockUtil::make_sockaddr(get_peer_ip().data(), rtcp_port); auto dst = SockUtil::make_sockaddr(get_peer_ip().data(), rtcp_port);
pRtcpSockRef->bindPeerAddr((struct sockaddr *)&(dst)); pRtcpSockRef->bindPeerAddr((struct sockaddr *)&(dst));
} else { } else {
createUdpSockIfNecessary(track_idx); createUdpSockIfNecessary(track_idx);
//udp单播 // udp单播
auto dst = SockUtil::make_sockaddr(get_peer_ip().data(), rtp_port); auto dst = SockUtil::make_sockaddr(get_peer_ip().data(), rtp_port);
pRtpSockRef->bindPeerAddr((struct sockaddr *)&(dst)); pRtpSockRef->bindPeerAddr((struct sockaddr *)&(dst));
//发送rtp打洞包 // 发送rtp打洞包
pRtpSockRef->send("\xce\xfa\xed\xfe", 4); pRtpSockRef->send("\xce\xfa\xed\xfe", 4);
dst = SockUtil::make_sockaddr(get_peer_ip().data(), rtcp_port); dst = SockUtil::make_sockaddr(get_peer_ip().data(), rtcp_port);
//设置rtcp发送目标，为后续发送rtcp做准备 // 设置rtcp发送目标，为后续发送rtcp做准备
pRtcpSockRef->bindPeerAddr((struct sockaddr *)&(dst)); pRtcpSockRef->bindPeerAddr((struct sockaddr *)&(dst));
} }
auto peer_ip = get_peer_ip(); auto peer_ip = get_peer_ip();
weak_ptr<RtspPlayer> weakSelf = static_pointer_cast<RtspPlayer>(shared_from_this()); weak_ptr<RtspPlayer> weakSelf = static_pointer_cast<RtspPlayer>(shared_from_this());
//设置rtp over udp接收回调处理函数 // 设置rtp over udp接收回调处理函数
pRtpSockRef->setOnRead([peer_ip, track_idx, weakSelf](const Buffer::Ptr &buf, struct sockaddr *addr , int addr_len) { pRtpSockRef->setOnRead([peer_ip, track_idx, weakSelf](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) {
auto strongSelf = weakSelf.lock(); auto strongSelf = weakSelf.lock();
if (!strongSelf) { if (!strongSelf) {
return; return;
@ -359,13 +354,13 @@ void RtspPlayer::handleResSETUP(const Parser &parser, unsigned int track_idx) {
WarnL << "收到其他地址的rtp数据:" << SockUtil::inet_ntoa(addr); WarnL << "收到其他地址的rtp数据:" << SockUtil::inet_ntoa(addr);
return; return;
} }
strongSelf->handleOneRtp(track_idx, strongSelf->_sdp_track[track_idx]->_type, strongSelf->handleOneRtp(
strongSelf->_sdp_track[track_idx]->_samplerate, (uint8_t *) buf->data(), buf->size()); track_idx, strongSelf->_sdp_track[track_idx]->_type, strongSelf->_sdp_track[track_idx]->_samplerate, (uint8_t *)buf->data(), buf->size());
}); });
if(pRtcpSockRef) { if (pRtcpSockRef) {
//设置rtcp over udp接收回调处理函数 // 设置rtcp over udp接收回调处理函数
pRtcpSockRef->setOnRead([peer_ip, track_idx, weakSelf](const Buffer::Ptr &buf, struct sockaddr *addr , int addr_len) { pRtcpSockRef->setOnRead([peer_ip, track_idx, weakSelf](const Buffer::Ptr &buf, struct sockaddr *addr, int addr_len) {
auto strongSelf = weakSelf.lock(); auto strongSelf = weakSelf.lock();
if (!strongSelf) { if (!strongSelf) {
return; return;
@ -374,68 +369,66 @@ void RtspPlayer::handleResSETUP(const Parser &parser, unsigned int track_idx) {
WarnL << "收到其他地址的rtcp数据:" << SockUtil::inet_ntoa(addr); WarnL << "收到其他地址的rtcp数据:" << SockUtil::inet_ntoa(addr);
return; return;
} }
strongSelf->onRtcpPacket(track_idx, strongSelf->_sdp_track[track_idx], (uint8_t *) buf->data(), buf->size()); strongSelf->onRtcpPacket(track_idx, strongSelf->_sdp_track[track_idx], (uint8_t *)buf->data(), buf->size());
}); });
} }
} }
if (track_idx < _sdp_track.size() - 1) { if (track_idx < _sdp_track.size() - 1) {
//需要继续发送SETUP命令 // 需要继续发送SETUP命令
sendSetup(track_idx + 1); sendSetup(track_idx + 1);
return; return;
} }
//所有setup命令发送完毕 // 所有setup命令发送完毕
//发送play命令 // 发送play命令
sendPause(type_play, 0); sendPause(type_play, 0);
} }
void RtspPlayer::sendDescribe() { void RtspPlayer::sendDescribe() {
//发送DESCRIBE命令后处理函数:handleResDESCRIBE // 发送DESCRIBE命令后处理函数:handleResDESCRIBE
_on_response = std::bind(&RtspPlayer::handleResDESCRIBE, this, placeholders::_1); _on_response = std::bind(&RtspPlayer::handleResDESCRIBE, this, placeholders::_1);
sendRtspRequest("DESCRIBE", _play_url, {"Accept", "application/sdp"}); sendRtspRequest("DESCRIBE", _play_url, { "Accept", "application/sdp" });
} }
void RtspPlayer::sendOptions(){ void RtspPlayer::sendOptions() {
_on_response = [this](const Parser& parser){ _on_response = [this](const Parser &parser) {
if (!handleResponse("OPTIONS", parser)) { if (!handleResponse("OPTIONS", parser)) {
return; return;
} }
//获取服务器支持的命令 // 获取服务器支持的命令
_supported_cmd.clear(); _supported_cmd.clear();
auto public_val = split(parser["Public"],","); auto public_val = split(parser["Public"], ",");
for(auto &cmd : public_val){ for (auto &cmd : public_val) {
trim(cmd); trim(cmd);
_supported_cmd.emplace(cmd); _supported_cmd.emplace(cmd);
} }
//发送Describe请求，获取sdp // 发送Describe请求，获取sdp
sendDescribe(); sendDescribe();
}; };
sendRtspRequest("OPTIONS", _play_url); sendRtspRequest("OPTIONS", _play_url);
} }
void RtspPlayer::sendKeepAlive(){ void RtspPlayer::sendKeepAlive() {
_on_response = [](const Parser &parser) {}; _on_response = [](const Parser &parser) {};
if(_supported_cmd.find("GET_PARAMETER") != _supported_cmd.end()){ if (_supported_cmd.find("GET_PARAMETER") != _supported_cmd.end()) {
//支持GET_PARAMETER，用此命令保活 // 支持GET_PARAMETER，用此命令保活
sendRtspRequest("GET_PARAMETER", _content_base); sendRtspRequest("GET_PARAMETER", _content_base);
}else{ } else {
//不支持GET_PARAMETER，用OPTIONS命令保活 // 不支持GET_PARAMETER，用OPTIONS命令保活
sendRtspRequest("OPTIONS", _play_url); sendRtspRequest("OPTIONS", _play_url);
} }
} }
void RtspPlayer::sendPause(int type , uint32_t seekMS){ void RtspPlayer::sendPause(int type, uint32_t seekMS) {
_on_response = std::bind(&RtspPlayer::handleResPAUSE, this, placeholders::_1, type); _on_response = std::bind(&RtspPlayer::handleResPAUSE, this, placeholders::_1, type);
//开启或暂停rtsp // 开启或暂停rtsp
switch (type){ switch (type) {
case type_pause: case type_pause: sendRtspRequest("PAUSE", _content_base); break;
sendRtspRequest("PAUSE", _content_base);
break;
case type_play: case type_play:
sendRtspRequest("PLAY", _content_base); // sendRtspRequest("PLAY", _content_base);
break; // break;
case type_seek: case type_seek:
sendRtspRequest("PLAY", _content_base, {"Range",StrPrinter << "npt=" << setiosflags(ios::fixed) << setprecision(2) << seekMS / 1000.0 << "-"}); sendRtspRequest("PLAY", _content_base, { "Range", StrPrinter << "npt=" << setiosflags(ios::fixed) << setprecision(2) << seekMS / 1000.0 << "-" });
break; break;
default: default:
WarnL << "unknown type : " << type; WarnL << "unknown type : " << type;
@ -449,35 +442,31 @@ void RtspPlayer::pause(bool bPause) {
} }
void RtspPlayer::speed(float speed) { void RtspPlayer::speed(float speed) {
sendRtspRequest("PLAY", _content_base, {"Scale", StrPrinter << speed}); sendRtspRequest("PLAY", _content_base, { "Scale", StrPrinter << speed });
} }
void RtspPlayer::handleResPAUSE(const Parser& parser,int type) { void RtspPlayer::handleResPAUSE(const Parser &parser, int type) {
if (parser.Url() != "200") { if (parser.Url() != "200") {
switch (type) { switch (type) {
case type_pause: case type_pause: WarnL << "Pause failed:" << parser.Url() << " " << parser.Tail() << endl; break;
WarnL << "Pause failed:" << parser.Url() << " " << parser.Tail() << endl;
break;
case type_play: case type_play:
WarnL << "Play failed:" << parser.Url() << " " << parser.Tail() << endl; WarnL << "Play failed:" << parser.Url() << " " << parser.Tail() << endl;
onPlayResult_l(SockException(Err_shutdown, StrPrinter << "rtsp play failed:" << parser.Url() << " " << parser.Tail() ), !_play_check_timer); onPlayResult_l(SockException(Err_shutdown, StrPrinter << "rtsp play failed:" << parser.Url() << " " << parser.Tail()), !_play_check_timer);
break;
case type_seek:
WarnL << "Seek failed:" << parser.Url() << " " << parser.Tail() << endl;
break; break;
case type_seek: WarnL << "Seek failed:" << parser.Url() << " " << parser.Tail() << endl; break;
} }
return; return;
} }
if (type == type_pause) { if (type == type_pause) {
//暂停成功！ // 暂停成功！
_rtp_check_timer.reset(); _rtp_check_timer.reset();
return; return;
} }
//play或seek成功 // play或seek成功
uint32_t iSeekTo = 0; uint32_t iSeekTo = 0;
//修正时间轴 // 修正时间轴
auto strRange = parser["Range"]; auto strRange = parser["Range"];
if (strRange.size()) { if (strRange.size()) {
auto strStart = FindField(strRange.data(), "npt=", "-"); auto strStart = FindField(strRange.data(), "npt=", "-");
@ -495,54 +484,56 @@ void RtspPlayer::onWholeRtspPacket(Parser &parser) {
try { try {
decltype(_on_response) func; decltype(_on_response) func;
_on_response.swap(func); _on_response.swap(func);
if(func){ if (func) {
func(parser); func(parser);
} }
parser.Clear(); parser.Clear();
} catch (std::exception &err) { } catch (std::exception &err) {
//定时器_pPlayTimer为空后表明握手结束了 // 定时器_pPlayTimer为空后表明握手结束了
onPlayResult_l(SockException(Err_other, err.what()),!_play_check_timer); onPlayResult_l(SockException(Err_other, err.what()), !_play_check_timer);
} }
} }
void RtspPlayer::onRtpPacket(const char *data, size_t len) { void RtspPlayer::onRtpPacket(const char *data, size_t len) {
int trackIdx = -1; int trackIdx = -1;
uint8_t interleaved = data[1]; uint8_t interleaved = data[1];
if(interleaved %2 == 0){ if (interleaved % 2 == 0) {
trackIdx = getTrackIndexByInterleaved(interleaved); trackIdx = getTrackIndexByInterleaved(interleaved);
if (trackIdx == -1) { if (trackIdx == -1) {
return; return;
} }
handleOneRtp(trackIdx, _sdp_track[trackIdx]->_type, _sdp_track[trackIdx]->_samplerate, (uint8_t *)data + RtpPacket::kRtpTcpHeaderSize, len - RtpPacket::kRtpTcpHeaderSize); handleOneRtp(
}else{ trackIdx, _sdp_track[trackIdx]->_type, _sdp_track[trackIdx]->_samplerate, (uint8_t *)data + RtpPacket::kRtpTcpHeaderSize,
len - RtpPacket::kRtpTcpHeaderSize);
} else {
trackIdx = getTrackIndexByInterleaved(interleaved - 1); trackIdx = getTrackIndexByInterleaved(interleaved - 1);
if (trackIdx == -1) { if (trackIdx == -1) {
return; return;
} }
onRtcpPacket(trackIdx, _sdp_track[trackIdx], (uint8_t *) data + RtpPacket::kRtpTcpHeaderSize, len - RtpPacket::kRtpTcpHeaderSize); onRtcpPacket(trackIdx, _sdp_track[trackIdx], (uint8_t *)data + RtpPacket::kRtpTcpHeaderSize, len - RtpPacket::kRtpTcpHeaderSize);
} }
} }
//此处预留rtcp处理函数 // 此处预留rtcp处理函数
void RtspPlayer::onRtcpPacket(int track_idx, SdpTrack::Ptr &track, uint8_t *data, size_t len){ void RtspPlayer::onRtcpPacket(int track_idx, SdpTrack::Ptr &track, uint8_t *data, size_t len) {
auto rtcp_arr = RtcpHeader::loadFromBytes((char *) data, len); auto rtcp_arr = RtcpHeader::loadFromBytes((char *)data, len);
for (auto &rtcp : rtcp_arr) { for (auto &rtcp : rtcp_arr) {
_rtcp_context[track_idx]->onRtcp(rtcp); _rtcp_context[track_idx]->onRtcp(rtcp);
if ((RtcpType) rtcp->pt == RtcpType::RTCP_SR) { if ((RtcpType)rtcp->pt == RtcpType::RTCP_SR) {
auto sr = (RtcpSR *) (rtcp); auto sr = (RtcpSR *)(rtcp);
//设置rtp时间戳与ntp时间戳的对应关系 // 设置rtp时间戳与ntp时间戳的对应关系
setNtpStamp(track_idx, sr->rtpts, sr->getNtpUnixStampMS()); setNtpStamp(track_idx, sr->rtpts, sr->getNtpUnixStampMS());
} }
} }
} }
void RtspPlayer::onRtpSorted(RtpPacket::Ptr rtppt, int trackidx){ void RtspPlayer::onRtpSorted(RtpPacket::Ptr rtppt, int trackidx) {
_stamp[trackidx] = rtppt->getStampMS(); _stamp[trackidx] = rtppt->getStampMS();
_rtp_recv_ticker.resetTime(); _rtp_recv_ticker.resetTime();
onRecvRTP(std::move(rtppt), _sdp_track[trackidx]); onRecvRTP(std::move(rtppt), _sdp_track[trackidx]);
} }
float RtspPlayer::getPacketLossRate(TrackType type) const{ float RtspPlayer::getPacketLossRate(TrackType type) const {
size_t lost = 0, expected = 0; size_t lost = 0, expected = 0;
try { try {
auto track_idx = getTrackIndexByTrackType(type); auto track_idx = getTrackIndexByTrackType(type);
@ -558,32 +549,32 @@ float RtspPlayer::getPacketLossRate(TrackType type) const{
if (!expected) { if (!expected) {
return 0; return 0;
} }
return (float) (double(lost) / double(expected)); return (float)(double(lost) / double(expected));
} }
uint32_t RtspPlayer::getProgressMilliSecond() const{ uint32_t RtspPlayer::getProgressMilliSecond() const {
return MAX(_stamp[0],_stamp[1]); return MAX(_stamp[0], _stamp[1]);
} }
void RtspPlayer::seekToMilliSecond(uint32_t ms) { void RtspPlayer::seekToMilliSecond(uint32_t ms) {
sendPause(type_seek,ms); sendPause(type_seek, ms);
} }
void RtspPlayer::sendRtspRequest(const string &cmd, const string &url, const std::initializer_list<string> &header) { void RtspPlayer::sendRtspRequest(const string &cmd, const string &url, const std::initializer_list<string> &header) {
string key; string key;
StrCaseMap header_map; StrCaseMap header_map;
int i = 0; int i = 0;
for(auto &val : header){ for (auto &val : header) {
if(++i % 2 == 0){ if (++i % 2 == 0) {
header_map.emplace(key,val); header_map.emplace(key, val);
}else{ } else {
key = val; key = val;
} }
} }
sendRtspRequest(cmd,url,header_map); sendRtspRequest(cmd, url, header_map);
} }
void RtspPlayer::sendRtspRequest(const string &cmd, const string &url,const StrCaseMap &header_const) { void RtspPlayer::sendRtspRequest(const string &cmd, const string &url, const StrCaseMap &header_const) {
auto header = header_const; auto header = header_const;
header.emplace("CSeq", StrPrinter << _cseq_send++); header.emplace("CSeq", StrPrinter << _cseq_send++);
header.emplace("User-Agent", kServerName); header.emplace("User-Agent", kServerName);
@ -624,44 +615,49 @@ void RtspPlayer::sendRtspRequest(const string &cmd, const string &url,const StrC
} }
_StrPrinter printer; _StrPrinter printer;
if (cmd == "PLAY") {
printer << cmd << " " << _play_url << " RTSP/1.0\r\n";
} else {
printer << cmd << " " << url << " RTSP/1.0\r\n"; printer << cmd << " " << url << " RTSP/1.0\r\n";
for (auto &pr : header){ }
for (auto &pr : header) {
printer << pr.first << ": " << pr.second << "\r\n"; printer << pr.first << ": " << pr.second << "\r\n";
} }
printer << "\r\n"; printer << "\r\n";
SockSender::send(std::move(printer)); SockSender::send(std::move(printer));
} }
void RtspPlayer::onBeforeRtpSorted(const RtpPacket::Ptr &rtp, int track_idx){ void RtspPlayer::onBeforeRtpSorted(const RtpPacket::Ptr &rtp, int track_idx) {
auto &rtcp_ctx = _rtcp_context[track_idx]; auto &rtcp_ctx = _rtcp_context[track_idx];
rtcp_ctx->onRtp(rtp->getSeq(), rtp->getStamp(), rtp->ntp_stamp, rtp->sample_rate, rtp->size() - RtpPacket::kRtpTcpHeaderSize); rtcp_ctx->onRtp(rtp->getSeq(), rtp->getStamp(), rtp->ntp_stamp, rtp->sample_rate, rtp->size() - RtpPacket::kRtpTcpHeaderSize);
auto &ticker = _rtcp_send_ticker[track_idx]; auto &ticker = _rtcp_send_ticker[track_idx];
if (ticker.elapsedTime() < 3 * 1000) { if (ticker.elapsedTime() < 3 * 1000) {
//时间未到 // 时间未到
return; return;
} }
auto &rtcp_flag = _send_rtcp[track_idx]; auto &rtcp_flag = _send_rtcp[track_idx];
//每3秒发送一次心跳，rtcp与rtsp信令轮流心跳，该特性用于兼容issue:642 // 每3秒发送一次心跳，rtcp与rtsp信令轮流心跳，该特性用于兼容issue:642
//有些rtsp服务器需要rtcp保活，有些需要发送信令保活 // 有些rtsp服务器需要rtcp保活，有些需要发送信令保活
//发送信令保活 // 发送信令保活
if (!rtcp_flag) { if (!rtcp_flag) {
if (track_idx == 0) { if (track_idx == 0) {
sendKeepAlive(); sendKeepAlive();
} }
ticker.resetTime(); ticker.resetTime();
//下次发送rtcp保活 // 下次发送rtcp保活
rtcp_flag = true; rtcp_flag = true;
return; return;
} }
//发送rtcp // 发送rtcp
static auto send_rtcp = [](RtspPlayer *thiz, int index, Buffer::Ptr ptr) { static auto send_rtcp = [](RtspPlayer *thiz, int index, Buffer::Ptr ptr) {
if (thiz->_rtp_type == Rtsp::RTP_TCP) { if (thiz->_rtp_type == Rtsp::RTP_TCP) {
auto &track = thiz->_sdp_track[index]; auto &track = thiz->_sdp_track[index];
thiz->send(makeRtpOverTcpPrefix((uint16_t) (ptr->size()), track->_interleaved + 1)); thiz->send(makeRtpOverTcpPrefix((uint16_t)(ptr->size()), track->_interleaved + 1));
thiz->send(std::move(ptr)); thiz->send(std::move(ptr));
} else { } else {
thiz->_rtcp_sock[index]->send(std::move(ptr)); thiz->_rtcp_sock[index]->send(std::move(ptr));
@ -670,39 +666,39 @@ void RtspPlayer::onBeforeRtpSorted(const RtpPacket::Ptr &rtp, int track_idx){
auto ssrc = rtp->getSSRC(); auto ssrc = rtp->getSSRC();
auto rtcp = rtcp_ctx->createRtcpRR(ssrc + 1, ssrc); auto rtcp = rtcp_ctx->createRtcpRR(ssrc + 1, ssrc);
auto rtcp_sdes = RtcpSdes::create({kServerName}); auto rtcp_sdes = RtcpSdes::create({ kServerName });
rtcp_sdes->chunks.type = (uint8_t) SdesType::RTCP_SDES_CNAME; rtcp_sdes->chunks.type = (uint8_t)SdesType::RTCP_SDES_CNAME;
rtcp_sdes->chunks.ssrc = htonl(ssrc); rtcp_sdes->chunks.ssrc = htonl(ssrc);
send_rtcp(this, track_idx, std::move(rtcp)); send_rtcp(this, track_idx, std::move(rtcp));
send_rtcp(this, track_idx, RtcpHeader::toBuffer(rtcp_sdes)); send_rtcp(this, track_idx, RtcpHeader::toBuffer(rtcp_sdes));
ticker.resetTime(); ticker.resetTime();
//下次发送信令保活 // 下次发送信令保活
rtcp_flag = false; rtcp_flag = false;
} }
void RtspPlayer::onPlayResult_l(const SockException &ex , bool handshake_done) { void RtspPlayer::onPlayResult_l(const SockException &ex, bool handshake_done) {
if (ex.getErrCode() == Err_shutdown) { if (ex.getErrCode() == Err_shutdown) {
//主动shutdown的，不触发回调 // 主动shutdown的，不触发回调
return; return;
} }
WarnL << ex.getErrCode() << " " << ex.what(); WarnL << ex.getErrCode() << " " << ex.what();
if (!handshake_done) { if (!handshake_done) {
//开始播放阶段 // 开始播放阶段
_play_check_timer.reset(); _play_check_timer.reset();
onPlayResult(ex); onPlayResult(ex);
//是否为性能测试模式 // 是否为性能测试模式
_benchmark_mode = (*this)[Client::kBenchmarkMode].as<int>(); _benchmark_mode = (*this)[Client::kBenchmarkMode].as<int>();
} else if (ex) { } else if (ex) {
//播放成功后异常断开回调 // 播放成功后异常断开回调
onShutdown(ex); onShutdown(ex);
} else { } else {
//恢复播放 // 恢复播放
onResume(); onResume();
} }
if (!ex) { if (!ex) {
//播放成功，恢复rtp接收超时定时器 // 播放成功，恢复rtp接收超时定时器
_rtp_recv_ticker.resetTime(); _rtp_recv_ticker.resetTime();
auto timeoutMS = (*this)[Client::kMediaTimeoutMS].as<uint64_t>(); auto timeoutMS = (*this)[Client::kMediaTimeoutMS].as<uint64_t>();
weak_ptr<RtspPlayer> weakSelf = static_pointer_cast<RtspPlayer>(shared_from_this()); weak_ptr<RtspPlayer> weakSelf = static_pointer_cast<RtspPlayer>(shared_from_this());
@ -712,13 +708,13 @@ void RtspPlayer::onPlayResult_l(const SockException &ex , bool handshake_done) {
return false; return false;
} }
if (strongSelf->_rtp_recv_ticker.elapsedTime() > timeoutMS) { if (strongSelf->_rtp_recv_ticker.elapsedTime() > timeoutMS) {
//接收rtp媒体数据包超时 // 接收rtp媒体数据包超时
strongSelf->onPlayResult_l(SockException(Err_timeout, "receive rtp timeout"), true); strongSelf->onPlayResult_l(SockException(Err_timeout, "receive rtp timeout"), true);
return false; return false;
} }
return true; return true;
}; };
//创建rtp数据接收超时检测定时器 // 创建rtp数据接收超时检测定时器
_rtp_check_timer = std::make_shared<Timer>(timeoutMS / 2000.0f, lam, getPoller()); _rtp_check_timer = std::make_shared<Timer>(timeoutMS / 2000.0f, lam, getPoller());
} else { } else {
sendTeardown(); sendTeardown();
@ -752,8 +748,7 @@ int RtspPlayer::getTrackIndexByTrackType(TrackType track_type) const {
/////////////////////////////////////////////////// ///////////////////////////////////////////////////
// RtspPlayerImp // RtspPlayerImp
float RtspPlayerImp::getDuration() const float RtspPlayerImp::getDuration() const {
{
return _demuxer ? _demuxer->getDuration() : 0; return _demuxer ? _demuxer->getDuration() : 0;
} }
@ -770,13 +765,11 @@ void RtspPlayerImp::addTrackCompleted() {
} }
} }
std::vector<Track::Ptr> RtspPlayerImp::getTracks(bool ready /*= true*/) const std::vector<Track::Ptr> RtspPlayerImp::getTracks(bool ready /*= true*/) const {
{
return _demuxer ? _demuxer->getTracks(ready) : Super::getTracks(ready); return _demuxer ? _demuxer->getTracks(ready) : Super::getTracks(ready);
} }
bool RtspPlayerImp::onCheckSDP(const std::string &sdp) bool RtspPlayerImp::onCheckSDP(const std::string &sdp) {
{
_rtsp_media_src = std::dynamic_pointer_cast<RtspMediaSource>(_media_src); _rtsp_media_src = std::dynamic_pointer_cast<RtspMediaSource>(_media_src);
if (_rtsp_media_src) { if (_rtsp_media_src) {
_rtsp_media_src->setSdp(sdp); _rtsp_media_src->setSdp(sdp);
@ -788,7 +781,7 @@ bool RtspPlayerImp::onCheckSDP(const std::string &sdp)
} }
void RtspPlayerImp::onRecvRTP(RtpPacket::Ptr rtp, const SdpTrack::Ptr &track) { void RtspPlayerImp::onRecvRTP(RtpPacket::Ptr rtp, const SdpTrack::Ptr &track) {
//rtp解复用时可以判断是否为关键帧起始位置 // rtp解复用时可以判断是否为关键帧起始位置
auto key_pos = _demuxer->inputRtp(rtp); auto key_pos = _demuxer->inputRtp(rtp);
if (_rtsp_media_src) { if (_rtsp_media_src) {
_rtsp_media_src->onWrite(std::move(rtp), key_pos); _rtsp_media_src->onWrite(std::move(rtp), key_pos);