Merge pull request #17 from libxengine/develop

V3.9.0.1001 Merge

Author: xengine-qyt

CHANGELOG

@@ -1,3 +1,31 @@
+XEngine_MQService V3.9.0.1001
+
+增加:消息属性的通知所有和通知自身的支持
+增加:为指定用户发送消息
+增加:http管理接口获取在线用户列表
+优化:内存的消息通知处理方法
+优化:查询数据库的方法
+修正:当消息通知的时候客户端网络类型不正确的问题
+修正:无论用户是否在线都进行消息通知的问题
+修正:DBModule_MQUser_KeyList函数列举数据不正确的问题
+修正:未读协议内存溢出
+修正:未读协议序列号处理错误
+删除:获取消息次数支持
+删除:定时发布的break字段
+
+added:message attritube supported with notify all and notify self
+added:send a message for a specific user
+added:get online user list for http restful
+improved:memory and process way for msg notify
+improved:query sql point handle way
+fixed:client nettype is incorrent when message notify
+fixed:Doesn't handle whether the client is online or not when a message notification is made
+fixed:DBModule_MQUser_KeyList function list is incorrent
+fixed:unread protocol memory leak
+fixed:unread protocol serial handle is incorrent
+delete:get time number supported
+delete:time publish break field
+======================================================================================
 XEngine_MQService V3.8.0.1001
 
 增加:websocket的二进制消息负载能力支持

README.en.md

@@ -32,6 +32,7 @@ this software support following features
 16. Access control(planning)
 17. binary message
 18. http interface management
+19. message attritube
 
 ## install
 

README.md

@@ -31,6 +31,7 @@ c c++Message Service
 16. 权限控制(planning)
 17. 二进制消息
 18. http接口管理
+19. 消息属性
 
 ## 安装教程
 
@@ -118,9 +119,8 @@ make FLAGS=CleanAll 清理编译
 扩展获取消息的内容  
 数据库缓存功能  
 用户消息  
-消息属性生效  
+完善消息属性生效  
 公众号和小程序  
-消息强推送(不限订阅,强制推送)  
 QOS(支持强回复确认)  
 功能开关  
 MQTT支持  
@@ -129,6 +129,9 @@ MQTT支持
 完善HTTP_CALL  
 高性能内存队列  
 扩充HTTP管理接口  
+消息变量,支持消息变量设置  
+序列号重置与优化  
+完善公用KEY的使用
 
 ## 提交问题
 

XEngine_Apps/MQCore_HTTPApp/MQCore_HTTPApp.cpp

@@ -20,7 +20,7 @@
 #include <XEngine_Include/XEngine_NetHelp/APIClient_Error.h>
 #include "../../XEngine_Source/XQueue_ProtocolHdr.h"
 
-//g++ -std=c++17 -Wall -g MQCore_HTTPApp.cpp -o MQCore_HTTPApp.exe -I ../../XEngine_Source/XEngine_ThirdPart/jsoncpp -L /usr/local/lib/XEngine_Release/XEngine_BaseLib -L /usr/local/lib/XEngine_Release/XEngine_NetHelp -L ../../XEngine_Source/XEngine_ThirdPart/jsoncpp -lXEngine_BaseLib -lNetHelp_APIClient -ljsoncpp -Wl,-rpath=../../XEngine_Source/XEngine_ThirdPart/jsoncpp,--disable-new-dtags
+//g++ -std=c++17 -Wall -g MQCore_HTTPApp.cpp -o MQCore_HTTPApp.exe -I ../../XEngine_Source/XEngine_ThirdPart/jsoncpp -L /usr/local/lib/XEngine_Release/XEngine_BaseLib -L /usr/local/lib/XEngine_Release/XEngine_NetHelp -L ../../XEngine_Source/XEngine_ThirdPart/jsoncpp -lXEngine_BaseLib -lNetHelp_APIClient -ljsoncpp
 void MQ_GetUserList()
 {
 	LPCXSTR lpszPostUrl = _X("http://127.0.0.1:5202/api?function=get&method=user");
@@ -35,6 +35,20 @@ void MQ_GetUserList()
 	_xtprintf("MQ_GetUserList:%s\n", ptszMsgBody);
 	BaseLib_OperatorMemory_FreeCStyle((XPPMEM)&ptszMsgBody);
 }
