added:write and read to message attr of the database supported

xengine-qyt · xengine-qyt · commit 65d98faf2c85 · 2023-12-25T10:11:29.000+08:00
diff --git a/XEngine_Source/MQCore_DBModule/DBModule_Define.h b/XEngine_Source/MQCore_DBModule/DBModule_Define.h
@@ -25,7 +25,8 @@ typedef struct
 	__int64x nQueueSerial;                                                //包序列号
 	__int64x nQueueGetTime;                                               //可以获取的次数
 	int nMsgLen;                                                          //消息大小
-	XBYTE byMsgType;                                                       //消息类型,参考:ENUM_XENGINE_PROTOCOLHDR_PAYLOAD_TYPE
+	XBYTE byMsgType;                                                      //消息类型,参考:ENUM_XENGINE_PROTOCOLHDR_PAYLOAD_TYPE
+	XBYTE byMsgAttr;                                                      //消息属性
 }XENGINE_DBMESSAGEQUEUE;
 typedef struct
 {
diff --git a/XEngine_Source/MQCore_DBModule/DBModule_MQData/DBModule_MQData.cpp b/XEngine_Source/MQCore_DBModule/DBModule_MQData/DBModule_MQData.cpp
@@ -100,12 +100,12 @@ bool CDBModule_MQData::DBModule_MQData_Insert(XENGINE_DBMESSAGEQUEUE* pSt_DBInfo
 	memset(tszSQLCoder, '\0', sizeof(tszSQLCoder));
 
 	DataBase_MySQL_Coder(xhDBSQL, pSt_DBInfo->tszMsgBuffer, tszSQLCoder, &pSt_DBInfo->nMsgLen);
-	__int64u nRet = _xstprintf(tszSQLStatement, _X("INSERT INTO `%s` (tszUserName,tszQueueName,nQueueSerial,nQueueGetTime,tszQueueLeftTime,tszQueuePublishTime,tszQueueData,nDataType,tszQueueCreateTime) VALUES('%s','%s',%lld,%lld,'%s','%s','"), pSt_DBInfo->tszQueueName, pSt_DBInfo->tszUserName, pSt_DBInfo->tszQueueName, pSt_DBInfo->nQueueSerial, pSt_DBInfo->nQueueGetTime, pSt_DBInfo->tszQueueLeftTime, pSt_DBInfo->tszQueuePublishTime);
+	__int64u nRet = _xstprintf(tszSQLStatement, _X("INSERT INTO `%s` (tszUserName,tszQueueName,nQueueSerial,nQueueGetTime,tszQueueLeftTime,tszQueuePublishTime,tszQueueData,nDataType,nDataAttr,tszQueueCreateTime) VALUES('%s','%s',%lld,%lld,'%s','%s','"), pSt_DBInfo->tszQueueName, pSt_DBInfo->tszUserName, pSt_DBInfo->tszQueueName, pSt_DBInfo->nQueueSerial, pSt_DBInfo->nQueueGetTime, pSt_DBInfo->tszQueueLeftTime, pSt_DBInfo->tszQueuePublishTime);
 	memcpy(tszSQLStatement + nRet, tszSQLCoder, pSt_DBInfo->nMsgLen);
 	nRet += pSt_DBInfo->nMsgLen;
 
 	memset(tszSQLCoder, '\0', sizeof(tszSQLCoder));
-	int nLen = _xstprintf(tszSQLCoder, _X("',%d,now())"), pSt_DBInfo->byMsgType);
+	int nLen = _xstprintf(tszSQLCoder, _X("',%d,%d,now())"), pSt_DBInfo->byMsgType, pSt_DBInfo->byMsgAttr);
 	memcpy(tszSQLStatement + nRet, tszSQLCoder, nLen);
 	nRet += nLen;
 
@@ -199,7 +199,11 @@ bool CDBModule_MQData::DBModule_MQData_Query(XENGINE_DBMESSAGEQUEUE* pSt_DBInfo)
 	}
 	if (NULL != pptszResult[9])
 	{
-		_tcsxcpy(pSt_DBInfo->tszQueueCreateTime, pptszResult[9]);
+		pSt_DBInfo->byMsgAttr = _ttxoi(pptszResult[9]);
+	}
+	if (NULL != pptszResult[10])
+	{
+		_tcsxcpy(pSt_DBInfo->tszQueueCreateTime, pptszResult[10]);
 	}
 	DataBase_MySQL_FreeResult(xhDBSQL, xhTable);
 	return true;
@@ -240,7 +244,7 @@ bool CDBModule_MQData::DBModule_MQData_Modify(XENGINE_DBMESSAGEQUEUE* pSt_DBInfo
 	nRet += pSt_DBInfo->nMsgLen;
 
 	memset(tszSQLCoder, '\0', sizeof(tszSQLCoder));
-	int nLen = _xstprintf(tszSQLCoder, _X("',nDataType = %d WHERE tszUserName = '%s' AND tszQueueName = '%s' AND nQueueSerial = %lld"), pSt_DBInfo->byMsgType, pSt_DBInfo->tszUserName, pSt_DBInfo->tszQueueName, pSt_DBInfo->nQueueSerial);
+	int nLen = _xstprintf(tszSQLCoder, _X("',nDataType = %d,nDataAttr = %d WHERE tszUserName = '%s' AND tszQueueName = '%s' AND nQueueSerial = %lld"), pSt_DBInfo->byMsgType, pSt_DBInfo->byMsgAttr, pSt_DBInfo->tszUserName, pSt_DBInfo->tszQueueName, pSt_DBInfo->nQueueSerial);
 	memcpy(tszSQLStatement + nRet, tszSQLCoder, nLen);
 	nRet += nLen;
 
@@ -359,7 +363,11 @@ bool CDBModule_MQData::DBModule_MQData_List(LPCXSTR lpszQueueName, __int64x nSer
 		}
 		if (NULL != pptszResult[9])
 		{
-			_tcsxcpy((*pppSt_DBMessage)[i]->tszQueueCreateTime, pptszResult[9]);
+			(*pppSt_DBMessage)[i]->byMsgAttr = _ttxoi(pptszResult[9]);
+		}
+		if (NULL != pptszResult[10])
+		{
+			_tcsxcpy((*pppSt_DBMessage)[i]->tszQueueCreateTime, pptszResult[10]);
 		}
 	}
 	DataBase_MySQL_FreeResult(xhDBSQL, xhTable);
@@ -465,7 +473,11 @@ bool CDBModule_MQData::DBModule_MQData_GetSerial(LPCXSTR lpszName, __int64x* pIn
 		}
 		if (NULL != pptszResult[9])
 		{
-			_tcsxcpy(pSt_DBStart->tszQueueCreateTime, pptszResult[9]);
+			pSt_DBStart->byMsgAttr = _ttxoi(pptszResult[9]);
+		}
+		if (NULL != pptszResult[10])
+		{
+			_tcsxcpy(pSt_DBStart->tszQueueCreateTime, pptszResult[10]);
 		}
 		DataBase_MySQL_FreeResult(xhDBSQL, xhTable);
 	}
