Merge pull request #10 from libxengine/develop

V3.3.0.1001 Merge

Author: xengine-qyt

CHANGELOG

@@ -1,3 +1,23 @@
+XEngine_MQService V3.3.0.1001
+
+添加:定时发布功能
+添加:为会话模块添加获取用户网络类型的功能
+修复:某些时候程序退出崩溃
+修复:重复订阅现在不允许了
+修复:用户离开订阅还存在的问题
+修复:订阅的消息会发送给自己的问题
+修复:登录打印不正确的问题
+删除:DDS模块和相关代码
+
+added:Timed Release
+added:get user net type for session module
+fixed:sometime is crashed when program exit
+fixed:Duplicate subscriptions are not allowed
+fixed:The problem that the user leaves the subscription is work
+fixed:the problem that subscribe messages sent to yourself
+fixed:login printf is incorrent
+delete:dds module and Delete related functions
+======================================================================================
 XEngine_MQService V3.2.0.1001
 
 添加:为会话模块添加心跳

README.en.md

@@ -28,7 +28,7 @@ this software support following features
 12. Active delivery acquisition mode or passive subscription notification module  
 13. get order and start serial pos setting  
 14. Unlimited load message types  
-15. data distribution service(DDS),Support WAN and LAN
+15. timed message
 16. Access control(planning)
 
 ## install

README.md

@@ -27,7 +27,7 @@ c c++Message Service
 12. 主动投递获取模式或者被动订阅通知模块  
 13. 获取顺序与开始序列号设置  
 14. 不限制负载的消息类型  
-15. 消息分发服务(DDS),支持广域网和局域网
+15. 定时消息
 16. 权限控制(planning)
 
 ## 安装教程
@@ -95,7 +95,6 @@ make FLAGS=CleanAll 清理编译
 如果你觉得这个软件对你有帮助,请你给我们一个START吧
 
 ## 开发计划
-DDS消息订阅与发布  
 扩展获取消息的内容  
 消息属性生效  
 删除包协议  

XEngine_Apps/MQCore_TCPApp/MQCore_TCPApp.cpp

@@ -320,6 +320,58 @@ void MQ_Get()
 		std::this_thread::sleep_for(std::chrono::milliseconds(1));
 	}
 }
+
+void MQ_TimePublish()
+{
+	int nLen = 0;
+	LPCTSTR lpszMsgBuffer = _T("hello world");
+	XENGINE_PROTOCOLHDR st_ProtocolHdr;
+	XENGINE_PROTOCOL_XMQ st_XMQProtocol;
+	TCHAR 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_REQPOST;
+	st_ProtocolHdr.byVersion = 1;
+	st_ProtocolHdr.byIsReply = TRUE;                  //必须为真
+	st_ProtocolHdr.wTail = XENGIEN_COMMUNICATION_PACKET_PROTOCOL_TAIL;
+
+	st_ProtocolHdr.unPacketSize = sizeof(XENGINE_PROTOCOL_XMQ) + _tcslen(lpszMsgBuffer);
+	st_XMQProtocol.nSerial = 0;          //要获取的序列号,如果为0,服务会自动处理
+	st_XMQProtocol.nPubTime = time(NULL) + 60; //当前时间+60秒
+	strcpy(st_XMQProtocol.tszMQKey, lpszKey);
+
+	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));
+	memcpy(tszMsgBuffer + sizeof(XENGINE_PROTOCOLHDR) + sizeof(XENGINE_PROTOCOL_XMQ), lpszMsgBuffer, _tcslen(lpszMsgBuffer));
+
+	if (!XClient_TCPSelect_SendMsg(m_Socket, tszMsgBuffer, nLen))
+	{
+		printf("发送投递失败！\n");
+		return;
+	}
+	nLen = 0;
+	CHAR* ptszMsgBuffer = NULL;
+	if (!XClient_TCPSelect_RecvPkt(m_Socket, &ptszMsgBuffer, &nLen, &st_ProtocolHdr))
+	{
+		printf("接受数据失败！\n");
+		return;
+	}
+	BaseLib_OperatorMemory_FreeCStyle((XPPMEM)&ptszMsgBuffer);
+
+	if (!XClient_TCPSelect_RecvPkt(m_Socket, &ptszMsgBuffer, &nLen, &st_ProtocolHdr, 60))
+	{
+		printf("接受数据失败！\n");
+		return;
+	}
+	BaseLib_OperatorMemory_FreeCStyle((XPPMEM)&ptszMsgBuffer);
+}
+
 void MQ_GetNumber()
 {
 	int nLen = 0;
@@ -393,7 +445,7 @@ void MQ_GetSerial()
 
 	st_ProtocolHdr.unPacketSize = sizeof(XENGINE_PROTOCOL_XMQ);
 
-	st_XMQProtocol.nSerial = 10; //设置为10开始读取
+	st_XMQProtocol.nSerial = 5; //设置为5开始读取
 	strcpy(st_XMQProtocol.tszMQKey, lpszKey);
 
 	nLen = sizeof(XENGINE_PROTOCOLHDR) + st_ProtocolHdr.unPacketSize;
@@ -492,29 +544,14 @@ void MQ_Subscribe()
 		printf("发送投递失败！\n");
 		return;
 	}
