Skip to content

Z
ZLMediaKit
Commit e6d511cc by xiongziliang
Options

解决播放成功与中途断开事件触发紊乱的问题:#143

parent 85cd4a7d
正在显示 包含 140 行增加 和 127 行删除
media-server @ 678678b2
Subproject commit 40edf6243d9d99676062062efdec203b24a178aa
Subproject commit 678678b2ee7118d21d33a90a303698e0da02790c
src/Rtmp/RtmpPlayer.cpp
......@@ -34,44 +34,35 @@ using namespace mediakit::Client;
namespace mediakit {
unordered_map<string, RtmpPlayer::rtmpCMDHandle> RtmpPlayer::g_mapCmd;
RtmpPlayer::RtmpPlayer(const EventPoller::Ptr &poller) : TcpClient(poller) {
static onceToken token([]() {
g_mapCmd.emplace("_error",&RtmpPlayer::onCmd_result);
g_mapCmd.emplace("_result",&RtmpPlayer::onCmd_result);
g_mapCmd.emplace("onStatus",&RtmpPlayer::onCmd_onStatus);
g_mapCmd.emplace("onMetaData",&RtmpPlayer::onCmd_onMetaData);
}, []() {});
}
RtmpPlayer::~RtmpPlayer() {
DebugL << endl;
}
void RtmpPlayer::teardown() {
if (alive()) {
_strApp.clear();
_strStream.clear();
_strTcUrl.clear();
{
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
_mapOnResultCB.clear();
}
{
lock_guard<recursive_mutex> lck(_mtxOnStatusCB);
_dqOnStatusCB.clear();
}
_pBeatTimer.reset();
_pPlayTimer.reset();
_pMediaTimer.reset();
_iSeekTo = 0;
CLEAR_ARR(_aiFistStamp);
CLEAR_ARR(_aiNowStamp);
reset();
shutdown(SockException(Err_shutdown,"teardown"));
shutdown(SockException(Err_shutdown,"teardown"));
}
_strApp.clear();
_strStream.clear();
_strTcUrl.clear();
_pBeatTimer.reset();
_pPlayTimer.reset();
_pMediaTimer.reset();
_iSeekTo = 0;
RtmpProtocol::reset();
CLEAR_ARR(_aiFistStamp);
CLEAR_ARR(_aiNowStamp);
lock_guard<recursive_mutex> lck(_mtxOnResultCB);
_mapOnResultCB.clear();
lock_guard<recursive_mutex> lck2(_mtxOnStatusCB);
_dqOnStatusCB.clear();
}
void RtmpPlayer::play(const string &strUrl) {
teardown();
string strHost = FindField(strUrl.data(), "://", "/");
......@@ -80,7 +71,7 @@ void RtmpPlayer::play(const string &strUrl) {
_strTcUrl = string("rtmp://") + strHost + "/" + _strApp;
if (!_strApp.size() || !_strStream.size()) {
onPlayResult_l(SockException(Err_other,"rtmp url非法"));
onPlayResult_l(SockException(Err_other,"rtmp url非法"),false);
return;
}
DebugL << strHost << " " << _strApp << " " << _strStream;
......@@ -104,7 +95,7 @@ void RtmpPlayer::play(const string &strUrl) {
if(!strongSelf) {
return false;
}
strongSelf->onPlayResult_l(SockException(Err_timeout,"play rtmp timeout"));
strongSelf->onPlayResult_l(SockException(Err_timeout,"play rtmp timeout"),false);
return false;
},getPoller()));
......@@ -112,53 +103,52 @@ void RtmpPlayer::play(const string &strUrl) {
startConnect(strHost, iPort , playTimeOutSec);
}
void RtmpPlayer::onErr(const SockException &ex){
onPlayResult_l(ex);
//定时器_pPlayTimer为空后表明握手结束了
onPlayResult_l(ex, !_pPlayTimer);
}
void RtmpPlayer::onPlayResult_l(const SockException &ex) {
void RtmpPlayer::onPlayResult_l(const SockException &ex , bool handshakeCompleted) {
WarnL << ex.getErrCode() << " " << ex.what();
if(!ex){
//恢复rtmp接收超时定时器
//播放成功，恢复rtmp接收超时定时器
_mediaTicker.resetTime();
weak_ptr<RtmpPlayer> weakSelf = dynamic_pointer_cast<RtmpPlayer>(shared_from_this());
int timeoutMS = (*this)[kMediaTimeoutMS].as<int>();
//创建rtmp数据接收超时检测定时器
_pMediaTimer.reset( new Timer(timeoutMS / 2000.0, [weakSelf,timeoutMS]() {
auto strongSelf=weakSelf.lock();
if(!strongSelf) {
return false;
}
if(strongSelf->_mediaTicker.elapsedTime()> timeoutMS) {
//recv media timeout!
strongSelf->onPlayResult_l(SockException(Err_timeout,"recv rtmp timeout"));
//接收rtmp媒体数据超时
strongSelf->onPlayResult_l(SockException(Err_timeout,"receive rtmp timeout"),true);
return false;
}
return true;
},getPoller()));
}
if (_pPlayTimer) {
if (!handshakeCompleted) {
//开始播放阶段
_pPlayTimer.reset();
onPlayResult(ex);
}else {
//播放中途阶段
if (ex) {
//播放成功后异常断开回调
onShutdown(ex);
}else{
//恢复播放
onResume();
}
}
} else if (ex) {
//播放成功后异常断开回调
onShutdown(ex);
} else {
//恢复播放
onResume();
}
if(ex){
teardown();
}
}
void RtmpPlayer::onConnect(const SockException &err){
if(err.getErrCode()!=Err_success) {
onPlayResult_l(err);
if(err.getErrCode() != Err_success) {
onPlayResult_l(err, false);
return;
}
weak_ptr<RtmpPlayer> weakSelf= dynamic_pointer_cast<RtmpPlayer>(shared_from_this());
......@@ -175,7 +165,8 @@ void RtmpPlayer::onRecv(const Buffer::Ptr &pBuf){
onParseRtmp(pBuf->data(), pBuf->size());
} catch (exception &e) {
SockException ex(Err_other, e.what());
onPlayResult_l(ex);
//定时器_pPlayTimer为空后表明握手结束了
onPlayResult_l(ex, !_pPlayTimer);
}
}
......@@ -253,7 +244,7 @@ inline void RtmpPlayer::send_pause(bool bPause) {
}else{
_bPaused = bPause;
if(!bPause){
onPlayResult_l(SockException(Err_success, "rtmp resum success"));
onPlayResult_l(SockException(Err_success, "resum rtmp success"), true);
}else{
//暂停播放
_pMediaTimer.reset();
......@@ -327,7 +318,7 @@ void RtmpPlayer::onCmd_onMetaData(AMFDecoder &dec) {
void RtmpPlayer::onStreamDry(uint32_t ui32StreamId) {
//TraceL << ui32StreamId;
onPlayResult_l(SockException(Err_other,"rtmp stream dry"));
onPlayResult_l(SockException(Err_other,"rtmp stream dry"), true);
}
void RtmpPlayer::onMediaData_l(const RtmpPacket::Ptr &packet) {
......@@ -343,7 +334,7 @@ void RtmpPlayer::onMediaData_l(const RtmpPacket::Ptr &packet) {
onMediaData(packet);
}else{
//先触发onPlayResult事件，这个时候解码器才能初始化完毕
onPlayResult_l(SockException(Err_success,"play rtmp success"));
onPlayResult_l(SockException(Err_success,"play rtmp success"), false);
//触发onPlayResult事件后，再把帧数据输入到解码器
onMediaData(packet);
}
......@@ -351,6 +342,15 @@ void RtmpPlayer::onMediaData_l(const RtmpPacket::Ptr &packet) {
void RtmpPlayer::onRtmpChunk(RtmpPacket &chunkData) {
typedef void (RtmpPlayer::*rtmp_func_ptr)(AMFDecoder &dec);
static unordered_map<string, rtmp_func_ptr> s_func_map;
static onceToken token([]() {
s_func_map.emplace("_error",&RtmpPlayer::onCmd_result);
s_func_map.emplace("_result",&RtmpPlayer::onCmd_result);
s_func_map.emplace("onStatus",&RtmpPlayer::onCmd_onStatus);
s_func_map.emplace("onMetaData",&RtmpPlayer::onCmd_onMetaData);
}, []() {});
switch (chunkData.typeId) {
case MSG_CMD:
case MSG_CMD3:
......@@ -358,8 +358,8 @@ void RtmpPlayer::onRtmpChunk(RtmpPacket &chunkData) {
case MSG_DATA3: {
AMFDecoder dec(chunkData.strBuf, 0);
std::string type = dec.load<std::string>();
auto it = g_mapCmd.find(type);
if(it != g_mapCmd.end()){
auto it = s_func_map.find(type);
if(it != s_func_map.end()){
auto fun = it->second;
(this->*fun)(dec);
}else{
......
src/Rtmp/RtmpPlayer.h
......@@ -61,7 +61,7 @@ protected:
void seekToMilliSecond(uint32_t ms);
protected:
void onMediaData_l(const RtmpPacket::Ptr &chunkData);
void onPlayResult_l(const SockException &ex);
void onPlayResult_l(const SockException &ex, bool handshakeCompleted);
//form Tcpclient
void onRecv(const Buffer::Ptr &pBuf) override;
......@@ -104,9 +104,6 @@ private:
deque<function<void(AMFValue &dec)> > _dqOnStatusCB;
recursive_mutex _mtxOnStatusCB;
typedef void (RtmpPlayer::*rtmpCMDHandle)(AMFDecoder &dec);
static unordered_map<string, rtmpCMDHandle> g_mapCmd;
//超时功能实现
Ticker _mediaTicker;
std::shared_ptr<Timer> _pMediaTimer;
......
src/Rtsp/RtspPlayer.cpp
......@@ -42,11 +42,17 @@ using namespace mediakit::Client;
namespace mediakit {
enum PlayType {
type_play = 0,
type_pause,
type_seek
};
RtspPlayer::RtspPlayer(const EventPoller::Ptr &poller) : TcpClient(poller){
RtpReceiver::setPoolSize(64);
}
RtspPlayer::~RtspPlayer(void) {
DebugL<<endl;
DebugL << endl;
}
void RtspPlayer::teardown(){
if (alive()) {
......@@ -56,7 +62,6 @@ void RtspPlayer::teardown(){
_rtspMd5Nonce.clear();
_rtspRealm.clear();
_aTrackInfo.clear();
_strSession.clear();
_strContentBase.clear();
......@@ -135,7 +140,7 @@ void RtspPlayer::play(bool isSSL,const string &strUrl, const string &strUser, co
}
if(ip.empty()){
onPlayResult_l(SockException(Err_other,StrPrinter << "illegal rtsp url:" << strUrl));
onPlayResult_l(SockException(Err_other,StrPrinter << "illegal rtsp url:" << strUrl),false);
return;
}
......@@ -148,7 +153,7 @@ void RtspPlayer::play(bool isSSL,const string &strUrl, const string &strUser, co
if(!strongSelf) {
return false;
}
strongSelf->onPlayResult_l(SockException(Err_timeout,"play rtsp timeout"));
strongSelf->onPlayResult_l(SockException(Err_timeout,"play rtsp timeout"),false);
return false;
},getPoller()));
......@@ -158,8 +163,8 @@ void RtspPlayer::play(bool isSSL,const string &strUrl, const string &strUser, co
startConnect(ip, port , playTimeOutSec);
}
void RtspPlayer::onConnect(const SockException &err){
if(err.getErrCode()!=Err_success) {
onPlayResult_l(err);
if(err.getErrCode() != Err_success) {
onPlayResult_l(err,false);
return;
}
......@@ -170,7 +175,8 @@ void RtspPlayer::onRecv(const Buffer::Ptr& pBuf) {
input(pBuf->data(),pBuf->size());
}
void RtspPlayer::onErr(const SockException &ex) {
onPlayResult_l(ex);
//定时器_pPlayTimer为空后表明握手结束了
onPlayResult_l(ex,!_pPlayTimer);
}
// from live555
bool RtspPlayer::handleAuthenticationFailure(const string &paramsStr) {
......@@ -403,8 +409,7 @@ void RtspPlayer::handleResSETUP(const Parser &parser, unsigned int uiTrackIndex)
WarnL << "收到其他地址的rtcp数据:" << inet_ntoa(((struct sockaddr_in *) addr)->sin_addr);
return;
}
strongSelf->onRtcpPacket(uiTrackIndex, strongSelf->_aTrackInfo[uiTrackIndex],
(unsigned char *) buf->data(), buf->size());
strongSelf->onRtcpPacket(uiTrackIndex, strongSelf->_aTrackInfo[uiTrackIndex], (unsigned char *) buf->data(), buf->size());
});
}
}
......@@ -416,14 +421,7 @@ void RtspPlayer::handleResSETUP(const Parser &parser, unsigned int uiTrackIndex)
}
//所有setup命令发送完毕
//发送play命令
sendPause(false, 0,false);
}
void RtspPlayer::sendOptions() {
_onHandshake = [](const Parser& parser){
// DebugL << "options response";
};
sendRtspRequest("OPTIONS",_strContentBase);
sendPause(type_play, 0);
}
void RtspPlayer::sendDescribe() {
......@@ -432,46 +430,67 @@ void RtspPlayer::sendDescribe() {
sendRtspRequest("DESCRIBE",_strUrl,{"Accept","application/sdp"});
}
void RtspPlayer::sendPause(bool bPause,uint32_t seekMS,bool range){
void RtspPlayer::sendPause(int type , uint32_t seekMS){
_onHandshake = std::bind(&RtspPlayer::handleResPAUSE,this, placeholders::_1,type);
//开启或暂停rtsp
_onHandshake = std::bind(&RtspPlayer::handleResPAUSE,this, placeholders::_1,bPause);
if(!bPause && range){
sendRtspRequest(bPause ? "PAUSE" : "PLAY", _strContentBase,
{"Range",StrPrinter << "npt=" << setiosflags(ios::fixed) << setprecision(2) << seekMS / 1000.0 << "-"});
} else{
sendRtspRequest(bPause ? "PAUSE" : "PLAY", _strContentBase);
}
switch (type){
case type_pause:
sendRtspRequest("PAUSE", _strContentBase);
break;
case type_play:
sendRtspRequest("PLAY", _strContentBase);
break;
case type_seek:
sendRtspRequest("PLAY", _strContentBase, {"Range",StrPrinter << "npt=" << setiosflags(ios::fixed) << setprecision(2) << seekMS / 1000.0 << "-"});
break;
default:
WarnL << "unknown type : " << type;
_onHandshake = nullptr;
break;
}
}
void RtspPlayer::pause(bool bPause) {
sendPause(bPause, getProgressMilliSecond(),false);
sendPause(bPause ? type_pause : type_seek, getProgressMilliSecond());
}
void RtspPlayer::handleResPAUSE(const Parser& parser, bool bPause) {
void RtspPlayer::handleResPAUSE(const Parser& parser,int type) {
if (parser.Url() != "200") {
WarnL <<(bPause ? "Pause" : "Play") << " failed:" << parser.Url() << " " << parser.Tail() << endl;
switch (type) {
case type_pause:
WarnL << "Pause failed:" << parser.Url() << " " << parser.Tail() << endl;
break;
case type_play:
WarnL << "Play failed:" << parser.Url() << " " << parser.Tail() << endl;
break;
case type_seek:
WarnL << "Seek failed:" << parser.Url() << " " << parser.Tail() << endl;
break;
}
return;
}
if (!bPause) {
uint32_t iSeekTo = 0;
//修正时间轴
auto strRange = parser["Range"];
if (strRange.size()) {
auto strStart = FindField(strRange.data(), "npt=", "-");
if (strStart == "now") {
strStart = "0";
}
iSeekTo = 1000 * atof(strStart.data());
DebugL << "seekTo(ms):" << iSeekTo ;
}
//设置相对时间戳
_stamp[0].setRelativeStamp(iSeekTo);
_stamp[1].setRelativeStamp(iSeekTo);
onPlayResult_l(SockException(Err_success, "rtsp play success"));
} else {
if (type == type_pause) {
//暂停成功！
_pRtpTimer.reset();
return;
}
//play或seek成功
uint32_t iSeekTo = 0;
//修正时间轴
auto strRange = parser["Range"];
if (strRange.size()) {
auto strStart = FindField(strRange.data(), "npt=", "-");
if (strStart == "now") {
strStart = "0";
}
iSeekTo = 1000 * atof(strStart.data());
DebugL << "seekTo(ms):" << iSeekTo;
}
//设置相对时间戳
_stamp[0].setRelativeStamp(iSeekTo);
_stamp[1].setRelativeStamp(iSeekTo);
onPlayResult_l(SockException(Err_success, type == type_seek ? "resum rtsp success" : "rtsp play success"), type == type_seek);
}
void RtspPlayer::onWholeRtspPacket(Parser &parser) {
......@@ -483,8 +502,8 @@ void RtspPlayer::onWholeRtspPacket(Parser &parser) {
}
parser.Clear();
} catch (std::exception &err) {
SockException ex(Err_other, err.what());
onPlayResult_l(ex);
//定时器_pPlayTimer为空后表明握手结束了
onPlayResult_l(SockException(Err_other, err.what()),!_pPlayTimer);
}
}
......@@ -674,7 +693,7 @@ uint32_t RtspPlayer::getProgressMilliSecond() const{
return MAX(_stamp[0].getRelativeStamp(),_stamp[1].getRelativeStamp());
}
void RtspPlayer::seekToMilliSecond(uint32_t ms) {
sendPause(false,ms, true);
sendPause(type_seek,ms);
}
void RtspPlayer::sendRtspRequest(const string &cmd, const string &url, const std::initializer_list<string> &header) {
......@@ -764,7 +783,7 @@ void RtspPlayer::onRecvRTP_l(const RtpPacket::Ptr &pkt, const SdpTrack::Ptr &tra
}
void RtspPlayer::onPlayResult_l(const SockException &ex) {
void RtspPlayer::onPlayResult_l(const SockException &ex , bool handshakeCompleted) {
WarnL << ex.getErrCode() << " " << ex.what();
if(!ex){
......@@ -772,33 +791,31 @@ void RtspPlayer::onPlayResult_l(const SockException &ex) {
_rtpTicker.resetTime();
weak_ptr<RtspPlayer> weakSelf = dynamic_pointer_cast<RtspPlayer>(shared_from_this());
int timeoutMS = (*this)[kMediaTimeoutMS].as<int>();
_pRtpTimer.reset( new Timer(timeoutMS / 2000.0, [weakSelf,timeoutMS]() {
//创建rtp数据接收超时检测定时器
_pRtpTimer.reset( new Timer(timeoutMS / 2000.0, [weakSelf,timeoutMS]() {
auto strongSelf=weakSelf.lock();
if(!strongSelf) {
return false;
}
if(strongSelf->_rtpTicker.elapsedTime()> timeoutMS) {
//recv rtp timeout!
strongSelf->onPlayResult_l(SockException(Err_timeout,"recv rtp timeout"));
//接收rtp媒体数据包超时
strongSelf->onPlayResult_l(SockException(Err_timeout,"receive rtp timeout"), true);
return false;
}
return true;
},getPoller()));
}
if (_pPlayTimer) {
if (!handshakeCompleted) {
//开始播放阶段
_pPlayTimer.reset();
onPlayResult(ex);
}else {
//播放中途阶段
if (ex) {
//播放成功后异常断开回调
onShutdown(ex);
}else{
//恢复播放
onResume();
}
} else if (ex) {
//播放成功后异常断开回调
onShutdown(ex);
} else {
//恢复播放
onResume();
}
if(ex){
......
src/Rtsp/RtspPlayer.h
......@@ -101,7 +101,7 @@ protected:
void onErr(const SockException &ex) override;
private:
void onRecvRTP_l(const RtpPacket::Ptr &pRtppt, const SdpTrack::Ptr &track);
void onPlayResult_l(const SockException &ex);
void onPlayResult_l(const SockException &ex , bool handshakeCompleted);
int getTrackIndexByControlSuffix(const string &controlSuffix) const;
int getTrackIndexByInterleaved(int interleaved) const;
......@@ -111,12 +111,11 @@ private:
void handleResSETUP(const Parser &parser, unsigned int uiTrackIndex);
void handleResDESCRIBE(const Parser &parser);
bool handleAuthenticationFailure(const string &wwwAuthenticateParamsStr);
void handleResPAUSE(const Parser &parser, bool bPause);
void handleResPAUSE(const Parser &parser, int type);
//发送SETUP命令
void sendSetup(unsigned int uiTrackIndex);
void sendPause(bool bPause,uint32_t ms, bool range);
void sendOptions();
void sendPause(int type , uint32_t ms);
void sendDescribe();
void sendRtspRequest(const string &cmd, const string &url ,const StrCaseMap &header = StrCaseMap());
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论