Merge pull request #2 from libxengine/develop

V1.3版本分支发布合并

Author: xengine-qyt

README.en.md

@@ -21,7 +21,8 @@ this software support following features
 8. message get time  
 9. message subcribe  
 10. support Permission Validation(planning)  
-11. Active delivery acquisition mode or passive subscription notification module  
+11. support multi protocol(TCP,HTTP,WEBSOCKET)  
+12. Active delivery acquisition mode or passive subscription notification module  
 
 ## install
 

README.md

@@ -20,7 +20,8 @@ c c++Message Queue Service
 8. 消息次数  
 9. 消息订阅  
 10. 支持权限验证(planning)  
-11. 主动投递获取模式或者被动订阅通知模块  
+11. 支持多种协议(TCP,HTTP,WEBSOCKET)  
+12. 主动投递获取模式或者被动订阅通知模块  
 
 ## 安装教程
 

XEngine_Docment/Docment_en.docx

@@ -4,19 +4,22 @@
   "bDeamon": 0,
   "nTCPPort": 5200,
   "nHttpPort": 5201,
+  "nWSPort": 5202,
   "XMax": {
     "nMaxClient": 10000,
     "nMaxQueue": 10000,
     "nIOThread": 2,
     "nTCPThread": 2,
-    "nHttpThread": 2
+    "nHttpThread": 2,
+    "nWSThread": 2
   },
   "XTime": {
     "bHBTime": 0,
     "nDBMonth": 3,
     "nTimeCheck": 3,
     "nTCPTimeOut": 5,
-    "nHttpTimeOut": 5
+    "nHttpTimeOut": 5,
+    "nWSTimeOut": 5
   },
   "XLog": {
     "MaxSize": 1024000,

XEngine_Docment/Docment_zh.docx

@@ -20,13 +20,15 @@ typedef struct tag_XEngine_ServerConfig
     BOOL bDeamon;
     int nTCPPort;
     int nHttpPort;
+    int nWSPort;
     struct
     {
         int nMaxClient;
         int nMaxQueue;
         int nIOThread;
         int nTCPThread;
         int nHttpThread;
+        int nWSThread;
     }st_XMax;
     struct
     {
@@ -35,6 +37,7 @@ typedef struct tag_XEngine_ServerConfig
         int nTimeCheck;
         int nTCPTimeOut;
         int nHttpTimeOut;
+        int nWSTimeOut;
     }st_XTime;
     struct
     {

XEngine_Release/XEngine_Config/XEngine_Config.json

@@ -68,8 +68,9 @@ BOOL CConfig_Json::Config_Json_File(LPCTSTR lpszConfigFile,XENGINE_SERVERCONFIG
     pSt_ServerConfig->bDeamon = st_JsonRoot["bDeamon"].asInt();
     pSt_ServerConfig->nTCPPort = st_JsonRoot["nTCPPort"].asInt();
     pSt_ServerConfig->nHttpPort = st_JsonRoot["nHttpPort"].asInt();
+    pSt_ServerConfig->nWSPort = st_JsonRoot["nWSPort"].asInt();
 
-    if (st_JsonRoot["XMax"].empty() || (5 != st_JsonRoot["XMax"].size()))
+    if (st_JsonRoot["XMax"].empty() || (6 != st_JsonRoot["XMax"].size()))
     {
         Config_IsErrorOccur = TRUE;
         Config_dwErrorCode = ERROR_MQ_MODULE_CONFIG_JSON_XMAX;
@@ -81,8 +82,9 @@ BOOL CConfig_Json::Config_Json_File(LPCTSTR lpszConfigFile,XENGINE_SERVERCONFIG
     pSt_ServerConfig->st_XMax.nIOThread = st_JsonXMax["nIOThread"].asInt();
     pSt_ServerConfig->st_XMax.nTCPThread = st_JsonXMax["nTCPThread"].asInt();
     pSt_ServerConfig->st_XMax.nHttpThread = st_JsonXMax["nHttpThread"].asInt();
+    pSt_ServerConfig->st_XMax.nWSThread = st_JsonXMax["nWSThread"].asInt();
 
-    if (st_JsonRoot["XTime"].empty() || (5 != st_JsonRoot["XTime"].size()))
+    if (st_JsonRoot["XTime"].empty() || (6 != st_JsonRoot["XTime"].size()))
     {
         Config_IsErrorOccur = TRUE;
         Config_dwErrorCode = ERROR_MQ_MODULE_CONFIG_JSON_XTIME;
@@ -94,6 +96,7 @@ BOOL CConfig_Json::Config_Json_File(LPCTSTR lpszConfigFile,XENGINE_SERVERCONFIG
     pSt_ServerConfig->st_XTime.nTimeCheck = st_JsonXTime["nTimeCheck"].asInt();
     pSt_ServerConfig->st_XTime.nTCPTimeOut = st_JsonXTime["nTCPTimeOut"].asInt();
     pSt_ServerConfig->st_XTime.nHttpTimeOut = st_JsonXTime["nHttpTimeOut"].asInt();
+    pSt_ServerConfig->st_XTime.nWSTimeOut = st_JsonXTime["nWSTimeOut"].asInt();
 
     if (st_JsonRoot["XLog"].empty() || (3 != st_JsonRoot["XLog"].size()))
     {

XEngine_Source/MQCore_ConfigModule/Config_Define.h

@@ -126,6 +126,7 @@ BOOL CProtocolModule_Packet::ProtocolModule_Packet_HttpCommon(XENGINE_PROTOCOLHD
     Json::Value st_JsonRoot;
     Json::Value st_JsonMQProtocol;
     Json::Value st_JsonPayload;
+	Json::StreamWriterBuilder st_JsonBuilder;
 
     st_JsonMQProtocol["tszMQKey"] = pSt_MQProtocol->tszMQKey;
     st_JsonMQProtocol["nSerial"] = pSt_MQProtocol->nSerial;
@@ -145,9 +146,10 @@ BOOL CProtocolModule_Packet::ProtocolModule_Packet_HttpCommon(XENGINE_PROTOCOLHD
 
         st_JsonRoot["st_Payload"] = st_JsonPayload;
     }
+    st_JsonBuilder["emitUTF8"] = true;
 
-    *pInt_MsgLen = st_JsonRoot.toStyledString().length();
-    memcpy(ptszMsgBuffer, st_JsonRoot.toStyledString().c_str(), *pInt_MsgLen);
+    *pInt_MsgLen = Json::writeString(st_JsonBuilder, st_JsonRoot).length();
+    memcpy(ptszMsgBuffer, Json::writeString(st_JsonBuilder, st_JsonRoot).c_str(), *pInt_MsgLen);
 
     return TRUE;
 }

XEngine_Source/MQCore_ConfigModule/Config_Json/Config_Json.cpp

@@ -28,6 +28,8 @@ using namespace std;
 #include <XEngine_Include/XEngine_HelpComponents/Packets_Error.h>
 #include <XEngine_Include/XEngine_RfcComponents/HttpServer_Define.h>
 #include <XEngine_Include/XEngine_RfcComponents/HttpServer_Error.h>
+#include <XEngine_Include/XEngine_RfcComponents/WSProtocol_Define.h>
+#include <XEngine_Include/XEngine_RfcComponents/WSProtocol_Error.h>
 #include "../XQueue_ProtocolHdr.h"
 #include "../MQCore_ConfigModule/Config_Define.h"
 #include "../MQCore_ConfigModule/Config_Error.h"
@@ -40,16 +42,24 @@ using namespace std;
 
 #define XENGINE_MQAPP_NETTYPE_TCP 1
 #define XENGINE_MQAPP_NETTYPE_HTTP 2
+#define XENGINE_MQAPP_NETTYPE_WEBSOCKET 3
 
 extern BOOL bIsRun;
 extern XLOG xhLog;
 extern XNETHANDLE xhTCPSocket;
-extern XNETHANDLE xhTCPPacket;
-extern XNETHANDLE xhTCPHeart;
 extern XNETHANDLE xhHTTPSocket;
+extern XNETHANDLE xhWSSocket;
+
+extern XNETHANDLE xhTCPHeart;
+extern XNETHANDLE xhWSHeart;
+
+extern XNETHANDLE xhTCPPacket;
 extern XHANDLE xhHTTPPacket;
+
 extern XNETHANDLE xhTCPPool;
 extern XNETHANDLE xhHttpPool;
+extern XNETHANDLE xhWSPool;
+
 extern XENGINE_SERVERCONFIG st_ServiceCfg;
 
 void ServiceApp_Stop(int signo);
@@ -58,6 +68,7 @@ void ServiceApp_Stop(int signo);
 #include "MQService_Net.h"
 #include "MQService_TCPTask.h"
 #include "MQService_HttpTask.h"
+#include "MQService_WSTask.h"
 
 #ifdef _WINDOWS
 #ifdef _DEBUG
@@ -71,6 +82,7 @@ void ServiceApp_Stop(int signo);
 #pragma comment(lib,"x86/XEngine_HelpComponents/HelpComponents_XLog.lib")
 #pragma comment(lib,"x86/XEngine_HelpComponents/HelpComponents_Packets.lib")
 #pragma comment(lib,"x86/XEngine_RfcComponents/RfcComponents_HttpServer.lib")
+#pragma comment(lib,"x86/XEngine_RfcComponents/RfcComponents_WSProtocol.lib")
 #else
 #ifdef WIN64
 #pragma comment(lib,"../x64/Release/MQCore_ConfigModule.lib")
@@ -83,6 +95,7 @@ void ServiceApp_Stop(int signo);
 #pragma comment(lib,"x64/XEngine_HelpComponents/HelpComponents_XLog.lib")
 #pragma comment(lib,"x64/XEngine_HelpComponents/HelpComponents_Packets.lib")
 #pragma comment(lib,"x64/XEngine_RfcComponents/RfcComponents_HttpServer.lib")
+#pragma comment(lib,"x64/XEngine_RfcComponents/RfcComponents_WSProtocol.lib")
 #else
 #pragma comment(lib,"../Release/MQCore_ConfigModule.lib")
 #pragma comment(lib,"../Release/MQCore_ProtocolModule.lib")
@@ -94,6 +107,7 @@ void ServiceApp_Stop(int signo);
 #pragma comment(lib,"x86/XEngine_HelpComponents/HelpComponents_XLog.lib")
 #pragma comment(lib,"x86/XEngine_HelpComponents/HelpComponents_Packets.lib")
 #pragma comment(lib,"x86/XEngine_RfcComponents/RfcComponents_HttpServer.lib")
+#pragma comment(lib,"x86/XEngine_RfcComponents/RfcComponents_WSProtocol.lib")
 #endif
 #endif
 #pragma comment(lib,"Ws2_32.lib")

XEngine_Source/MQCore_ProtocolModule/ProtocolModule_Packet/ProtocolModule_Packet.cpp

@@ -45,6 +45,48 @@ void __stdcall MessageQueue_Callback_HttpLeave(LPCTSTR lpszClientAddr, SOCKET hS
     XEngine_MQXService_Close(lpszClientAddr, XENGINE_MQAPP_NETTYPE_HTTP, FALSE);
 }
 //////////////////////////////////////////////////////////////////////////
+BOOL __stdcall MessageQueue_Callback_WSLogin(LPCTSTR lpszClientAddr, SOCKET hSocket, LPVOID lParam)
+{
+    SessionModule_Client_Create(lpszClientAddr);
+    SocketOpt_HeartBeat_InsertAddrEx(xhWSHeart, lpszClientAddr);
+	RfcComponents_WSPacket_Create(lpszClientAddr, 0);
+	XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _T("Websocket客户端连接，Websocket客户端地址：%s"), lpszClientAddr);
+	return TRUE;
+}
+void __stdcall MessageQueue_Callback_WSRecv(LPCTSTR lpszClientAddr, SOCKET hSocket, LPCTSTR lpszRecvMsg, int nMsgLen, LPVOID lParam)
+{
+	BOOL bLogin = FALSE;
+	RfcComponents_WSPacket_GetLogin(lpszClientAddr, &bLogin);
+	if (bLogin)
+	{
+		if (!RfcComponents_WSPacket_Post(lpszClientAddr, lpszRecvMsg, nMsgLen))
+		{
+            XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _T("投递Websocket数据包到消息队列失败，错误：%lX"), WSFrame_GetLastError());
+            return;
+		}
+        SocketOpt_HeartBeat_ActiveAddrEx(xhWSHeart, lpszClientAddr);
+	}
+	else
+	{
+		int nSDLen = nMsgLen;
+		TCHAR tszHandsBuffer[1024];
+		memset(tszHandsBuffer, '\0', sizeof(tszHandsBuffer));
+
+		RfcComponents_WSConnector_HandShake(lpszRecvMsg, &nSDLen, tszHandsBuffer);
+		RfcComponents_WSPacket_SetLogin(lpszClientAddr);
+        NetCore_TCPXCore_SendEx(xhWSSocket, lpszClientAddr, tszHandsBuffer, nSDLen);
+        XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _T("WEBSOCKET客户端:%s 与服务器握手成功"), lpszClientAddr);
+	}
+}
+void __stdcall MessageQueue_Callback_WSLeave(LPCTSTR lpszClientAddr, SOCKET hSocket, LPVOID lParam)
+{
+    XEngine_MQXService_Close(lpszClientAddr, XENGINE_MQAPP_NETTYPE_WEBSOCKET, FALSE);
+}
+void __stdcall MessageQueue_Callback_WSHeart(LPCSTR lpszClientAddr, SOCKET hSocket, int nStatus, LPVOID lParam)
+{
+    XEngine_MQXService_Close(lpszClientAddr, XENGINE_MQAPP_NETTYPE_WEBSOCKET, TRUE);
+}
+//////////////////////////////////////////////////////////////////////////
 void XEngine_MQXService_Close(LPCTSTR lpszClientAddr, int nIPProto, BOOL bHeart)
 {
     if (XENGINE_MQAPP_NETTYPE_TCP == nIPProto)
@@ -62,6 +104,21 @@ void XEngine_MQXService_Close(LPCTSTR lpszClientAddr, int nIPProto, BOOL bHeart)
         }
         XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _T("TCP客户端离开，TCP客户端地址：%s"), lpszClientAddr);
     }
+    else if (XENGINE_MQAPP_NETTYPE_WEBSOCKET == nIPProto)
+    {
+		RfcComponents_WSPacket_Delete(lpszClientAddr);
+        SessionModule_Client_Delete(lpszClientAddr);
+
+		if (bHeart)
+		{
+			NetCore_TCPXCore_CloseForClientEx(xhWSSocket, lpszClientAddr);
+		}
+		else
+		{
+            SocketOpt_HeartBeat_DeleteAddrEx(xhWSHeart, lpszClientAddr);
+		}
+		XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _T("Websocket客户端离开，Websocket客户端地址：%s"), lpszClientAddr);
+    }
     else
 	{
 		RfcComponents_HttpServer_CloseClinetEx(xhHTTPPacket, lpszClientAddr);
@@ -100,5 +157,18 @@ BOOL XEngine_MQXService_Send(LPCTSTR lpszClientAddr, LPCTSTR lpszMsgBuffer, int
 			return FALSE;
 		}
     }
+    else if (XENGINE_MQAPP_NETTYPE_WEBSOCKET == nIPProto)
+    {
+		TCHAR tszPKTBuffer[8196];
+		memset(tszPKTBuffer, '\0', sizeof(tszPKTBuffer));
+
+        RfcComponents_WSCodec_EncodeMsg(lpszMsgBuffer, tszPKTBuffer, &nMsgLen, ENUM_XENGINE_RFCOMPONENTS_WEBSOCKET_OPCODE_TEXT);
+		if (!NetCore_TCPXCore_SendEx(xhWSSocket, lpszClientAddr, tszPKTBuffer, nMsgLen))
+		{
+			XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _T("发送数据给Websocket客户端：%s，失败，错误：%lX"), lpszClientAddr, NetCore_GetLastError());
+			return FALSE;
+		}
+        SocketOpt_HeartBeat_ActiveAddrEx(xhWSHeart, lpszClientAddr);
+    }
     return TRUE;
 }