Skip to content

Commit f1aa6dc

Browse files
xengine-qytxengine-qyt
committed
improved:mqtt task
1 parent e8badfc commit f1aa6dc

File tree

4 files changed

+103
-36
lines changed

4 files changed

+103
-36
lines changed

‎XEngine_Source/XEngine_MQServiceApp/MQService_MQTTTask.cpp‎

Lines changed: 89 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -18,19 +18,22 @@ XHTHREAD CALLBACK MessageQueue_MQTTThread(XPVOID lParam)
1818
{
1919
for (int j = 0; j < ppSst_ListAddr[i]->nPktCount; j++)
2020
{
21-
MQTTPROTOCOL_INFORMATION st_MQTTProtcol = {};
21+
int nMSGLen = 0;
22+
XCHAR *ptszMSGBuffer = NULL;
23+
MQTTPROTOCOL_FIXEDHEADER st_MQTTHdr = {};
2224

23-
if (MQTTProtocol_Parse_Recv(ppSst_ListAddr[i]->tszClientAddr, &st_MQTTProtcol))
25+
if (MQTTProtocol_Parse_Recv(ppSst_ListAddr[i]->tszClientAddr, &st_MQTTHdr, &ptszMSGBuffer, &nMSGLen))
2426
{
25-
MQService_MQTT_Handle(ppSst_ListAddr[i]->tszClientAddr, &st_MQTTProtcol);
27+
MQService_MQTT_Handle(ppSst_ListAddr[i]->tszClientAddr, &st_MQTTHdr, ptszMSGBuffer, nMSGLen);
28+
BaseLib_OperatorMemory_FreeCStyle((XPPMEM)&ptszMSGBuffer);
2629
}
2730
}
2831
}
2932
BaseLib_OperatorMemory_Free((XPPPMEM)&ppSst_ListAddr, nListCount);
3033
}
3134
return 0;
3235
}
33-
void Packet_Property(XCHAR* ptszMsgBuffer, int* pInt_Len, MQTTPROTOCOL_HDRPROPERTY*** pppSt_HDRProperty, int nListCount)
36+
void Packet_Property(MQTTPROTOCOL_HDRPROPERTY*** pppSt_HDRProperty, int nListCount)
3437
{
3538
BaseLib_OperatorMemory_Malloc((XPPPMEM)pppSt_HDRProperty, nListCount, sizeof(MQTTPROTOCOL_HDRPROPERTY));
3639

@@ -58,7 +61,7 @@ void Packet_Property(XCHAR* ptszMsgBuffer, int* pInt_Len, MQTTPROTOCOL_HDRPROPER
5861
(*pppSt_HDRProperty)[5]->st_unValue.byValue = 1;
5962
(*pppSt_HDRProperty)[5]->byProFlag = XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_PROPERTY_WILDCARDSUBAVAI;
6063
}
61-
bool MQService_MQTT_Handle(LPCXSTR lpszClientAddr, MQTTPROTOCOL_INFORMATION* pSt_MQTTProtcol)
64+
bool MQService_MQTT_Handle(LPCXSTR lpszClientAddr, MQTTPROTOCOL_FIXEDHEADER* pSt_MQTTHdr, LPCXSTR lpszMSGBuffer, int nMSGLen)
6265
{
6366
int nSDLen = 0;
6467
int nRVLen = 0;
@@ -68,54 +71,110 @@ bool MQService_MQTT_Handle(LPCXSTR lpszClientAddr, MQTTPROTOCOL_INFORMATION* pSt
6871
memset(tszSDBuffer, '\0', sizeof(tszSDBuffer));
6972
memset(tszRVBuffer, '\0', sizeof(tszRVBuffer));
7073
//是不是连接
71-
if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_CONNECT == pSt_MQTTProtcol->st_FixedHdr.byMsgType)
74+
if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_CONNECT == pSt_MQTTHdr->byMsgType)
7275
{
73-
int nListCount = 6;
76+
int nListCount = 0;
7477
MQTTPROTOCOL_HDRPROPERTY** ppSt_HDRProperty;
78+
MQTTPROTOCOL_HDRCONNNECT st_HDRConnect = {};
79+
MQTTPROTOCOL_USERINFO st_USerInfo = {};
80+
81+
if (!MQTTProtocol_Parse_Connect(lpszMSGBuffer, nMSGLen, &st_HDRConnect, &st_USerInfo, &ppSt_HDRProperty, &nListCount))
82+
{
83+
//错误断开连接
84+
MQTTProtocol_Packet_DisConnect(tszRVBuffer, &nRVLen);
85+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_DISCONN, tszRVBuffer, nRVLen);
86+
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
87+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求链接失败,错误码:%lX"), lpszClientAddr, MQTTProtocol_GetLastError());
88+
return false;
89+
}
90+
BaseLib_OperatorMemory_Free((XPPPMEM)&ppSt_HDRProperty, nListCount);
7591

76-
Packet_Property(tszSDBuffer, &nSDLen, &ppSt_HDRProperty, nListCount);
92+
nListCount = 6;
93+
Packet_Property(&ppSt_HDRProperty, nListCount);
7794
MQTTProtocol_Packet_REPConnect(tszRVBuffer, &nRVLen, 0, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_REASON_SUCCESS, &ppSt_HDRProperty, nListCount);
78-
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, nRVLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_CONNACK, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_FLAG_CONNACK);
79-
memcpy(tszSDBuffer + nSDLen, tszRVBuffer, nRVLen);
80-
nSDLen += nRVLen;
95+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_CONNACK, tszRVBuffer, nRVLen);
8196

