Skip to content

Commit cfd3827

Browse files
xengine-qytxengine-qyt
committed
update:CHANGELOG
modify:mqtt protcol bind and unbind does not post tcp handle and mqtt protocol process modify
1 parent f95754b commit cfd3827

2 files changed

Lines changed: 125 additions & 0 deletions

File tree

‎CHANGELOG‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,39 @@
1+
XEngine_MQService V3.19.0.1001
2+
3+
增加:权限控制开关
4+
增加:数据传输加密支持
5+
更新:依赖库
6+
修改:mqtt协议的订阅和取消订阅不在交给tcp处理并且mqtt协议流程代码修改
7+
修改:当tcp登录的时候会创建http的token了
8+
修改:重构了tcp和http协议,现在大部分消息处理都通过http实现,tcp只处理投递订阅和获取消息
9+
修改:绑定消息协议字符串
10+
修改:协议消息统一
11+
修改:权限控制和处理
12+
修改:tcp登录权限处理
13+
修正:某些时候协议没有响应和解析问题
14+
修正:当post的协议找不到的时候没有响应问题
15+
修正:当没有找到http协议的时候无响应问题
16+
删除:http公用消息打包函数
17+
删除:多余的协议错误定义
18+
19+
added:permission control of build switch
20+
added:data transfer crypt
21+
update:depend library
22+
modify:mqtt protcol bind and unbind does not post tcp handle and mqtt protocol process modify
23+
modify:create token of http when tcp login
24+
modify:Reconstruct the TCP and HTTP protocols,now the data processed by the HTTP protocol is no longer handed over to the TCP for processing
25+
modify:message and topic modify and request message number to http protocol
26+
modify:does not case-sensitive
27+
modify:bind message protocol string
28+
modify:protocol user unified
29+
modify:permission management and process
30+
modify:tcp login permission handle
31+
fixed:some protocol not reply and parse process failure
32+
fixed:no reply when not found post protocol
33+
fixed:http protocol not reply when not found protocol
34+
delete:http comm message packet
35+
delete:more protocol error define
36+
======================================================================================
137
XEngine_MQService V3.18.0.1001
238

339
ci:增加CentOS和Alma以及debian13系统支持

