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/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..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 @@ -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; @@ -33,11 +31,14 @@ 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 io.netty.handler.codec.mqtt.MqttVersion; 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; @@ -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; @@ -76,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; @@ -83,6 +84,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; @@ -95,10 +97,16 @@ public abstract class AbstractGatewaySessionHandler deviceCreationLockMap; private final ConcurrentMap devices; @@ -193,10 +201,6 @@ 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) { @@ -368,34 +366,25 @@ 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().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(); - } - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device telemetry command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); + validateJsonObject(json); + for (Map.Entry deviceEntry : json.getAsJsonObject().entrySet()) { + if (!deviceEntry.getValue().isJsonArray()) { + log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); + continue; } - } else { - throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); + String deviceName = deviceEntry.getKey(); + process(deviceName, deviceCtx -> processPostTelemetryMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), + t -> failedToProcessLog(deviceName, TELEMETRY, t)); + } + } + + private void processPostTelemetryMsg(T deviceCtx, JsonElement msg, String deviceName, int msgId) { + try { + TransportProtos.PostTelemetryMsg postTelemetryMsg = JsonConverter.convertToTelemetryProto(msg.getAsJsonArray()); + 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); + ackOrClose(msgId); } } @@ -403,40 +392,29 @@ 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()); + process(deviceName, deviceCtx -> processPostTelemetryMsg(deviceCtx, telemetryMsg.getMsg(), deviceName, msgId), + t -> failedToProcessLog(deviceName, TELEMETRY, t)); + }); } 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); + ackOrClose(msgId); + } } public TransportProtos.PostTelemetryMsg postTelemetryMsgCreated(TransportProtos.KeyValueProto keyValueProto, long ts) { @@ -452,34 +430,27 @@ 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); - } - } - - @Override - public void onFailure(Throwable t) { - log.debug("[{}][{}][{}] Failed to process device claiming command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, t); - } - }, context.getExecutor()); + validateJsonObject(json); + for (Map.Entry deviceEntry : json.getAsJsonObject().entrySet()) { + if (!deviceEntry.getValue().isJsonObject()) { + log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); + continue; } - } else { - throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); + + String deviceName = deviceEntry.getKey(); + process(deviceName, deviceCtx -> processClaimDeviceMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), + t -> failedToProcessLog(deviceName, CLAIMING, t)); + } + } + + 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); + ackOrClose(msgId); } } @@ -487,70 +458,54 @@ 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()); + process(deviceName, deviceCtx -> processClaimDeviceMsg(deviceCtx, claimDeviceMsg.getClaimRequest(), deviceName, msgId), + t -> failedToProcessLog(deviceName, CLAIMING, t)); + }); } 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); + ackOrClose(msgId); + } } 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()); + validateJsonObject(json); + for (Map.Entry deviceEntry : json.getAsJsonObject().entrySet()) { + if (!deviceEntry.getValue().isJsonObject()) { + log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); + continue; } - } else { - throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); + + String deviceName = deviceEntry.getKey(); + process(deviceName, deviceCtx -> processPostAttributesMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), + t -> failedToProcessLog(deviceName, ATTRIBUTE, t)); + } + } + + 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); } } @@ -558,66 +513,50 @@ public abstract class AbstractGatewaySessionHandler 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()); + process(deviceName, deviceCtx -> processPostAttributesMsg(deviceCtx, attributesMsg.getMsg(), deviceName, msgId), + t -> failedToProcessLog(deviceName, ATTRIBUTE, t)); + }); } 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, 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); + ackOrClose(msgId); + } } private void onDeviceAttributesRequestJson(MqttPublishMessage msg) throws AdaptorException { JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, msg.payload()); - if (json.isJsonObject()) { - 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 { - 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 { @@ -637,26 +576,16 @@ 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); - } + validateJsonObject(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); + } - @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 { + private static void validateJsonObject(JsonElement json) { + if (!json.isJsonObject()) { throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json); } } @@ -665,46 +594,36 @@ public abstract class AbstractGatewaySessionHandler() { - @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) { + process(deviceName, deviceCtx -> processRpcResponseMsg(deviceCtx, requestId, data, deviceName, msgId), + t -> failedToProcessLog(deviceName, RPC_RESPONSE, t)); + } + + 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)); - } + process(deviceName, deviceCtx -> processGetAttributeRequestMessage(deviceCtx, requestMsg, deviceName, msgId), + t -> { + failedToProcessLog(deviceName, ATTRIBUTES_REQUEST, t); + ack(msgId, ReturnCode.IMPLEMENTATION_SPECIFIC); + }); + } - @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) { @@ -719,16 +638,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!"); @@ -756,11 +665,23 @@ public abstract class AbstractGatewaySessionHandler 0) { writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(deviceSessionCtx, msgId, returnCode)); } } + protected 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(), @@ -798,4 +719,22 @@ 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()); + } + } + + protected void failedToProcessLog(String deviceName, String msgType, Throwable t) { + log.debug("[{}][{}][{}] Failed to process device {} command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, msgType, deviceName, t); + } + } 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..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; @@ -26,15 +25,16 @@ 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; -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; @@ -43,7 +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; @@ -51,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; @@ -68,7 +66,9 @@ import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopi @Slf4j public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler { + @Getter private final SparkplugTopic sparkplugTopicNode; + @Getter private final Map nodeBirthMetrics; private final MqttTransportHandler parent; @@ -85,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()); @@ -135,69 +131,46 @@ 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(); - } - } - - @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); - } + 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); + } + } + }, + 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)) { - attributesMsgList.forEach(attributesMsg -> { - Futures.addCallback(contextListenableFuture, - new FutureCallback<>() { - @Override - public void onSuccess(@Nullable MqttDeviceAwareSessionContext deviceCtx) { - 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); } } public void handleSparkplugSubscribeMsg(List 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) { @@ -238,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))); } } @@ -304,10 +275,6 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler createSparkplugMqttPublishMsg(TransportProtos.TsKvProto tsKvProto, String sparkplugTopic, SparkplugBProto.Payload.Metric metricBirth) { @@ -328,7 +295,6 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler 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))); @@ -833,6 +838,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; @@ -844,13 +857,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;