97+
BaseLib_OperatorMemory_Free((XPPPMEM)&ppSt_HDRProperty, nListCount);
8298
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
99+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求链接成功,客户端ID:%s,用户名:%s"), lpszClientAddr, st_USerInfo.tszClientID, st_USerInfo.tszClientUser);
83100
}
84-
else if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_SUBSCRIBE == pSt_MQTTProtcol->st_FixedHdr.byMsgType)
101+
else if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_SUBSCRIBE == pSt_MQTTHdr->byMsgType)
85102
{
86-
MQTTProtocol_Packet_REPComm(tszRVBuffer, &nRVLen, pSt_MQTTProtcol->wMsgID, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_REASON_SUCCESS);
87-
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, nRVLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_SUBACK, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_FLAG_SUBACK);
88-
memcpy(tszSDBuffer + nSDLen, tszRVBuffer, nRVLen);
89-
nSDLen += nRVLen;
103+
XSHOT wMsgID = 0;
104+
XCHAR tszTopicName[MAX_PATH] = {};
105+
int nListCount = 0;
106+
MQTTPROTOCOL_HDRPROPERTY** ppSt_HDRProperty;
107+
MQTTPROTOCOL_HDRSUBSCRIBE st_SubScribe = {};
108+
109+
if (!MQTTProtocol_Parse_Subscribe(lpszMSGBuffer, nMSGLen, &wMsgID, tszTopicName, &st_SubScribe, &ppSt_HDRProperty, &nListCount))
110+
{
111+
MQTTProtocol_Packet_DisConnect(tszRVBuffer, &nRVLen);
112+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_DISCONN, tszRVBuffer, nRVLen);
113+
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
114+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求订阅失败,错误码:%lX"), lpszClientAddr, MQTTProtocol_GetLastError());
115+
return false;
116+
}
117+
118+
MQTTProtocol_Packet_REPComm(tszRVBuffer, &nRVLen, wMsgID, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_REASON_SUCCESS);
119+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_SUBACK, tszRVBuffer, nRVLen);
90120

91121
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
122+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求订阅成功,主题名称:%s"), lpszClientAddr, tszTopicName);
92123
}
93-
else if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_UNSUBSCRIBE == pSt_MQTTProtcol->st_FixedHdr.byMsgType)
124+
else if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_UNSUBSCRIBE == pSt_MQTTHdr->byMsgType)
94125
{
95-
MQTTProtocol_Packet_REPComm(tszRVBuffer, &nRVLen, pSt_MQTTProtcol->wMsgID, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_REASON_SUCCESS);
96-
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, nRVLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_UNSUBACK, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_FLAG_UNSUBACK);
97-
memcpy(tszSDBuffer + nSDLen, tszRVBuffer, nRVLen);
98-
nSDLen += nRVLen;
126+
XSHOT wMsgID = 0;
127+
XCHAR tszTopicName[MAX_PATH] = {};
128+
int nListCount = 0;
129+
MQTTPROTOCOL_HDRPROPERTY** ppSt_HDRProperty;
130+
131+
if (!MQTTProtocol_Parse_UNSubcribe(lpszMSGBuffer, nMSGLen, &wMsgID, tszTopicName, &ppSt_HDRProperty, &nListCount))
132+
{
133+
MQTTProtocol_Packet_DisConnect(tszRVBuffer, &nRVLen);
134+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_DISCONN, tszRVBuffer, nRVLen);
135+
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
136+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求取消订阅失败,错误码:%lX"), lpszClientAddr, MQTTProtocol_GetLastError());
137+
return false;
138+
}
139+
MQTTProtocol_Packet_REPComm(tszRVBuffer, &nRVLen, wMsgID, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_REASON_SUCCESS);
140+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_UNSUBACK, tszRVBuffer, nRVLen);
99141

