diff --git a/src/Common/config.cpp b/src/Common/config.cpp index 16172cbb..496922fe 100644 --- a/src/Common/config.cpp +++ b/src/Common/config.cpp @@ -89,7 +89,7 @@ const char kSendBufSize[] = HTTP_FIELD"sendBufSize"; const char kMaxReqSize[] = HTTP_FIELD"maxReqSize"; //http keep-alive秒数 -#define HTTP_KEEP_ALIVE_SECOND 5 +#define HTTP_KEEP_ALIVE_SECOND 10 const char kKeepAliveSecond[] = HTTP_FIELD"keepAliveSecond"; //http keep-alive最大请求数 @@ -194,10 +194,6 @@ const char kVideoMtuSize[] = RTP_FIELD"videoMtuSize"; #define RTP_Audio_MTU_SIZE 600 const char kAudioMtuSize[] = RTP_FIELD"audioMtuSize"; -//udp方式接受RTP包的最大缓存 -#define RTP_UDP_BUFSIZE (128 * 1024) -const char kUdpBufSize[] = RTP_FIELD"udpBufSize"; - //RTP排序缓存最大个数 #define RTP_MAX_RTP_COUNT 50 const char kMaxRtpCount[] = RTP_FIELD"maxRtpCount"; @@ -214,7 +210,6 @@ const char kCycleMS[] = RTP_FIELD"cycleMS"; onceToken token([](){ mINI::Instance()[kVideoMtuSize] = RTP_VIDOE_MTU_SIZE; mINI::Instance()[kAudioMtuSize] = RTP_Audio_MTU_SIZE; - mINI::Instance()[kUdpBufSize] = RTP_UDP_BUFSIZE; mINI::Instance()[kMaxRtpCount] = RTP_MAX_RTP_COUNT; mINI::Instance()[kClearCount] = RTP_CLEAR_COUNT; mINI::Instance()[kCycleMS] = RTP_CYCLE_MS; diff --git a/src/Common/config.h b/src/Common/config.h index a2f01c09..dab730ab 100644 --- a/src/Common/config.h +++ b/src/Common/config.h @@ -192,8 +192,6 @@ namespace Rtp { extern const char kVideoMtuSize[]; //RTP打包最大MTU,公网情况下更小 extern const char kAudioMtuSize[]; -//udp方式接受RTP包的最大缓存 -extern const char kUdpBufSize[]; //RTP排序缓存最大个数 extern const char kMaxRtpCount[]; //如果RTP序列正确次数累计达到该数字就启动清空排序缓存 diff --git a/src/Http/HttpSession.cpp b/src/Http/HttpSession.cpp index e3a22139..4c42036e 100644 --- a/src/Http/HttpSession.cpp +++ b/src/Http/HttpSession.cpp @@ -49,7 +49,6 @@ namespace Http { static int sock_flags = SOCKET_DEFAULE_FLAGS | FLAG_MORE; -unordered_map HttpSession::g_mapCmdIndex; string dateStr() { char buf[64]; time_t tt = time(NULL); @@ -102,13 +101,14 @@ get_mime_type(const char* name) { HttpSession::HttpSession(const std::shared_ptr &pTh, const Socket::Ptr &pSock) : TcpSession(pTh, pSock) { - GET_CONFIG_AND_REGISTER(string,rootPath,Config::Http::kRootPath); + //设置10秒发送缓存 + pSock->setSendBufSecond(10); + //设置15秒发送超时时间 + pSock->setSendTimeOutSecond(15); + + GET_CONFIG_AND_REGISTER(string,rootPath,Config::Http::kRootPath); m_strPath = rootPath; - static onceToken token([]() { - g_mapCmdIndex.emplace("GET",&HttpSession::Handle_Req_GET); - g_mapCmdIndex.emplace("POST",&HttpSession::Handle_Req_POST); - }, nullptr); } HttpSession::~HttpSession() { @@ -154,6 +154,14 @@ inline HttpSession::HttpCode HttpSession::parserHttpReq(const string &str) { m_parser.Parse(str.data()); urlDecode(m_parser); string cmd = m_parser.Method(); + + typedef HttpSession::HttpCode (HttpSession::*HttpCMDHandle)(); + static unordered_map g_mapCmdIndex; + static onceToken token([]() { + g_mapCmdIndex.emplace("GET",&HttpSession::Handle_Req_GET); + g_mapCmdIndex.emplace("POST",&HttpSession::Handle_Req_POST); + }, nullptr); + auto it = g_mapCmdIndex.find(cmd); if (it == g_mapCmdIndex.end()) { WarnL << cmd; diff --git a/src/Http/HttpSession.h b/src/Http/HttpSession.h index 758c08e8..21a34b3a 100644 --- a/src/Http/HttpSession.h +++ b/src/Http/HttpSession.h @@ -64,8 +64,6 @@ private: Http_failed = 1, Http_moreData = 2, } HttpCode; - typedef HttpSession::HttpCode (HttpSession::*HttpCMDHandle)(); - static unordered_map g_mapCmdIndex; Parser m_parser; string m_strPath; diff --git a/src/Rtmp/RtmpSession.cpp b/src/Rtmp/RtmpSession.cpp index ea4d8e68..90d0b51f 100644 --- a/src/Rtmp/RtmpSession.cpp +++ b/src/Rtmp/RtmpSession.cpp @@ -32,23 +32,21 @@ namespace ZL { namespace Rtmp { -unordered_map RtmpSession::g_mapCmd; RtmpSession::RtmpSession(const std::shared_ptr &pTh, const Socket::Ptr &pSock) : TcpSession(pTh, pSock) { - static onceToken token([]() { - g_mapCmd.emplace("connect",&RtmpSession::onCmd_connect); - g_mapCmd.emplace("createStream",&RtmpSession::onCmd_createStream); - g_mapCmd.emplace("publish",&RtmpSession::onCmd_publish); - g_mapCmd.emplace("deleteStream",&RtmpSession::onCmd_deleteStream); - g_mapCmd.emplace("play",&RtmpSession::onCmd_play); - g_mapCmd.emplace("play2",&RtmpSession::onCmd_play2); - g_mapCmd.emplace("seek",&RtmpSession::onCmd_seek); - g_mapCmd.emplace("pause",&RtmpSession::onCmd_pause);}, []() {}); DebugL << get_peer_ip(); + //设置10秒发送缓存 + pSock->setSendBufSecond(10); + //设置15秒发送超时时间 + pSock->setSendTimeOutSecond(15); } RtmpSession::~RtmpSession() { DebugL << get_peer_ip(); + if(m_delayTask){ + m_delayTask(); + m_delayTask = nullptr; + } } void RtmpSession::onError(const SockException& err) { @@ -63,7 +61,7 @@ void RtmpSession::onError(const SockException& err) { } void RtmpSession::onManager() { - if (m_ticker.createdTime() > 10 * 1000) { + if (m_ticker.createdTime() > 15 * 1000) { if (!m_pRingReader && !m_pPublisherSrc) { WarnL << "非法链接:" << get_peer_ip(); shutdown(); @@ -71,7 +69,7 @@ void RtmpSession::onManager() { } if (m_pPublisherSrc) { //publisher - if (m_ticker.elapsedTime() > 10 * 1000) { + if (m_ticker.elapsedTime() > 15 * 1000) { WarnL << "数据接收超时:" << get_peer_ip(); shutdown(); } @@ -139,10 +137,14 @@ void RtmpSession::onCmd_createStream(AMFDecoder &dec) { } void RtmpSession::onCmd_publish(AMFDecoder &dec) { + std::shared_ptr pTicker(new Ticker); + std::shared_ptr pToken(new onceToken(nullptr,[pTicker](){ + DebugL << "publish 回复时间:" << pTicker->elapsedTime() << "ms"; + })); dec.load();/* NULL */ m_mediaInfo.parse(m_strTcUrl + "/" + dec.load()); - auto onRes = [this](const string &err){ + auto onRes = [this,pToken](const string &err){ auto src = dynamic_pointer_cast(MediaSource::find(RTMP_SCHEMA, m_mediaInfo.m_vhost, m_mediaInfo.m_app, @@ -171,12 +173,12 @@ void RtmpSession::onCmd_publish(AMFDecoder &dec) { }; weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); - Broadcast::AuthInvoker invoker = [weakSelf,onRes](const string &err){ + Broadcast::AuthInvoker invoker = [weakSelf,onRes,pToken](const string &err){ auto strongSelf = weakSelf.lock(); if(!strongSelf){ return; } - strongSelf->async([weakSelf,onRes,err](){ + strongSelf->async([weakSelf,onRes,err,pToken](){ auto strongSelf = weakSelf.lock(); if(!strongSelf){ return; @@ -203,7 +205,7 @@ void RtmpSession::onCmd_deleteStream(AMFDecoder &dec) { throw std::runtime_error(StrPrinter << "Stop publishing." << endl); } -void RtmpSession::doPlayResponse(const string &err,bool tryDelay) { +void RtmpSession::doPlayResponse(const string &err,bool tryDelay,const std::shared_ptr &pToken) { //获取流对象 auto src = dynamic_pointer_cast(MediaSource::find(RTMP_SCHEMA, m_mediaInfo.m_vhost, @@ -220,10 +222,52 @@ void RtmpSession::doPlayResponse(const string &err,bool tryDelay) { //是否鉴权成功 bool authSuccess = err.empty(); if(authSuccess && !src && tryDelay ){ - //校验成功,但是流不存在而导致的不能播放,我们看看该流延时几秒后是否确实不能播放 - doDelay(3,[this](){ - //延时后就不再延时重试了 - doPlayResponse("",false); + //校验成功,但是流不存在而导致的不能播放 + //所以我们注册rtmp注册事件,等rtmp推流端推流成功后再告知播放器开始播放 + auto task_id = this; + auto media_info = m_mediaInfo; + weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); + + NoticeCenter::Instance().addListener(task_id,Broadcast::kBroadcastMediaChanged, + [task_id,weakSelf,media_info,pToken](BroadcastMediaChangedArgs){ + if(bRegist && + schema == media_info.m_schema && + vhost == media_info.m_vhost && + app == media_info.m_app && + stream == media_info.m_schema){ + //播发器请求的rtmp流终于注册上了 + auto strongSelf = weakSelf.lock(); + if(!strongSelf) { + return; + } + //切换到自己的线程再回复 + strongSelf->async([task_id,weakSelf,pToken](){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf) { + return; + } + //回复播放器 + strongSelf->doPlayResponse("",false,pToken); + //取消延时任务,防止多次回复 + strongSelf->cancelDelyaTask(); + //取消事件监听 + NoticeCenter::Instance().delListener(task_id,Broadcast::kBroadcastMediaChanged); + }); + } + }); + + //5秒后执行延时任务 + doDelay(5,[task_id,weakSelf,pToken](){ + //取消监听该事件,该延时任务可以在本对象析构时或到达指定延时后调用 + //所以该对象在销毁前一定会被取消事件监听 + NoticeCenter::Instance().delListener(task_id,Broadcast::kBroadcastMediaChanged); + + auto strongSelf = weakSelf.lock(); + if(!strongSelf) { + return; + } + //5秒后,我们不管流有没有注册上都回复播放器 + strongSelf->doPlayResponse("",false,pToken); }); return; } @@ -325,24 +369,28 @@ void RtmpSession::doPlayResponse(const string &err,bool tryDelay) { } void RtmpSession::doPlay(AMFDecoder &dec){ + std::shared_ptr pTicker(new Ticker); + std::shared_ptr pToken(new onceToken(nullptr,[pTicker](){ + DebugL << "play 回复时间:" << pTicker->elapsedTime() << "ms"; + })); weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); - Broadcast::AuthInvoker invoker = [weakSelf](const string &err){ + Broadcast::AuthInvoker invoker = [weakSelf,pToken](const string &err){ auto strongSelf = weakSelf.lock(); if(!strongSelf){ return; } - strongSelf->async([weakSelf,err](){ + strongSelf->async([weakSelf,err,pToken](){ auto strongSelf = weakSelf.lock(); if(!strongSelf){ return; } - strongSelf->doPlayResponse(err,true); + strongSelf->doPlayResponse(err,true,pToken); }); }; auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed,m_mediaInfo,invoker,*this); if(!flag){ //该事件无人监听,默认不鉴权 - doPlayResponse("",true); + doPlayResponse("",true,pToken); } } void RtmpSession::onCmd_play2(AMFDecoder &dec) { @@ -401,7 +449,19 @@ void RtmpSession::setMetaData(AMFDecoder &dec) { } void RtmpSession::onProcessCmd(AMFDecoder &dec) { - std::string method = dec.load(); + typedef void (RtmpSession::*rtmpCMDHandle)(AMFDecoder &dec); + static unordered_map g_mapCmd; + static onceToken token([]() { + g_mapCmd.emplace("connect",&RtmpSession::onCmd_connect); + g_mapCmd.emplace("createStream",&RtmpSession::onCmd_createStream); + g_mapCmd.emplace("publish",&RtmpSession::onCmd_publish); + g_mapCmd.emplace("deleteStream",&RtmpSession::onCmd_deleteStream); + g_mapCmd.emplace("play",&RtmpSession::onCmd_play); + g_mapCmd.emplace("play2",&RtmpSession::onCmd_play2); + g_mapCmd.emplace("seek",&RtmpSession::onCmd_seek); + g_mapCmd.emplace("pause",&RtmpSession::onCmd_pause);}, []() {}); + + std::string method = dec.load(); auto it = g_mapCmd.find(method); if (it == g_mapCmd.end()) { TraceL << "can not support cmd:" << method; @@ -487,10 +547,17 @@ void RtmpSession::onSendMedia(const RtmpPacket::Ptr &pkt) { } void RtmpSession::doDelay(int delaySec, const std::function &fun) { + if(m_delayTask){ + m_delayTask(); + } m_delayTask = fun; m_iTaskTimeLine = time(NULL) + delaySec; } +void RtmpSession::cancelDelyaTask(){ + m_delayTask = nullptr; +} + } /* namespace Rtmp */ } /* namespace ZL */ diff --git a/src/Rtmp/RtmpSession.h b/src/Rtmp/RtmpSession.h index 7ab2f6ee..eaeed8f0 100644 --- a/src/Rtmp/RtmpSession.h +++ b/src/Rtmp/RtmpSession.h @@ -53,21 +53,6 @@ public: void onError(const SockException &err) override; void onManager() override; private: - std::string m_strTcUrl; - MediaInfo m_mediaInfo; - double m_dNowReqID = 0; - Ticker m_ticker;//数据接收时间 - SmoothTicker m_stampTicker[2];//时间戳生产器 - typedef void (RtmpSession::*rtmpCMDHandle)(AMFDecoder &dec); - static unordered_map g_mapCmd; - - RingBuffer::RingReader::Ptr m_pRingReader; - std::shared_ptr m_pPublisherSrc; - bool m_bPublisherSrcRegisted = false; - std::weak_ptr m_pPlayerSrc; - uint32_t m_aui32FirstStamp[2] = {0}; - //消耗的总流量 - uint64_t m_ui64TotalBytes = 0; void onProcessCmd(AMFDecoder &dec); void onCmd_connect(AMFDecoder &dec); void onCmd_createStream(AMFDecoder &dec); @@ -78,7 +63,7 @@ private: void onCmd_play(AMFDecoder &dec); void onCmd_play2(AMFDecoder &dec); void doPlay(AMFDecoder &dec); - void doPlayResponse(const string &err,bool tryDelay); + void doPlayResponse(const string &err,bool tryDelay,const std::shared_ptr &token); void onCmd_seek(AMFDecoder &dec); void onCmd_pause(AMFDecoder &dec); void setMetaData(AMFDecoder &dec); @@ -108,6 +93,20 @@ private: } void doDelay(int delaySec,const std::function &fun); + void cancelDelyaTask(); +private: + std::string m_strTcUrl; + MediaInfo m_mediaInfo; + double m_dNowReqID = 0; + Ticker m_ticker;//数据接收时间 + SmoothTicker m_stampTicker[2];//时间戳生产器 + RingBuffer::RingReader::Ptr m_pRingReader; + std::shared_ptr m_pPublisherSrc; + bool m_bPublisherSrcRegisted = false; + std::weak_ptr m_pPlayerSrc; + uint32_t m_aui32FirstStamp[2] = {0}; + //消耗的总流量 + uint64_t m_ui64TotalBytes = 0; std::function m_delayTask; uint32_t m_iTaskTimeLine = 0; diff --git a/src/Rtsp/RtspPlayer.cpp b/src/Rtsp/RtspPlayer.cpp index 90a5fbbe..99f59d57 100644 --- a/src/Rtsp/RtspPlayer.cpp +++ b/src/Rtsp/RtspPlayer.cpp @@ -50,6 +50,8 @@ namespace Rtsp { onRecvRTP_l(it->second, trackidx); \ m_amapRtpSort[trackidx].erase(it); +#define RTP_BUF_SIZE (4 * 1024) + const char kRtspMd5Nonce[] = "rtsp_md5_nonce"; const char kRtspRealm[] = "rtsp_realm"; @@ -133,8 +135,7 @@ void RtspPlayer::play(const char* strUrl, const char *strUser, const char *strPw m_eType = eType; if (m_eType == RTP_TCP && !m_pucRtpBuf) { - GET_CONFIG_AND_REGISTER(uint32_t,rtpSize,Config::Rtp::kUdpBufSize); - m_pucRtpBuf = new uint8_t[rtpSize]; + m_pucRtpBuf = new uint8_t[RTP_BUF_SIZE]; } auto ip = FindField(strUrl, "://", "/"); if (!ip.size()) { @@ -207,8 +208,7 @@ void RtspPlayer::onRecv(const Buffer::Ptr& pBuf) { if (m_eType == RTP_TCP && m_pucRtpBuf) { //RTP data while (size > 0) { - GET_CONFIG_AND_REGISTER(uint32_t,rtpSize,Config::Rtp::kUdpBufSize); - int added = rtpSize - m_uiRtpBufLen; + int added = RTP_BUF_SIZE - m_uiRtpBufLen; added = (added > size ? size : added); memcpy(m_pucRtpBuf + m_uiRtpBufLen, buf, added); m_uiRtpBufLen += added; diff --git a/src/Rtsp/RtspSession.cpp b/src/Rtsp/RtspSession.cpp index cf9baaf5..3e454df9 100644 --- a/src/Rtsp/RtspSession.cpp +++ b/src/Rtsp/RtspSession.cpp @@ -54,21 +54,12 @@ unordered_map > RtspSession::g_mapGetter; unordered_map > RtspSession::g_mapPostter; recursive_mutex RtspSession::g_mtxGetter; //对quicktime上锁保护 recursive_mutex RtspSession::g_mtxPostter; //对quicktime上锁保护 -unordered_map RtspSession::g_mapCmd; RtspSession::RtspSession(const std::shared_ptr &pTh, const Socket::Ptr &pSock) : TcpSession(pTh, pSock), m_pSender(pSock) { - static onceToken token( []() { - g_mapCmd.emplace("OPTIONS",&RtspSession::handleReq_Options); - g_mapCmd.emplace("DESCRIBE",&RtspSession::handleReq_Describe); - g_mapCmd.emplace("SETUP",&RtspSession::handleReq_Setup); - g_mapCmd.emplace("PLAY",&RtspSession::handleReq_Play); - g_mapCmd.emplace("PAUSE",&RtspSession::handleReq_Pause); - g_mapCmd.emplace("TEARDOWN",&RtspSession::handleReq_Teardown); - g_mapCmd.emplace("GET",&RtspSession::handleReq_Get); - g_mapCmd.emplace("POST",&RtspSession::handleReq_Post); - g_mapCmd.emplace("SET_PARAMETER",&RtspSession::handleReq_SET_PARAMETER); - g_mapCmd.emplace("GET_PARAMETER",&RtspSession::handleReq_SET_PARAMETER); - }, []() {}); + //设置10秒发送缓存 + pSock->setSendBufSecond(10); + //设置15秒发送超时时间 + pSock->setSendTimeOutSecond(15); DebugL << get_peer_ip(); } @@ -128,7 +119,7 @@ void RtspSession::onError(const SockException& err) { } void RtspSession::onManager() { - if (m_ticker.createdTime() > 10 * 1000) { + if (m_ticker.createdTime() > 15 * 1000) { if (m_strSession.size() == 0) { WarnL << "非法链接:" << get_peer_ip(); shutdown(); @@ -162,6 +153,22 @@ void RtspSession::onRecv(const Buffer::Ptr &pBuf) { m_iCseq = atoi(m_parser["CSeq"].data()); bool ret = false; + + typedef bool (RtspSession::*rtspCMDHandle)(); + static unordered_map g_mapCmd; + static onceToken token( []() { + g_mapCmd.emplace("OPTIONS",&RtspSession::handleReq_Options); + g_mapCmd.emplace("DESCRIBE",&RtspSession::handleReq_Describe); + g_mapCmd.emplace("SETUP",&RtspSession::handleReq_Setup); + g_mapCmd.emplace("PLAY",&RtspSession::handleReq_Play); + g_mapCmd.emplace("PAUSE",&RtspSession::handleReq_Pause); + g_mapCmd.emplace("TEARDOWN",&RtspSession::handleReq_Teardown); + g_mapCmd.emplace("GET",&RtspSession::handleReq_Get); + g_mapCmd.emplace("POST",&RtspSession::handleReq_Post); + g_mapCmd.emplace("SET_PARAMETER",&RtspSession::handleReq_SET_PARAMETER); + g_mapCmd.emplace("GET_PARAMETER",&RtspSession::handleReq_SET_PARAMETER); + }, []() {}); + auto it = g_mapCmd.find(strCmd); if (it != g_mapCmd.end()) { auto fun = it->second; diff --git a/src/Rtsp/RtspSession.h b/src/Rtsp/RtspSession.h index 19f99ac9..2ed20b75 100644 --- a/src/Rtsp/RtspSession.h +++ b/src/Rtsp/RtspSession.h @@ -81,7 +81,6 @@ public: void onError(const SockException &err) override; void onManager() override; private: - typedef bool (RtspSession::*rtspCMDHandle)(); int send(const string &strBuf) override { m_ui64TotalBytes += strBuf.size(); return m_pSender->send(strBuf); @@ -162,7 +161,6 @@ private: bool m_bFirstPlay = true; MediaInfo m_mediaInfo; std::weak_ptr m_pMediaSrc; - static unordered_map g_mapCmd; //RTP缓冲 weak_ptr > m_pWeakRing;