added:topic owner for db

xengine-qyt · xengine-qyt · commit 7e64d41af5cb · 2023-02-09T15:24:48.000+08:00
diff --git a/XEngine_Source/MQCore_DBModule/DBModule_Define.h b/XEngine_Source/MQCore_DBModule/DBModule_Define.h
@@ -46,6 +46,12 @@ typedef struct
 	TCHAR tszCreateTime[64];                                              //创建的时间
 	__int64x nKeySerial;                                                  //包序列号
 }XENGINE_DBUSERKEY;
+//主题所有者
+typedef struct
+{
+	TCHAR tszQueueName[256];                                              //主题名称
+	TCHAR tszUserName[256];                                               //主题所有者
+}XENGINE_DBTOPICOWNER;
 //////////////////////////////////////////////////////////////////////////
 //                       导出的函数
 //////////////////////////////////////////////////////////////////////////
@@ -466,4 +472,49 @@ extern "C" BOOL DBModule_MQUser_TimeDelete(XENGINE_DBTIMERELEASE * pSt_DBInfo);
   意思：是否成功
 备注：
 *********************************************************************/
-extern "C" BOOL DBModule_MQUser_TimeClaer(time_t nTime = 0);
+extern "C" BOOL DBModule_MQUser_TimeClaer(time_t nTime = 0);
+/*************************************************************************
+						主题所有者导出函数
+**************************************************************************/
+/********************************************************************
+函数名称：DBModule_MQUser_OwnerInsert
+函数功能：主题所有者插入
+ 参数.一：pSt_DBOwner
+  In/Out：In
+  类型：数据结构指针
+  可空：N
+  意思：要操作的数据
+返回值
+  类型：逻辑型
+  意思：是否成功
+备注：
+*********************************************************************/
+extern "C" BOOL DBModule_MQUser_OwnerInsert(XENGINE_DBTOPICOWNER* pSt_DBOwner);
+/********************************************************************
+函数名称：DBModule_MQUser_OwnerDelete
+函数功能：主题所有者删除
+ 参数.一：pSt_DBOwner
+  In/Out：In
+  类型：数据结构指针
+  可空：N
+  意思：要操作的数据
+返回值
+  类型：逻辑型
+  意思：是否成功
+备注：
+*********************************************************************/
+extern "C" BOOL DBModule_MQUser_OwnerDelete(XENGINE_DBTOPICOWNER* pSt_DBOwner);
+/********************************************************************
+函数名称：DBModule_MQUser_OwnerQuery
+函数功能：主题所有者查询
+ 参数.一：pSt_DBOwner
+  In/Out：In
+  类型：数据结构指针
+  可空：N
+  意思：要操作的数据
+返回值
+  类型：逻辑型
+  意思：是否成功
+备注：
+*********************************************************************/
+extern "C" BOOL DBModule_MQUser_OwnerQuery(XENGINE_DBTOPICOWNER* pSt_DBOwner);
diff --git a/XEngine_Source/MQCore_DBModule/DBModule_Error.h b/XEngine_Source/MQCore_DBModule/DBModule_Error.h
@@ -16,4 +16,5 @@
 #define ERROR_XENGINE_MQCORE_DATABASE_PARAMENT 0xF0001              //参数错误
 #define ERROR_XENGINE_MQCORE_DATABASE_EMPTY 0xF0002                 //数据为空
 #define ERROR_XENGINE_MQCORE_DATABASE_EXIST 0xF0003                 //数据存在
-#define ERROR_XENGINE_MQCORE_DATABASE_CREATETHREAD 0xF0004          //创建线程失败
+#define ERROR_XENGINE_MQCORE_DATABASE_CREATETHREAD 0xF0004          //创建线程失败
+#define ERROR_XENGINE_MQCORE_DATABASE_NOTFOUND 0xF0005              //没有找到
diff --git a/XEngine_Source/MQCore_DBModule/DBModule_MQUser/DBModule_MQUser.cpp b/XEngine_Source/MQCore_DBModule/DBModule_MQUser/DBModule_MQUser.cpp
@@ -738,6 +738,125 @@ BOOL CDBModule_MQUser::DBModule_MQUser_TimeClaer(time_t nTime /* = 0 */)
 	}
 	return TRUE;
 }
