dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Ack messages that we ignore

Browse Source
This commit is contained in:
Andrii Shvaika 2020-11-23 16:24:07 +02:00
parent 331c4d7b13
commit 4f0253a33a
1 changed files with 12 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
View File

@ -232,7 +232,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
if (topicName.startsWith(MqttTopics.BASE_GATEWAY_API_TOPIC)) { if (topicName.startsWith(MqttTopics.BASE_GATEWAY_API_TOPIC)) {
if (gatewaySessionHandler != null) { if (gatewaySessionHandler != null) {
handleGatewayPublishMsg(topicName, msgId, mqttMsg); handleGatewayPublishMsg(ctx, topicName, msgId, mqttMsg);
transportService.reportActivity(deviceSessionCtx.getSessionInfo()); transportService.reportActivity(deviceSessionCtx.getSessionInfo());
} }
} else { } else {
@ -240,7 +240,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
} }
} }
private void handleGatewayPublishMsg(String topicName, int msgId, MqttPublishMessage mqttMsg) { private void handleGatewayPublishMsg(ChannelHandlerContext ctx, String topicName, int msgId, MqttPublishMessage mqttMsg) {
try { try {
switch (topicName) { switch (topicName) {
case MqttTopics.GATEWAY_TELEMETRY_TOPIC: case MqttTopics.GATEWAY_TELEMETRY_TOPIC:
@ -264,6 +264,8 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
case MqttTopics.GATEWAY_DISCONNECT_TOPIC: case MqttTopics.GATEWAY_DISCONNECT_TOPIC:
gatewaySessionHandler.onDeviceDisconnect(mqttMsg); gatewaySessionHandler.onDeviceDisconnect(mqttMsg);
break; break;
default:
ack(ctx, msgId);
} }
} catch (RuntimeException | AdaptorException e) { } catch (RuntimeException | AdaptorException e) {
log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e); log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
@ -293,6 +295,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
transportService.process(deviceSessionCtx.getSessionInfo(), claimDeviceMsg, getPubAckCallback(ctx, msgId, claimDeviceMsg)); transportService.process(deviceSessionCtx.getSessionInfo(), claimDeviceMsg, getPubAckCallback(ctx, msgId, claimDeviceMsg));
} else { } else {
transportService.reportActivity(deviceSessionCtx.getSessionInfo()); transportService.reportActivity(deviceSessionCtx.getSessionInfo());
ack(ctx, msgId);
} }
} catch (AdaptorException e) { } catch (AdaptorException e) {
log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e); log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
@ -301,15 +304,18 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
} }
} }
private void ack(ChannelHandlerContext ctx, int msgId) {
if (msgId > 0) {
ctx.writeAndFlush(createMqttPubAckMsg(msgId));
}
}
private <T> TransportServiceCallback<Void> getPubAckCallback(final ChannelHandlerContext ctx, final int msgId, final T msg) { private <T> TransportServiceCallback<Void> getPubAckCallback(final ChannelHandlerContext ctx, final int msgId, final T msg) {
return new TransportServiceCallback<Void>() { return new TransportServiceCallback<Void>() {
@Override @Override
public void onSuccess(Void dummy) { public void onSuccess(Void dummy) {
log.trace("[{}] Published msg: {}", sessionId, msg); log.trace("[{}] Published msg: {}", sessionId, msg);
if (msgId > 0) { ack(ctx, msgId);
ctx.writeAndFlush(createMqttPubAckMsg(msgId));
}
} }
@Override @Override
@ -334,9 +340,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
@Override @Override
public void onSuccess(TransportProtos.ProvisionDeviceResponseMsg provisionResponseMsg) { public void onSuccess(TransportProtos.ProvisionDeviceResponseMsg provisionResponseMsg) {
log.trace("[{}] Published msg: {}", sessionId, msg); log.trace("[{}] Published msg: {}", sessionId, msg);
if (msgId > 0) { ack(ctx, msgId);
ctx.writeAndFlush(createMqttPubAckMsg(msgId));
}
try { try {
if (deviceSessionCtx.getProvisionPayloadType().equals(TransportPayloadType.JSON)) { if (deviceSessionCtx.getProvisionPayloadType().equals(TransportPayloadType.JSON)) {
deviceSessionCtx.getContext().getJsonMqttAdaptor().convertToPublish(deviceSessionCtx, provisionResponseMsg).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush); deviceSessionCtx.getContext().getJsonMqttAdaptor().convertToPublish(deviceSessionCtx, provisionResponseMsg).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);