diff --git a/src/Player/PlayerProxy.cpp b/src/Player/PlayerProxy.cpp index ec865788..b2b567ff 100644 --- a/src/Player/PlayerProxy.cpp +++ b/src/Player/PlayerProxy.cpp @@ -8,29 +8,31 @@ * may be found in the AUTHORS file in the root of the source tree. */ -#include "Common/config.h" #include "PlayerProxy.h" -#include "Util/mini.h" -#include "Util/MD5.h" -#include "Util/logger.h" +#include "Common/config.h" #include "Extension/AAC.h" #include "Rtmp/RtmpMediaSource.h" -#include "Rtsp/RtspMediaSource.h" #include "Rtmp/RtmpPlayer.h" +#include "Rtsp/RtspMediaSource.h" #include "Rtsp/RtspPlayer.h" +#include "Util/MD5.h" +#include "Util/logger.h" +#include "Util/mini.h" using namespace toolkit; using namespace std; namespace mediakit { -PlayerProxy::PlayerProxy(const string &vhost, const string &app, const string &stream_id, const ProtocolOption &option, - int retry_count, const EventPoller::Ptr &poller) : MediaPlayer(poller) , _option(option) { +PlayerProxy::PlayerProxy( + const string &vhost, const string &app, const string &stream_id, const ProtocolOption &option, int retry_count, const EventPoller::Ptr &poller) + : MediaPlayer(poller) + , _option(option) { _vhost = vhost; _app = app; _stream_id = stream_id; _retry_count = retry_count; - + _live_secs = 0; _live_status = 1; _repull_count = 0; @@ -48,7 +50,7 @@ void PlayerProxy::setOnClose(const function &cb) void PlayerProxy::play(const string &strUrlTmp) { weak_ptr weakSelf = shared_from_this(); - std::shared_ptr piFailedCnt(new int(0)); //连续播放失败次数 + std::shared_ptr piFailedCnt(new int(0)); // 连续播放失败次数 setOnPlayResult([weakSelf, strUrlTmp, piFailedCnt](const SockException &err) { auto strongSelf = weakSelf.lock(); if (!strongSelf) { @@ -66,7 +68,7 @@ void PlayerProxy::play(const string &strUrlTmp) { strongSelf->_live_ticker.resetTime(); strongSelf->_live_status = 0; // 播放成功 - *piFailedCnt = 0;//连续播放失败次数清0 + *piFailedCnt = 0; // 连续播放失败次数清0 strongSelf->onPlaySuccess(); InfoL << "play " << strUrlTmp << " success"; @@ -74,7 +76,7 @@ void PlayerProxy::play(const string &strUrlTmp) { // 播放失败,延时重试播放 strongSelf->rePlay(strUrlTmp, (*piFailedCnt)++); } else { - //达到了最大重试次数,回调关闭 + // 达到了最大重试次数,回调关闭 strongSelf->_on_close(err); } }); @@ -84,7 +86,7 @@ void PlayerProxy::play(const string &strUrlTmp) { return; } - //注销直接拉流代理产生的流:#532 + // 注销直接拉流代理产生的流:#532 strongSelf->setMediaSource(nullptr); if (strongSelf->_muxer) { @@ -101,19 +103,19 @@ void PlayerProxy::play(const string &strUrlTmp) { } } - if(*piFailedCnt == 0){ + if (*piFailedCnt == 0) { // 第一次重拉更新时长 - strongSelf->_live_secs += strongSelf->_live_ticker.elapsedTime()/1000; + strongSelf->_live_secs += strongSelf->_live_ticker.elapsedTime() / 1000; strongSelf->_live_ticker.resetTime(); - TraceL<<" live secs "<_live_secs; + TraceL << " live secs " << strongSelf->_live_secs; } - //播放异常中断,延时重试播放 + // 播放异常中断,延时重试播放 if (*piFailedCnt < strongSelf->_retry_count || strongSelf->_retry_count < 0) { strongSelf->_repull_count++; strongSelf->rePlay(strUrlTmp, (*piFailedCnt)++); } else { - //达到了最大重试次数,回调关闭 + // 达到了最大重试次数,回调关闭 strongSelf->_on_close(err); } }); @@ -131,13 +133,13 @@ void PlayerProxy::play(const string &strUrlTmp) { void PlayerProxy::setDirectProxy() { MediaSource::Ptr mediaSource; if (dynamic_pointer_cast(_delegate)) { - //rtsp拉流 + // rtsp拉流 GET_CONFIG(bool, directProxy, Rtsp::kDirectProxy); if (directProxy) { mediaSource = std::make_shared(_vhost, _app, _stream_id); } } else if (dynamic_pointer_cast(_delegate)) { - //rtmp拉流,rtmp强制直接代理 + // rtmp拉流,rtmp强制直接代理 mediaSource = std::make_shared(_vhost, _app, _stream_id); } if (mediaSource) { @@ -148,7 +150,7 @@ void PlayerProxy::setDirectProxy() { PlayerProxy::~PlayerProxy() { _timer.reset(); // 避免析构时, 忘记回调api请求 - if(_on_play) { + if (_on_play) { _on_play(SockException(Err_shutdown, "player proxy close")); _on_play = nullptr; } @@ -157,21 +159,24 @@ PlayerProxy::~PlayerProxy() { void PlayerProxy::rePlay(const string &strUrl, int iFailedCnt) { auto iDelay = MAX(2 * 1000, MIN(iFailedCnt * 3000, 60 * 1000)); weak_ptr weakSelf = shared_from_this(); - _timer = std::make_shared(iDelay / 1000.0f, [weakSelf, strUrl, iFailedCnt]() { - //播放失败次数越多,则延时越长 - auto strongPlayer = weakSelf.lock(); - if (!strongPlayer) { + _timer = std::make_shared( + iDelay / 1000.0f, + [weakSelf, strUrl, iFailedCnt]() { + // 播放失败次数越多,则延时越长 + auto strongPlayer = weakSelf.lock(); + if (!strongPlayer) { + return false; + } + WarnL << "重试播放[" << iFailedCnt << "]:" << strUrl; + strongPlayer->MediaPlayer::play(strUrl); + strongPlayer->setDirectProxy(); return false; - } - WarnL << "重试播放[" << iFailedCnt << "]:" << strUrl; - strongPlayer->MediaPlayer::play(strUrl); - strongPlayer->setDirectProxy(); - return false; - }, getPoller()); + }, + getPoller()); } bool PlayerProxy::close(MediaSource &sender) { - //通知其停止推流 + // 通知其停止推流 weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); getPoller()->async_first([weakSelf]() { auto strongSelf = weakSelf.lock(); @@ -214,19 +219,19 @@ float PlayerProxy::getLossRate(MediaSource &sender, TrackType type) { void PlayerProxy::onPlaySuccess() { GET_CONFIG(bool, reset_when_replay, General::kResetWhenRePlay); if (dynamic_pointer_cast(_media_src)) { - //rtsp拉流代理 + // rtsp拉流代理 if (reset_when_replay || !_muxer) { _option.enable_rtsp = false; _muxer = std::make_shared(_vhost, _app, _stream_id, getDuration(), _option); } } else if (dynamic_pointer_cast(_media_src)) { - //rtmp拉流代理 + // rtmp拉流代理 if (reset_when_replay || !_muxer) { _option.enable_rtmp = false; _muxer = std::make_shared(_vhost, _app, _stream_id, getDuration(), _option); } } else { - //其他拉流代理 + // 其他拉流代理 if (reset_when_replay || !_muxer) { _muxer = std::make_shared(_vhost, _app, _stream_id, getDuration(), _option); } @@ -235,25 +240,25 @@ void PlayerProxy::onPlaySuccess() { auto videoTrack = getTrack(TrackVideo, false); if (videoTrack) { - //添加视频 + // 添加视频 _muxer->addTrack(videoTrack); - //视频数据写入_mediaMuxer + // 视频数据写入_mediaMuxer videoTrack->addDelegate(_muxer); } auto audioTrack = getTrack(TrackAudio, false); if (audioTrack) { - //添加音频 + // 添加音频 _muxer->addTrack(audioTrack); - //音频数据写入_mediaMuxer + // 音频数据写入_mediaMuxer audioTrack->addDelegate(_muxer); } - //添加完毕所有track,防止单track情况下最大等待3秒 + // 添加完毕所有track,防止单track情况下最大等待3秒 _muxer->addTrackCompleted(); if (_media_src) { - //让_muxer对象拦截一部分事件(比如说录像相关事件) + // 让_muxer对象拦截一部分事件(比如说录像相关事件) _media_src->setListener(_muxer); } } @@ -262,14 +267,14 @@ int PlayerProxy::getStatus() { return _live_status.load(); } uint64_t PlayerProxy::getLiveSecs() { - if(_live_status == 0){ - return _live_secs + _live_ticker.elapsedTime()/1000; - }else{ + if (_live_status == 0) { + return _live_secs + _live_ticker.elapsedTime() / 1000; + } else { return _live_secs; } } -uint64_t PlayerProxy::getRePullCount(){ +uint64_t PlayerProxy::getRePullCount() { return _repull_count; } diff --git a/src/Player/PlayerProxy.h b/src/Player/PlayerProxy.h index 5ab23481..d828f696 100644 --- a/src/Player/PlayerProxy.h +++ b/src/Player/PlayerProxy.h @@ -11,21 +11,25 @@ #ifndef SRC_DEVICE_PLAYERPROXY_H_ #define SRC_DEVICE_PLAYERPROXY_H_ -#include #include "Common/MultiMediaSourceMuxer.h" #include "Player/MediaPlayer.h" #include "Util/TimeTicker.h" +#include namespace mediakit { -class PlayerProxy : public MediaPlayer, public MediaSourceEvent, public std::enable_shared_from_this { +class PlayerProxy + : public MediaPlayer + , public MediaSourceEvent + , public std::enable_shared_from_this { public: using Ptr = std::shared_ptr; - //如果retry_count<0,则一直重试播放;否则重试retry_count次数 - //默认一直重试 - PlayerProxy(const std::string &vhost, const std::string &app, const std::string &stream_id, - const ProtocolOption &option, int retry_count = -1, const toolkit::EventPoller::Ptr &poller = nullptr); + // 如果retry_count<0,则一直重试播放;否则重试retry_count次数 + // 默认一直重试 + PlayerProxy( + const std::string &vhost, const std::string &app, const std::string &stream_id, const ProtocolOption &option, int retry_count = -1, + const toolkit::EventPoller::Ptr &poller = nullptr); ~PlayerProxy() override; @@ -50,15 +54,14 @@ public: /** * 获取观看总人数 */ - int totalReaderCount() ; - + int totalReaderCount(); int getStatus(); uint64_t getLiveSecs(); uint64_t getRePullCount(); private: - //MediaSourceEvent override + // MediaSourceEvent override bool close(MediaSource &sender) override; int totalReaderCount(MediaSource &sender) override; MediaOriginType getOriginType(MediaSource &sender) const override; @@ -66,7 +69,7 @@ private: std::shared_ptr getOriginSock(MediaSource &sender) const override; float getLossRate(MediaSource &sender, TrackType type) override; - void rePlay(const std::string &strUrl,int iFailedCnt); + void rePlay(const std::string &strUrl, int iFailedCnt); void onPlaySuccess(); void setDirectProxy(); @@ -83,8 +86,8 @@ private: MultiMediaSourceMuxer::Ptr _muxer; toolkit::Ticker _live_ticker; - //0 表示正常 1 表示正在尝试拉流 - std::atomic _live_status; + // 0 表示正常 1 表示正在尝试拉流 + std::atomic _live_status; std::atomic _live_secs; std::atomic _repull_count; diff --git a/src/Pusher/PusherProxy.cpp b/src/Pusher/PusherProxy.cpp index b936842a..ffe1a500 100644 --- a/src/Pusher/PusherProxy.cpp +++ b/src/Pusher/PusherProxy.cpp @@ -16,7 +16,7 @@ using namespace std; namespace mediakit { PusherProxy::PusherProxy(const MediaSource::Ptr &src, int retry_count, const EventPoller::Ptr &poller) - : MediaPusher(src, poller){ + : MediaPusher(src, poller) { _retry_count = retry_count; _on_close = [](const SockException &) {}; _weak_src = src; @@ -65,7 +65,7 @@ void PusherProxy::publish(const string &dst_url) { strong_self->_live_status = 1; strong_self->rePublish(dst_url, (*failed_cnt)++); } else { - //如果媒体源已经注销, 或达到了最大重试次数,回调关闭 + // 如果媒体源已经注销, 或达到了最大重试次数,回调关闭 strong_self->_on_close(err); } }); @@ -76,20 +76,20 @@ void PusherProxy::publish(const string &dst_url) { return; } - if(*failed_cnt == 0){ + if (*failed_cnt == 0) { // 第一次重推更新时长 - strong_self->_live_secs += strong_self->_live_ticker.elapsedTime()/1000; + strong_self->_live_secs += strong_self->_live_ticker.elapsedTime() / 1000; strong_self->_live_ticker.resetTime(); - TraceL<<" live secs "<_live_secs; + TraceL << " live secs " << strong_self->_live_secs; } auto src = strong_self->_weak_src.lock(); - //推流异常中断,延时重试播放 + // 推流异常中断,延时重试播放 if (src && (*failed_cnt < strong_self->_retry_count || strong_self->_retry_count < 0)) { strong_self->_republish_count++; strong_self->rePublish(dst_url, (*failed_cnt)++); } else { - //如果媒体源已经注销, 或达到了最大重试次数,回调关闭 + // 如果媒体源已经注销, 或达到了最大重试次数,回调关闭 strong_self->_on_close(err); } }); @@ -100,30 +100,33 @@ void PusherProxy::publish(const string &dst_url) { void PusherProxy::rePublish(const string &dst_url, int failed_cnt) { auto delay = MAX(2 * 1000, MIN(failed_cnt * 3000, 60 * 1000)); weak_ptr weak_self = shared_from_this(); - _timer = std::make_shared(delay / 1000.0f, [weak_self, dst_url, failed_cnt]() { - //推流失败次数越多,则延时越长 - auto strong_self = weak_self.lock(); - if (!strong_self) { + _timer = std::make_shared( + delay / 1000.0f, + [weak_self, dst_url, failed_cnt]() { + // 推流失败次数越多,则延时越长 + auto strong_self = weak_self.lock(); + if (!strong_self) { + return false; + } + WarnL << "推流重试[" << failed_cnt << "]:" << dst_url; + strong_self->MediaPusher::publish(dst_url); return false; - } - WarnL << "推流重试[" << failed_cnt << "]:" << dst_url; - strong_self->MediaPusher::publish(dst_url); - return false; - }, getPoller()); + }, + getPoller()); } int PusherProxy::getStatus() { return _live_status.load(); } uint64_t PusherProxy::getLiveSecs() { - if(_live_status == 0){ - return _live_secs + _live_ticker.elapsedTime()/1000; - }else{ + if (_live_status == 0) { + return _live_secs + _live_ticker.elapsedTime() / 1000; + } else { return _live_secs; } } -uint64_t PusherProxy::getRePublishCount(){ +uint64_t PusherProxy::getRePublishCount() { return _republish_count; } diff --git a/src/Pusher/PusherProxy.h b/src/Pusher/PusherProxy.h index baeef029..e0146514 100644 --- a/src/Pusher/PusherProxy.h +++ b/src/Pusher/PusherProxy.h @@ -16,7 +16,9 @@ namespace mediakit { -class PusherProxy : public MediaPusher, public std::enable_shared_from_this { +class PusherProxy + : public MediaPusher + , public std::enable_shared_from_this { public: using Ptr = std::shared_ptr; @@ -41,7 +43,7 @@ public: * 开始拉流播放 * @param dstUrl 目标推流地址 */ - void publish(const std::string& dstUrl) override; + void publish(const std::string &dstUrl) override; int getStatus(); uint64_t getLiveSecs(); @@ -55,8 +57,8 @@ private: int _retry_count; toolkit::Timer::Ptr _timer; toolkit::Ticker _live_ticker; - //0 表示正常 1 表示正在尝试推流 - std::atomic _live_status; + // 0 表示正常 1 表示正在尝试推流 + std::atomic _live_status; std::atomic _live_secs; std::atomic _republish_count; std::weak_ptr _weak_src; @@ -66,4 +68,4 @@ private: } /* namespace mediakit */ -#endif //SRC_DEVICE_PUSHERPROXY_H +#endif // SRC_DEVICE_PUSHERPROXY_H