+/********************************************************************
+函数名称：DBModule_MQUser_OwnerInsert
+函数功能：主题所有者插入
+ 参数.一：pSt_DBOwner
+  In/Out：In
+  类型：数据结构指针
+  可空：N
+  意思：要操作的数据
+返回值
+  类型：逻辑型
+  意思：是否成功
+备注：
+*********************************************************************/
+BOOL CDBModule_MQUser::DBModule_MQUser_OwnerInsert(XENGINE_DBTOPICOWNER* pSt_DBOwner)
+{
+	DBModule_IsErrorOccur = FALSE;
+
+	if (NULL == pSt_DBOwner)
+	{
+		DBModule_IsErrorOccur = TRUE;
+		DBModule_dwErrorCode = ERROR_XENGINE_MQCORE_DATABASE_PARAMENT;
+		return FALSE;
+	}
+	TCHAR tszSQLStatement[10240];
+	memset(tszSQLStatement, '\0', sizeof(tszSQLStatement));
+
+	_stprintf(tszSQLStatement, _T("INSERT IGNORE INTO `KeyOwner` (tszUserName,tszKeyName) VALUES('%s','%s')"), pSt_DBOwner->tszUserName, pSt_DBOwner->tszQueueName);
+	if (!DataBase_MySQL_Execute(xhDBSQL, tszSQLStatement))
+	{
+		DBModule_IsErrorOccur = TRUE;
+		DBModule_dwErrorCode = DataBase_GetLastError();
+		return FALSE;
+	}
+	return TRUE;
+}
+/********************************************************************
+函数名称：DBModule_MQUser_OwnerDelete
+函数功能：主题所有者删除
+ 参数.一：pSt_DBOwner
+  In/Out：In
+  类型：数据结构指针
+  可空：N
+  意思：要操作的数据
+返回值
+  类型：逻辑型
+  意思：是否成功
+备注：
+*********************************************************************/
+BOOL CDBModule_MQUser::DBModule_MQUser_OwnerDelete(XENGINE_DBTOPICOWNER* pSt_DBOwner)
+{
+	DBModule_IsErrorOccur = FALSE;
+
+	if (NULL == pSt_DBOwner)
+	{
+		DBModule_IsErrorOccur = TRUE;
+		DBModule_dwErrorCode = ERROR_XENGINE_MQCORE_DATABASE_PARAMENT;
+		return FALSE;
+	}
+	__int64u nAffectRow = 0;
+	TCHAR tszSQLStatement[10240];
+	memset(tszSQLStatement, '\0', sizeof(tszSQLStatement));
+
+	_stprintf(tszSQLStatement, _T("DELETE FROM `KeyOwner` WHERE tszUserName = '%s' AND tszKeyName = '%s'"), pSt_DBOwner->tszUserName, pSt_DBOwner->tszQueueName);
+	if (!DataBase_MySQL_Execute(xhDBSQL, tszSQLStatement, &nAffectRow))
+	{
+		DBModule_IsErrorOccur = TRUE;
+		DBModule_dwErrorCode = DataBase_GetLastError();
+		return FALSE;
+	}
+	if (nAffectRow <= 0)
+	{
+		DBModule_IsErrorOccur = TRUE;
+		DBModule_dwErrorCode = ERROR_XENGINE_MQCORE_DATABASE_NOTFOUND;
+		return FALSE;
+	}
+	return TRUE;
+}
+/********************************************************************
+函数名称：DBModule_MQUser_OwnerQuery
+函数功能：主题所有者查询
+ 参数.一：pSt_DBOwner
+  In/Out：In
+  类型：数据结构指针
+  可空：N
+  意思：要操作的数据
+返回值
+  类型：逻辑型
+  意思：是否成功
+备注：
+*********************************************************************/
+BOOL CDBModule_MQUser::DBModule_MQUser_OwnerQuery(XENGINE_DBTOPICOWNER* pSt_DBOwner)
+{
+	DBModule_IsErrorOccur = FALSE;
+
+	//查询
+	XNETHANDLE xhTable = 0;
+	__int64u nllLine = 0;
+	__int64u nllRow = 0;
+	TCHAR tszSQLStatement[1024];
+
+	memset(tszSQLStatement, '\0', sizeof(tszSQLStatement));
+
+	_stprintf(tszSQLStatement, _T("SELECT * FROM `KeyOwner` WHERE tszUserName = '%s' AND tszKeyName = '%s'"), pSt_DBOwner->tszUserName, pSt_DBOwner->tszQueueName);
+
+	if (!DataBase_MySQL_ExecuteQuery(xhDBSQL, &xhTable, tszSQLStatement, &nllLine, &nllRow))
+	{
+		DBModule_IsErrorOccur = TRUE;
+		DBModule_dwErrorCode = DataBase_GetLastError();
+		return FALSE;
+	}
+	DataBase_MySQL_FreeResult(xhDBSQL, xhTable);
+	if (nllLine <= 0)
+	{
+		DBModule_IsErrorOccur = TRUE;
+		DBModule_dwErrorCode = ERROR_XENGINE_MQCORE_DATABASE_EMPTY;
+		return FALSE;
+	}
+	return TRUE;
+}
 //////////////////////////////////////////////////////////////////////////
 //                         线程函数
 //////////////////////////////////////////////////////////////////////////