‎XEngine_Source/XEngine_MQServiceApp/MQService_MQTTTask.cpp‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,95 @@ bool MQService_MQTT_Handle(LPCXSTR lpszClientAddr, MQTTPROTOCOL_FIXEDHEADER* pSt
108108

109109
MessageQueue_TCP_Handle(&st_ProtocolHdr, lpszClientAddr, (LPCXSTR)&st_ProtocolAuth, sizeof(XENGINE_PROTOCOL_USERAUTH), XENGINE_MQAPP_NETTYPE_MQTT);
110110
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求链接成功,客户端ID:%s,用户名:%s"), lpszClientAddr, st_USerInfo.tszClientID, st_USerInfo.tszClientUser);
111+
return true;
112+
}
113+
114+
XCHAR tszUserName[XPATH_MIN] = {};
115+
if (!SessionModule_Client_GetUser(lpszClientAddr, tszUserName))
116+
{
117+
MQTTProtocol_Packet_DisConnect(tszRVBuffer, &nRVLen);
118+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_DISCONN, tszRVBuffer, nRVLen);
119+
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
120+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求指定消息:%d 失败,用户未登录,错误码:%lX"), lpszClientAddr, pSt_MQTTHdr->byMsgType, MQTTProtocol_GetLastError());
121+
return false;
122+
}
123+
if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_SUBSCRIBE == pSt_MQTTHdr->byMsgType)
124+
{
125+
XSHOT wMsgID = 0;
126+
XCHAR tszTopicName[XPATH_MAX] = {};
127+
int nListCount = 0;
128+
MQTTPROTOCOL_HDRPROPERTY** ppSt_HDRProperty;
129+
MQTTPROTOCOL_HDRSUBSCRIBE st_SubScribe = {};
130+
131+
if (!MQTTProtocol_Parse_Subscribe(lpszMSGBuffer, nMSGLen, &wMsgID, tszTopicName, &st_SubScribe, &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+
XENGINE_DBUSERKEY st_Userkey = {};
140+
141+
_tcsxcpy(st_Userkey.tszUserName, tszUserName);
142+
_tcsxcpy(st_Userkey.tszKeyName, tszTopicName);
143+
//先查询有没有
144+
if (DBModule_MQUser_KeyQuery(&st_Userkey))
145+
{
146+
//有就更新
147+
st_Userkey.nKeySerial = wMsgID;
148+
if (!DBModule_MQUser_KeyUPDate(&st_Userkey))
149+
{
150+
MQTTProtocol_Packet_DisConnect(tszRVBuffer, &nRVLen);
151+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_DISCONN, tszRVBuffer, nRVLen);
152+
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
153+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _X("MQTT消息端:%s,设置消息队列主题更新失败,主题名称:%s,序列号:%d,错误：%lX"), lpszClientAddr, tszTopicName, wMsgID, DBModule_GetLastError());
154+
return false;
155+
}
156+
}
157+
else
158+
{
159+
//没有就创建
160+
st_Userkey.nKeySerial = wMsgID;
161+
if (!DBModule_MQUser_KeyInsert(&st_Userkey))
162+
{
163+
ProtocolModule_Packet_Http(tszSDBuffer, &nSDLen, ERROR_XENGINE_MESSAGE_HTTP_FAILURE, _X("insert bind topic is failure"));
164+
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_HTTP);
165+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _X("MQTT消息端:%s,设置消息队列主题创建失败,主题名称:%s,序列号:%d,错误：%lX"), lpszClientAddr, tszTopicName, wMsgID, DBModule_GetLastError());
166+
return false;
167+
}
168+
}
169+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求订阅成功,主题名称:%s"), lpszClientAddr, tszTopicName);
170+
}
171+
else if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_UNSUBSCRIBE == pSt_MQTTHdr->byMsgType)
172+
{
173+
XSHOT wMsgID = 0;
174+
XCHAR tszTopicName[XPATH_MAX] = {};
175+
int nListCount = 0;
176+
MQTTPROTOCOL_HDRPROPERTY** ppSt_HDRProperty;
177+
178+
if (!MQTTProtocol_Parse_UNSubcribe(lpszMSGBuffer, nMSGLen, &wMsgID, tszTopicName, &ppSt_HDRProperty, &nListCount))
179+
{
180+
MQTTProtocol_Packet_DisConnect(tszRVBuffer, &nRVLen);
181+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_DISCONN, tszRVBuffer, nRVLen);
182+
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
183+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求取消订阅失败,错误码:%lX"), lpszClientAddr, MQTTProtocol_GetLastError());
184+
return false;
185+
}
186+
XENGINE_DBUSERKEY st_Userkey = {};
187+
188+
_tcsxcpy(st_Userkey.tszUserName, tszUserName);
189+
_tcsxcpy(st_Userkey.tszKeyName, tszTopicName);
190+
191+
if (!DBModule_MQUser_KeyDelete(&st_Userkey))
192+
{
193+
MQTTProtocol_Packet_DisConnect(tszRVBuffer, &nRVLen);
194+
MQTTProtocol_Packet_Header(tszSDBuffer, &nSDLen, XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_DISCONN, tszRVBuffer, nRVLen);
195+
XEngine_MQXService_Send(lpszClientAddr, tszSDBuffer, nSDLen, XENGINE_MQAPP_NETTYPE_MQTT);
196+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_ERROR, _X("MQTT消息端:%s,解除消息绑定订阅失败,可能没有找到主题用户,主题名称:%s,用户名称:%s,错误：%lX"), lpszClientAddr, tszTopicName, tszUserName, DBModule_GetLastError());
197+
return false;
198+
}
199+
XLOG_PRINT(xhLog, XENGINE_HELPCOMPONENTS_XLOG_IN_LOGLEVEL_INFO, _X("MQTT客户端:%s,请求取消订阅,主题名称:%s"), lpszClientAddr, tszTopicName);
111200
}
112201
else if (XENGINE_RFCCOMPONENTS_MQTT_PROTOCOL_TYPE_PUBLISH == pSt_MQTTHdr->byMsgType)
113202
{

0 commit comments

Comments
 (0)