@@ -526,7 +538,11 @@ bool CDBModule_MQData::DBModule_MQData_GetSerial(LPCXSTR lpszName, __int64x* pIn
 		}
 		if (NULL != pptszResult[9])
 		{
-			_tcsxcpy(pSt_DBEnd->tszQueueCreateTime, pptszResult[9]);
+			pSt_DBEnd->byMsgAttr = _ttxoi(pptszResult[9]);
+		}
+		if (NULL != pptszResult[10])
+		{
+			_tcsxcpy(pSt_DBEnd->tszQueueCreateTime, pptszResult[10]);
 		}
 		DataBase_MySQL_FreeResult(xhDBSQL, xhTable);
 	}
@@ -592,6 +608,7 @@ bool CDBModule_MQData::DBModule_MQData_CreateTable(LPCXSTR lpszQueueName)
         "`tszQueuePublishTime` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '发布时间',"
         "`tszQueueData` longblob NOT NULL COMMENT '保存数据',"
 		"`nDataType` tinyint NOT NULL COMMENT '数据类型',"
+		"`nDataAttr` tinyint NULL DEFAULT NULL COMMENT '消息属性',"
         "`tszQueueCreateTime` datetime NOT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '插入时间',"
         "PRIMARY KEY (`ID`) USING BTREE"
         ") ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci ROW_FORMAT = Dynamic;"
diff --git a/XEngine_Source/XQueue_ProtocolHdr.h b/XEngine_Source/XQueue_ProtocolHdr.h
@@ -54,14 +54,26 @@
 ///////////////////////////////////////////////////////////////////////////
 #pragma pack(push)
 #pragma pack(1)