+void MQ_GetOnlineList()
+{
+	LPCXSTR lpszPostUrl = _X("http://127.0.0.1:5202/api?function=get&method=online&type=0");
+
+	int nLen = 0;
+	XCHAR* ptszMsgBody = NULL;
+	if (!APIClient_Http_Request(_X("GET"), lpszPostUrl, NULL, NULL, &ptszMsgBody, &nLen))
+	{
+		_xtprintf("发送投递失败！\n");
+		return;
+	}
+	_xtprintf("MQ_GetOnlineList:%s\n", ptszMsgBody);
+	BaseLib_OperatorMemory_FreeCStyle((XPPMEM)&ptszMsgBody);
+}
 void MQ_GetTopicList()
 {
 	LPCXSTR lpszPostUrl = _X("http://127.0.0.1:5202/api?function=get&method=topic");
@@ -58,6 +72,7 @@ int main()
 #endif
 
 	MQ_GetUserList();
+	MQ_GetOnlineList();
 	MQ_GetTopicList();
 
 #ifdef _MSC_BUILD

XEngine_Apps/MQCore_TCPApp/MQCore_TCPApp.cpp

@@ -121,6 +121,58 @@ void MQ_Authorize()
 		BaseLib_OperatorMemory_FreeCStyle((XPPMEM)&ptszMsgBuffer);
 	}
 }
+//获取未读消息
+void MQ_GetUNRead(int nType = 0)
+{
+	int nLen = 0;
+	XENGINE_PROTOCOLHDR st_ProtocolHdr;
+	XENGINE_PROTOCOL_XMQ st_XMQProtocol;
+	XCHAR tszMsgBuffer[2048];
+
+	memset(tszMsgBuffer, '\0', sizeof(tszMsgBuffer));
+	memset(&st_ProtocolHdr, '\0', sizeof(XENGINE_PROTOCOLHDR));
+	memset(&st_XMQProtocol, '\0', sizeof(XENGINE_PROTOCOL_XMQ));
+
+	st_ProtocolHdr.wHeader = XENGIEN_COMMUNICATION_PACKET_PROTOCOL_HEADER;
+	st_ProtocolHdr.unOperatorType = ENUM_XENGINE_COMMUNICATION_PROTOCOL_TYPE_XMQ;
+	st_ProtocolHdr.unOperatorCode = XENGINE_COMMUNICATION_PROTOCOL_OPERATOR_CODE_MQ_REQUNREAD;
+	st_ProtocolHdr.byVersion = 1;
+	st_ProtocolHdr.byIsReply = true;                  //必须为真
+	st_ProtocolHdr.wTail = XENGIEN_COMMUNICATION_PACKET_PROTOCOL_TAIL;
+	st_ProtocolHdr.unPacketSize = sizeof(XENGINE_PROTOCOL_XMQ);
+
+	nLen = sizeof(XENGINE_PROTOCOLHDR) + st_ProtocolHdr.unPacketSize;
+	memcpy(tszMsgBuffer, &st_ProtocolHdr, sizeof(XENGINE_PROTOCOLHDR));
+	memcpy(tszMsgBuffer + sizeof(XENGINE_PROTOCOLHDR), &st_XMQProtocol, sizeof(XENGINE_PROTOCOL_XMQ));
+
+	if (!XClient_TCPSelect_SendMsg(m_Socket, tszMsgBuffer, nLen))
+	{
+		_xtprintf("发送投递失败！\n");
+		return;
+	}
+
+	while (true)
+	{
+		nLen = 0;
+		XCHAR* ptszMsgBuffer;
+
+		memset(&st_ProtocolHdr, '\0', sizeof(XENGINE_PROTOCOLHDR));
+		if (XClient_TCPSelect_RecvPkt(m_Socket, &ptszMsgBuffer, &nLen, &st_ProtocolHdr))
+		{
+			if (0 == st_ProtocolHdr.wReserve)
+			{
+				_xtprintf("接受到数据,长度：%d，内容：%s\n", st_ProtocolHdr.unPacketSize, ptszMsgBuffer);
+			}
+			else
+			{
+				_xtprintf("获取消息队列数据失败,错误码:%d\n", st_ProtocolHdr.wReserve);
+			}
+			BaseLib_OperatorMemory_FreeCStyle((XPPMEM)&ptszMsgBuffer);
+			break;
+		}
+		std::this_thread::sleep_for(std::chrono::milliseconds(1));
+	}
+}
 //删除用户
 void MQ_DeleteUser()
 {
@@ -222,7 +274,7 @@ typedef struct
 	int a;
 	int b;
 }XENGINE_SQLBINARY;
