added:whether break self for time release

Author: xengine-qyt

XEngine_Source/MQCore_DBModule/DBModule_Define.h

@@ -11,10 +11,6 @@
 //    History:
 *********************************************************************/
 //////////////////////////////////////////////////////////////////////////
-//                       导出的回调
-//////////////////////////////////////////////////////////////////////////
-typedef void(CALLBACK* CALLBACK_MESSAGEQUEUE_MODULE_DATABASE_TIMEPUBLISH)(LPCXSTR lpszQueueName, __int64x nIDMsg, __int64x nIDTime, XPVOID lParam);
-//////////////////////////////////////////////////////////////////////////
 //                       导出的数据结构
 //////////////////////////////////////////////////////////////////////////
 //消息队列
@@ -38,6 +34,7 @@ typedef struct
 	__int64x nIDMsg;                                                      //消息ID
 	__int64x nIDTime;                                                     //发布时间
 	bool bActive;                                                         //是否激活过
+	bool bBreak;                                                          //跳过自身
 }XENGINE_DBTIMERELEASE;
 //用户消息
 typedef struct
@@ -56,6 +53,10 @@ typedef struct
 	XCHAR tszCreateTime[64];                                              //创建时间
 }XENGINE_DBTOPICOWNER;
 //////////////////////////////////////////////////////////////////////////
+//                       导出的回调
+//////////////////////////////////////////////////////////////////////////
+typedef void(CALLBACK* CALLBACK_MESSAGEQUEUE_MODULE_DATABASE_TIMEPUBLISH)(XENGINE_DBTIMERELEASE* pSt_DBInfo, XPVOID lParam);
+//////////////////////////////////////////////////////////////////////////
 //                       导出的函数
 //////////////////////////////////////////////////////////////////////////
 extern "C" XLONG DBModule_GetLastError(int *pInt_SysError = NULL);

XEngine_Source/MQCore_DBModule/DBModule_MQUser/DBModule_MQUser.cpp