100142
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
143+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求取消订阅,主题名称:%s"), lpszClientAddr, tszTopicName);
101144
}
102-
else if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_PUBLISH == pSt_MQTTProtcol->st_FixedHdr.byMsgType)
145+
else if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_PUBLISH == pSt_MQTTHdr->byMsgType)
103146
{
104-
if ((XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_FLAG_PUBLISH_QOS1 == pSt_MQTTProtcol->st_FixedHdr.byMsgFlag) || (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_FLAG_PUBLISH_QOS2 == pSt_MQTTProtcol->st_FixedHdr.byMsgFlag))
147+
int nPLen = 0;
148+
int nListCount = 0;
149+
XSHOT wMsgID = 0;
150+
XCHAR tszTopicName[MAX_PATH] = {};
151+
XENGINE_PROTOCOLHDR st_ProtocolHdr = {};
152+
MQTTPROTOCOL_HDRPROPERTY** ppSt_HDRProperty;
153+
154+
if (!MQTTProtocol_Parse_Publish(lpszMSGBuffer, nMSGLen, pSt_MQTTHdr, tszTopicName, &wMsgID, tszRVBuffer, &nRVLen, &ppSt_HDRProperty, &nListCount))
155+
{
156+
MQTTProtocol_Packet_DisConnect(tszRVBuffer, &nRVLen);
157+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_DISCONN, tszRVBuffer, nRVLen);
158+
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
159+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求取消订阅失败,错误码:%lX"), lpszClientAddr, MQTTProtocol_GetLastError());
160+
return false;
161+
}
162+
if ((XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_FLAG_PUBLISH_QOS1 == pSt_MQTTHdr->byMsgFlag) || (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_FLAG_PUBLISH_QOS2 == pSt_MQTTHdr->byMsgFlag))
105163
{
106164
//需要回复
107-
MQTTProtocol_Packet_REPPublish(tszRVBuffer, &nRVLen, pSt_MQTTProtcol->wMsgID, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_REASON_SUCCESS);
108-
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, nRVLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_PUBACK, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_FLAG_PUBACK);
109-
memcpy(tszSDBuffer + nSDLen, tszRVBuffer, nRVLen);
110-
nSDLen += nRVLen;
111-
165+
MQTTProtocol_Packet_REPPublish(tszRVBuffer, &nRVLen, wMsgID, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_REASON_SUCCESS);
166+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_PUBACK, tszRVBuffer, nRVLen);
112167
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
113168
}
169+
//ProtocolModule_Parse_Websocket(tszRVBuffer, nRVLen, &st_ProtocolHdr, tszSDBuffer, &nSDLen);
170+
//MessageQueue_TCP_Handle(&st_ProtocolHdr, lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_WEBSOCKET);
171+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,发布消息,主题名称:%s,推送大小:%d"), lpszClientAddr, tszTopicName, nRVLen);
114172
}
115-
else if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_PINGREQ == pSt_MQTTProtcol->st_FixedHdr.byMsgType)
173+
else if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_PINGREQ == pSt_MQTTHdr->byMsgType)
116174
{
117-
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, 0, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_PINGREP, 0);
175+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_PINGREP);
118176
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
177+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求了心跳PING协议成功"));
119178
}
120179
return true;
121180
}

‎XEngine_Source/XEngine_MQServiceApp/MQService_MQTTTask.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
11
#pragma once
22

33
XHTHREAD CALLBACK MessageQueue_MQTTThread(XPVOID lParam);
4-
bool MQService_MQTT_Handle(LPCXSTR lpszClientAddr, MQTTPROTOCOL_INFORMATION *pSt_MQTTProtcol);
4+
bool MQService_MQTT_Handle(LPCXSTR lpszClientAddr, MQTTPROTOCOL_FIXEDHEADER* pSt_MQTTHdr, LPCXSTR lpszMSGBuffer, int nMSGLen);

