Skip to content

Commit 320529b

Browse files
xengine-qytxengine-qyt
committed
modify:supported notify message to all client
1 parent 65d98fa commit 320529b

File tree

7 files changed

+114
-21
lines changed

7 files changed

+114
-21
lines changed

‎XEngine_Source/MQCore_SessionModule/MQCore_SessionModule.def‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,4 +11,5 @@ EXPORTS
1111
SessionModule_Client_GetUser
1212
SessionModule_Client_GetAddr
1313
SessionModule_Client_GetInfoByUser
14-
SessionModule_Client_GetType
14+
SessionModule_Client_GetType
15+
SessionModule_Client_GetListAddr

‎XEngine_Source/MQCore_SessionModule/SessionModule_Client/SessionModule_Client.cpp‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -339,4 +339,38 @@ bool CSessionModule_Client::SessionModule_Client_GetType(LPCXSTR lpszSessionStr,
339339
*pInt_NetType = stl_MapIterator->second.nNetType;
340340
st_Locker.unlock_shared();
341341
return true;
342+
}
343+
/********************************************************************
344+
函数名称：SessionModule_Client_GetList
345+
函数功能：获取客户端地址列表
346+
参数.一：ppptszClientList
347+
In/Out：Out
348+
类型：三级指针
349+
可空：N
350+
意思：输出客户端列表
351+
参数.二：pInt_ListCount
352+
In/Out：Out
353+
类型：整数型指针
354+
可空：N
355+
意思：输出列表个数
356+
返回值
357+
类型：逻辑型
358+
意思：是否成功
359+
备注：
360+
*********************************************************************/
361+
bool CSessionModule_Client::SessionModule_Client_GetListAddr(XCHAR*** ppptszClientList, int* pInt_ListCount)
362+
{
363+
Session_IsErrorOccur = false;
364+
365+
*pInt_ListCount = stl_MapSession.size();
366+
BaseLib_OperatorMemory_Malloc((XPPPMEM)ppptszClientList, stl_MapSession.size(), 128);
367+
368+
st_Locker.lock_shared();
369+
unordered_map<tstring, XENGINE_SESSIONINFO>::iterator stl_MapIterator = stl_MapSession.begin();
370+
for (int i = 0; stl_MapIterator != stl_MapSession.end(); stl_MapIterator++, i++)
371+
{
372+
_tcsxcpy((*ppptszClientList)[i], stl_MapIterator->second.tszUserAddr);
373+
}
374+
st_Locker.unlock_shared();
375+
return true;
342376
}

‎XEngine_Source/MQCore_SessionModule/SessionModule_Client/SessionModule_Client.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ class CSessionModule_Client
3333
bool SessionModule_Client_GetAddr(LPCXSTR lpszUserName, XCHAR* ptszUserAddr = NULL);
3434
bool SessionModule_Client_GetInfoByUser(LPCXSTR lpszUserName, XENGINE_PROTOCOL_USERINFO* pSt_UserInfo);
3535
bool SessionModule_Client_GetType(LPCXSTR lpszSessionStr, int* pInt_NetType);
36+
bool SessionModule_Client_GetListAddr(XCHAR*** ppptszClientList, int* pInt_ListCount);
3637
private:
3738
shared_mutex st_Locker;
3839
private:

‎XEngine_Source/MQCore_SessionModule/Session_Define.h‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -172,4 +172,23 @@ extern "C" bool SessionModule_Client_GetInfoByUser(LPCXSTR lpszUserName, XENGINE
172172
意思：是否成功
173173
备注：
174174
*********************************************************************/
175-
extern "C" bool SessionModule_Client_GetType(LPCXSTR lpszSessionStr, int* pInt_NetType);
175+
extern "C" bool SessionModule_Client_GetType(LPCXSTR lpszSessionStr, int* pInt_NetType);
176+
/********************************************************************
177+
函数名称：SessionModule_Client_GetList
178+
函数功能：获取客户端地址列表
179+
参数.一：ppptszClientList
180+
In/Out：Out
181+
类型：三级指针
182+
可空：N
183+
意思：输出客户端列表
184+
参数.二：pInt_ListCount
185+
In/Out：Out
186+
类型：整数型指针
187+
可空：N
188+
意思：输出列表个数
189+
返回值
190+
类型：逻辑型
191+
意思：是否成功
192+
备注：
193+
*********************************************************************/
194+
extern "C" bool SessionModule_Client_GetListAddr(XCHAR*** ppptszClientList, int* pInt_ListCount);

‎XEngine_Source/MQCore_SessionModule/pch.cpp‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,4 +64,8 @@ extern "C" bool SessionModule_Client_GetInfoByUser(LPCXSTR lpszUserName, XENGINE
6464
extern "C" bool SessionModule_Client_GetType(LPCXSTR lpszSessionStr, int* pInt_NetType)
6565
{
6666
return m_SessionClient.SessionModule_Client_GetType(lpszSessionStr, pInt_NetType);
67+
}
68+
extern "C" bool SessionModule_Client_GetListAddr(XCHAR * **ppptszClientList, int* pInt_ListCount)
69+
{
70+
return m_SessionClient.SessionModule_Client_GetListAddr(ppptszClientList, pInt_ListCount);
6771
}

‎XEngine_Source/MQCore_SessionModule/pch.h‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@
2323
#include <XEngine_Include/XEngine_CommHdr.h>
2424
#include <XEngine_Include/XEngine_ProtocolHdr.h>
2525
#include <XEngine_Include/XEngine_Types.h>
26+
#include <XEngine_Include/XEngine_BaseLib/BaseLib_Define.h>
27+
#include <XEngine_Include/XEngine_BaseLib/BaseLib_Error.h>
2628
#include "../XQueue_ProtocolHdr.h"
2729
#include "Session_Define.h"
2830
#include "Session_Error.h"
@@ -44,4 +46,8 @@ typedef std::string tstring;
4446
// History:
4547
*********************************************************************/
4648
extern bool Session_IsErrorOccur;
47-
extern XLONG Session_dwErrorCode;
49+
extern XLONG Session_dwErrorCode;
50+
51+
#ifdef _MSC_BUILD
52+
#pragma comment(lib,"XEngine_BaseLib/XEngine_BaseLib.lib")
53+
#endif

‎XEngine_Source/XEngine_MQServiceApp/MQService_TCPTask.cpp‎

Lines changed: 46 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -415,37 +415,65 @@ bool MessageQueue_TCP_Handle(XENGINE_PROTOCOLHDR* pSt_ProtocolHdr, LPCXSTR lpszC
415415
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, nNetType);
416416
}
417417
//是否需要通知
418-
int nListCount = 0;
419418
if (0 == st_MQProtocol.nPubTime)
420419
{
421-
//定时任务不通知
422-
XENGINE_DBUSERKEY** ppSt_ListUser;
423-
if (DBModule_MQUser_KeyList(NULL, st_MQProtocol.tszMQKey, &ppSt_ListUser, &nListCount))
420+
//设置为0,不是定时发布
421+
if (1 == st_MQProtocol.st_MSGAttr.byAttrAll)
424422
{
425-
int nTCPLen = 0;
426-
XCHAR tszTCPBuffer[4096];
427-
memset(tszTCPBuffer, '\0', sizeof(tszTCPBuffer));
428-
429-
pSt_ProtocolHdr->unOperatorCode = XENGINE_COMMUNICATION_PROTOCOL_OPERATOR_CODE_MQ_MSGNOTIFY;
430-
ProtocolModule_Packet_Common(nNetType, pSt_ProtocolHdr, &st_MQProtocol, tszTCPBuffer, &nTCPLen, lpszMsgBuffer + sizeof(XENGINE_PROTOCOL_XMQ), nMsgLen - sizeof(XENGINE_PROTOCOL_XMQ));
423+
int nListCount = 0;
424+
XCHAR** pptszListAddr;
425+
SessionModule_Client_GetListAddr(&pptszListAddr, &nListCount);
431426
for (int i = 0; i < nListCount; i++)
432427
{
433428
//跳过自己
434-
if (0 == _tcsxncmp(tszUserName, ppSt_ListUser[i]->tszUserName, _tcsxlen(tszUserName)))
429+
if (0 == _tcsxncmp(lpszClientAddr, pptszListAddr[i], _tcsxlen(lpszClientAddr)))
435430
{
436431
continue;
437432
}
438-
XCHAR tszUserAddr[128];
439-
memset(tszUserAddr, '\0', sizeof(tszUserAddr));
433+
nSDLen = 0;
434+
int nMSGLen = 0;
435+
int nClientType = 0;
436+
memset(tszSDBuffer, '\0', sizeof(tszSDBuffer));
437+
438+
pSt_ProtocolHdr->unOperatorCode = XENGINE_COMMUNICATION_PROTOCOL_OPERATOR_CODE_MQ_MSGNOTIFY;
439+
440+
SessionModule_Client_GetType(pptszListAddr[i], &nClientType);
441+
ProtocolModule_Packet_Common(nClientType, pSt_ProtocolHdr, &st_MQProtocol, tszSDBuffer, &nSDLen, lpszMsgBuffer + sizeof(XENGINE_PROTOCOL_XMQ), nMsgLen - sizeof(XENGINE_PROTOCOL_XMQ));
442+
XEngine_MQXService_Send(pptszListAddr[i], tszSDBuffer, nSDLen, nClientType);
443+
BaseLib_OperatorMemory_Free((XPPPMEM)&pptszListAddr, nListCount);
444+
}
445+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("%s消息端:%s,主题:%s,序列:%lld,投递数据到消息队列成功,通知客户端个数:%d"), lpszClientType, lpszClientAddr, st_DBQueue.tszQueueName, st_DBQueue.nQueueSerial, nListCount);
446+
}
447+
else
448+
{
449+
int nListCount = 0;
450+
XENGINE_DBUSERKEY** ppSt_ListUser;
451+
if (DBModule_MQUser_KeyList(NULL, st_MQProtocol.tszMQKey, &ppSt_ListUser, &nListCount))
452+
{
453+
int nTCPLen = 0;
454+
XCHAR tszTCPBuffer[4096];
455+
memset(tszTCPBuffer, '\0', sizeof(tszTCPBuffer));
440456

441-
SessionModule_Client_GetAddr(ppSt_ListUser[i]->tszUserName, tszUserAddr);
442-
XEngine_MQXService_Send(tszUserAddr, tszTCPBuffer, nTCPLen, nNetType);
457+
pSt_ProtocolHdr->unOperatorCode = XENGINE_COMMUNICATION_PROTOCOL_OPERATOR_CODE_MQ_MSGNOTIFY;
458+
ProtocolModule_Packet_Common(nNetType, pSt_ProtocolHdr, &st_MQProtocol, tszTCPBuffer, &nTCPLen, lpszMsgBuffer + sizeof(XENGINE_PROTOCOL_XMQ), nMsgLen - sizeof(XENGINE_PROTOCOL_XMQ));
459+
for (int i = 0; i < nListCount; i++)
460+
{
461+
//跳过自己
462+
if (0 == _tcsxncmp(tszUserName, ppSt_ListUser[i]->tszUserName, _tcsxlen(tszUserName)))
463+
{
464+
continue;
465+
}
466+
XCHAR tszUserAddr[128];
467+
memset(tszUserAddr, '\0', sizeof(tszUserAddr));
468+
469+
SessionModule_Client_GetAddr(ppSt_ListUser[i]->tszUserName, tszUserAddr);
470+
XEngine_MQXService_Send(tszUserAddr, tszTCPBuffer, nTCPLen, nNetType);
471+
}
472+
BaseLib_OperatorMemory_Free((XPPPMEM)&ppSt_ListUser, nListCount);
443473
}
444-
BaseLib_OperatorMemory_Free((XPPPMEM)&ppSt_ListUser, nListCount);
474+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("%s消息端:%s,主题:%s,序列:%lld,投递数据到消息队列成功,通知客户端个数:%d"), lpszClientType, lpszClientAddr, st_DBQueue.tszQueueName, st_DBQueue.nQueueSerial, nListCount);
445475
}
446476
}
447-
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("%s消息端:%s,主题:%s,序列:%lld,投递数据到消息队列成功,通知客户端个数:%d"), lpszClientType, lpszClientAddr, st_DBQueue.tszQueueName, st_DBQueue.nQueueSerial, nListCount);
448-
449477
}
450478
else if (XENGINE_COMMUNICATION_PROTOCOL_OPERATOR_CODE_MQ_REQGET == pSt_ProtocolHdr->unOperatorCode)
451479
{

0 commit comments

Comments
 (0)