-	while (TRUE)
+	nLen = 0;
+	CHAR* ptszMsgBuffer = NULL;
+	if (!XClient_TCPSelect_RecvPkt(m_Socket, &ptszMsgBuffer, &nLen, &st_ProtocolHdr))
 	{
-		nLen = 2048;
-		memset(tszMsgBuffer, '\0', sizeof(tszMsgBuffer));
-		if (XClient_TCPSelect_RecvMsg(m_Socket, tszMsgBuffer, &nLen))
-		{
-			memset(&st_ProtocolHdr, '\0', sizeof(XENGINE_PROTOCOLHDR));
-			memset(&st_XMQProtocol, '\0', sizeof(XENGINE_PROTOCOL_XMQ));
-
-			memcpy(&st_ProtocolHdr, tszMsgBuffer, sizeof(XENGINE_PROTOCOLHDR));
-			memcpy(&st_XMQProtocol, tszMsgBuffer + sizeof(XENGINE_PROTOCOLHDR), sizeof(XENGINE_PROTOCOL_XMQ));
-
-			if (0 == st_ProtocolHdr.wReserve)
-			{
-				printf("接受到数据,主题:%s,序列:%lld,长度：%d，内容：%s\n", st_XMQProtocol.tszMQKey, st_XMQProtocol.nSerial, st_ProtocolHdr.unPacketSize - sizeof(XENGINE_PROTOCOL_XMQ), tszMsgBuffer + sizeof(XENGINE_PROTOCOLHDR) + sizeof(XENGINE_PROTOCOL_XMQ));
-			}
-			else
-			{
-				printf("获取消息队列数据失败,错误码:%d\n", st_ProtocolHdr.wReserve);
-			}
-		}
-		std::this_thread::sleep_for(std::chrono::milliseconds(1));
+		printf("接受数据失败！\n");
+		return;
 	}
