added:heart support for mqtt

Author: xengine-qyt

XEngine_Release/XEngine_Config/XEngine_Config.json

@@ -18,7 +18,8 @@
         "nHeartCheck": 3,
         "nTCPTime": 5,
         "nWSTime": 5,
-        "nHTTPTime": 5
+        "nHTTPTime": 5,
+        "nMQTTTime": 5
     },
     "XVerification": {
         "nTokenTimeout": 3600,

XEngine_Source/AuthorizeModule_Configure/Config_Define.h

@@ -38,6 +38,7 @@ typedef struct
 		int nHeartCheck;                                    //检测次数
 		int nTCPTime;                                       //TCP检测时间
 		int nWSTime;                                        //WEBSOCKET检测时间
+		int nMQTime;                                        //MQTT检测时间
 		int nHTTPTime;                                      //HTTP检测时间
 	}st_XTime;
 	struct   

XEngine_Source/AuthorizeModule_Configure/ModuleConfigure_Json/ModuleConfigure_Json.cpp

@@ -94,7 +94,7 @@ bool CModuleConfigure_Json::ModuleConfigure_Json_File(LPCXSTR lpszConfigFile, XE
 	pSt_ServerConfig->st_XMax.nHTTPThread = st_JsonXMax["nHTTPThread"].asInt();
 	pSt_ServerConfig->st_XMax.nMQTTThread = st_JsonXMax["nMQTTThread"].asInt();
 	//时间配置
-	if (st_JsonRoot["XTime"].empty() || (4 != st_JsonRoot["XTime"].size()))
+	if (st_JsonRoot["XTime"].empty() || (5 != st_JsonRoot["XTime"].size()))
 	{
 		Config_IsErrorOccur = true;
 		Config_dwErrorCode = ERROR_AUTHORIZE_MODULE_CONFIGURE_XTIME;
@@ -105,6 +105,7 @@ bool CModuleConfigure_Json::ModuleConfigure_Json_File(LPCXSTR lpszConfigFile, XE
 	pSt_ServerConfig->st_XTime.nTCPTime = st_JsonXTime["nTCPTime"].asInt();
 	pSt_ServerConfig->st_XTime.nWSTime = st_JsonXTime["nWSTime"].asInt();
 	pSt_ServerConfig->st_XTime.nHTTPTime = st_JsonXTime["nHTTPTime"].asInt();
+	pSt_ServerConfig->st_XTime.nMQTime = st_JsonXTime["nMQTTTime"].asInt();
 	//验证配置
 	if (st_JsonRoot["XVerification"].empty() || (8 != st_JsonRoot["XVerification"].size()))
 	{

XEngine_Source/XEngine_APPService/XEngine_AuthorizeService/Authorize_Hdr.h

@@ -79,6 +79,7 @@ extern XHANDLE xhHttpPacket;
 extern XHANDLE xhTCPHeart;
 extern XHANDLE xhWSHeart;
 extern XHANDLE xhHTTPHeart;
+extern XHANDLE xhMQTTHeart;
 
 extern XHANDLE xhMemPool;
 extern XHANDLE xhTCPPool;

XEngine_Source/XEngine_APPService/XEngine_AuthorizeService/Authorize_Net.cpp

@@ -101,12 +101,17 @@ void XCALLBACK XEngine_Client_MQTTRecv(LPCXSTR lpszClientAddr, XSOCKET hSocket,
 		XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _X("MQTT客户端:%s,投递MQTT数据包到消息队列失败，错误：%lX"), lpszClientAddr, MQTTProtocol_GetLastError());
 		return;
 	}
+	SocketOpt_HeartBeat_ActiveAddrEx(xhMQTTSocket, lpszClientAddr);
 	XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_DEBUG, _X("MQTT客户端:%s,投递MQTT数据包到消息队列成功,%d"), lpszClientAddr, nMsgLen);
 }
 void XCALLBACK XEngine_Client_MQTTLeave(LPCXSTR lpszClientAddr, XSOCKET hSocket, XPVOID lParam)
 {
 	XEngine_CloseClient(lpszClientAddr, 0);
 }
+void XCALLBACK XEngine_Client_MQTTHeart(LPCXSTR lpszClientAddr, XSOCKET hSocket, int nStatus, XPVOID lParam)
+{
+	XEngine_CloseClient(lpszClientAddr, 2);
+}
 //////////////////////////////////////////////////////////////////////////
 bool XEngine_CloseClient(LPCXSTR lpszClientAddr, int nLeaveType)
 {
@@ -207,6 +212,7 @@ bool XEngine_Client_TaskSend(LPCXSTR lpszClientAddr, LPCXSTR lpszMsgBuffer, int
 			XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _X("发送数据给MQTT客户端：%s，失败，错误：%lX"), lpszClientAddr, NetCore_GetLastError());
 			return false;
 		}
+		SocketOpt_HeartBeat_ActiveAddrEx(xhMQTTSocket, lpszClientAddr);
 	}
 	else
 	{

XEngine_Source/XEngine_APPService/XEngine_AuthorizeService/Authorize_Net.h

@@ -18,6 +18,7 @@ void XCALLBACK XEngine_Client_HttpHeart(LPCXSTR lpszClientAddr, XSOCKET hSocket,
 bool XCALLBACK XEngine_Client_MQTTLogin(LPCXSTR lpszClientAddr, XSOCKET hSocket, XPVOID lParam);
 void XCALLBACK XEngine_Client_MQTTRecv(LPCXSTR lpszClientAddr, XSOCKET hSocket, LPCXSTR lpszRecvMsg, int nMsgLen, XPVOID lParam);
 void XCALLBACK XEngine_Client_MQTTLeave(LPCXSTR lpszClientAddr, XSOCKET hSocket, XPVOID lParam);
+void XCALLBACK XEngine_Client_MQTTHeart(LPCXSTR lpszClientAddr, XSOCKET hSocket, int nStatus, XPVOID lParam);
 
 bool XEngine_CloseClient(LPCXSTR lpszClientAddr, int nLeaveType = 0);
 bool XEngine_Client_TaskSend(LPCXSTR lpszClientAddr, LPCXSTR lpszMsgBuffer, int nMsgLen, int nNetType);

XEngine_Source/XEngine_APPService/XEngine_AuthorizeService/XEngine_AuthorizeService.cpp

@@ -16,6 +16,7 @@ XHANDLE xhHttpPacket = NULL;
 XHANDLE xhTCPHeart = NULL;
 XHANDLE xhWSHeart = NULL;
 XHANDLE xhHTTPHeart = NULL;
+XHANDLE xhMQTTHeart = NULL;
 
 XHANDLE xhMemPool = NULL;
 XHANDLE xhTCPPool = NULL;
@@ -46,6 +47,7 @@ void ServiceApp_Stop(int signo)
 		SocketOpt_HeartBeat_DestoryEx(xhTCPHeart);
 		SocketOpt_HeartBeat_DestoryEx(xhWSHeart);
 		SocketOpt_HeartBeat_DestoryEx(xhHTTPHeart);
+		SocketOpt_HeartBeat_DestoryEx(xhMQTTHeart);
 
 		ManagePool_Thread_NQDestroy(xhTCPPool);
 		ManagePool_Thread_NQDestroy(xhWSPool);
@@ -398,6 +400,21 @@ int main(int argc, char** argv)
 		NetCore_TCPXCore_RegisterCallBackEx(xhMQTTSocket, XEngine_Client_MQTTLogin, XEngine_Client_MQTTRecv, XEngine_Client_MQTTLeave);
 		XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("启动服务中，注册MQTT网络事件成功"));
 
+		if (st_AuthConfig.st_XTime.nMQTime > 0)
+		{
+			xhMQTTHeart = SocketOpt_HeartBeat_InitEx(st_AuthConfig.st_XTime.nHeartCheck, st_AuthConfig.st_XTime.nMQTime, XEngine_Client_MQTTHeart);
+			if (NULL == xhMQTTHeart)
+			{
+				XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _X("启动服务中，初始化MQTT心跳服务失败，错误：%lX"), NetCore_GetLastError());
+				goto XENGINE_EXITAPP;
+			}
+			XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("启动服务中，初始化MQTT心跳服务成功,检测次数:%d,检测时间:%d"), st_AuthConfig.st_XTime.nHeartCheck, st_AuthConfig.st_XTime.nMQTime);
+		}
+		else
+		{
+			XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_WARN, _X("启动服务中，检测到MQTT心跳服务没有启用"));
+		}
+
 		BaseLib_Memory_Malloc((XPPPMEM)&ppSt_ListMQTTParam, st_AuthConfig.st_XMax.nMQTTThread, sizeof(THREADPOOL_PARAMENT));
 		for (int i = 0; i < st_AuthConfig.st_XMax.nMQTTThread; i++)
 		{
@@ -506,6 +523,7 @@ int main(int argc, char** argv)
 		SocketOpt_HeartBeat_DestoryEx(xhTCPHeart);
 		SocketOpt_HeartBeat_DestoryEx(xhWSHeart);
 		SocketOpt_HeartBeat_DestoryEx(xhHTTPHeart);
+		SocketOpt_HeartBeat_DestoryEx(xhMQTTHeart);
 
 		ManagePool_Thread_NQDestroy(xhTCPPool);
 		ManagePool_Thread_NQDestroy(xhWSPool);