@@ -642,7 +642,7 @@ bool CDBModule_MQUser::DBModule_MQUser_TimeInsert(XENGINE_DBTIMERELEASE* pSt_DBI
 	XCHAR tszSQLStatement[10240];
 	memset(tszSQLStatement, '\0', sizeof(tszSQLStatement));
 
-	_xstprintf(tszSQLStatement, _X("INSERT INTO `UserTime` (tszQueueName,nIDMsg,nIDTime,bActive,tszCreateTime) VALUES('%s',%lld,%lld,0,now())"), pSt_DBInfo->tszQueueName, pSt_DBInfo->nIDMsg, pSt_DBInfo->nIDTime);
+	_xstprintf(tszSQLStatement, _X("INSERT INTO `UserTime` (tszQueueName,nIDMsg,nIDTime,bActive,bBreak,tszCreateTime) VALUES('%s',%lld,%lld,0,%d,now())"), pSt_DBInfo->tszQueueName, pSt_DBInfo->nIDMsg, pSt_DBInfo->nIDTime, pSt_DBInfo->bBreak);
 	if (!DataBase_MySQL_Execute(xhDBSQL, tszSQLStatement))
 	{
 		DBModule_IsErrorOccur = true;
@@ -715,7 +715,11 @@ bool CDBModule_MQUser::DBModule_MQUser_TimeQuery(XENGINE_DBTIMERELEASE*** pppSt_
 		}
 		if (NULL != pptszResult[4])
 		{
-			_tcsxcpy((*pppSt_DBInfo)[i]->tszCreateTime, pptszResult[4]);
+			(*pppSt_DBInfo)[i]->bBreak = _ttoi(pptszResult[4]);
+		}
+		if (NULL != pptszResult[5])
+		{
+			_tcsxcpy((*pppSt_DBInfo)[i]->tszCreateTime, pptszResult[5]);
 		}
 	}
 	DataBase_MySQL_FreeResult(xhDBSQL, xhTable);
@@ -1050,7 +1054,7 @@ XHTHREAD CALLBACK CDBModule_MQUser::DBModule_MQUser_TimeThread(XPVOID lParam)
 		pClass_This->DBModule_MQUser_TimeQuery(&ppSt_DBInfo, &nListCount);
 		for (int i = 0; i < nListCount; i++)
 		{
-			pClass_This->lpCall_TimePublish(ppSt_DBInfo[i]->tszQueueName, ppSt_DBInfo[i]->nIDMsg, ppSt_DBInfo[i]->nIDTime, pClass_This->m_lParam);
+			pClass_This->lpCall_TimePublish(ppSt_DBInfo[i], pClass_This->m_lParam);
 		}
 		BaseLib_OperatorMemory_Free((XPPPMEM)&ppSt_DBInfo, nListCount);
 		std::this_thread::sleep_for(std::chrono::seconds(1));

XEngine_Source/XEngine_MQServiceApp/MQService_TCPTask.cpp

@@ -373,6 +373,7 @@ bool MessageQueue_TCP_Handle(XENGINE_PROTOCOLHDR* pSt_ProtocolHdr, LPCXSTR lpszC
 
 				st_DBTime.nIDMsg = st_DBQueue.nQueueSerial;
 				st_DBTime.nIDTime = st_MQProtocol.nPubTime;
+				st_DBTime.bBreak = pSt_ProtocolHdr->wReserve;
 				_tcsxcpy(st_DBTime.tszQueueName, st_DBQueue.tszQueueName);
 
 				BaseLib_OperatorTime_TTimeToStuTime(st_MQProtocol.nPubTime, &st_LibTime);

XEngine_Source/XEngine_MQServiceApp/MQService_Task.cpp

@@ -1,19 +1,17 @@
 #include "MQService_Hdr.h"
 
-void CALLBACK MessageQueue_CBTask_TimePublish(LPCXSTR lpszQueueName, __int64x nIDMsg, __int64x nIDTime, XPVOID lParam)
+void CALLBACK MessageQueue_CBTask_TimePublish(XENGINE_DBTIMERELEASE* pSt_DBInfo, XPVOID lParam)
 {
 	int nMsgLen = 0;
 	XCHAR tszMsgBuffer[4096];
 	XENGINE_PROTOCOLHDR st_ProtocolHdr;
 	XENGINE_PROTOCOL_XMQ st_MQProtocol;
 	XENGINE_DBMESSAGEQUEUE st_DBInfo;
-	XENGINE_DBTIMERELEASE st_DBTime;
 
 	memset(tszMsgBuffer, '\0', sizeof(tszMsgBuffer));
 	memset(&st_ProtocolHdr, '\0', sizeof(XENGINE_PROTOCOLHDR));
 	memset(&st_MQProtocol, '\0', sizeof(XENGINE_PROTOCOL_XMQ));
 	memset(&st_DBInfo, '\0', sizeof(XENGINE_DBMESSAGEQUEUE));
-	memset(&st_DBTime, '\0', sizeof(XENGINE_DBTIMERELEASE));
 
 	st_ProtocolHdr.wHeader = XENGIEN_COMMUNICATION_PACKET_PROTOCOL_HEADER;
 	st_ProtocolHdr.unOperatorType = ENUM_XENGINE_COMMUNICATION_PROTOCOL_TYPE_XMQ;
@@ -23,13 +21,12 @@ void CALLBACK MessageQueue_CBTask_TimePublish(LPCXSTR lpszQueueName, __int64x nI
 	st_ProtocolHdr.unPacketSize = sizeof(XENGINE_PROTOCOL_XMQ);
 	st_ProtocolHdr.wTail = XENGIEN_COMMUNICATION_PACKET_PROTOCOL_TAIL;
 
-	st_MQProtocol.nPubTime = nIDTime;
-	st_MQProtocol.nSerial = nIDMsg;
-	_tcsxcpy(st_MQProtocol.tszMQKey, lpszQueueName);
+	st_MQProtocol.nPubTime = pSt_DBInfo->nIDTime;
+	st_MQProtocol.nSerial = pSt_DBInfo->nIDMsg;
+	_tcsxcpy(st_MQProtocol.tszMQKey, pSt_DBInfo->tszQueueName);
 
-	st_DBTime.nIDMsg = nIDMsg;
-	st_DBInfo.nQueueSerial = nIDMsg;
-	_tcsxcpy(st_DBInfo.tszQueueName, lpszQueueName);
+	st_DBInfo.nQueueSerial = pSt_DBInfo->nIDMsg;
+	_tcsxcpy(st_DBInfo.tszQueueName, pSt_DBInfo->tszQueueName);
 	DBModule_MQData_Query(&st_DBInfo);
 	//是否需要通知
 	int nListCount = 0;
@@ -40,10 +37,14 @@ void CALLBACK MessageQueue_CBTask_TimePublish(LPCXSTR lpszQueueName, __int64x nI
 		int nNetType = 0;
 		XCHAR tszUserAddr[128];
 		memset(tszUserAddr, '\0', sizeof(tszUserAddr));
-		//跳过自己
-		if (0 == _tcsxncmp(st_DBInfo.tszUserName, ppSt_ListUser[i]->tszUserName, _tcsxlen(st_DBInfo.tszUserName)))
+
+		if (pSt_DBInfo->bBreak)
 		{
-			continue;
+			//跳过自己
+			if (0 == _tcsxncmp(st_DBInfo.tszUserName, ppSt_ListUser[i]->tszUserName, _tcsxlen(st_DBInfo.tszUserName)))
+			{
+				continue;
+			}
 		}
 		SessionModule_Client_GetAddr(ppSt_ListUser[i]->tszUserName, tszUserAddr);
 		SessionModule_Client_GetType(tszUserAddr, &nNetType);
@@ -55,12 +56,12 @@ void CALLBACK MessageQueue_CBTask_TimePublish(LPCXSTR lpszQueueName, __int64x nI
 	if (st_DBConfig.st_MQUser.st_UserTime.bPubClear)
 	{
 		//移除这条消息
-		DBModule_MQUser_TimeDelete(&st_DBTime);
+		DBModule_MQUser_TimeDelete(pSt_DBInfo);
 	}
 	else
 	{
 		//更新
-		DBModule_MQUser_TimeUPDate(&st_DBTime);
+		DBModule_MQUser_TimeUPDate(pSt_DBInfo);
 	}
-	XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("定时任务,消息主题:%s,序列:%lld,定时任务分发成功,客户端个数:%d"), lpszQueueName, nIDMsg, nListCount);
+	XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("定时任务,消息主题:%s,序列:%lld,定时任务分发成功,客户端个数:%d"), pSt_DBInfo->tszQueueName, pSt_DBInfo->nIDMsg, nListCount);
 }

XEngine_Source/XEngine_MQServiceApp/MQService_Task.h

@@ -1,3 +1,3 @@
 #pragma once
 
-void CALLBACK MessageQueue_CBTask_TimePublish(LPCXSTR lpszQueueName, __int64x nIDMsg, __int64x nIDTime, XPVOID lParam);
+void CALLBACK MessageQueue_CBTask_TimePublish(XENGINE_DBTIMERELEASE* pSt_DBInfo, XPVOID lParam);