+typedef struct
+{
+	XBYTE byAttrAll : 1;                                                 //通知所有
+	XBYTE byAttrDisk : 1;                                                //消息存储标记
+	XBYTE byAttrResver2 : 1;
+	XBYTE byAttrResver3 : 1;
+	XBYTE byAttrResver4 : 1;
+	XBYTE byAttrResver5 : 1;
+	XBYTE byAttrResver6 : 1;
+	XBYTE byAttrResver7 : 1;
+}XENGINE_PROTOCOL_MSGATTR, * LPXENGINE_PROTOCOL_MSGATTR;
 //消息队列服务协议
 typedef struct 
 {
-	XCHAR tszMQKey[256];                                                   //此消息的KEY，不能为空
+	XCHAR tszMQKey[256];                                                  //此消息的KEY，不能为空
 	__int64x nSerial;                                                     //包序列号
 	__int64x nPubTime;                                                    //发布时间，根据自己需求配置时区
 	int nKeepTime;                                                        //保持时间,单位秒,-1 永久存在 0 一次就结束,>0 保存秒数
 	int nGetTimer;                                                        //可以获取的次数
+	XENGINE_PROTOCOL_MSGATTR st_MSGAttr;                                  //消息属性
 }XENGINE_PROTOCOL_XMQ, * LPXENGINE_PROTOCOL_XMQ;
 typedef struct 
 {

Original file line number	Diff line number	Diff line change
`@@ -100,12 +100,12 @@ bool CDBModule_MQData::DBModule_MQData_Insert(XENGINE_DBMESSAGEQUEUE* pSt_DBInfo`
`100`	`100`	`memset(tszSQLCoder, '\0', sizeof(tszSQLCoder));`
`101`	`101`
`102`	`102`	`DataBase_MySQL_Coder(xhDBSQL, pSt_DBInfo->tszMsgBuffer, tszSQLCoder, &pSt_DBInfo->nMsgLen);`
`103`		- __int64u nRet = _xstprintf(tszSQLStatement, _X("INSERT INTO `%s` (tszUserName,tszQueueName,nQueueSerial,nQueueGetTime,tszQueueLeftTime,tszQueuePublishTime,tszQueueData,nDataType,tszQueueCreateTime) VALUES('%s','%s',%lld,%lld,'%s','%s','"), pSt_DBInfo->tszQueueName, pSt_DBInfo->tszUserName, pSt_DBInfo->tszQueueName, pSt_DBInfo->nQueueSerial, pSt_DBInfo->nQueueGetTime, pSt_DBInfo->tszQueueLeftTime, pSt_DBInfo->tszQueuePublishTime);
	`103`	+ __int64u nRet = _xstprintf(tszSQLStatement, _X("INSERT INTO `%s` (tszUserName,tszQueueName,nQueueSerial,nQueueGetTime,tszQueueLeftTime,tszQueuePublishTime,tszQueueData,nDataType,nDataAttr,tszQueueCreateTime) VALUES('%s','%s',%lld,%lld,'%s','%s','"), pSt_DBInfo->tszQueueName, pSt_DBInfo->tszUserName, pSt_DBInfo->tszQueueName, pSt_DBInfo->nQueueSerial, pSt_DBInfo->nQueueGetTime, pSt_DBInfo->tszQueueLeftTime, pSt_DBInfo->tszQueuePublishTime);
`104`	`104`	`memcpy(tszSQLStatement + nRet, tszSQLCoder, pSt_DBInfo->nMsgLen);`
`105`	`105`	`nRet += pSt_DBInfo->nMsgLen;`
`106`	`106`
`107`	`107`	`memset(tszSQLCoder, '\0', sizeof(tszSQLCoder));`
`108`		`- int nLen = _xstprintf(tszSQLCoder, _X("',%d,now())"), pSt_DBInfo->byMsgType);`
	`108`	`+ int nLen = _xstprintf(tszSQLCoder, _X("',%d,%d,now())"), pSt_DBInfo->byMsgType, pSt_DBInfo->byMsgAttr);`
`109`	`109`	`memcpy(tszSQLStatement + nRet, tszSQLCoder, nLen);`
`110`	`110`	`nRet += nLen;`
`111`	`111`
`@@ -199,7 +199,11 @@ bool CDBModule_MQData::DBModule_MQData_Query(XENGINE_DBMESSAGEQUEUE* pSt_DBInfo)`
`199`	`199`	`}`
`200`	`200`	`if (NULL != pptszResult[9])`
`201`	`201`	`{`
`202`		`- _tcsxcpy(pSt_DBInfo->tszQueueCreateTime, pptszResult[9]);`
	`202`	`+ pSt_DBInfo->byMsgAttr = _ttxoi(pptszResult[9]);`
	`203`	`+ }`
	`204`	`+ if (NULL != pptszResult[10])`
	`205`	`+ {`
	`206`	`+ _tcsxcpy(pSt_DBInfo->tszQueueCreateTime, pptszResult[10]);`
`203`	`207`	`}`
`204`	`208`	`DataBase_MySQL_FreeResult(xhDBSQL, xhTable);`
`205`	`209`	`return true;`
`@@ -240,7 +244,7 @@ bool CDBModule_MQData::DBModule_MQData_Modify(XENGINE_DBMESSAGEQUEUE* pSt_DBInfo`
`240`	`244`	`nRet += pSt_DBInfo->nMsgLen;`
`241`	`245`
`242`	`246`	`memset(tszSQLCoder, '\0', sizeof(tszSQLCoder));`
`243`		`- int nLen = _xstprintf(tszSQLCoder, _X("',nDataType = %d WHERE tszUserName = '%s' AND tszQueueName = '%s' AND nQueueSerial = %lld"), pSt_DBInfo->byMsgType, pSt_DBInfo->tszUserName, pSt_DBInfo->tszQueueName, pSt_DBInfo->nQueueSerial);`
	`247`	`+ int nLen = _xstprintf(tszSQLCoder, _X("',nDataType = %d,nDataAttr = %d WHERE tszUserName = '%s' AND tszQueueName = '%s' AND nQueueSerial = %lld"), pSt_DBInfo->byMsgType, pSt_DBInfo->byMsgAttr, pSt_DBInfo->tszUserName, pSt_DBInfo->tszQueueName, pSt_DBInfo->nQueueSerial);`
`244`	`248`	`memcpy(tszSQLStatement + nRet, tszSQLCoder, nLen);`
`245`	`249`	`nRet += nLen;`
`246`	`250`
`@@ -359,7 +363,11 @@ bool CDBModule_MQData::DBModule_MQData_List(LPCXSTR lpszQueueName, __int64x nSer`
`359`	`363`	`}`
`360`	`364`	`if (NULL != pptszResult[9])`
`361`	`365`	`{`
`362`		`- _tcsxcpy((*pppSt_DBMessage)[i]->tszQueueCreateTime, pptszResult[9]);`
	`366`	`+ (*pppSt_DBMessage)[i]->byMsgAttr = _ttxoi(pptszResult[9]);`
	`367`	`+ }`
	`368`	`+ if (NULL != pptszResult[10])`
	`369`	`+ {`
	`370`	`+ _tcsxcpy((*pppSt_DBMessage)[i]->tszQueueCreateTime, pptszResult[10]);`
`363`	`371`	`}`
`364`	`372`	`}`
`365`	`373`	`DataBase_MySQL_FreeResult(xhDBSQL, xhTable);`
`@@ -465,7 +473,11 @@ bool CDBModule_MQData::DBModule_MQData_GetSerial(LPCXSTR lpszName, __int64x* pIn`
`465`	`473`	`}`
`466`	`474`	`if (NULL != pptszResult[9])`
`467`	`475`	`{`
`468`		`- _tcsxcpy(pSt_DBStart->tszQueueCreateTime, pptszResult[9]);`
	`476`	`+ pSt_DBStart->byMsgAttr = _ttxoi(pptszResult[9]);`
	`477`	`+ }`
	`478`	`+ if (NULL != pptszResult[10])`
	`479`	`+ {`
	`480`	`+ _tcsxcpy(pSt_DBStart->tszQueueCreateTime, pptszResult[10]);`
`469`	`481`	`}`
`470`	`482`	`DataBase_MySQL_FreeResult(xhDBSQL, xhTable);`
`471`	`483`	`}`
`@@ -526,7 +538,11 @@ bool CDBModule_MQData::DBModule_MQData_GetSerial(LPCXSTR lpszName, __int64x* pIn`
`526`	`538`	`}`
`527`	`539`	`if (NULL != pptszResult[9])`
`528`	`540`	`{`
`529`		`- _tcsxcpy(pSt_DBEnd->tszQueueCreateTime, pptszResult[9]);`
	`541`	`+ pSt_DBEnd->byMsgAttr = _ttxoi(pptszResult[9]);`
	`542`	`+ }`
	`543`	`+ if (NULL != pptszResult[10])`
	`544`	`+ {`
	`545`	`+ _tcsxcpy(pSt_DBEnd->tszQueueCreateTime, pptszResult[10]);`
`530`	`546`	`}`
`531`	`547`	`DataBase_MySQL_FreeResult(xhDBSQL, xhTable);`
`532`	`548`	`}`
`@@ -592,6 +608,7 @@ bool CDBModule_MQData::DBModule_MQData_CreateTable(LPCXSTR lpszQueueName)`
`592`	`608`	"`tszQueuePublishTime` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '发布时间',"
`593`	`609`	"`tszQueueData` longblob NOT NULL COMMENT '保存数据',"
`594`	`610`	"`nDataType` tinyint NOT NULL COMMENT '数据类型',"
	`611`	+ "`nDataAttr` tinyint NULL DEFAULT NULL COMMENT '消息属性',"
`595`	`612`	"`tszQueueCreateTime` datetime NOT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '插入时间',"
`596`	`613`	"PRIMARY KEY (`ID`) USING BTREE"
`597`	`614`	`") ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci ROW_FORMAT = Dynamic;"`