From 5c6928558bb455e82d2663d7b30dfe92ce986b29 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Thu, 16 May 2024 12:45:17 +0200 Subject: [PATCH 1/5] mqtt gateway refactoring due to OOM --- .../AbstractGatewaySessionHandler.java | 431 ++++++++++-------- .../session/SparkplugNodeSessionHandler.java | 77 ++-- 2 files changed, 264 insertions(+), 244 deletions(-) diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index d033774e93..49f2f1cc63 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -33,6 +33,7 @@ import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.mqtt.MqttMessage; import io.netty.handler.codec.mqtt.MqttPublishMessage; +import jakarta.annotation.Nullable; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; @@ -62,7 +63,6 @@ import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor; import org.thingsboard.server.transport.mqtt.util.ReturnCode; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState; -import jakarta.annotation.Nullable; import java.util.ArrayList; import java.util.Collections; import java.util.Date; @@ -99,6 +99,7 @@ public abstract class AbstractGatewaySessionHandler deviceCreationLockMap; private final ConcurrentMap devices; @@ -193,10 +194,6 @@ public abstract class AbstractGatewaySessionHandler deviceEntry : jsonObj.entrySet()) { - String deviceName = deviceEntry.getKey(); + if (!json.isJsonObject()) { + throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); + } + + for (Map.Entry deviceEntry : json.getAsJsonObject().entrySet()) { + if (!deviceEntry.getValue().isJsonArray()) { + log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); + continue; + } + + String deviceName = deviceEntry.getKey(); + T deviceCtx = devices.get(deviceName); + if (deviceCtx != null) { + processPostTelemetryMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId); + } else { Futures.addCallback(checkDeviceConnected(deviceName), new FutureCallback<>() { @Override public void onSuccess(@Nullable T deviceCtx) { - if (!deviceEntry.getValue().isJsonArray()) { - throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); - } - try { - TransportProtos.PostTelemetryMsg postTelemetryMsg = JsonConverter.convertToTelemetryProto(deviceEntry.getValue().getAsJsonArray()); - processPostTelemetryMsg(deviceCtx, postTelemetryMsg, deviceName, msgId); - } catch (Throwable e) { - log.warn("[{}][{}][{}] Failed to convert telemetry: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, deviceEntry.getValue(), e); - channel.close(); - } + processPostTelemetryMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId); } @Override @@ -394,8 +393,16 @@ public abstract class AbstractGatewaySessionHandler deviceMsgList = telemetryMsgProto.getMsgList(); - if (!CollectionUtils.isEmpty(deviceMsgList)) { - deviceMsgList.forEach(telemetryMsg -> { - String deviceName = checkDeviceName(telemetryMsg.getDeviceName()); - Futures.addCallback(checkDeviceConnected(deviceName), - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - TransportProtos.PostTelemetryMsg msg = telemetryMsg.getMsg(); - try { - TransportProtos.PostTelemetryMsg postTelemetryMsg = ProtoConverter.validatePostTelemetryMsg(msg.toByteArray()); - processPostTelemetryMsg(deviceCtx, postTelemetryMsg, deviceName, msgId); - } catch (Throwable e) { - log.warn("[{}][{}][{}] Failed to convert telemetry: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, msg, e); - channel.close(); - } - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device telemetry command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - }); - } else { + if (CollectionUtils.isEmpty(deviceMsgList)) { log.debug("[{}][{}][{}] Devices telemetry messages is empty", gateway.getTenantId(), gateway.getDeviceId(), sessionId); throw new IllegalArgumentException("[" + sessionId + "] Devices telemetry messages is empty for [" + gateway.getDeviceId() + "]"); } + + deviceMsgList.forEach(telemetryMsg -> { + String deviceName = checkDeviceName(telemetryMsg.getDeviceName()); + T deviceCtx = devices.get(deviceName); + if (deviceCtx != null) { + processPostTelemetryMsg(deviceCtx, telemetryMsg.getMsg(), deviceName, msgId); + } else { + Futures.addCallback(checkDeviceConnected(deviceName), new FutureCallback<>() { + @Override + public void onSuccess(@Nullable T deviceCtx) { + processPostTelemetryMsg(deviceCtx, telemetryMsg.getMsg(), deviceName, msgId); + } + + @Override + public void onFailure(Throwable t) { + log.debug("[{}][{}][{}] Failed to process device telemetry command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); + } + }, context.getExecutor()); + } + }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); } } - public void processPostTelemetryMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.PostTelemetryMsg postTelemetryMsg, String deviceName, int msgId) { - transportService.process(deviceCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(channel, deviceName, msgId, postTelemetryMsg)); + protected void processPostTelemetryMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.PostTelemetryMsg msg, String deviceName, int msgId) { + try { + TransportProtos.PostTelemetryMsg postTelemetryMsg = ProtoConverter.validatePostTelemetryMsg(msg.toByteArray()); + transportService.process(deviceCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(channel, deviceName, msgId, postTelemetryMsg)); + } catch (Throwable e) { + log.warn("[{}][{}][{}] Failed to convert telemetry: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, msg, e); + channel.close(); + } } public TransportProtos.PostTelemetryMsg postTelemetryMsgCreated(TransportProtos.KeyValueProto keyValueProto, long ts) { @@ -452,34 +462,44 @@ public abstract class AbstractGatewaySessionHandler deviceEntry : jsonObj.entrySet()) { - String deviceName = deviceEntry.getKey(); - Futures.addCallback(checkDeviceConnected(deviceName), - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - if (!deviceEntry.getValue().isJsonObject()) { - throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); - } - try { - DeviceId deviceId = deviceCtx.getDeviceId(); - TransportProtos.ClaimDeviceMsg claimDeviceMsg = JsonConverter.convertToClaimDeviceProto(deviceId, deviceEntry.getValue()); - processClaimDeviceMsg(deviceCtx, claimDeviceMsg, deviceName, msgId); - } catch (Throwable e) { - log.warn("[{}][{}][{}] Failed to convert claim message: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, deviceEntry.getValue(), e); - } - } + if (!json.isJsonObject()) { + throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); + } + JsonObject jsonObj = json.getAsJsonObject(); + for (Map.Entry deviceEntry : jsonObj.entrySet()) { + if (!deviceEntry.getValue().isJsonObject()) { + log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); + continue; + } - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device claiming command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); + String deviceName = deviceEntry.getKey(); + T deviceCtx = devices.get(deviceEntry.getKey()); + + if (deviceCtx != null) { + processClaimDeviceMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId); + } else { + Futures.addCallback(checkDeviceConnected(deviceName), new FutureCallback<>() { + @Override + public void onSuccess(@Nullable T deviceCtx) { + processClaimDeviceMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId); + } + + @Override + public void onFailure(Throwable t) { + log.debug("[{}][{}][{}] Failed to process device claiming command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); + } + }, context.getExecutor()); } - } else { - throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); + } + } + + private void processClaimDeviceMsg(MqttDeviceAwareSessionContext deviceCtx, JsonElement claimRequest, String deviceName, int msgId) { + try { + DeviceId deviceId = deviceCtx.getDeviceId(); + TransportProtos.ClaimDeviceMsg claimDeviceMsg = JsonConverter.convertToClaimDeviceProto(deviceId, claimRequest); + transportService.process(deviceCtx.getSessionInfo(), claimDeviceMsg, getPubAckCallback(channel, deviceName, msgId, claimDeviceMsg)); + } catch (Throwable e) { + log.warn("[{}][{}][{}] Failed to convert claim message: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, claimRequest, e); } } @@ -487,113 +507,120 @@ public abstract class AbstractGatewaySessionHandler claimMsgList = claimMsgProto.getMsgList(); - if (!CollectionUtils.isEmpty(claimMsgList)) { - claimMsgList.forEach(claimDeviceMsg -> { - String deviceName = checkDeviceName(claimDeviceMsg.getDeviceName()); - Futures.addCallback(checkDeviceConnected(deviceName), - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - TransportApiProtos.ClaimDevice claimRequest = claimDeviceMsg.getClaimRequest(); - if (claimRequest == null) { - throw new IllegalArgumentException("Claim request for device: " + deviceName + " is null!"); - } - try { - DeviceId deviceId = deviceCtx.getDeviceId(); - TransportProtos.ClaimDeviceMsg claimDeviceMsg = ProtoConverter.convertToClaimDeviceProto(deviceId, claimRequest.toByteArray()); - processClaimDeviceMsg(deviceCtx, claimDeviceMsg, deviceName, msgId); - } catch (Throwable e) { - log.warn("[{}][{}][{}] Failed to convert claim message: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, claimRequest, e); - } - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device claiming command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - }); - } else { + if (CollectionUtils.isEmpty(claimMsgList)) { log.debug("[{}][{}][{}] Devices claim messages is empty", gateway.getTenantId(), gateway.getDeviceId(), sessionId); throw new IllegalArgumentException("[" + sessionId + "] Devices claim messages is empty for [" + gateway.getDeviceId() + "]"); } + + claimMsgList.forEach(claimDeviceMsg -> { + String deviceName = checkDeviceName(claimDeviceMsg.getDeviceName()); + T deviceCtx = devices.get(deviceName); + if (deviceCtx != null) { + processClaimDeviceMsg(deviceCtx, claimDeviceMsg.getClaimRequest(), deviceName, msgId); + } else { + Futures.addCallback(checkDeviceConnected(deviceName), new FutureCallback<>() { + @Override + public void onSuccess(@Nullable T deviceCtx) { + processClaimDeviceMsg(deviceCtx, claimDeviceMsg.getClaimRequest(), deviceName, msgId); + } + + @Override + public void onFailure(Throwable t) { + log.debug("[{}][{}][{}] Failed to process device claiming command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); + } + }, context.getExecutor()); + } + }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); } } - private void processClaimDeviceMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.ClaimDeviceMsg claimDeviceMsg, String deviceName, int msgId) { - transportService.process(deviceCtx.getSessionInfo(), claimDeviceMsg, getPubAckCallback(channel, deviceName, msgId, claimDeviceMsg)); + private void processClaimDeviceMsg(MqttDeviceAwareSessionContext deviceCtx, TransportApiProtos.ClaimDevice claimRequest, String deviceName, int msgId) { + try { + DeviceId deviceId = deviceCtx.getDeviceId(); + TransportProtos.ClaimDeviceMsg claimDeviceMsg = ProtoConverter.convertToClaimDeviceProto(deviceId, claimRequest.toByteArray()); + transportService.process(deviceCtx.getSessionInfo(), claimDeviceMsg, getPubAckCallback(channel, deviceName, msgId, claimDeviceMsg)); + } catch (Throwable e) { + log.warn("[{}][{}][{}] Failed to convert claim message: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, claimRequest, e); + } } private void onDeviceAttributesJson(int msgId, ByteBuf payload) throws AdaptorException { JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, payload); - if (json.isJsonObject()) { - JsonObject jsonObj = json.getAsJsonObject(); - for (Map.Entry deviceEntry : jsonObj.entrySet()) { - String deviceName = deviceEntry.getKey(); - Futures.addCallback(checkDeviceConnected(deviceName), - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - if (!deviceEntry.getValue().isJsonObject()) { - throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); - } - TransportProtos.PostAttributeMsg postAttributeMsg = JsonConverter.convertToAttributesProto(deviceEntry.getValue().getAsJsonObject()); - processPostAttributesMsg(deviceCtx, postAttributeMsg, deviceName, msgId); - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - } - } else { + if (!json.isJsonObject()) { throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); } + JsonObject jsonObj = json.getAsJsonObject(); + for (Map.Entry deviceEntry : jsonObj.entrySet()) { + if (!deviceEntry.getValue().isJsonObject()) { + log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); + continue; + } + String deviceName = deviceEntry.getKey(); + Futures.addCallback(checkDeviceConnected(deviceName), new FutureCallback<>() { + @Override + public void onSuccess(@Nullable T deviceCtx) { + processPostAttributesMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId); + } + + @Override + public void onFailure(Throwable t) { + log.debug("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); + } + }, context.getExecutor()); + } } private void onDeviceAttributesProto(int msgId, ByteBuf payload) throws AdaptorException { try { TransportApiProtos.GatewayAttributesMsg attributesMsgProto = TransportApiProtos.GatewayAttributesMsg.parseFrom(getBytes(payload)); List attributesMsgList = attributesMsgProto.getMsgList(); - if (!CollectionUtils.isEmpty(attributesMsgList)) { - attributesMsgList.forEach(attributesMsg -> { - String deviceName = checkDeviceName(attributesMsg.getDeviceName()); - Futures.addCallback(checkDeviceConnected(deviceName), - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - TransportProtos.PostAttributeMsg kvListProto = attributesMsg.getMsg(); - if (kvListProto == null) { - throw new IllegalArgumentException("Attributes List for device: " + deviceName + " is empty!"); - } - try { - TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto); - processPostAttributesMsg(deviceCtx, postAttributeMsg, deviceName, msgId); - } catch (Throwable e) { - log.warn("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, kvListProto, e); - } - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - }); - } else { + if (CollectionUtils.isEmpty(attributesMsgList)) { log.debug("[{}][{}][{}] Devices attributes keys list is empty", gateway.getTenantId(), gateway.getDeviceId(), sessionId); throw new IllegalArgumentException("[" + sessionId + "] Devices attributes keys list is empty for [" + gateway.getDeviceId() + "]"); } + + attributesMsgList.forEach(attributesMsg -> { + String deviceName = checkDeviceName(attributesMsg.getDeviceName()); + T deviceCtx = devices.get(deviceName); + if (deviceCtx != null) { + processPostAttributesMsg(deviceCtx, attributesMsg.getMsg(), deviceName, msgId); + } else { + Futures.addCallback(checkDeviceConnected(deviceName), new FutureCallback<>() { + @Override + public void onSuccess(@Nullable T deviceCtx) { + processPostAttributesMsg(deviceCtx, attributesMsg.getMsg(), deviceName, msgId); + } + + @Override + public void onFailure(Throwable t) { + log.debug("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); + } + }, context.getExecutor()); + } + }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); } } - protected void processPostAttributesMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.PostAttributeMsg postAttributeMsg, String deviceName, int msgId) { - transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(channel, deviceName, msgId, postAttributeMsg)); + protected void processPostAttributesMsg(MqttDeviceAwareSessionContext deviceCtx, JsonElement msg, String deviceName, int msgId) { + try { + TransportProtos.PostAttributeMsg postAttributeMsg = JsonConverter.convertToAttributesProto(msg.getAsJsonObject()); + transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(channel, deviceName, msgId, postAttributeMsg)); + } catch (Throwable e) { + log.warn("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, msg, e); + } + } + + protected void processPostAttributesMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.PostAttributeMsg kvListProto, String deviceName, int msgId) { + try { + TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto); + transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(channel, deviceName, msgId, postAttributeMsg)); + } catch (Throwable e) { + log.warn("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, kvListProto, e); + } } private void onDeviceAttributesRequestJson(MqttPublishMessage msg) throws AdaptorException { @@ -637,74 +664,76 @@ public abstract class AbstractGatewaySessionHandler() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - Integer requestId = jsonObj.get("id").getAsInt(); - String data = jsonObj.get("data").toString(); - TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder() - .setRequestId(requestId).setPayload(data).build(); - processRpcResponseMsg(deviceCtx, rpcResponseMsg, deviceName, msgId); - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device Rpc response command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - } else { + if (!json.isJsonObject()) { throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); } + JsonObject jsonObj = json.getAsJsonObject(); + String deviceName = jsonObj.get(DEVICE_PROPERTY).getAsString(); + Integer requestId = jsonObj.get("id").getAsInt(); + String data = jsonObj.get("data").toString(); + onDeviceRpcResponse(requestId, data, deviceName, msgId); } private void onDeviceRpcResponseProto(int msgId, ByteBuf payload) throws AdaptorException { try { TransportApiProtos.GatewayRpcResponseMsg gatewayRpcResponseMsg = TransportApiProtos.GatewayRpcResponseMsg.parseFrom(getBytes(payload)); String deviceName = checkDeviceName(gatewayRpcResponseMsg.getDeviceName()); - Futures.addCallback(checkDeviceConnected(deviceName), - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - Integer requestId = gatewayRpcResponseMsg.getId(); - String data = gatewayRpcResponseMsg.getData(); - TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder() - .setRequestId(requestId).setPayload(data).build(); - processRpcResponseMsg(deviceCtx, rpcResponseMsg, deviceName, msgId); - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device Rpc response command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); + Integer requestId = gatewayRpcResponseMsg.getId(); + String data = gatewayRpcResponseMsg.getData(); + onDeviceRpcResponse(requestId, data, deviceName, msgId); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); } } - private void processRpcResponseMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg, String deviceName, int msgId) { + private void onDeviceRpcResponse(Integer requestId, String data, String deviceName, int msgId) { + T deviceCtx = devices.get(deviceName); + if (deviceCtx != null) { + processRpcResponseMsg(deviceCtx, requestId, data, deviceName, msgId); + } else { + Futures.addCallback(checkDeviceConnected(deviceName), new FutureCallback<>() { + @Override + public void onSuccess(@Nullable T deviceCtx) { + processRpcResponseMsg(deviceCtx, requestId, data, deviceName, msgId); + } + + @Override + public void onFailure(Throwable t) { + log.debug("[{}][{}][{}] Failed to process device Rpc response command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); + } + }, context.getExecutor()); + } + } + + private void processRpcResponseMsg(MqttDeviceAwareSessionContext deviceCtx, Integer requestId, String data, String deviceName, int msgId) { + TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder() + .setRequestId(requestId).setPayload(data).build(); transportService.process(deviceCtx.getSessionInfo(), rpcResponseMsg, getPubAckCallback(channel, deviceName, msgId, rpcResponseMsg)); } private void processGetAttributeRequestMessage(MqttPublishMessage mqttMsg, String deviceName, TransportProtos.GetAttributeRequestMsg requestMsg) { int msgId = getMsgId(mqttMsg); - Futures.addCallback(checkDeviceConnected(deviceName), - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - transportService.process(deviceCtx.getSessionInfo(), requestMsg, getPubAckCallback(channel, deviceName, msgId, requestMsg)); - } + T deviceCtx = devices.get(deviceName); + if (deviceCtx != null) { + processGetAttributeRequestMessage(deviceCtx, requestMsg, deviceName, msgId); + } else { + Futures.addCallback(checkDeviceConnected(deviceName), new FutureCallback<>() { + @Override + public void onSuccess(@Nullable T deviceCtx) { + processGetAttributeRequestMessage(deviceCtx, requestMsg, deviceName, msgId); + } - @Override - public void onFailure(Throwable t) { - ack(mqttMsg, ReturnCode.IMPLEMENTATION_SPECIFIC); - log.debug("[{}][{}][{}] Failed to process device attributes request command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); + @Override + public void onFailure(Throwable t) { + ack(mqttMsg, ReturnCode.IMPLEMENTATION_SPECIFIC); + log.debug("[{}][{}][{}] Failed to process device attributes request command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); + } + }, context.getExecutor()); + } + } + + private void processGetAttributeRequestMessage(T deviceCtx, TransportProtos.GetAttributeRequestMsg requestMsg, String deviceName, int msgId) { + transportService.process(deviceCtx.getSessionInfo(), requestMsg, getPubAckCallback(channel, deviceName, msgId, requestMsg)); } private TransportProtos.GetAttributeRequestMsg toGetAttributeRequestMsg(int requestId, boolean clientScope, Set keys) { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java index 1f389fced1..cbc515c1d0 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java @@ -26,6 +26,7 @@ import io.netty.handler.codec.mqtt.MqttMessage; import io.netty.handler.codec.mqtt.MqttPublishMessage; import io.netty.handler.codec.mqtt.MqttQoS; import io.netty.handler.codec.mqtt.MqttTopicSubscription; +import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.eclipse.leshan.core.ResponseCode; import org.springframework.util.CollectionUtils; @@ -44,6 +45,7 @@ import org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopic; import jakarta.annotation.Nullable; + import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -68,6 +70,7 @@ import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopi @Slf4j public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler { + @Getter private final SparkplugTopic sparkplugTopicNode; private final Map nodeBirthMetrics; private final MqttTransportHandler parent; @@ -136,41 +139,36 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler contextListenableFuture, int msgId, List postTelemetryMsgList, String deviceName) throws AdaptorException { - try { - int finalMsgId = msgId; - postTelemetryMsgList.forEach(telemetryMsg -> { - Futures.addCallback(contextListenableFuture, - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable MqttDeviceAwareSessionContext deviceCtx) { - try { - processPostTelemetryMsg(deviceCtx, telemetryMsg, deviceName, finalMsgId); - } catch (Throwable e) { - log.warn("[{}][{}] Failed to convert telemetry: {}", gateway.getDeviceId(), deviceName, telemetryMsg, e); - channel.close(); - } - } + Futures.addCallback(contextListenableFuture, new FutureCallback<>() { + @Override + public void onSuccess(@Nullable MqttDeviceAwareSessionContext deviceCtx) { + for (TransportProtos.PostTelemetryMsg telemetryMsg : postTelemetryMsgList) { + try { + processPostTelemetryMsg(deviceCtx, telemetryMsg, deviceName, msgId); + } catch (Throwable e) { + log.warn("[{}][{}] Failed to convert telemetry: {}", gateway.getDeviceId(), deviceName, telemetryMsg, e); + channel.close(); + break; + } + } + } - @Override - public void onFailure(Throwable t) { - log.debug("[{}] Failed to process device telemetry command: {}", sessionId, deviceName, t); - } - }, context.getExecutor()); - }); - } catch (RuntimeException e) { - throw new AdaptorException(e); - } + @Override + public void onFailure(Throwable t) { + log.debug("[{}] Failed to process device telemetry command: {}", sessionId, deviceName, t); + } + }, context.getExecutor()); } private void onDeviceAttributesProto(ListenableFuture contextListenableFuture, int msgId, List attributesMsgList, String deviceName) throws AdaptorException { try { if (!CollectionUtils.isEmpty(attributesMsgList)) { - attributesMsgList.forEach(attributesMsg -> { - Futures.addCallback(contextListenableFuture, - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable MqttDeviceAwareSessionContext deviceCtx) { + Futures.addCallback(contextListenableFuture, + new FutureCallback<>() { + @Override + public void onSuccess(@Nullable MqttDeviceAwareSessionContext deviceCtx) { + for (TransportApiProtos.AttributesMsg attributesMsg : attributesMsgList) { TransportProtos.PostAttributeMsg kvListProto = attributesMsg.getMsg(); try { TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto); @@ -179,13 +177,13 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler grantedQoSList, MqttTopicSubscription subscription, - MqttQoS reqQoS) throws ThingsboardException, AdaptorException, - ExecutionException, InterruptedException { - SparkplugTopic sparkplugTopic = parseTopicSubscribe(subscription.topicName()); + MqttQoS reqQoS) throws ThingsboardException { + SparkplugTopic sparkplugTopic = parseTopicSubscribe(subscription.topicFilter()); if (sparkplugTopic.getGroupId() == null) { // TODO SUBSCRIBE NameSpace } else if (sparkplugTopic.getType() == null) { @@ -304,10 +301,6 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler createSparkplugMqttPublishMsg(TransportProtos.TsKvProto tsKvProto, String sparkplugTopic, SparkplugBProto.Payload.Metric metricBirth) { @@ -328,7 +321,6 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler Date: Thu, 16 May 2024 18:45:50 +0200 Subject: [PATCH 2/5] added ackOrClose instead of chanel.close and refactoring --- .../AbstractGatewaySessionHandler.java | 54 ++++++++++++------- 1 file changed, 34 insertions(+), 20 deletions(-) diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index 49f2f1cc63..b4644ac3ac 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -32,7 +32,9 @@ import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.mqtt.MqttMessage; +import io.netty.handler.codec.mqtt.MqttMessageBuilders; import io.netty.handler.codec.mqtt.MqttPublishMessage; +import io.netty.handler.codec.mqtt.MqttVersion; import jakarta.annotation.Nullable; import lombok.Getter; import lombok.Setter; @@ -83,6 +85,7 @@ import static org.thingsboard.server.common.transport.service.DefaultTransportSe import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_OPEN; import static org.thingsboard.server.common.transport.service.DefaultTransportService.SUBSCRIBE_TO_ATTRIBUTE_UPDATES_ASYNC_MSG; import static org.thingsboard.server.common.transport.service.DefaultTransportService.SUBSCRIBE_TO_RPC_ASYNC_MSG; +import static org.thingsboard.server.transport.mqtt.util.ReturnCode.PAYLOAD_FORMAT_INVALID; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.OFFLINE; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.STATE; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.messageName; @@ -365,10 +368,7 @@ public abstract class AbstractGatewaySessionHandler deviceEntry : json.getAsJsonObject().entrySet()) { if (!deviceEntry.getValue().isJsonArray()) { log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); @@ -402,7 +402,7 @@ public abstract class AbstractGatewaySessionHandler deviceEntry : jsonObj.entrySet()) { + validateJsonObject(json); + for (Map.Entry deviceEntry : json.getAsJsonObject().entrySet()) { if (!deviceEntry.getValue().isJsonObject()) { log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); continue; @@ -500,6 +497,7 @@ public abstract class AbstractGatewaySessionHandler deviceEntry : jsonObj.entrySet()) { + validateJsonObject(json); + for (Map.Entry deviceEntry : json.getAsJsonObject().entrySet()) { if (!deviceEntry.getValue().isJsonObject()) { log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); continue; @@ -611,6 +607,7 @@ public abstract class AbstractGatewaySessionHandler 0) { writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(deviceSessionCtx, msgId, returnCode)); } } + private void ackOrClose(int msgId) { + if (MqttVersion.MQTT_5.equals(deviceSessionCtx.getMqttVersion())) { + ack(msgId, PAYLOAD_FORMAT_INVALID); + } else { + channel.close(); + } + } + private void deregisterSession(String deviceName, MqttDeviceAwareSessionContext deviceSessionCtx) { if (this.deviceSessionCtx.isSparkplug()) { sendSparkplugStateOnTelemetry(deviceSessionCtx.getSessionInfo(), From 0fe3be153751c343e595fa2d88dc36cfd9d65444 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Thu, 16 May 2024 20:16:04 +0200 Subject: [PATCH 3/5] Used DonAsynchron instead of addCallback --- .../AbstractGatewaySessionHandler.java | 245 +++++------------- .../session/SparkplugNodeSessionHandler.java | 84 +++--- 2 files changed, 95 insertions(+), 234 deletions(-) diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index b4644ac3ac..1eaf1f38e8 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -15,9 +15,7 @@ */ package org.thingsboard.server.transport.mqtt.session; - import com.fasterxml.jackson.databind.JsonNode; -import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.SettableFuture; @@ -32,15 +30,15 @@ import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.mqtt.MqttMessage; -import io.netty.handler.codec.mqtt.MqttMessageBuilders; import io.netty.handler.codec.mqtt.MqttPublishMessage; import io.netty.handler.codec.mqtt.MqttVersion; -import jakarta.annotation.Nullable; import lombok.Getter; import lombok.Setter; +import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.springframework.util.CollectionUtils; import org.springframework.util.ConcurrentReferenceHashMap; +import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.server.common.adaptor.AdaptorException; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.adaptor.ProtoConverter; @@ -78,6 +76,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Consumer; import static org.springframework.util.ConcurrentReferenceHashMap.ReferenceType; import static org.thingsboard.server.common.data.DataConstants.DEFAULT_DEVICE_TYPE; @@ -224,18 +223,12 @@ public abstract class AbstractGatewaySessionHandler() { - @Override - public void onSuccess(@Nullable T result) { - ack(msg, ReturnCode.SUCCESS); - log.trace("[{}][{}][{}] onDeviceConnectOk: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName); - } - - @Override - public void onFailure(Throwable t) { - logDeviceCreationError(t, deviceName); - } - }, context.getExecutor()); + process(onDeviceConnect(deviceName, deviceType), + result -> { + ack(msg, ReturnCode.SUCCESS); + log.trace("[{}][{}][{}] onDeviceConnectOk: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName); + }, + t -> logDeviceCreationError(t, deviceName)); } public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional deviceProfileOpt) { @@ -374,25 +367,9 @@ public abstract class AbstractGatewaySessionHandler() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - processPostTelemetryMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId); - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device telemetry command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - } + process(deviceName, deviceCtx -> processPostTelemetryMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), + t -> log.debug("[{}][{}][{}] Failed to process device telemetry command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); } } @@ -417,22 +394,8 @@ public abstract class AbstractGatewaySessionHandler { String deviceName = checkDeviceName(telemetryMsg.getDeviceName()); - T deviceCtx = devices.get(deviceName); - if (deviceCtx != null) { - processPostTelemetryMsg(deviceCtx, telemetryMsg.getMsg(), deviceName, msgId); - } else { - Futures.addCallback(checkDeviceConnected(deviceName), new FutureCallback<>() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - processPostTelemetryMsg(deviceCtx, telemetryMsg.getMsg(), deviceName, msgId); - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device telemetry command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - } + process(deviceName, deviceCtx -> processPostTelemetryMsg(deviceCtx, telemetryMsg.getMsg(), deviceName, msgId), + t -> log.debug("[{}][{}][{}] Failed to process device telemetry command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); @@ -470,23 +433,8 @@ public abstract class AbstractGatewaySessionHandler() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - processClaimDeviceMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId); - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device claiming command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - } + process(deviceName, deviceCtx -> processClaimDeviceMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), + t -> log.debug("[{}][{}][{}] Failed to process device claiming command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); } } @@ -512,22 +460,8 @@ public abstract class AbstractGatewaySessionHandler { String deviceName = checkDeviceName(claimDeviceMsg.getDeviceName()); - T deviceCtx = devices.get(deviceName); - if (deviceCtx != null) { - processClaimDeviceMsg(deviceCtx, claimDeviceMsg.getClaimRequest(), deviceName, msgId); - } else { - Futures.addCallback(checkDeviceConnected(deviceName), new FutureCallback<>() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - processClaimDeviceMsg(deviceCtx, claimDeviceMsg.getClaimRequest(), deviceName, msgId); - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device claiming command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - } + process(deviceName, deviceCtx -> processClaimDeviceMsg(deviceCtx, claimDeviceMsg.getClaimRequest(), deviceName, msgId), + t -> log.debug("[{}][{}][{}] Failed to process device claiming command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); @@ -553,18 +487,20 @@ public abstract class AbstractGatewaySessionHandler() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - processPostAttributesMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId); - } + process(deviceName, deviceCtx -> processPostAttributesMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), + t -> log.debug("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); + } + } - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); + private void processPostAttributesMsg(MqttDeviceAwareSessionContext deviceCtx, JsonElement msg, String deviceName, int msgId) { + try { + TransportProtos.PostAttributeMsg postAttributeMsg = JsonConverter.convertToAttributesProto(msg.getAsJsonObject()); + transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(channel, deviceName, msgId, postAttributeMsg)); + } catch (Throwable e) { + log.warn("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, msg, e); + ackOrClose(msgId); } } @@ -579,38 +515,14 @@ public abstract class AbstractGatewaySessionHandler { String deviceName = checkDeviceName(attributesMsg.getDeviceName()); - T deviceCtx = devices.get(deviceName); - if (deviceCtx != null) { - processPostAttributesMsg(deviceCtx, attributesMsg.getMsg(), deviceName, msgId); - } else { - Futures.addCallback(checkDeviceConnected(deviceName), new FutureCallback<>() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - processPostAttributesMsg(deviceCtx, attributesMsg.getMsg(), deviceName, msgId); - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - } + process(deviceName, deviceCtx -> processPostAttributesMsg(deviceCtx, attributesMsg.getMsg(), deviceName, msgId), + t -> log.debug("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); } } - protected void processPostAttributesMsg(MqttDeviceAwareSessionContext deviceCtx, JsonElement msg, String deviceName, int msgId) { - try { - TransportProtos.PostAttributeMsg postAttributeMsg = JsonConverter.convertToAttributesProto(msg.getAsJsonObject()); - transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(channel, deviceName, msgId, postAttributeMsg)); - } catch (Throwable e) { - log.warn("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, msg, e); - ackOrClose(msgId); - } - } - protected void processPostAttributesMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.PostAttributeMsg kvListProto, String deviceName, int msgId) { try { TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto); @@ -623,26 +535,23 @@ public abstract class AbstractGatewaySessionHandler keys; - if (jsonObj.has("key")) { - keys = Collections.singleton(jsonObj.get("key").getAsString()); - } else { - JsonArray keysArray = jsonObj.get("keys").getAsJsonArray(); - keys = new HashSet<>(); - for (JsonElement keyObj : keysArray) { - keys.add(keyObj.getAsString()); - } - } - TransportProtos.GetAttributeRequestMsg requestMsg = toGetAttributeRequestMsg(requestId, clientScope, keys); - processGetAttributeRequestMessage(msg, deviceName, requestMsg); + validateJsonObject(json); + JsonObject jsonObj = json.getAsJsonObject(); + int requestId = jsonObj.get("id").getAsInt(); + String deviceName = jsonObj.get(DEVICE_PROPERTY).getAsString(); + boolean clientScope = jsonObj.get("client").getAsBoolean(); + Set keys; + if (jsonObj.has("key")) { + keys = Collections.singleton(jsonObj.get("key").getAsString()); } else { - throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); + JsonArray keysArray = jsonObj.get("keys").getAsJsonArray(); + keys = new HashSet<>(); + for (JsonElement keyObj : keysArray) { + keys.add(keyObj.getAsString()); + } } + TransportProtos.GetAttributeRequestMsg requestMsg = toGetAttributeRequestMsg(requestId, clientScope, keys); + processGetAttributeRequestMessage(msg, deviceName, requestMsg); } private void onDeviceAttributesRequestProto(MqttPublishMessage mqttMsg) throws AdaptorException { @@ -689,22 +598,8 @@ public abstract class AbstractGatewaySessionHandler() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - processRpcResponseMsg(deviceCtx, requestId, data, deviceName, msgId); - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device Rpc response command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - } + process(deviceName, deviceCtx -> processRpcResponseMsg(deviceCtx, requestId, data, deviceName, msgId), + t -> log.debug("[{}][{}][{}] Failed to process device Rpc response command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); } private void processRpcResponseMsg(MqttDeviceAwareSessionContext deviceCtx, Integer requestId, String data, String deviceName, int msgId) { @@ -715,23 +610,11 @@ public abstract class AbstractGatewaySessionHandler() { - @Override - public void onSuccess(@Nullable T deviceCtx) { - processGetAttributeRequestMessage(deviceCtx, requestMsg, deviceName, msgId); - } - - @Override - public void onFailure(Throwable t) { + process(deviceName, deviceCtx -> processGetAttributeRequestMessage(deviceCtx, requestMsg, deviceName, msgId), + t -> { ack(msgId, ReturnCode.IMPLEMENTATION_SPECIFIC); log.debug("[{}][{}][{}] Failed to process device attributes request command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); - } + }); } private void processGetAttributeRequestMessage(T deviceCtx, TransportProtos.GetAttributeRequestMsg requestMsg, String deviceName, int msgId) { @@ -750,16 +633,6 @@ public abstract class AbstractGatewaySessionHandler checkDeviceConnected(String deviceName) { - T ctx = devices.get(deviceName); - if (ctx == null) { - log.debug("[{}][{}][{}] Missing device [{}] for the gateway session", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName); - return onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE); - } else { - return Futures.immediateFuture(ctx); - } - } - protected String checkDeviceName(String deviceName) { if (StringUtils.isEmpty(deviceName)) { throw new RuntimeException("Device name is empty!"); @@ -796,7 +669,7 @@ public abstract class AbstractGatewaySessionHandler onSuccess, Consumer onFailure) { + ListenableFuture deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE); + process(deviceCtxFuture, onSuccess, onFailure); + } + + @SneakyThrows + protected void process(ListenableFuture deviceCtxFuture, Consumer onSuccess, Consumer onFailure) { + if (deviceCtxFuture.isDone()) { + onSuccess.accept(deviceCtxFuture.get()); + } else { + DonAsynchron.withCallback(deviceCtxFuture, onSuccess, onFailure, context.getExecutor()); + } + } + } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java index cbc515c1d0..e06d954678 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.transport.mqtt.session; -import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; @@ -30,12 +29,12 @@ import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.eclipse.leshan.core.ResponseCode; import org.springframework.util.CollectionUtils; -import org.thingsboard.server.common.data.device.profile.MqttDeviceProfileTransportConfiguration; -import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; -import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.adaptor.AdaptorException; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.adaptor.ProtoConverter; +import org.thingsboard.server.common.data.device.profile.MqttDeviceProfileTransportConfiguration; +import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; +import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse; import org.thingsboard.server.gen.transport.TransportApiProtos; import org.thingsboard.server.gen.transport.TransportProtos; @@ -44,8 +43,6 @@ import org.thingsboard.server.transport.mqtt.MqttTransportHandler; import org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopic; -import jakarta.annotation.Nullable; - import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -53,7 +50,6 @@ import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.ONLINE; @@ -72,6 +68,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler nodeBirthMetrics; private final MqttTransportHandler parent; @@ -88,10 +85,6 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler metric))); } - public Map getNodeBirthMetrics() { - return this.nodeBirthMetrics; - } - public TransportProtos.PostTelemetryMsg convertToPostTelemetry(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException { DeviceSessionCtx deviceSessionCtx = (DeviceSessionCtx) ctx; byte[] bytes = getBytes(inbound.payload()); @@ -138,55 +131,38 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler contextListenableFuture, - int msgId, List postTelemetryMsgList, String deviceName) throws AdaptorException { - Futures.addCallback(contextListenableFuture, new FutureCallback<>() { - @Override - public void onSuccess(@Nullable MqttDeviceAwareSessionContext deviceCtx) { - for (TransportProtos.PostTelemetryMsg telemetryMsg : postTelemetryMsgList) { - try { - processPostTelemetryMsg(deviceCtx, telemetryMsg, deviceName, msgId); - } catch (Throwable e) { - log.warn("[{}][{}] Failed to convert telemetry: {}", gateway.getDeviceId(), deviceName, telemetryMsg, e); - channel.close(); - break; + int msgId, List postTelemetryMsgList, String deviceName) { + process(contextListenableFuture, deviceCtx -> { + for (TransportProtos.PostTelemetryMsg telemetryMsg : postTelemetryMsgList) { + try { + processPostTelemetryMsg(deviceCtx, telemetryMsg, deviceName, msgId); + } catch (Throwable e) { + log.warn("[{}][{}] Failed to convert telemetry: {}", gateway.getDeviceId(), deviceName, telemetryMsg, e); + ackOrClose(msgId); + } } - } - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}] Failed to process device telemetry command: {}", sessionId, deviceName, t); - } - }, context.getExecutor()); + }, + t -> log.debug("[{}] Failed to process device telemetry command: {}", sessionId, deviceName, t)); } private void onDeviceAttributesProto(ListenableFuture contextListenableFuture, int msgId, List attributesMsgList, String deviceName) throws AdaptorException { try { - if (!CollectionUtils.isEmpty(attributesMsgList)) { - Futures.addCallback(contextListenableFuture, - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable MqttDeviceAwareSessionContext deviceCtx) { - for (TransportApiProtos.AttributesMsg attributesMsg : attributesMsgList) { - TransportProtos.PostAttributeMsg kvListProto = attributesMsg.getMsg(); - try { - TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto); - processPostAttributesMsg(deviceCtx, postAttributeMsg, deviceName, msgId); - } catch (Throwable e) { - log.warn("[{}][{}] Failed to process device attributes command: {}", gateway.getDeviceId(), deviceName, kvListProto, e); - } - } - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}] Failed to process device attributes command: {}", sessionId, deviceName, t); - } - }, context.getExecutor()); - } else { + if (CollectionUtils.isEmpty(attributesMsgList)) { log.debug("[{}] Devices attributes keys list is empty for: [{}]", sessionId, gateway.getDeviceId()); } + process(contextListenableFuture, deviceCtx -> { + for (TransportApiProtos.AttributesMsg attributesMsg : attributesMsgList) { + TransportProtos.PostAttributeMsg kvListProto = attributesMsg.getMsg(); + try { + TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto); + processPostAttributesMsg(deviceCtx, postAttributeMsg, deviceName, msgId); + } catch (Throwable e) { + log.warn("[{}][{}] Failed to process device attributes command: {}", gateway.getDeviceId(), deviceName, kvListProto, e); + } + } + }, + t -> log.debug("[{}] Failed to process device attributes command: {}", sessionId, deviceName, t)); } catch (RuntimeException e) { throw new AdaptorException(e); } @@ -235,9 +211,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler keyValueProtoOpt = fromSparkplugBMetricToKeyValueProto(key, protoMetric); - if (keyValueProtoOpt.isPresent()) { - msgs.add(postTelemetryMsgCreated(keyValueProtoOpt.get(), ts)); - } + keyValueProtoOpt.ifPresent(kvProto -> msgs.add(postTelemetryMsgCreated(kvProto, ts))); } } From 70a424d9c4217ccf8c2c5d9da3c33f9eb8269831 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Thu, 16 May 2024 20:35:54 +0200 Subject: [PATCH 4/5] refactored logs --- .../AbstractGatewaySessionHandler.java | 25 +++++++++++++------ 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index 1eaf1f38e8..61d868f08e 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -97,6 +97,11 @@ public abstract class AbstractGatewaySessionHandler processPostTelemetryMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), - t -> log.debug("[{}][{}][{}] Failed to process device telemetry command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); + t -> failedToProcessLog(deviceName, TELEMETRY, t)); } } @@ -395,7 +400,7 @@ public abstract class AbstractGatewaySessionHandler { String deviceName = checkDeviceName(telemetryMsg.getDeviceName()); process(deviceName, deviceCtx -> processPostTelemetryMsg(deviceCtx, telemetryMsg.getMsg(), deviceName, msgId), - t -> log.debug("[{}][{}][{}] Failed to process device telemetry command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); + t -> failedToProcessLog(deviceName, TELEMETRY, t)); }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); @@ -434,7 +439,7 @@ public abstract class AbstractGatewaySessionHandler processClaimDeviceMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), - t -> log.debug("[{}][{}][{}] Failed to process device claiming command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); + t -> failedToProcessLog(deviceName, CLAIMING, t)); } } @@ -461,7 +466,7 @@ public abstract class AbstractGatewaySessionHandler { String deviceName = checkDeviceName(claimDeviceMsg.getDeviceName()); process(deviceName, deviceCtx -> processClaimDeviceMsg(deviceCtx, claimDeviceMsg.getClaimRequest(), deviceName, msgId), - t -> log.debug("[{}][{}][{}] Failed to process device claiming command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); + t -> failedToProcessLog(deviceName, CLAIMING, t)); }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); @@ -490,7 +495,7 @@ public abstract class AbstractGatewaySessionHandler processPostAttributesMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), - t -> log.debug("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); + t -> failedToProcessLog(deviceName, ATTRIBUTE, t)); } } @@ -516,7 +521,7 @@ public abstract class AbstractGatewaySessionHandler { String deviceName = checkDeviceName(attributesMsg.getDeviceName()); process(deviceName, deviceCtx -> processPostAttributesMsg(deviceCtx, attributesMsg.getMsg(), deviceName, msgId), - t -> log.debug("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); + t -> failedToProcessLog(deviceName, ATTRIBUTE, t)); }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); @@ -599,7 +604,7 @@ public abstract class AbstractGatewaySessionHandler processRpcResponseMsg(deviceCtx, requestId, data, deviceName, msgId), - t -> log.debug("[{}][{}][{}] Failed to process device Rpc response command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t)); + t -> failedToProcessLog(deviceName, RPC_RESPONSE, t)); } private void processRpcResponseMsg(MqttDeviceAwareSessionContext deviceCtx, Integer requestId, String data, String deviceName, int msgId) { @@ -612,8 +617,8 @@ public abstract class AbstractGatewaySessionHandler processGetAttributeRequestMessage(deviceCtx, requestMsg, deviceName, msgId), t -> { + failedToProcessLog(deviceName, ATTRIBUTES_REQUEST, t); ack(msgId, ReturnCode.IMPLEMENTATION_SPECIFIC); - log.debug("[{}][{}][{}] Failed to process device attributes request command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); }); } @@ -728,4 +733,8 @@ public abstract class AbstractGatewaySessionHandler Date: Wed, 22 May 2024 12:02:13 +0200 Subject: [PATCH 5/5] added check rate limits for CreateDeviceFromGateway --- .../mqtt/MqttGatewayRateLimitsTest.java | 3 ++- .../DefaultTransportRateLimitService.java | 2 +- .../service/DefaultTransportService.java | 24 +++++++++++++++---- 3 files changed, 23 insertions(+), 6 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/MqttGatewayRateLimitsTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/MqttGatewayRateLimitsTest.java index 794d93a44a..a1986ca2c2 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/MqttGatewayRateLimitsTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/MqttGatewayRateLimitsTest.java @@ -81,7 +81,8 @@ public class MqttGatewayRateLimitsTest extends AbstractControllerTest { @Test public void transportGatewayMsgRateLimitTest() throws Exception { - transportGatewayRateLimitTest(profileConfiguration -> profileConfiguration.setTransportGatewayMsgRateLimit("1:600")); + // Device A 2 msgs success ('create device', 'to device actor'), Device B 'create device success' , 'to device actor' - limited + transportGatewayRateLimitTest(profileConfiguration -> profileConfiguration.setTransportGatewayMsgRateLimit("3:600")); } @Test diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java index f0a5d73f11..975dd21fab 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java @@ -84,7 +84,7 @@ public class DefaultTransportRateLimitService implements TransportRateLimitServi return TbPair.of(EntityType.DEVICE, true); } - if (!checkEntityRateLimit(dataPoints, getDeviceRateLimits(tenantId, deviceId))) { + if (deviceId != null && !checkEntityRateLimit(dataPoints, getDeviceRateLimits(tenantId, deviceId))) { return TbPair.of(EntityType.DEVICE, false); } return null; diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 9312370c34..598a5aa7a6 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -452,8 +452,13 @@ public class DefaultTransportService extends TransportActivityManager implements @Override public void process(TenantId tenantId, TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg requestMsg, TransportServiceCallback callback) { - TbProtoQueueMsg protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setGetOrCreateDeviceRequestMsg(requestMsg).build()); log.trace("Processing msg: {}", requestMsg); + DeviceId gatewayid = new DeviceId(new UUID(requestMsg.getGatewayIdMSB(), requestMsg.getGatewayIdLSB())); + if (!checkLimits(tenantId, gatewayid, null, requestMsg.getDeviceName(), requestMsg, callback, 0)) { + return; + } + + TbProtoQueueMsg protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setGetOrCreateDeviceRequestMsg(requestMsg).build()); var key = new EntityLimitKey(tenantId, StringUtils.truncate(requestMsg.getDeviceName(), 256)); if (entityLimitsCache.get(key)) { transportCallbackExecutor.submit(() -> callback.onError(new RuntimeException(DataConstants.MAXIMUM_NUMBER_OF_DEVICES_REACHED))); @@ -857,6 +862,14 @@ public class DefaultTransportService extends TransportActivityManager implements gatewayId = new DeviceId(new UUID(sessionInfo.getGatewayIdMSB(), sessionInfo.getGatewayIdLSB())); } + return checkLimits(tenantId, gatewayId, deviceId, sessionInfo.getDeviceName(), msg, callback, dataPoints); + } + + private boolean checkLimits(TenantId tenantId, DeviceId gatewayId, DeviceId deviceId, String deviceName, Object msg, TransportServiceCallback callback, int dataPoints) { + if (log.isTraceEnabled()) { + log.trace("[{}][{}] Processing msg: {}", tenantId, deviceName, msg); + } + var rateLimitedPair = rateLimitService.checkLimits(tenantId, gatewayId, deviceId, dataPoints); if (rateLimitedPair == null) { return true; @@ -868,13 +881,16 @@ public class DefaultTransportService extends TransportActivityManager implements if (rateLimitedEntityType == EntityType.DEVICE || rateLimitedEntityType == EntityType.TENANT) { LimitedApi limitedApi = - rateLimitedEntityType == EntityType.TENANT ? LimitedApi.TRANSPORT_MESSAGES_PER_TENANT : + rateLimitedEntityType == EntityType.TENANT ? LimitedApi.TRANSPORT_MESSAGES_PER_TENANT : rateLimitedPair.getSecond() ? LimitedApi.TRANSPORT_MESSAGES_PER_GATEWAY : LimitedApi.TRANSPORT_MESSAGES_PER_DEVICE; + + EntityId limitLevel = rateLimitedEntityType == EntityType.DEVICE ? deviceId == null ? gatewayId : deviceId : tenantId; + notificationRuleProcessor.process(RateLimitsTrigger.builder() .tenantId(tenantId) .api(limitedApi) - .limitLevel(rateLimitedEntityType == EntityType.DEVICE ? deviceId : tenantId) - .limitLevelEntityName(rateLimitedEntityType == EntityType.DEVICE ? sessionInfo.getDeviceName() : null) + .limitLevel(limitLevel) + .limitLevelEntityName(rateLimitedEntityType == EntityType.DEVICE ? deviceName : null) .build()); } return false;