diff --git a/XEngine_Source/MQCore_DBModule/DBModule_MQUser/DBModule_MQUser.h b/XEngine_Source/MQCore_DBModule/DBModule_MQUser/DBModule_MQUser.h
@@ -34,6 +34,10 @@ class CDBModule_MQUser
 	BOOL DBModule_MQUser_TimeQuery(XENGINE_DBTIMERELEASE*** pppSt_DBInfo, int* pInt_ListCount);
 	BOOL DBModule_MQUser_TimeDelete(XENGINE_DBTIMERELEASE* pSt_DBInfo);
 	BOOL DBModule_MQUser_TimeClaer(time_t nTime = 0);
+public:
+	BOOL DBModule_MQUser_OwnerInsert(XENGINE_DBTOPICOWNER* pSt_DBOwner);
+	BOOL DBModule_MQUser_OwnerDelete(XENGINE_DBTOPICOWNER* pSt_DBOwner);
+	BOOL DBModule_MQUser_OwnerQuery(XENGINE_DBTOPICOWNER* pSt_DBOwner);
 protected:
 	static XHTHREAD CALLBACK DBModule_MQUser_TimeThread(LPVOID lParam);
 private:
diff --git a/XEngine_Source/MQCore_DBModule/MQCore_DBModule.def b/XEngine_Source/MQCore_DBModule/MQCore_DBModule.def
@@ -30,4 +30,8 @@ EXPORTS
 	DBModule_MQUser_TimeInsert
 	DBModule_MQUser_TimeQuery
 	DBModule_MQUser_TimeDelete
-	DBModule_MQUser_TimeClaer
+	DBModule_MQUser_TimeClaer
+
+	DBModule_MQUser_OwnerInsert
+	DBModule_MQUser_OwnerDelete
+	DBModule_MQUser_OwnerQuery
diff --git a/XEngine_Source/MQCore_DBModule/pch.cpp b/XEngine_Source/MQCore_DBModule/pch.cpp
@@ -139,4 +139,19 @@ extern "C" BOOL DBModule_MQUser_TimeDelete(XENGINE_DBTIMERELEASE * pSt_DBInfo)
 extern "C" BOOL DBModule_MQUser_TimeClaer(time_t nTime)
 {
 	return m_DBUser.DBModule_MQUser_TimeClaer(nTime);
+}
+/*************************************************************************
+						主题所有者导出函数
+**************************************************************************/
+extern "C" BOOL DBModule_MQUser_OwnerInsert(XENGINE_DBTOPICOWNER * pSt_DBOwner)
+{
+	return m_DBUser.DBModule_MQUser_OwnerInsert(pSt_DBOwner);
+}
+extern "C" BOOL DBModule_MQUser_OwnerDelete(XENGINE_DBTOPICOWNER * pSt_DBOwner)
+{
+	return m_DBUser.DBModule_MQUser_OwnerDelete(pSt_DBOwner);
+}
+extern "C" BOOL DBModule_MQUser_OwnerQuery(XENGINE_DBTOPICOWNER * pSt_DBOwner)
+{
+	return m_DBUser.DBModule_MQUser_OwnerQuery(pSt_DBOwner);
 }
diff --git a/XEngine_Source/XEngine_MQServiceApp/MQService_TCPTask.cpp b/XEngine_Source/XEngine_MQServiceApp/MQService_TCPTask.cpp
@@ -476,6 +476,7 @@ BOOL MessageQueue_TCP_Handle(XENGINE_PROTOCOLHDR* pSt_ProtocolHdr, LPCTSTR lpszC
 		else if (XENGINE_COMMUNICATION_PROTOCOL_OPERATOR_CODE_MQ_REQCREATE == pSt_ProtocolHdr->unOperatorCode)
 		{
 			pSt_ProtocolHdr->unOperatorCode = XENGINE_COMMUNICATION_PROTOCOL_OPERATOR_CODE_MQ_REPCREATE;
+			//创建表
 			if (!DBModule_MQData_CreateTable(st_MQProtocol.tszMQKey))
 			{
 				if (pSt_ProtocolHdr->byIsReply || (XENGINE_MQAPP_NETTYPE_HTTP == nNetType))
@@ -484,10 +485,30 @@ BOOL MessageQueue_TCP_Handle(XENGINE_PROTOCOLHDR* pSt_ProtocolHdr, LPCTSTR lpszC
 					ProtocolModule_Packet_Common(nNetType, pSt_ProtocolHdr, &st_MQProtocol, tszSDBuffer, &nSDLen);
 					XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, nNetType);
 				}
-				XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _T("%s消息端:%s,创建主题失败,主题名称:%s,无法继续,错误：%lX"), lpszClientType, lpszClientAddr, st_MQProtocol.tszMQKey, DBModule_GetLastError());
+				XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _T("%s消息端:%s,创建主题失败,创建表失败,主题名称:%s,无法继续,错误：%lX"), lpszClientType, lpszClientAddr, st_MQProtocol.tszMQKey, DBModule_GetLastError());
 				return FALSE;
 			}
