优化rtmp握手体验

优化代码结构
添加rtmp鉴权时间统计

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;

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序列正确次数累计达到该数字就启动清空排序缓存

src/Http/HttpSession.cpp

@@ -49,7 +49,6 @@ namespace Http {
 
 static int sock_flags = SOCKET_DEFAULE_FLAGS | FLAG_MORE;
 
-unordered_map<string, HttpSession::HttpCMDHandle> 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<ThreadPool> &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<string, HttpCMDHandle> 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;

src/Http/HttpSession.h

@@ -64,8 +64,6 @@ private:
 		Http_failed = 1,
 		Http_moreData = 2,
 	} HttpCode;
-	typedef HttpSession::HttpCode (HttpSession::*HttpCMDHandle)();
-	static unordered_map<string, HttpCMDHandle> g_mapCmdIndex;
 
 	Parser m_parser;
 	string m_strPath;

src/Rtmp/RtmpSession.cpp

@@ -32,23 +32,21 @@
 namespace ZL {
 namespace Rtmp {
 
-unordered_map<string, RtmpSession::rtmpCMDHandle> RtmpSession::g_mapCmd;
 RtmpSession::RtmpSession(const std::shared_ptr<ThreadPool> &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<Ticker> pTicker(new Ticker);
+    std::shared_ptr<onceToken> pToken(new onceToken(nullptr,[pTicker](){
+        DebugL << "publish 回复时间:" << pTicker->elapsedTime() << "ms";
+    }));
 	dec.load<AMFValue>();/* NULL */
     m_mediaInfo.parse(m_strTcUrl + "/" + dec.load<std::string>());
 
-    auto onRes = [this](const string &err){
+    auto onRes = [this,pToken](const string &err){
         auto src = dynamic_pointer_cast<RtmpMediaSource>(MediaSource::find(RTMP_SCHEMA,
                                                                            m_mediaInfo.m_vhost,
                                                                            m_mediaInfo.m_app,
@@ -171,12 +173,12 @@ void RtmpSession::onCmd_publish(AMFDecoder &dec) {
     };
 
     weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(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<onceToken> &pToken) {
     //获取流对象
     auto src = dynamic_pointer_cast<RtmpMediaSource>(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](){
+        //所以我们注册rtmp注册事件，等rtmp推流端推流成功后再告知播放器开始播放
-            //延时后就不再延时重试了
+        auto task_id = this;
-            doPlayResponse("",false);
+        auto media_info = m_mediaInfo;
+        weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(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<Ticker> pTicker(new Ticker);
+    std::shared_ptr<onceToken> pToken(new onceToken(nullptr,[pTicker](){
+        DebugL << "play 回复时间:" << pTicker->elapsedTime() << "ms";
+    }));
     weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(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<std::string>();
+    typedef void (RtmpSession::*rtmpCMDHandle)(AMFDecoder &dec);
+    static unordered_map<string, rtmpCMDHandle> 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<std::string>();
 	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<void()> &fun) {
+    if(m_delayTask){
+        m_delayTask();
+    }
     m_delayTask = fun;
     m_iTaskTimeLine = time(NULL) + delaySec;
 }
 
+void RtmpSession::cancelDelyaTask(){
+    m_delayTask = nullptr;
+}
+
 
 } /* namespace Rtmp */
 } /* namespace ZL */

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<string, rtmpCMDHandle> g_mapCmd;
-
-	RingBuffer<RtmpPacket::Ptr>::RingReader::Ptr m_pRingReader;
-	std::shared_ptr<RtmpMediaSource> m_pPublisherSrc;
-	bool m_bPublisherSrcRegisted = false;
-    std::weak_ptr<RtmpMediaSource> 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<onceToken> &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<void()> &fun);
+	void cancelDelyaTask();
+private:
+	std::string m_strTcUrl;
+	MediaInfo m_mediaInfo;
+	double m_dNowReqID = 0;
+	Ticker m_ticker;//数据接收时间
+	SmoothTicker m_stampTicker[2];//时间戳生产器
+	RingBuffer<RtmpPacket::Ptr>::RingReader::Ptr m_pRingReader;
+	std::shared_ptr<RtmpMediaSource> m_pPublisherSrc;
+	bool m_bPublisherSrcRegisted = false;
+	std::weak_ptr<RtmpMediaSource> m_pPlayerSrc;
+	uint32_t m_aui32FirstStamp[2] = {0};
+	//消耗的总流量
+	uint64_t m_ui64TotalBytes = 0;
     std::function<void()> m_delayTask;
     uint32_t m_iTaskTimeLine = 0;
 

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[RTP_BUF_SIZE];
-        m_pucRtpBuf = new uint8_t[rtpSize];
 	}
 	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 = RTP_BUF_SIZE - m_uiRtpBufLen;
-            int added = rtpSize - m_uiRtpBufLen;
 			added = (added > size ? size : added);
 			memcpy(m_pucRtpBuf + m_uiRtpBufLen, buf, added);
 			m_uiRtpBufLen += added;

src/Rtsp/RtspSession.cpp

@@ -54,21 +54,12 @@ unordered_map<string, weak_ptr<RtspSession> > RtspSession::g_mapGetter;
 unordered_map<void *, std::shared_ptr<RtspSession> > RtspSession::g_mapPostter;
 recursive_mutex RtspSession::g_mtxGetter; //对quicktime上锁保护
 recursive_mutex RtspSession::g_mtxPostter; //对quicktime上锁保护
-unordered_map<string, RtspSession::rtspCMDHandle> RtspSession::g_mapCmd;
 RtspSession::RtspSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock) :
 		TcpSession(pTh, pSock), m_pSender(pSock) {
-	static onceToken token( []() {
+	//设置10秒发送缓存
-		g_mapCmd.emplace("OPTIONS",&RtspSession::handleReq_Options);
+	pSock->setSendBufSecond(10);
-		g_mapCmd.emplace("DESCRIBE",&RtspSession::handleReq_Describe);
+	//设置15秒发送超时时间
-		g_mapCmd.emplace("SETUP",&RtspSession::handleReq_Setup);
+	pSock->setSendTimeOutSecond(15);
-		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);
-	}, []() {});
 
 	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<string, rtspCMDHandle> 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;

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<RtspMediaSource> m_pMediaSrc;
-	static unordered_map<string, rtspCMDHandle> g_mapCmd;
 
 	//RTP缓冲
 	weak_ptr<RingBuffer<RtpPacket::Ptr> > m_pWeakRing;