Http服务改成异步回复模式

src/Http/HttpSession.cpp

@@ -106,14 +106,14 @@ void HttpSession::onRecv(const Socket::Buffer::Ptr&pBuf) {
 		onePkt = m_strRcvBuf.substr(0, index + 4);
 		m_strRcvBuf.erase(0, index + 4);
 		switch (parserHttpReq(onePkt)) {
-		case 0:
+		case Http_failed:
 			//失败
 			shutdown();
 			return;
-		case 1:
+		case Http_success:
 			//成功
 			break;
-		case 2:
+		case Http_moreData:
 			//需要更多数据,恢复数据并退出
 			m_strRcvBuf = onePkt + m_strRcvBuf;
 			m_parser.Clear();
@@ -122,14 +122,14 @@ void HttpSession::onRecv(const Socket::Buffer::Ptr&pBuf) {
 	}
 	m_parser.Clear();
 }
-inline int HttpSession::parserHttpReq(const string &str) {
+inline HttpSession::HttpCode HttpSession::parserHttpReq(const string &str) {
 	m_parser.Parse(str.data());
 	string cmd = m_parser.Method();
 	auto it = g_mapCmdIndex.find(cmd);
 	if (it == g_mapCmdIndex.end()) {
 		WarnL << cmd;
 		sendResponse("403 Forbidden", makeHttpHeader(true), "");
-		return false;
+		return Http_failed;
 	}
 	auto fun = it->second;
 	return (this->*fun)();
@@ -147,34 +147,35 @@ void HttpSession::onManager() {
 	}
 }
 
-inline int HttpSession::Handle_Req_GET() {
+inline HttpSession::HttpCode HttpSession::Handle_Req_GET() {
 	string strUrl = strCoding::UrlUTF8Decode(m_parser.Url());
 	string strFile = m_strPath + strUrl;
 	string strConType = m_parser["Connection"];
 	static uint32_t reqCnt =  mINI::Instance()[Config::Http::kMaxReqCount].as<uint32_t>();
 	bool bClose = (strcasecmp(strConType.data(),"close") == 0) && ( ++m_iReqCnt < reqCnt);
+	HttpCode eHttpCode = bClose ? Http_failed : Http_success;
 	if (strFile.back() == '/') {
 		//index the folder
 		string strMeun;
 		if (!makeMeun(strFile, strMeun)) {
 			sendNotFound(bClose);
-			return !bClose;
+			return eHttpCode;
 		}
 		sendResponse("200 OK", makeHttpHeader(bClose,strMeun.size() ), strMeun);
-		return !bClose;
+		return eHttpCode;
 	}
 	//download the file
 	struct stat tFileStat;
 	if (0 != stat(strFile.data(), &tFileStat)) {
 		sendNotFound(bClose);
-		return !bClose;
+		return eHttpCode;
 	}
 
 	TimeTicker();
 	FILE *pFile = fopen(strFile.data(), "rb");
 	if (pFile == NULL) {
 		sendNotFound(bClose);
-		return !bClose;
+		return eHttpCode;
 	}
 
 	auto &strRange = m_parser["Range"];
@@ -200,7 +201,7 @@ inline int HttpSession::Handle_Req_GET() {
 	sendResponse(pcHttpResult, httpHeader, "");
 	if (iRangeEnd - iRangeStart < 0) {
 		//file is empty!
-		return !bClose;
+		return eHttpCode;
 	}
 
 	//send the file
@@ -251,7 +252,7 @@ inline int HttpSession::Handle_Req_GET() {
 	};
 	onFlush();
 	sock->setOnFlush(onFlush);
-	return true;
+	return Http_success;
 }
 
 inline bool HttpSession::makeMeun(const string &strFullPath, string &strRet) {
@@ -365,45 +366,63 @@ inline HttpSession::KeyValue HttpSession::makeHttpHeader(bool bClose, int64_t iC
 	}
 	return headerOut;
 }
-inline int HttpSession::Handle_Req_POST() {
+inline HttpSession::HttpCode HttpSession::Handle_Req_POST() {
 	int iContentLen = atoi(m_parser["Content-Length"].data());
 	if (!iContentLen) {
-		return false;
+		return Http_failed;
 	}
 	if ((int) m_strRcvBuf.size() < iContentLen) {
-		return 2; //需要更多数据
+		return Http_moreData; //需要更多数据
 	}
 	auto strContent = m_strRcvBuf.substr(0, iContentLen);
 	m_strRcvBuf.erase(0, iContentLen);
 
 	string strUrl = strCoding::UrlUTF8Decode(m_parser.Url());
 	string strConType = m_parser["Connection"];
-	static string charSet = mINI::Instance()[Config::Http::kCharSet];
 	static uint32_t reqCnt = mINI::Instance()[Config::Http::kMaxReqCount].as<uint32_t>();
-
 	bool bClose = (strcasecmp(strConType.data(),"close") == 0) && ( ++m_iReqCnt < reqCnt);
 	m_parser.setUrl(strUrl);
 	m_parser.setContent(strContent);
-	auto headerOut=makeHttpHeader(bClose);
 
-	static string notFound =  mINI::Instance()[Config::Http::kNotFound];
-	string strContentOut = notFound;
-	string strCodeOut = "404 Not Found";
-	NoticeCenter::Instance().emitEvent(
-			Config::Broadcast::kBroadcastHttpRequest,
-			m_parser,strCodeOut,headerOut,strContentOut);
-
-	if(strContentOut.size()){
-		headerOut.emplace("Content-Type",StrPrinter<<"text/json; charset=" << charSet <<endl);
-		headerOut.emplace("Content-Length",StrPrinter<<strContentOut.size()<<endl);
+	weak_ptr<HttpSession> weakSelf = dynamic_pointer_cast<HttpSession>(shared_from_this());
+	HttpResponseInvoker invoker = [weakSelf,bClose](const string &codeOut,
+													const KeyValue &headerOut,
+													const string &contentOut){
+		auto strongSelf = weakSelf.lock();
+		if(!strongSelf) {
+			return;
 		}
-	sendResponse(strCodeOut.data(), headerOut, strContentOut);
-	return !bClose;
+		strongSelf->async([weakSelf,bClose,codeOut,headerOut,contentOut]() {
+			auto strongSelf = weakSelf.lock();
+			if(!strongSelf) {
+				return;
+			}
+			strongSelf->responseDelay(bClose,
+					const_cast<string &>(codeOut),
+					const_cast<KeyValue &>(headerOut),
+					const_cast<string &>(contentOut));
+			if(bClose){
+				strongSelf->shutdown();
+			}
+		});
+	};
+	if(!NoticeCenter::Instance().emitEvent(Config::Broadcast::kBroadcastHttpRequest,m_parser,invoker)){
+		invoker("404 Not Found",KeyValue(),"");
+	}
+	return Http_success;
+}
+void HttpSession::responseDelay(bool bClose,string &codeOut,KeyValue &headerOut, string &contentOut){
+	if(codeOut.empty()){
+		sendNotFound(bClose);
+		return;
+	}
+	auto headerOther=makeHttpHeader(bClose,contentOut.size(),"text/json");
+	headerOut.insert(headerOther.begin(), headerOther.end());
+	sendResponse(codeOut.data(), headerOut, contentOut);
 }
-
 inline void HttpSession::sendNotFound(bool bClose) {
 	static string notFound =  mINI::Instance()[Config::Http::kNotFound];
-	sendResponse("404 Not Found", makeHttpHeader(bClose, notFound.size()), notFound.data());
+	sendResponse("404 Not Found", makeHttpHeader(bClose, notFound.size()), notFound);
 }
 
 } /* namespace Http */

src/Http/HttpSession.h

@@ -11,16 +11,23 @@
 #include "config.h"
 #include "Rtsp/Rtsp.h"
 #include "Network/TcpLimitedSession.h"
+#include <functional>
 
 using namespace std;
 using namespace ZL::Network;
+using namespace ZL::Network;
 
 namespace ZL {
 namespace Http {
 
+
 class HttpSession: public TcpLimitedSession<MAX_TCP_SESSION> {
 public:
 	typedef map<string,string> KeyValue;
+	typedef std::function<void(const string &codeOut,
+							   const KeyValue &headerOut,
+							   const string &contentOut)>  HttpResponseInvoker;
+
 	HttpSession(const std::shared_ptr<ThreadPool> &pTh, const Socket::Ptr &pSock);
 	virtual ~HttpSession();
 
@@ -28,21 +35,30 @@ public:
 	void onError(const SockException &err) override;
 	void onManager() override;
 private:
-	typedef int (HttpSession::*HttpCMDHandle)();
+	typedef enum
+	{
+		Http_success = 0,
+		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;
 	string m_strRcvBuf;
 	Ticker m_ticker;
 	uint32_t m_iReqCnt = 0;
 
-	inline int parserHttpReq(const string &);
-	inline int Handle_Req_GET();
-	inline int Handle_Req_POST();
+
+	inline HttpCode parserHttpReq(const string &);
+	inline HttpCode Handle_Req_GET();
+	inline HttpCode Handle_Req_POST();
 	inline bool makeMeun(const string &strFullPath, string &strRet);
 	inline void sendNotFound(bool bClose);
 	inline void sendResponse(const char *pcStatus,const KeyValue &header,const string &strContent);
-	inline KeyValue makeHttpHeader(bool bClose=false,int64_t iContentSize=-1,const char *pcContentType="text/html");
+	inline static KeyValue makeHttpHeader(bool bClose=false,int64_t iContentSize=-1,const char *pcContentType="text/html");
+	void responseDelay(bool bClose,string &codeOut,KeyValue &headerOut, string &contentOut);
 };
 
 } /* namespace Http */

src/MediaSender.h

@@ -13,7 +13,7 @@ using namespace ZL::Thread;
 class MediaSender {
 public:
 	static ThreadPool & sendThread() {
-		static ThreadPool pool(1, ThreadPool::PRIORITY_HIGHEST);
+		static ThreadPool pool(1);
 		return pool;
 	}
 private:

src/config.h

@@ -38,10 +38,20 @@ namespace Config {
 ////////////广播名称///////////
 namespace Broadcast {
 extern const char kBroadcastRtspSessionPlay[];
+#define BroadcastRtspSessionPlayArgs const char *app,const char *stream
+
 extern const char kBroadcastRtspSrcRegisted[];
+#define BroadcastRtspSrcRegistedArgs const char *app,const char *stream
+
 extern const char kBroadcastRtmpSrcRegisted[];
+#define BroadcastRtmpSrcRegistedArgs const char *app,const char *stream
+
 extern const char kBroadcastRecordMP4[];
+#define BroadcastRecordMP4Args const Mp4Info &info
+
 extern const char kBroadcastHttpRequest[];
+#define BroadcastHttpRequestArgs const Parser &parser,HttpSession::HttpResponseInvoker &invoker
+
 } //namespace Broadcast
 
 //代理失败最大重试次数