解决rtsp粘包问题

src/Rtsp/RtspSession.cpp

@@ -139,35 +139,19 @@ void RtspSession::onManager() {
 	}
 }
 
-void RtspSession::onRecv(const Buffer::Ptr &pBuf) {
-	m_ticker.resetTime();
+
+void RtspSession::onRecvContent(const string &content){
+
+}
+
+int64_t RtspSession::onRecvHeader(const string &header) {
     char tmp[2 * 1024];
     m_pcBuf = tmp;
 
-    m_ui64TotalBytes += pBuf->size();
-    if (m_bBase64need) {
-		//quicktime 加密后的rtsp请求，需要解密
-		av_base64_decode((uint8_t *) m_pcBuf, pBuf->data(), sizeof(tmp));
-		m_parser.Parse(m_pcBuf); //rtsp请求解析
-	} else {
-		m_parser.Parse(pBuf->data()); //rtsp请求解析
-		if(!m_parser.Content().empty()){
-			weak_ptr<TcpSession> weakSelf = shared_from_this();
-			BufferString::Ptr nextRecv(new BufferString(m_parser.Content()));
-			async([weakSelf,nextRecv](){
-				auto strongSelf = weakSelf.lock();
-				if(strongSelf){
-					strongSelf->onRecv(nextRecv);
-				}
-			}, false);
-		}
-	}
-
+	m_parser.Parse(header.data()); //rtsp请求解析
 	string strCmd = m_parser.Method(); //提取出请求命令字
 	m_iCseq = atoi(m_parser["CSeq"].data());
 
-	bool ret = false;
-
 	typedef bool (RtspSession::*rtspCMDHandle)();
 	static unordered_map<string, rtspCMDHandle> g_mapCmd;
 	static onceToken token( []() {
@@ -186,15 +170,36 @@ void RtspSession::onRecv(const Buffer::Ptr &pBuf) {
 	auto it = g_mapCmd.find(strCmd);
 	if (it != g_mapCmd.end()) {
 		auto fun = it->second;
-		ret = (this->*fun)();
-		m_parser.Clear();
-	} else {
-		ret = (m_rtpType == PlayerBase::RTP_TCP);
+		if(!(this->*fun)()){
+		    shutdown();
 		}
-	if (!ret) {
+	} else{
 		shutdown();
 		WarnL << "cmd=" << strCmd;
 	}
+
+    m_parser.Clear();
+    return 0;
+}
+
+
+void RtspSession::onRecv(const Buffer::Ptr &pBuf) {
+	m_ticker.resetTime();
+    m_ui64TotalBytes += pBuf->size();
+    if (m_bBase64need) {
+		//quicktime 加密后的rtsp请求，需要解密
+		inputRtspOrRtcp(decodeBase64(string(pBuf->data(),pBuf->size())));
+	} else {
+        inputRtspOrRtcp(string(pBuf->data(),pBuf->size()));
+	}
+}
+
+void RtspSession::inputRtspOrRtcp(const string &str) {
+	if(str[0] == '$' && m_rtpType == PlayerBase::RTP_TCP){
+		//这是rtcp
+		return;
+	}
+    input(str);
 }
 
 bool RtspSession::handleReq_Options() {

src/Rtsp/RtspSession.h

@@ -38,6 +38,7 @@
 #include "Util/util.h"
 #include "Util/logger.h"
 #include "Network/TcpSession.h"
+#include "Http/HttpRequestSplitter.h"
 
 using namespace std;
 using namespace ZL::Util;
@@ -67,7 +68,7 @@ private:
     uint32_t _offset;
 };
 
-class RtspSession: public TcpSession {
+class RtspSession: public TcpSession, public HttpRequestSplitter {
 public:
 	typedef std::shared_ptr<RtspSession> Ptr;
 	typedef std::function<void(const string &realm)> onGetRealm;
@@ -80,7 +81,27 @@ public:
 	void onRecv(const Buffer::Ptr &pBuf) override;
 	void onError(const SockException &err) override;
 	void onManager() override;
+
+protected:
+	//HttpRequestSplitter override
+	/**
+    * 收到请求头
+    * @param header 请求头
+    * @return 请求头后的content长度,
+    *  <0 : 代表后面所有数据都是content
+    *  0 : 代表为后面数据还是请求头,
+    *  >0 : 代表后面数据为固定长度content,
+    */
+	int64_t onRecvHeader(const string &header) override ;
+
+	/**
+     * 收到content分片或全部数据
+     * onRecvHeader函数返回>0,则为全部数据
+     * @param content
+     */
+	void onRecvContent(const string &content) override;
 private:
+	void inputRtspOrRtcp(const string &str);
 	int send(const string &strBuf) override {
         m_ui64TotalBytes += strBuf.size();
 		return m_pSender->send(strBuf);
@@ -153,6 +174,7 @@ private:
     static void onAuthBasic(const weak_ptr<RtspSession> &weakSelf,const string &realm,const string &strBase64);
     static void onAuthDigest(const weak_ptr<RtspSession> &weakSelf,const string &realm,const string &strMd5);
 
+private:
 	char *m_pcBuf = nullptr;
 	Ticker m_ticker;
 	Parser m_parser; //rtsp解析类