+			//插入所有者
+			XENGINE_DBTOPICOWNER st_DBOwner;
+			memset(&st_DBOwner, '\0', sizeof(XENGINE_DBTOPICOWNER));
+
+			_tcscpy(st_DBOwner.tszUserName, tszUserName);
+			_tcscpy(st_DBOwner.tszQueueName, st_MQProtocol.tszMQKey);
+
+			if (!DBModule_MQUser_OwnerInsert(&st_DBOwner))
+			{
+				if (pSt_ProtocolHdr->byIsReply || (XENGINE_MQAPP_NETTYPE_HTTP == nNetType))
+				{
+					pSt_ProtocolHdr->wReserve = 762;
+					ProtocolModule_Packet_Common(nNetType, pSt_ProtocolHdr, &st_MQProtocol, tszSDBuffer, &nSDLen);
+					XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, nNetType);
+				}
+				XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _T("%s消息端:%s,创建主题失败,插入所有者失败,主题名称:%s,无法继续,错误：%lX"), lpszClientType, lpszClientAddr, st_MQProtocol.tszMQKey, DBModule_GetLastError());
+				return FALSE;
+			}
+			//创建通知
 			SessionModule_Notify_Create(st_MQProtocol.tszMQKey);
+			//回复
 			if (pSt_ProtocolHdr->byIsReply || (XENGINE_MQAPP_NETTYPE_HTTP == nNetType))
 			{
 				pSt_ProtocolHdr->wReserve = 0;
@@ -499,15 +520,33 @@ BOOL MessageQueue_TCP_Handle(XENGINE_PROTOCOLHDR* pSt_ProtocolHdr, LPCTSTR lpszC
 		else if (XENGINE_COMMUNICATION_PROTOCOL_OPERATOR_CODE_MQ_REQDELETE == pSt_ProtocolHdr->unOperatorCode)
 		{
 			pSt_ProtocolHdr->unOperatorCode = XENGINE_COMMUNICATION_PROTOCOL_OPERATOR_CODE_MQ_REPDELETE;
-			
+			//清理所有者
+			XENGINE_DBTOPICOWNER st_DBOwner;
+			memset(&st_DBOwner, '\0', sizeof(XENGINE_DBTOPICOWNER));
+
+			_tcscpy(st_DBOwner.tszUserName, tszUserName);
+			_tcscpy(st_DBOwner.tszQueueName, st_MQProtocol.tszMQKey);
+
+			if (!DBModule_MQUser_OwnerDelete(&st_DBOwner))
+			{
+				if (pSt_ProtocolHdr->byIsReply || (XENGINE_MQAPP_NETTYPE_HTTP == nNetType))
+				{
+					pSt_ProtocolHdr->wReserve = 0;
+					ProtocolModule_Packet_Common(nNetType, pSt_ProtocolHdr, &st_MQProtocol, tszSDBuffer, &nSDLen);
+					XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, nNetType);
+				}
+				XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _T("%s消息端:%s,删除主题失败,删除所有者失败,主题名称:%s,无法继续,错误：%lX"), lpszClientType, lpszClientAddr, st_MQProtocol.tszMQKey, DBModule_GetLastError());
+				return FALSE;
+			}
+			//清楚数据库
+			SessionModule_Notify_Destory(st_MQProtocol.tszMQKey);
+			DBModule_MQData_DeleteTable(st_MQProtocol.tszMQKey);
 			if (pSt_ProtocolHdr->byIsReply || (XENGINE_MQAPP_NETTYPE_HTTP == nNetType))
 			{
 				pSt_ProtocolHdr->wReserve = 0;
 				ProtocolModule_Packet_Common(nNetType, pSt_ProtocolHdr, &st_MQProtocol, tszSDBuffer, &nSDLen);
 				XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, nNetType);
 			}
-			DBModule_MQData_DeleteTable(st_MQProtocol.tszMQKey);
-			SessionModule_Notify_Destory(st_MQProtocol.tszMQKey);
 			XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _T("%s消息端:%s,主题:%s,删除主题成功"), lpszClientType, lpszClientAddr, st_MQProtocol.tszMQKey);
 		}
 		else if (XENGINE_COMMUNICATION_PROTOCOL_OPERATOR_CODE_MQ_REQNOTIFY == pSt_ProtocolHdr->unOperatorCode)