‎XEngine_Source/XEngine_MQServiceApp/MQService_Net.cpp‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,7 @@ void XEngine_MQXService_Close(LPCXSTR lpszClientAddr, int nIPProto, bool bHeart)
117117
{
118118
MQTTProtocol_Parse_Delete(lpszClientAddr);
119119
NetCore_TCPXCore_CloseForClientEx(xhTCPSocket, lpszClientAddr);
120-
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("TCP客户端离开，TCP客户端地址：%s"), lpszClientAddr);
120+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端离开，MQTT客户端地址：%s"), lpszClientAddr);
121121
}
122122
XENGINE_PROTOCOL_USERINFO st_UserInfo;
123123
memset(&st_UserInfo, '\0', sizeof(XENGINE_PROTOCOL_USERINFO));
@@ -195,5 +195,13 @@ bool XEngine_MQXService_Send(LPCXSTR lpszClientAddr, LPCXSTR lpszMsgBuffer, int
195195
return false;
196196
}
197197
}
198+
else if (XENGINE_MQAPP_NETTYPE_MQTT == nIPProto)
199+
{
200+
if (!NetCore_TCPXCore_SendEx(xhMQTTSocket, lpszClientAddr, lpszMsgBuffer, nMsgLen))
201+
{
202+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _X("发送数据给MQTT客户端：%s，失败，错误：%lX"), lpszClientAddr, NetCore_GetLastError());
203+
return false;
204+
}
205+
}
198206
return true;
199207
}

‎XEngine_Source/XEngine_MQServiceApp/XEngine_MQServiceApp.cpp‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -296,14 +296,14 @@ int main(int argc, char** argv)
296296
}
297297
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("启动服务中，初始化MQTT组包成功，IO线程个数:%d"), st_ServiceCfg.st_XMax.nMQTTThread);
298298

299-
xhWSSocket = NetCore_TCPXCore_StartEx(st_ServiceCfg.nMQTTPort, st_ServiceCfg.st_XMax.nMaxClient, st_ServiceCfg.st_XMax.nMQTTThread);
300-
if (NULL == xhWSSocket)
299+
xhMQTTSocket = NetCore_TCPXCore_StartEx(st_ServiceCfg.nMQTTPort, st_ServiceCfg.st_XMax.nMaxClient, st_ServiceCfg.st_XMax.nMQTTThread);
300+
if (NULL == xhMQTTSocket)
301301
{
302302
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _X("启动MQTT网络服务器失败，错误：%lX"), NetCore_GetLastError());
303303
goto NETSERVICEEXIT;
304304
}
305305
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("启动服务中，启动MQTT网络服务器成功,MQTT端口:%d,IO:%d"), st_ServiceCfg.nMQTTPort, st_ServiceCfg.st_XMax.nMQTTThread);
306-
NetCore_TCPXCore_RegisterCallBackEx(xhWSSocket, MessageQueue_Callback_MQTTLogin, MessageQueue_Callback_MQTTRecv, MessageQueue_Callback_MQTTLeave);
306+
NetCore_TCPXCore_RegisterCallBackEx(xhMQTTSocket, MessageQueue_Callback_MQTTLogin, MessageQueue_Callback_MQTTRecv, MessageQueue_Callback_MQTTLeave);
307307
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("启动服务中，注册MQTT网络事件成功"));
308308

309309
BaseLib_OperatorMemory_Malloc((XPPPMEM)&ppSt_ListMQTTParam, st_ServiceCfg.st_XMax.nMQTTThread, sizeof(THREADPOOL_PARAMENT));
@@ -313,7 +313,7 @@ int main(int argc, char** argv)
313313

314314
*pInt_Pos = i;
315315
ppSt_ListMQTTParam[i]->lParam = pInt_Pos;
316-
ppSt_ListMQTTParam[i]->fpCall_ThreadsTask = MessageQueue_WebsocketThread;
316+
ppSt_ListMQTTParam[i]->fpCall_ThreadsTask = MessageQueue_MQTTThread;
317317
}
318318
xhWSPool = ManagePool_Thread_NQCreate(&ppSt_ListMQTTParam, st_ServiceCfg.st_XMax.nMQTTThread);
319319
if (NULL == xhWSPool)

0 commit comments

Comments
 (0)