-void MQ_Post(LPCXSTR lpszMsgBuffer, int nType = 0)
+void MQ_Post(LPCXSTR lpszMsgBuffer, int nType = 0, int nPubTime = -1, bool bSelf = false)
 {
 	int nLen = 0;
 	XENGINE_PROTOCOLHDR st_ProtocolHdr;
@@ -240,9 +292,11 @@ void MQ_Post(LPCXSTR lpszMsgBuffer, int nType = 0)
 	st_ProtocolHdr.wTail = XENGIEN_COMMUNICATION_PACKET_PROTOCOL_TAIL;
 
 	st_XMQProtocol.nSerial = 0;          //序列号,0服务会自动处理
-	st_XMQProtocol.nKeepTime = -1;        //保存时间，单位秒，如果为0，获取一次后被抛弃。-1 永久存在，PacketKey不能为空
+	st_XMQProtocol.nKeepTime = 0;      
+	st_XMQProtocol.nPubTime = nPubTime;
 	strcpy(st_XMQProtocol.tszMQKey, lpszKey);
 
+	st_XMQProtocol.st_MSGAttr.byAttrSelf = bSelf;
 	if (ENUM_XENGINE_PROTOCOLHDR_PAYLOAD_TYPE_BIN == nType)
 	{
 		XENGINE_SQLBINARY st_SQLBinary;
@@ -379,6 +433,7 @@ void MQ_TimePublish()
 	st_ProtocolHdr.unPacketSize = sizeof(XENGINE_PROTOCOL_XMQ) + _tcsxlen(lpszMsgBuffer);
 	st_XMQProtocol.nSerial = 0;          //要获取的序列号,如果为0,服务会自动处理
 	st_XMQProtocol.nPubTime = time(NULL) + 30; //当前时间+60秒
+	st_XMQProtocol.st_MSGAttr.byAttrSelf = 1;
 	strcpy(st_XMQProtocol.tszMQKey, lpszKey);
 
 	nLen = sizeof(XENGINE_PROTOCOLHDR) + st_ProtocolHdr.unPacketSize;
@@ -571,14 +626,16 @@ int main(int argc, char** argv)
 
 	MQ_Register();
 	MQ_Authorize();
+	MQ_GetUNRead();
 	MQ_Create();
-	MQ_Post(NULL, ENUM_XENGINE_PROTOCOLHDR_PAYLOAD_TYPE_BIN);
+	MQ_Post(NULL, ENUM_XENGINE_PROTOCOLHDR_PAYLOAD_TYPE_BIN, -1, true);
 	MQ_BindTopic();
 	MQ_Get(ENUM_XENGINE_PROTOCOLHDR_PAYLOAD_TYPE_BIN);
 
+	MQ_Post(lpszMsgBuffer, ENUM_XENGINE_PROTOCOLHDR_PAYLOAD_TYPE_STRING, 0);
 	for (int i = 0; i < 10; i++)
 	{
-		MQ_Post(lpszMsgBuffer, ENUM_XENGINE_PROTOCOLHDR_PAYLOAD_TYPE_STRING);
+		MQ_Post(lpszMsgBuffer, ENUM_XENGINE_PROTOCOLHDR_PAYLOAD_TYPE_STRING, -1, true);
 	}
 	MQ_GetNumber();
 	MQ_Get(ENUM_XENGINE_PROTOCOLHDR_PAYLOAD_TYPE_STRING);

XEngine_Apps/MQCore_WSApp/MQCore_WSApp.cpp

@@ -21,7 +21,7 @@
 #include <XEngine_Include/XEngine_RfcComponents/WSProtocol_Error.h>
 #include "../../XEngine_Source/XQueue_ProtocolHdr.h"
 
-//g++ -std=c++17 -Wall -g MQCore_WSApp.cpp -o MQCore_WSApp.exe -I ../../XEngine_Source/XEngine_ThirdPart/jsoncpp -L /usr/local/lib/XEngine_Release/XEngine_BaseLib -L /usr/local/lib/XEngine_Release/XEngine_Client -L /usr/local/lib/XEngine_Release/XEngine_RfcComponents -L ../../XEngine_Source/XEngine_ThirdPart/jsoncpp -lXEngine_BaseLib -lXEngine_Algorithm -lXClient_Socket -lRfcComponents_WSProtocol -ljsoncpp -Wl,-rpath=../../XEngine_Source/XEngine_ThirdPart/jsoncpp,--disable-new-dtags
+//g++ -std=c++17 -Wall -g MQCore_WSApp.cpp -o MQCore_WSApp.exe -I ../../XEngine_Source/XEngine_ThirdPart/jsoncpp -L /usr/local/lib/XEngine_Release/XEngine_BaseLib -L /usr/local/lib/XEngine_Release/XEngine_Client -L /usr/local/lib/XEngine_Release/XEngine_RfcComponents -L ../../XEngine_Source/XEngine_ThirdPart/jsoncpp -lXEngine_BaseLib -lXEngine_Algorithm -lXClient_Socket -lRfcComponents_WSProtocol -ljsoncpp
 
 XSOCKET m_Socket;
 LPCXSTR lpszKey = _X("XEngine_Notify");  //主题
@@ -149,10 +149,19 @@ void MQ_Post(LPCXSTR lpszMsgBuffer)
 	st_JsonRoot["byVersion"] = ENUM_XENGINE_PROTOCOLHDR_PAYLOAD_TYPE_BIN;
 	st_JsonRoot["byIsReply"] = 1;
 
+	XSHOT nMSGAttr = 0;
+	XENGINE_PROTOCOL_MSGATTR st_MSGAttr;
+	memset(&st_MSGAttr, '\0', sizeof(XENGINE_PROTOCOL_MSGATTR));
+
+	st_MSGAttr.byAttrSelf = 1;
+	memcpy(&nMSGAttr, &st_MSGAttr, sizeof(XENGINE_PROTOCOL_MSGATTR));
+
 	st_JsonMQProtocol["tszMQKey"] = lpszKey;
 	st_JsonMQProtocol["nSerial"] = 0;             //序列号,0服务会自动处理
-	st_JsonMQProtocol["nKeepTime"] = -1;          //保存时间，单位秒，如果为0，获取一次后被抛弃。-1 永久存在，PacketKey不能为空
+	st_JsonMQProtocol["nKeepTime"] = 0;          //生效时间
+	st_JsonMQProtocol["nPubTime"] = -1;         
 	st_JsonMQProtocol["nGetTimer"] = 0;
+	st_JsonMQProtocol["nMSGAttr"] = nMSGAttr;
 
 	st_JsonPayload["nPayLen"] = (Json::Value::UInt)strlen(lpszMsgBuffer);
 	st_JsonPayload["tszPayData"] = lpszMsgBuffer;

XEngine_Apps/VSCopy-x86.bat

@@ -4,10 +4,4 @@ copy /y "%XEngine_Lib32%\XEngine_Core\XEngine_OPenSsl.dll" "./"
 copy /y "%XEngine_Lib32%\XEngine_Core\XEngine_ManagePool.dll" "./"
 copy /y "%XEngine_Lib32%\XEngine_RfcComponents\RfcComponents_WSProtocol.dll" "./"
 copy /y "%XEngine_Lib32%\XEngine_NetHelp\NetHelp_APIClient.dll" "./"
-copy /y "%XEngine_Lib32%\XEngine_Client\XClient_Socket.dll" "./"
-
-copy /y "%XEngine_Lib32%\XEngine_LibEx\libcrypto-3.dll" "./"
-copy /y "%XEngine_Lib32%\XEngine_LibEx\libssl-3.dll" "./"
-copy /y "%XEngine_Lib32%\XEngine_NetHelp\libcurl.dll" "./"
-copy /y "%XEngine_Lib32%\XEngine_NetHelp\nghttp2.dll" "./"
-copy /y "%XEngine_Lib32%\XEngine_HelpComponents\zlib1.dll" "./"
+copy /y "%XEngine_Lib32%\XEngine_Client\XClient_Socket.dll" "./"

XEngine_Docment/Docment_en.docx

@@ -25,7 +25,7 @@
         "SQLAddr":"10.0.3.155",
         "SQLPort":3306,
         "SQLUser":"root",
-        "SQLPass":"123123Qyt"
+        "SQLPass":"123123"
     },
     "XPass":{
         "nTimeout":2,
@@ -35,6 +35,7 @@
         "tszPassLogout":""
     },
     "XVer":[
+        "3.9.0.1001 Build20241227",
         "3.8.0.1001 Build20230816",
         "3.7.0.1001 Build20230426",
         "3.6.0.1001 Build20230222",