+	BaseLib_OperatorMemory_FreeCStyle((XPPMEM)&ptszMsgBuffer);
 }
 int main(int argc, char** argv)
 {
@@ -533,26 +570,20 @@ int main(int argc, char** argv)
 
 	MQ_Register();
 	MQ_Authorize();
-	if (argc > 1)
+	MQ_Create();
+	for (int i = 0; i < 10; i++)
 	{
-		MQ_Create();
-		MQ_Subscribe();
-	}
-	else
-	{
-		MQ_Create();
-		for (int i = 0; i < 1000; i++)
-		{
-			MQ_Post(lpszMsgBuffer);
-		}
-		MQ_GetSerial();
-		MQ_GetNumber();
-		MQ_Get();
-		MQ_Get();
-		MQ_Get();
-		MQ_DeleteTopic();
-		MQ_DeleteUser();
+		MQ_Post(lpszMsgBuffer);
 	}
+	MQ_GetSerial();
+	MQ_GetNumber();
+	MQ_Get();
+	MQ_Get();
+	MQ_Get();
+	MQ_Subscribe();
+	MQ_TimePublish();
+	MQ_DeleteTopic();
+	MQ_DeleteUser();
 
 	XClient_TCPSelect_Close(m_Socket);
 #ifdef _WINDOWS

XEngine_Docment/Docment_en.docx

@@ -6,8 +6,6 @@
     "nHttpPort":5201,
     "nWSPort":5202,
     "nRPCPort":5203,
-    "nBroadRVPort":5210,
-    "nBroadSDPort":5211,
     "XMax":{
         "nMaxClient":10000,
         "nMaxQueue":10000,
@@ -32,6 +30,7 @@
         "SQLPass":"123123Ruiyue"
     },
     "XVer":[
+        "3.3.0.1001 Build20220805",
         "3.2.0.1001 Build20220715",
         "3.1.0.1001 Build20220408",
         "3.0.0.1001 Build20220318",

XEngine_Docment/Docment_zh.docx

@@ -3,15 +3,15 @@
 
  Source Server         : Test
  Source Server Type    : MySQL
- Source Server Version : 80028
+ Source Server Version : 80030
  Source Host           : 192.168.1.12:3306
  Source Schema         : XEngine_MQData
 
  Target Server Type    : MySQL
- Target Server Version : 80028
+ Target Server Version : 80030
  File Encoding         : 65001
 
- Date: 01/04/2022 11:30:02
+ Date: 01/08/2022 13:15:31
 */
 
 SET NAMES utf8mb4;
@@ -23,15 +23,25 @@ SET FOREIGN_KEY_CHECKS = 0;
 DROP TABLE IF EXISTS `XEngine_CommKey`;
 CREATE TABLE `XEngine_CommKey`  (
   `ID` int NOT NULL AUTO_INCREMENT,
-  `tszQueueName` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '所属队列',
+  `tszQueueName` varchar(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NOT NULL COMMENT '所属队列',
   `nQueueSerial` bigint NOT NULL COMMENT '消息序列',
   `nQueueGetTime` bigint NOT NULL COMMENT '获取次数',
-  `tszQueueLeftTime` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '过期时间',
-  `tszQueuePublishTime` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '发布时间',
-  `tszQueueData` varchar(8192) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '保存数据',
+  `tszQueueLeftTime` varchar(128) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '过期时间',
+  `tszQueuePublishTime` varchar(128) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '发布时间',
+  `tszQueueData` varchar(8192) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NOT NULL COMMENT '保存数据',
   `nDataLen` int NOT NULL COMMENT '数据大小',
   `tszQueueCreateTime` datetime NOT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '插入时间',
   PRIMARY KEY (`ID`) USING BTREE
-) ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = DYNAMIC;
+) ENGINE = InnoDB CHARACTER SET = utf8mb3 COLLATE = utf8mb3_general_ci ROW_FORMAT = DYNAMIC;
+
+-- ----------------------------
+-- Table structure for XEngine_TimeRelease
+-- ----------------------------
+DROP TABLE IF EXISTS `XEngine_TimeRelease`;
+CREATE TABLE `XEngine_TimeRelease`  (
+  `tszQueueName` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '队列名称',
+  `nIDMsg` bigint NOT NULL COMMENT '队列唯一ID',
+  `nIDTime` bigint NOT NULL COMMENT '发布时间'
+) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;
 
 SET FOREIGN_KEY_CHECKS = 1;

XEngine_Release/XEngine_Config/XEngine_Config.json

@@ -21,8 +21,6 @@ typedef struct tag_XEngine_ServerConfig
     int nTCPPort;
     int nHttpPort;
     int nWSPort;
-    int nBroadRVPort;
-    int nBroadSDPort;
     struct
     {
         int nMaxClient;

XEngine_SQLFile/XEngine_MQData.sql

@@ -69,8 +69,6 @@ BOOL CConfig_Json::Config_Json_File(LPCTSTR lpszConfigFile,XENGINE_SERVERCONFIG
     pSt_ServerConfig->nTCPPort = st_JsonRoot["nTCPPort"].asInt();
     pSt_ServerConfig->nHttpPort = st_JsonRoot["nHttpPort"].asInt();
     pSt_ServerConfig->nWSPort = st_JsonRoot["nWSPort"].asInt();
-    pSt_ServerConfig->nBroadRVPort = st_JsonRoot["nBroadRVPort"].asInt();
-    pSt_ServerConfig->nBroadSDPort = st_JsonRoot["nBroadSDPort"].asInt();
 
     if (st_JsonRoot["XMax"].empty() || (6 != st_JsonRoot["XMax"].size()))
     {