tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge pull request #10804 from YevhenBondarenko/fix/mqtt-transport-oom 

mqtt gateway refactoring due to OOM
pull/10856/head
Andrew Shvayka 2 years ago
committed by GitHub
parent
eeb72de939 415c99147e
commit
3ab5a67152
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
5 changed files with 250 additions and 329 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 3
      application/src/test/java/org/thingsboard/server/transport/mqtt/MqttGatewayRateLimitsTest.java
  2. 447
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
  3. 103
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java
  4. 2
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java
  5. 24
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java

3
application/src/test/java/org/thingsboard/server/transport/mqtt/MqttGatewayRateLimitsTest.java
View File

@ -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

447
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
View File

@ -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<T extends AbstractGatewayDev
private static final String CAN_T_PARSE_VALUE = "Can't parse value: ";
private static final String DEVICE_PROPERTY = "device";
public static final String TELEMETRY = "telemetry";
public static final String CLAIMING = "claiming";
public static final String ATTRIBUTE = "attribute";
public static final String RPC_RESPONSE = "Rpc response";
public static final String ATTRIBUTES_REQUEST = "attributes request";
protected final MqttTransportContext context;
protected final TransportService transportService;
protected final TransportDeviceInfo gateway;
@Getter
protected final UUID sessionId;
private final ConcurrentMap<String, Lock> deviceCreationLockMap;
private final ConcurrentMap<String, T> devices;
@ -193,10 +201,6 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
return context.getNodeId();
}
public UUID getSessionId() {
return sessionId;
}
public MqttTransportAdaptor getPayloadAdaptor() {
return deviceSessionCtx.getPayloadAdaptor();
}
@ -224,18 +228,12 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
protected void processOnConnect(MqttPublishMessage msg, String deviceName, String deviceType) {
log.trace("[{}][{}][{}] onDeviceConnect: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName);
Futures.addCallback(onDeviceConnect(deviceName, deviceType), new FutureCallback<>() {
@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<DeviceProfile> deviceProfileOpt) {
@ -368,34 +366,25 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
protected void onDeviceTelemetryJson(int msgId, ByteBuf payload) throws AdaptorException {
JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, payload);
if (json.isJsonObject()) {
JsonObject jsonObj = json.getAsJsonObject();
for (Map.Entry<String, JsonElement> 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<String, JsonElement> 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<T extends AbstractGatewayDev
try {
TransportApiProtos.GatewayTelemetryMsg telemetryMsgProto = TransportApiProtos.GatewayTelemetryMsg.parseFrom(getBytes(payload));
List<TransportApiProtos.TelemetryMsg> 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<T extends AbstractGatewayDev
private void onDeviceClaimJson(int msgId, ByteBuf payload) throws AdaptorException {
JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, payload);
if (json.isJsonObject()) {
JsonObject jsonObj = json.getAsJsonObject();
for (Map.Entry<String, JsonElement> 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<String, JsonElement> 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<T extends AbstractGatewayDev
try {
TransportApiProtos.GatewayClaimMsg claimMsgProto = TransportApiProtos.GatewayClaimMsg.parseFrom(getBytes(payload));
List<TransportApiProtos.ClaimDeviceMsg> 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<String, JsonElement> 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<String, JsonElement> 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<T extends AbstractGatewayDev
try {
TransportApiProtos.GatewayAttributesMsg attributesMsgProto = TransportApiProtos.GatewayAttributesMsg.parseFrom(getBytes(payload));
List<TransportApiProtos.AttributesMsg> 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<String> 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<String> 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<T extends AbstractGatewayDev
private void onDeviceRpcResponseJson(int msgId, ByteBuf payload) throws AdaptorException {
JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, payload);
if (json.isJsonObject()) {
JsonObject jsonObj = json.getAsJsonObject();
String deviceName = jsonObj.get(DEVICE_PROPERTY).getAsString();
Futures.addCallback(checkDeviceConnected(deviceName),
new FutureCallback<>() {
@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<T extends AbstractGatewayDev
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) {
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<String> keys) {
@ -719,16 +638,6 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
return result.build();
}
protected ListenableFuture<T> 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<T extends AbstractGatewayDev
protected void ack(MqttPublishMessage msg, ReturnCode returnCode) {
int msgId = getMsgId(msg);
ack(msgId, returnCode);
}
protected void ack(int msgId, ReturnCode returnCode) {
if (msgId > 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<T extends AbstractGatewayDev
};
}
protected void process(String deviceName, Consumer<T> onSuccess, Consumer<Throwable> onFailure) {
ListenableFuture<T> deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE);
process(deviceCtxFuture, onSuccess, onFailure);
}
@SneakyThrows
protected <T> void process(ListenableFuture<T> deviceCtxFuture, Consumer<T> onSuccess, Consumer<Throwable> 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);
}
}

103
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java
View File

@ -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<SparkplugDeviceSessionContext> {
@Getter
private final SparkplugTopic sparkplugTopicNode;
@Getter
private final Map<String, SparkplugBProto.Payload.Metric> nodeBirthMetrics;
private final MqttTransportHandler parent;
@ -85,10 +85,6 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
.collect(Collectors.toMap(SparkplugBProto.Payload.Metric::getName, metric -> metric)));
}
public Map<String, SparkplugBProto.Payload.Metric> 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<S
}
public void onDeviceTelemetryProto(ListenableFuture<MqttDeviceAwareSessionContext> contextListenableFuture,
int msgId, List<TransportProtos.PostTelemetryMsg> 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<TransportProtos.PostTelemetryMsg> 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<MqttDeviceAwareSessionContext> contextListenableFuture, int msgId,
List<TransportApiProtos.AttributesMsg> 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<Integer> 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<S
String key = "bdSeq".equals(protoMetric.getName()) ?
topicTypeName + " " + protoMetric.getName() : protoMetric.getName();
Optional<TransportProtos.KeyValueProto> 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<S
return Optional.empty();
}
public SparkplugTopic getSparkplugTopicNode() {
return this.sparkplugTopicNode;
}
public Optional<MqttPublishMessage> createSparkplugMqttPublishMsg(TransportProtos.TsKvProto tsKvProto,
String sparkplugTopic,
SparkplugBProto.Payload.Metric metricBirth) {
@ -328,7 +295,6 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
} catch (Exception e) {
log.trace("DeviceId: [{}] tenantId: [{}] sessionId:[{}] Failed to convert device attributes response to MQTT sparkplug msg",
deviceSessionCtx.getDeviceInfo().getDeviceId(), deviceSessionCtx.getDeviceInfo().getTenantId(), sessionId, e);
return Optional.empty();
}
return Optional.empty();
}
@ -350,5 +316,4 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
parent.sendSuccessRpcResponse(sessionInfo, requestId, result, successMsg);
}
}

2
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java
View File

@ -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;

24
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View File

@ -428,8 +428,13 @@ public class DefaultTransportService extends TransportActivityManager implements
@Override
public void process(TenantId tenantId, TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg requestMsg, TransportServiceCallback<GetOrCreateDeviceFromGatewayResponse> callback) {
TbProtoQueueMsg<TransportApiRequestMsg> 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<TransportApiRequestMsg> 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;

Write Preview
Loading…
Cancel
Save