Skip to content

mica-mqtt
mica-mqtt

java胎教生 / mica-mqtt
与 Fork 源项目一致

Fork自 mica / mica-mqtt

通知 5
Star 0
Fork 0
mica-mqtt
mica-mqtt

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 e1f6c182 编写于 作者: 浅梦2013's avatar 浅梦2013
浏览文件
操作

✨ mqtt 客户端日志优化。

上级 beca6336
隐藏空白更改
内联 并排
Showing with 44 addition and 28 deletion
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/DefaultMqttClientProcessor.java
浏览文件 @ e1f6c182
......@@ -22,7 +22,9 @@ import net.dreamlu.iot.mqtt.core.common.MqttPendingQos2Publish;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
import org.tio.core.Node;
import org.tio.core.Tio;
import org.tio.core.TioConfig;
import java.nio.ByteBuffer;
import java.util.List;
......@@ -60,7 +62,10 @@ public class DefaultMqttClientProcessor implements IMqttClientProcessor {
switch (returnCode) {
case CONNECTION_ACCEPTED:
connLatch.countDown();
logger.info("MqttClient connection succeeded!");
if (logger.isInfoEnabled()) {
Node node = context.getServerNode();
logger.info("MqttClient contextId:{} connection:{}:{} succeeded!", context.getId(), node.getIp(), node.getPort());
}
break;
case CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD:
case CONNECTION_REFUSED_IDENTIFIER_REJECTED:
......@@ -82,6 +87,9 @@ public class DefaultMqttClientProcessor implements IMqttClientProcessor {
if (paddingSubscribe == null) {
return;
}
if (logger.isInfoEnabled()) {
logger.info("MQTT Topic:{} successfully subscribed messageId:{}", paddingSubscribe.getTopicFilter(), messageId);
}
paddingSubscribe.onSubAckReceived();
clientStore.removePaddingSubscribe(messageId);
clientStore.addSubscription(paddingSubscribe.toSubscription());
......@@ -130,6 +138,9 @@ public class DefaultMqttClientProcessor implements IMqttClientProcessor {
if (pendingUnSubscription == null) {
return;
}
if (logger.isInfoEnabled()) {
logger.info("MQTT Topic:{} successfully unSubscribed messageId:{}", pendingUnSubscription.getTopic(), messageId);
}
pendingUnSubscription.onUnSubAckReceived();
clientStore.removePaddingUnSubscribe(messageId);
clientStore.removeSubscriptions(pendingUnSubscription.getTopic());
......@@ -143,6 +154,10 @@ public class DefaultMqttClientProcessor implements IMqttClientProcessor {
if (pendingPublish == null) {
return;
}
if (logger.isInfoEnabled()) {
String topicName = pendingPublish.getMessage().variableHeader().topicName();
logger.info("MQTT Topic:{} successfully PubAck messageId:{}", topicName, messageId);
}
pendingPublish.onPubAckReceived();
clientStore.removePendingPublish(messageId);
pendingPublish.getPayload().clear();
......@@ -188,6 +203,10 @@ public class DefaultMqttClientProcessor implements IMqttClientProcessor {
if (pendingPublish == null) {
return;
}
if (logger.isInfoEnabled()) {
String topicName = pendingPublish.getMessage().variableHeader().topicName();
logger.info("MQTT Topic:{} successfully PubComp", topicName);
}
pendingPublish.getPayload().clear();
pendingPublish.onPubCompReceived();
clientStore.removePendingPublish(messageId);
......
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClient.java
浏览文件 @ e1f6c182
......@@ -105,7 +105,7 @@ public final class MqttClient {
.build();
MqttPendingSubscription pendingSubscription = new MqttPendingSubscription(mqttQoS, topicFilter, listener, message);
Boolean result = Tio.send(context, message);
logger.debug("MQTT subscribe topicFilter:{} mqttQoS:{} messageId:{} result:{}", topicFilter, mqttQoS, messageId, result);
logger.info("MQTT Topic:{} mqttQoS:{} messageId:{} subscribing result:{}", topicFilter, mqttQoS, messageId, result);
pendingSubscription.startRetransmitTimer(executor, (msg) -> Tio.send(context, message));
clientStore.addPaddingSubscribe(messageId, pendingSubscription);
return this;
......@@ -125,7 +125,7 @@ public final class MqttClient {
.build();
MqttPendingUnSubscription pendingUnSubscription = new MqttPendingUnSubscription(topicFilter, message);
Boolean result = Tio.send(context, message);
logger.debug("MQTT unSubscribe topicFilter:{} messageId:{} result:{}", topicFilter, messageId, result);
logger.info("MQTT Topic:{} messageId:{} unSubscribing result:{}", topicFilter, messageId, result);
// 解绑 subManage listener
clientStore.addPaddingUnSubscribe(messageId, pendingUnSubscription);
pendingUnSubscription.startRetransmissionTimer(executor, msg -> Tio.send(context, msg));
......@@ -187,7 +187,7 @@ public final class MqttClient {
.messageId(messageId)
.build();
boolean result = Tio.send(context, message);
logger.debug("MQTT publish topic:{} qos:{} retain:{} result:{}", topic, qos, retain, result);
logger.info("MQTT Topic:{} qos:{} retain:{} publish result:{}", topic, qos, retain, result);
if (isHighLevelQoS) {
MqttPendingPublish pendingPublish = new MqttPendingPublish(payload, message, qos);
clientStore.addPendingPublish(messageId, pendingPublish);
......
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClientAioListener.java
浏览文件 @ e1f6c182
......@@ -102,7 +102,7 @@ public class MqttClientAioListener extends DefaultClientAioListener {
.build();
MqttPendingSubscription pendingSubscription = new MqttPendingSubscription(mqttQoS, topicFilter, subscription.getListener(), message);
Boolean result = Tio.send(context, message);
logger.info("MQTT reconnect subscribe topicFilter:{} mqttQoS:{} messageId:{} result:{}", topicFilter, mqttQoS, messageId, result);
logger.info("MQTT Topic:{} mqttQoS:{} messageId:{} resubscribing result:{}", topicFilter, mqttQoS, messageId, result);
pendingSubscription.startRetransmitTimer(executor, (msg) -> Tio.send(context, message));
clientStore.addPaddingSubscribe(messageId, pendingSubscription);
}
......
mica-mqtt-websocket/src/test/java/net/dreamlu/iot/mqtt/websocket/test/MqttWebServerAioHandler.java
浏览文件 @ e1f6c182
......@@ -223,11 +223,13 @@ public class MqttWebServerAioHandler implements ServerAioHandler {
*/
private static final String NOT_FINAL_WEBSOCKET_PACKET_PARTS = "TIO_N_F_W_P_P";
/**
* SEC_WEBSOCKET_KEY后缀
* SEC_WEBSOCKET_KEY 后缀
*/
private static final String SEC_WEBSOCKET_KEY_SUFFIX = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
private static final byte[] SEC_WEBSOCKET_KEY_SUFFIX_BYTES = SEC_WEBSOCKET_KEY_SUFFIX.getBytes();
/**
* websocket子协议
*/
private static final String Sec_Websocket_Protocol = "sec-websocket-protocol";
private static final HeaderName Header_Name_Sec_Websocket_Protocol = HeaderName.from(Sec_Websocket_Protocol);
......@@ -254,7 +256,7 @@ public class MqttWebServerAioHandler implements ServerAioHandler {
return null;
}
HttpResponse httpResponse = updateWebSocketProtocol(request);
// 并非 websocket
// 普通 http 非 websocket
if (httpResponse == null) {
return request;
}
......@@ -270,16 +272,16 @@ public class MqttWebServerAioHandler implements ServerAioHandler {
if (websocketPacket != null) {
// 数据包尚未完成
if (!websocketPacket.isWsEof()) {
List<WsRequest> parts = (List<WsRequest>) channelContext.getAttribute(NOT_FINAL_WEBSOCKET_PACKET_PARTS);
List<WsRequest> parts = (List<WsRequest>) channelContext.get(NOT_FINAL_WEBSOCKET_PACKET_PARTS);
if (parts == null) {
parts = new ArrayList<>();
channelContext.setAttribute(NOT_FINAL_WEBSOCKET_PACKET_PARTS, parts);
channelContext.set(NOT_FINAL_WEBSOCKET_PACKET_PARTS, parts);
}
parts.add(websocketPacket);
} else {
List<WsRequest> parts = (List<WsRequest>) channelContext.getAttribute(NOT_FINAL_WEBSOCKET_PACKET_PARTS);
List<WsRequest> parts = (List<WsRequest>) channelContext.get(NOT_FINAL_WEBSOCKET_PACKET_PARTS);
if (parts != null) {
channelContext.setAttribute(NOT_FINAL_WEBSOCKET_PACKET_PARTS, null);
channelContext.set(NOT_FINAL_WEBSOCKET_PACKET_PARTS, null);
parts.add(websocketPacket);
WsRequest first = parts.get(0);
......@@ -335,7 +337,6 @@ public class MqttWebServerAioHandler implements ServerAioHandler {
wsResponse = wsMsgHandler.encodeSubProtocol(packet, tioConfig, channelContext);
Objects.requireNonNull(wsResponse, "IWsMsgHandler encodeSubProtocol WsResponse is null.");
}
// 握手包
if (wsResponse.isHandShake()) {
WsSessionContext imSessionContext = (WsSessionContext) channelContext.get();
......@@ -350,7 +351,7 @@ public class MqttWebServerAioHandler implements ServerAioHandler {
return WsServerEncoder.encode(wsResponse, tioConfig, channelContext);
}
private WsResponse h(WsRequest websocketPacket, byte[] bytes, Opcode opcode, ChannelContext channelContext) throws Exception {
private WsResponse handlerWs(WsRequest websocketPacket, byte[] bytes, Opcode opcode, ChannelContext channelContext) throws Exception {
if (opcode == Opcode.TEXT) {
if (bytes == null || bytes.length == 0) {
Tio.remove(channelContext, "错误的websocket包，body为空");
......@@ -423,17 +424,13 @@ public class MqttWebServerAioHandler implements ServerAioHandler {
wsResponse.setHandShake(true);
Tio.send(channelContext, wsResponse);
wsSessionContext.setHandshaked(true);
wsMsgHandler.onAfterHandshaked(request, httpResponse, channelContext);
return;
}
if (!wsRequest.isWsEof()) {
return;
}
WsResponse wsResponse = h(wsRequest, wsRequest.getBody(), wsRequest.getWsOpcode(), channelContext);
WsResponse wsResponse = handlerWs(wsRequest, wsRequest.getBody(), wsRequest.getWsOpcode(), channelContext);
if (wsResponse != null) {
Tio.send(channelContext, wsResponse);
}
......@@ -468,21 +465,21 @@ public class MqttWebServerAioHandler implements ServerAioHandler {
*/
public HttpResponse updateWebSocketProtocol(HttpRequest request) {
Map<String, String> headers = request.getHeaders();
String Sec_WebSocket_Key = headers.get(HttpConst.RequestHeaderKey.Sec_WebSocket_Key);
String secWebSocketKey = headers.get(HttpConst.RequestHeaderKey.Sec_WebSocket_Key);
if (StrUtil.isNotBlank(Sec_WebSocket_Key)) {
byte[] Sec_WebSocket_Key_Bytes;
if (StrUtil.isNotBlank(secWebSocketKey)) {
byte[] secWebSocketKeyBytes;
try {
Sec_WebSocket_Key_Bytes = Sec_WebSocket_Key.getBytes(request.getCharset());
secWebSocketKeyBytes = secWebSocketKey.getBytes(request.getCharset());
} catch (UnsupportedEncodingException e) {
throw new RuntimeException(e);
}
byte[] allBs = new byte[Sec_WebSocket_Key_Bytes.length + SEC_WEBSOCKET_KEY_SUFFIX_BYTES.length];
System.arraycopy(Sec_WebSocket_Key_Bytes, 0, allBs, 0, Sec_WebSocket_Key_Bytes.length);
System.arraycopy(SEC_WEBSOCKET_KEY_SUFFIX_BYTES, 0, allBs, Sec_WebSocket_Key_Bytes.length, SEC_WEBSOCKET_KEY_SUFFIX_BYTES.length);
byte[] allBs = new byte[secWebSocketKeyBytes.length + SEC_WEBSOCKET_KEY_SUFFIX_BYTES.length];
System.arraycopy(secWebSocketKeyBytes, 0, allBs, 0, secWebSocketKeyBytes.length);
System.arraycopy(SEC_WEBSOCKET_KEY_SUFFIX_BYTES, 0, allBs, secWebSocketKeyBytes.length, SEC_WEBSOCKET_KEY_SUFFIX_BYTES.length);
byte[] key_array = SHA1Util.SHA1(allBs);
String acceptKey = BASE64Util.byteArrayToBase64(key_array);
byte[] keyArray = SHA1Util.SHA1(allBs);
String acceptKey = BASE64Util.byteArrayToBase64(keyArray);
HttpResponse httpResponse = new HttpResponse(request);
httpResponse.setStatus(HttpResponseStatus.C101);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或