diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 8609a7e506..3afcf551ef 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -17,9 +17,12 @@ package org.thingsboard.server.actors.ruleChain; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; import io.netty.channel.EventLoopGroup; import lombok.extern.slf4j.Slf4j; import org.bouncycastle.util.Arrays; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ListeningExecutor; import org.thingsboard.rule.engine.api.MailService; import org.thingsboard.rule.engine.api.RuleEngineAlarmService; @@ -40,6 +43,7 @@ import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.HasRuleEngineProfile; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.alarm.Alarm; @@ -53,6 +57,7 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.rule.RuleNode; @@ -91,6 +96,7 @@ import org.thingsboard.server.service.script.RuleNodeJsScriptEngine; import org.thingsboard.server.service.script.RuleNodeMvelScriptEngine; import java.util.Collections; +import java.util.List; import java.util.Set; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -333,58 +339,62 @@ class DefaultTbContext implements TbContext { } public TbMsg deviceCreatedMsg(Device device, RuleNodeId ruleNodeId) { - RuleChainId ruleChainId = null; - String queueName = null; + DeviceProfile deviceProfile = null; if (device.getDeviceProfileId() != null) { - DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().find(device.getDeviceProfileId()); - if (deviceProfile == null) { - log.warn("[{}] Device profile is null!", device.getDeviceProfileId()); - } else { - ruleChainId = deviceProfile.getDefaultRuleChainId(); - queueName = deviceProfile.getDefaultQueueName(); - } + deviceProfile = mainCtx.getDeviceProfileCache().find(device.getDeviceProfileId()); } - return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, queueName, ruleChainId); + return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, deviceProfile); } public TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId) { - RuleChainId ruleChainId = null; - String queueName = null; + AssetProfile assetProfile = null; if (asset.getAssetProfileId() != null) { - AssetProfile assetProfile = mainCtx.getAssetProfileCache().find(asset.getAssetProfileId()); - if (assetProfile == null) { - log.warn("[{}] Asset profile is null!", asset.getAssetProfileId()); - } else { - ruleChainId = assetProfile.getDefaultRuleChainId(); - queueName = assetProfile.getDefaultQueueName(); - } + assetProfile = mainCtx.getAssetProfileCache().find(asset.getAssetProfileId()); } - return entityActionMsg(asset, asset.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, queueName, ruleChainId); + return entityActionMsg(asset, asset.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, assetProfile); } public TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action) { - RuleChainId ruleChainId = null; - String queueName = null; + HasRuleEngineProfile profile = null; if (EntityType.DEVICE.equals(alarm.getOriginator().getEntityType())) { DeviceId deviceId = new DeviceId(alarm.getOriginator().getId()); - DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); - if (deviceProfile == null) { - log.warn("[{}] Device profile is null!", deviceId); - } else { - ruleChainId = deviceProfile.getDefaultRuleChainId(); - queueName = deviceProfile.getDefaultQueueName(); - } + profile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); } else if (EntityType.ASSET.equals(alarm.getOriginator().getEntityType())) { AssetId assetId = new AssetId(alarm.getOriginator().getId()); - AssetProfile assetProfile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId); - if (assetProfile == null) { - log.warn("[{}] Asset profile is null!", assetId); - } else { - ruleChainId = assetProfile.getDefaultRuleChainId(); - queueName = assetProfile.getDefaultQueueName(); - } + profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId); + } + return entityActionMsg(alarm, alarm.getOriginator(), ruleNodeId, action, profile); + } + + public TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes) { + ObjectNode entityNode = JacksonUtil.newObjectNode(); + if (attributes != null) { + attributes.forEach(attributeKvEntry -> JacksonUtil.addKvEntry(entityNode, attributeKvEntry)); } - return entityActionMsg(alarm, alarm.getId(), ruleNodeId, action, queueName, ruleChainId); + return attributesActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_UPDATED, JacksonUtil.toString(entityNode)); + } + + public TbMsg attributesDeletedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List keys) { + ObjectNode entityNode = JacksonUtil.newObjectNode(); + ArrayNode attrsArrayNode = entityNode.putArray("attributes"); + if (keys != null) { + keys.forEach(attrsArrayNode::add); + } + return attributesActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_DELETED, JacksonUtil.toString(entityNode)); + } + + private TbMsg attributesActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, String action, String msgData) { + TbMsgMetaData tbMsgMetaData = getActionMetaData(ruleNodeId); + tbMsgMetaData.putValue("scope", scope); + HasRuleEngineProfile profile = null; + if (EntityType.DEVICE.equals(originator.getEntityType())) { + DeviceId deviceId = new DeviceId(originator.getId()); + profile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); + } else if (EntityType.ASSET.equals(originator.getEntityType())) { + AssetId assetId = new AssetId(originator.getId()); + profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId); + } + return entityActionMsg(originator, tbMsgMetaData, msgData, action, profile); } @Override @@ -393,17 +403,27 @@ class DefaultTbContext implements TbContext { } public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action) { - return entityActionMsg(entity, id, ruleNodeId, action, null, null); + return entityActionMsg(entity, id, ruleNodeId, action, null); } - public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, String queueName, RuleChainId ruleChainId) { + public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, K profile) { try { - return TbMsg.newMsg(queueName, action, id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), ruleChainId, null); + return entityActionMsg(id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), action, profile); } catch (JsonProcessingException | IllegalArgumentException e) { throw new RuntimeException("Failed to process " + id.getEntityType().name().toLowerCase() + " " + action + " msg: " + e); } } + private TbMsg entityActionMsg(I id, TbMsgMetaData msgMetaData, String msgData, String action, K profile) { + String defaultQueueName = null; + RuleChainId defaultRuleChainId = null; + if (profile != null) { + defaultQueueName = profile.getDefaultQueueName(); + defaultRuleChainId = profile.getDefaultRuleChainId(); + } + return TbMsg.newMsg(defaultQueueName, action, id, msgMetaData, msgData, defaultRuleChainId, null); + } + @Override public RuleNodeId getSelfId() { return nodeCtx.getSelf().getId(); diff --git a/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java b/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java index b962961513..7057f903be 100644 --- a/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java +++ b/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java @@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityType; @@ -34,8 +35,6 @@ import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; -import org.thingsboard.server.common.data.kv.DataType; -import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.msg.TbMsg; @@ -180,7 +179,7 @@ public class EntityActionService { metaData.putValue(DataConstants.SCOPE, scope); if (attributes != null) { for (AttributeKvEntry attr : attributes) { - addKvEntry(entityNode, attr); + JacksonUtil.addKvEntry(entityNode, attr); } } } else if (actionType == ActionType.ATTRIBUTES_DELETED) { @@ -258,27 +257,11 @@ public class EntityActionService { element.put("ts", entry.getKey()); ObjectNode values = element.putObject("values"); for (TsKvEntry tsKvEntry : entry.getValue()) { - addKvEntry(values, tsKvEntry); + JacksonUtil.addKvEntry(values, tsKvEntry); } result.add(element); } } } - private void addKvEntry(ObjectNode entityNode, KvEntry kvEntry) throws Exception { - if (kvEntry.getDataType() == DataType.BOOLEAN) { - kvEntry.getBooleanValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); - } else if (kvEntry.getDataType() == DataType.DOUBLE) { - kvEntry.getDoubleValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); - } else if (kvEntry.getDataType() == DataType.LONG) { - kvEntry.getLongValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); - } else if (kvEntry.getDataType() == DataType.JSON) { - if (kvEntry.getJsonValue().isPresent()) { - entityNode.set(kvEntry.getKey(), json.readTree(kvEntry.getJsonValue().get())); - } - } else { - entityNode.put(kvEntry.getKey(), kvEntry.getValueAsString()); - } - } - } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 9eb5c09d2d..66a603a1a1 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -268,7 +268,7 @@ sql: audit_logs: partition_size: "${SQL_AUDIT_LOGS_PARTITION_SIZE_HOURS:168}" # Default value - 1 week # Specify whether to sort entities before batch update. Should be enabled for cluster mode to avoid deadlocks - batch_sort: "${SQL_BATCH_SORT:false}" + batch_sort: "${SQL_BATCH_SORT:true}" # Specify whether to remove null characters from strValue of attributes and timeseries before insert remove_null_chars: "${SQL_REMOVE_NULL_CHARS:true}" # Specify whether to log database queries and their parameters generated by entity query repository @@ -602,12 +602,16 @@ mvel: max_total_args_size: "${MVEL_MAX_TOTAL_ARGS_SIZE:100000}" max_result_size: "${MVEL_MAX_RESULT_SIZE:300000}" max_script_body_size: "${MVEL_MAX_SCRIPT_BODY_SIZE:50000}" + # Maximum allowed MVEL script execution memory + max_memory_limit_mb: "${MVEL_MAX_MEMORY_LIMIT_MB: 8}" # Maximum allowed MVEL script execution errors before it will be blacklisted max_errors: "${MVEL_MAX_ERRORS:3}" # MVEL Eval max request timeout in milliseconds. 0 - no timeout max_requests_timeout: "${MVEL_MAX_REQUEST_TIMEOUT:500}" # Maximum time in seconds for black listed function to stay in the list. max_black_list_duration_sec: "${MVEL_MAX_BLACKLIST_DURATION_SEC:60}" + # Specify thread pool size for javascript executor service + thread_pool_size: "${MVEL_THREAD_POOL_SIZE:50}" stats: enabled: "${TB_MVEL_STATS_ENABLED:false}" print_interval_ms: "${TB_MVEL_STATS_PRINT_INTERVAL_MS:10000}" diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java index da8edc695e..a66f3a40d9 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java @@ -45,7 +45,7 @@ import static org.thingsboard.server.common.data.SearchTextBasedWithAdditionalIn @ToString(exclude = {"image", "profileDataBytes"}) @EqualsAndHashCode(callSuper = true) @Slf4j -public class DeviceProfile extends SearchTextBased implements HasName, HasTenantId, HasOtaPackage, ExportableEntity { +public class DeviceProfile extends SearchTextBased implements HasName, HasTenantId, HasOtaPackage, HasRuleEngineProfile, ExportableEntity { private static final long serialVersionUID = 6998485460273302018L; @@ -141,7 +141,7 @@ public class DeviceProfile extends SearchTextBased implements H } @ApiModelProperty(position = 5, value = "Used to mark the default profile. Default profile is used when the device profile is not specified during device creation.") - public boolean isDefault(){ + public boolean isDefault() { return isDefault; } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/HasRuleEngineProfile.java b/common/data/src/main/java/org/thingsboard/server/common/data/HasRuleEngineProfile.java new file mode 100644 index 0000000000..450a2235a7 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/HasRuleEngineProfile.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data; + +import org.thingsboard.server.common.data.id.RuleChainId; + +public interface HasRuleEngineProfile { + + RuleChainId getDefaultRuleChainId(); + + String getDefaultQueueName(); + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/StringUtils.java b/common/data/src/main/java/org/thingsboard/server/common/data/StringUtils.java index 18d3dfff1e..9acbb15646 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/StringUtils.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/StringUtils.java @@ -146,6 +146,16 @@ public class StringUtils { return org.apache.commons.lang3.StringUtils.substringAfterLast(str, sep); } + public static boolean containedByAny(String searchString, String... strings) { + if (searchString == null) return false; + for (String string : strings) { + if (string != null && string.contains(searchString)) { + return true; + } + } + return false; + } + public static String randomNumeric(int length) { return RandomStringUtils.randomNumeric(length); } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/asset/AssetProfile.java b/common/data/src/main/java/org/thingsboard/server/common/data/asset/AssetProfile.java index 1bd9d9c6a5..01657feed6 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/asset/AssetProfile.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/asset/AssetProfile.java @@ -23,6 +23,7 @@ import lombok.ToString; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.ExportableEntity; import org.thingsboard.server.common.data.HasName; +import org.thingsboard.server.common.data.HasRuleEngineProfile; import org.thingsboard.server.common.data.HasTenantId; import org.thingsboard.server.common.data.SearchTextBased; import org.thingsboard.server.common.data.id.AssetProfileId; @@ -37,7 +38,7 @@ import org.thingsboard.server.common.data.validation.NoXss; @ToString(exclude = {"image"}) @EqualsAndHashCode(callSuper = true) @Slf4j -public class AssetProfile extends SearchTextBased implements HasName, HasTenantId, ExportableEntity { +public class AssetProfile extends SearchTextBased implements HasName, HasTenantId, HasRuleEngineProfile, ExportableEntity { private static final long serialVersionUID = 6998485460273302018L; diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/AbstractScriptInvokeService.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/AbstractScriptInvokeService.java index db376f8581..5f2b6d0b67 100644 --- a/common/script/script-api/src/main/java/org/thingsboard/script/api/AbstractScriptInvokeService.java +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/AbstractScriptInvokeService.java @@ -91,7 +91,7 @@ public abstract class AbstractScriptInvokeService implements ScriptInvokeService protected abstract ListenableFuture doEvalScript(TenantId tenantId, ScriptType scriptType, String scriptBody, UUID scriptId, String[] argNames); - protected abstract ListenableFuture doInvokeFunction(UUID scriptId, Object[] args); + protected abstract TbScriptExecutionTask doInvokeFunction(UUID scriptId, Object[] args); protected abstract void doRelease(UUID scriptId) throws Exception; @@ -129,7 +129,8 @@ public abstract class AbstractScriptInvokeService implements ScriptInvokeService } UUID scriptId = UUID.randomUUID(); pushedMsgs.incrementAndGet(); - return withTimeoutAndStatsCallback(scriptId, doEvalScript(tenantId, scriptType, scriptBody, scriptId, argNames), evalCallback, getMaxEvalRequestsTimeout()); + return withTimeoutAndStatsCallback(scriptId, null, + doEvalScript(tenantId, scriptType, scriptBody, scriptId, argNames), evalCallback, getMaxEvalRequestsTimeout()); } else { return error("Script Execution is disabled due to API limits!"); } @@ -146,13 +147,14 @@ public abstract class AbstractScriptInvokeService implements ScriptInvokeService TbScriptException t = new TbScriptException(scriptId, TbScriptException.ErrorCode.OTHER, null, new IllegalArgumentException( format("Script input arguments exceed maximum allowed total args size of %s symbols", getMaxTotalArgsSize()) )); - handleScriptException(scriptId, t); - return Futures.immediateFailedFuture(t); + return Futures.immediateFailedFuture(handleScriptException(scriptId, null, t)); } apiUsageReportClient.ifPresent(client -> client.report(tenantId, customerId, ApiUsageRecordKey.JS_EXEC_COUNT, 1)); pushedMsgs.incrementAndGet(); log.trace("InvokeScript uuid {} with timeout {}ms", scriptId, getMaxInvokeRequestsTimeout()); - var resultFuture = Futures.transformAsync(doInvokeFunction(scriptId, args), output -> { + var task = doInvokeFunction(scriptId, args); + + var resultFuture = Futures.transformAsync(task.getResultFuture(), output -> { String result = JacksonUtil.toString(output); if (resultSizeExceeded(result)) { throw new TbScriptException(scriptId, TbScriptException.ErrorCode.OTHER, null, new RuntimeException( @@ -162,7 +164,7 @@ public abstract class AbstractScriptInvokeService implements ScriptInvokeService return Futures.immediateFuture(output); }, MoreExecutors.directExecutor()); - return withTimeoutAndStatsCallback(scriptId, resultFuture, invokeCallback, getMaxInvokeRequestsTimeout()); + return withTimeoutAndStatsCallback(scriptId, task, resultFuture, invokeCallback, getMaxInvokeRequestsTimeout()); } else { String message = "Script invocation is blocked due to maximum error count " + getMaxErrors() + ", scriptId " + scriptId + "!"; @@ -174,19 +176,22 @@ public abstract class AbstractScriptInvokeService implements ScriptInvokeService } } - private ListenableFuture withTimeoutAndStatsCallback(UUID scriptId, ListenableFuture future, FutureCallback statsCallback, long timeout) { + private ListenableFuture withTimeoutAndStatsCallback(UUID scriptId, TbScriptExecutionTask task, ListenableFuture future, FutureCallback statsCallback, long timeout) { if (timeout > 0) { future = Futures.withTimeout(future, timeout, TimeUnit.MILLISECONDS, timeoutExecutorService); } Futures.addCallback(future, statsCallback, getCallbackExecutor()); - return Futures.catchingAsync(future, Exception.class, input -> { - handleScriptException(scriptId, input); - return Futures.immediateFailedFuture(input); - }, MoreExecutors.directExecutor()); + return Futures.catchingAsync(future, Exception.class, + input -> Futures.immediateFailedFuture(handleScriptException(scriptId, task, input)), + MoreExecutors.directExecutor()); } - private void handleScriptException(UUID scriptId, Throwable t) { - boolean blockList = t instanceof TimeoutException || (t.getCause() != null && t.getCause() instanceof TimeoutException); + private Throwable handleScriptException(UUID scriptId, TbScriptExecutionTask task, Throwable t) { + boolean timeout = t instanceof TimeoutException || (t.getCause() != null && t.getCause() instanceof TimeoutException); + if (timeout && task != null) { + task.stop(); + } + boolean blockList = timeout; String scriptBody = null; if (t instanceof TbScriptException) { var scriptException = (TbScriptException) t; @@ -204,7 +209,7 @@ public abstract class AbstractScriptInvokeService implements ScriptInvokeService log.debug("[{}] Failed to execute script: {}", scriptId, scriptException.getBody(), cause); break; } - blockList = blockList || scriptException.getErrorCode() != TbScriptException.ErrorCode.RUNTIME; + blockList = timeout || scriptException.getErrorCode() != TbScriptException.ErrorCode.RUNTIME; } if (blockList) { BlockedScriptInfo disableListInfo = disabledScripts.computeIfAbsent(scriptId, key -> new BlockedScriptInfo(getMaxBlackListDurationSec())); @@ -216,7 +221,11 @@ public abstract class AbstractScriptInvokeService implements ScriptInvokeService log.warn("Script has exception counter {} on disabledFunctions for id {}, exception {}", counter, scriptId, t.getMessage()); } - + } + if(timeout){ + return new TimeoutException("Script timeout!"); + } else { + return t; } } diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/TbScriptExecutionTask.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/TbScriptExecutionTask.java new file mode 100644 index 0000000000..df67d9896f --- /dev/null +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/TbScriptExecutionTask.java @@ -0,0 +1,30 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.script.api; + +import com.google.common.util.concurrent.ListenableFuture; +import lombok.Getter; +import lombok.RequiredArgsConstructor; + + +@RequiredArgsConstructor +public abstract class TbScriptExecutionTask { + + @Getter + private final ListenableFuture resultFuture; + + public abstract void stop(); +} diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/js/AbstractJsInvokeService.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/js/AbstractJsInvokeService.java index 66a2e3aec2..1bfccad510 100644 --- a/common/script/script-api/src/main/java/org/thingsboard/script/api/js/AbstractJsInvokeService.java +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/js/AbstractJsInvokeService.java @@ -61,8 +61,8 @@ public abstract class AbstractJsInvokeService extends AbstractScriptInvokeServic } @Override - protected ListenableFuture doInvokeFunction(UUID scriptId, Object[] args) { - return doInvokeFunction(scriptId, scriptInfoMap.get(scriptId), args); + protected JsScriptExecutionTask doInvokeFunction(UUID scriptId, Object[] args) { + return new JsScriptExecutionTask(doInvokeFunction(scriptId, scriptInfoMap.get(scriptId), args)); } @Override diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/js/JsScriptExecutionTask.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/js/JsScriptExecutionTask.java new file mode 100644 index 0000000000..5e493bc663 --- /dev/null +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/js/JsScriptExecutionTask.java @@ -0,0 +1,31 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.script.api.js; + +import com.google.common.util.concurrent.ListenableFuture; +import org.thingsboard.script.api.TbScriptExecutionTask; + +public class JsScriptExecutionTask extends TbScriptExecutionTask { + + public JsScriptExecutionTask(ListenableFuture resultFuture) { + super(resultFuture); + } + + @Override + public void stop() { + // do nothing + } +} diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/DefaultMvelInvokeService.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/DefaultMvelInvokeService.java index 706ef2ed46..54917b7eb5 100644 --- a/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/DefaultMvelInvokeService.java +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/DefaultMvelInvokeService.java @@ -26,6 +26,7 @@ import org.mvel2.ExecutionContext; import org.mvel2.MVEL; import org.mvel2.ParserContext; import org.mvel2.SandboxedParserContext; +import org.mvel2.ScriptMemoryOverflowException; import org.mvel2.optimizers.OptimizerFactory; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; @@ -42,7 +43,6 @@ import org.thingsboard.server.common.stats.TbApiUsageStateClient; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.io.Serializable; -import java.lang.reflect.Field; import java.util.Map; import java.util.Optional; import java.util.UUID; @@ -86,6 +86,12 @@ public class DefaultMvelInvokeService extends AbstractScriptInvokeService implem @Value("${mvel.stats.enabled:false}") private boolean statsEnabled; + @Value("${mvel.thread_pool_size:50}") + private int threadPoolSize; + + @Value("${mvel.max_memory_limit_mb:8}") + private long maxMemoryLimitMb; + private ListeningExecutorService executor; protected DefaultMvelInvokeService(Optional apiUsageStateClient, Optional apiUsageReportClient) { @@ -104,7 +110,7 @@ public class DefaultMvelInvokeService extends AbstractScriptInvokeService implem OptimizerFactory.setDefaultOptimizer(OptimizerFactory.SAFE_REFLECTIVE); parserContext = new SandboxedParserContext(); parserContext.addImport("JSON", TbJson.class); - executor = MoreExecutors.listeningDecorator(ThingsBoardExecutors.newWorkStealingPool(2, "mvel-executor")); + executor = MoreExecutors.listeningDecorator(ThingsBoardExecutors.newWorkStealingPool(threadPoolSize, "mvel-executor")); } @PreDestroy @@ -149,23 +155,21 @@ public class DefaultMvelInvokeService extends AbstractScriptInvokeService implem } @Override - protected ListenableFuture doInvokeFunction(UUID scriptId, Object[] args) { - return executor.submit(() -> { + protected MvelScriptExecutionTask doInvokeFunction(UUID scriptId, Object[] args) { + ExecutionContext executionContext = new ExecutionContext(maxMemoryLimitMb * 1024 * 1024); + return new MvelScriptExecutionTask(executionContext, executor.submit(() -> { MvelScript script = scriptMap.get(scriptId); if (script == null) { throw new TbScriptException(scriptId, TbScriptException.ErrorCode.OTHER, null, new RuntimeException("Script not found!")); } try { - ExecutionContext executionContext = new ExecutionContext(); - // TODO: return MVEL.executeExpression(script.getCompiledScript(), executionContext, script.createVars(args)); - } catch (OutOfMemoryError e) { - Runtime.getRuntime().gc(); + } catch (ScriptMemoryOverflowException e) { throw new TbScriptException(scriptId, TbScriptException.ErrorCode.OTHER, script.getScriptBody(), new RuntimeException("Memory error!")); } catch (Exception e) { throw new TbScriptException(scriptId, TbScriptException.ErrorCode.RUNTIME, script.getScriptBody(), e); } - }); + })); } @Override diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/MvelScriptExecutionTask.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/MvelScriptExecutionTask.java new file mode 100644 index 0000000000..e95b5b7b0c --- /dev/null +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/MvelScriptExecutionTask.java @@ -0,0 +1,38 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.script.api.mvel; + +import com.google.common.util.concurrent.ListenableFuture; +import lombok.Data; +import lombok.Getter; +import org.mvel2.ExecutionContext; +import org.thingsboard.script.api.TbScriptExecutionTask; + + +public class MvelScriptExecutionTask extends TbScriptExecutionTask { + + private final ExecutionContext context; + + public MvelScriptExecutionTask(ExecutionContext context, ListenableFuture resultFuture) { + super(resultFuture); + this.context = context; + } + + @Override + public void stop(){ + context.stop(); + } +} diff --git a/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java b/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java index f0904701c5..d703c53a7d 100644 --- a/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java +++ b/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java @@ -24,6 +24,8 @@ import com.fasterxml.jackson.databind.SerializationFeature; import com.fasterxml.jackson.databind.json.JsonMapper; import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; +import org.thingsboard.server.common.data.kv.DataType; +import org.thingsboard.server.common.data.kv.KvEntry; import java.io.IOException; import java.util.ArrayList; @@ -212,4 +214,20 @@ public class JacksonUtil { } } } + + public static void addKvEntry(ObjectNode entityNode, KvEntry kvEntry) { + if (kvEntry.getDataType() == DataType.BOOLEAN) { + kvEntry.getBooleanValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); + } else if (kvEntry.getDataType() == DataType.DOUBLE) { + kvEntry.getDoubleValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); + } else if (kvEntry.getDataType() == DataType.LONG) { + kvEntry.getLongValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); + } else if (kvEntry.getDataType() == DataType.JSON) { + if (kvEntry.getJsonValue().isPresent()) { + entityNode.set(kvEntry.getKey(), JacksonUtil.toJsonNode(kvEntry.getJsonValue().get())); + } + } else { + entityNode.put(kvEntry.getKey(), kvEntry.getValueAsString()); + } + } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/aspect/SqlDaoCallsAspect.java b/dao/src/main/java/org/thingsboard/server/dao/aspect/SqlDaoCallsAspect.java index ec8c97e19f..9c4ca8c536 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/aspect/SqlDaoCallsAspect.java +++ b/dao/src/main/java/org/thingsboard/server/dao/aspect/SqlDaoCallsAspect.java @@ -21,7 +21,6 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.ArrayUtils; -import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.exception.ExceptionUtils; import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.annotation.Around; @@ -29,9 +28,11 @@ import org.aspectj.lang.annotation.Aspect; import org.aspectj.lang.reflect.MethodSignature; import org.checkerframework.checker.nullness.qual.Nullable; import org.hibernate.exception.JDBCConnectionException; +import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.id.TenantId; import java.util.Arrays; @@ -46,6 +47,8 @@ import java.util.concurrent.ConcurrentMap; import java.util.function.Consumer; import java.util.stream.Collectors; +import static org.apache.commons.lang3.StringUtils.join; + @Aspect @ConditionalOnProperty(prefix = "sql", value = "log_tenant_stats", havingValue = "true") @Component @@ -55,6 +58,12 @@ public class SqlDaoCallsAspect { private final Set invalidTenantDbCallMethods = ConcurrentHashMap.newKeySet(); private final ConcurrentMap statsMap = new ConcurrentHashMap<>(); + @Value("${sql.batch_sort:true}") + private boolean batchSortEnabled; + + private static final String DEADLOCK_DETECTED_ERROR = "deadlock detected"; + + @Scheduled(initialDelayString = "${sql.log_tenant_stats_interval_ms:60000}", fixedDelayString = "${sql.log_tenant_stats_interval_ms:60000}") public void printStats() { @@ -135,7 +144,7 @@ public class SqlDaoCallsAspect { @SuppressWarnings({"rawtypes", "unchecked"}) @Around("@within(org.thingsboard.server.dao.util.SqlDao)") - public Object logExecutionTime(ProceedingJoinPoint joinPoint) throws Throwable { + public Object handleSqlCall(ProceedingJoinPoint joinPoint) throws Throwable { MethodSignature signature = (MethodSignature) joinPoint.getSignature(); var methodName = signature.toShortString(); if (invalidTenantDbCallMethods.contains(methodName)) { @@ -155,29 +164,47 @@ public class SqlDaoCallsAspect { new FutureCallback<>() { @Override public void onSuccess(@Nullable Object result) { - logTenantMethodExecution(tenantId, methodName, true, startTime, null); + reportSuccessfulMethodExecution(tenantId, methodName, startTime); } @Override public void onFailure(Throwable t) { - logTenantMethodExecution(tenantId, methodName, false, startTime, t); + reportFailedMethodExecution(tenantId, methodName, startTime, t, joinPoint); } }, MoreExecutors.directExecutor()); } else { - logTenantMethodExecution(tenantId, methodName, true, startTime, null); + reportSuccessfulMethodExecution(tenantId, methodName, startTime); } return result; } catch (Throwable t) { - logTenantMethodExecution(tenantId, methodName, false, startTime, t); + reportFailedMethodExecution(tenantId, methodName, startTime, t, joinPoint); throw t; } } - private void logTenantMethodExecution(TenantId tenantId, String method, boolean success, long startTime, Throwable t) { - if (!success && ExceptionUtils.indexOfThrowable(t, JDBCConnectionException.class) >= 0) { - return; + private void reportFailedMethodExecution(TenantId tenantId, String method, long startTime, Throwable t, ProceedingJoinPoint joinPoint) { + if (t != null) { + if (ExceptionUtils.indexOfThrowable(t, JDBCConnectionException.class) >= 0) { + return; + } + if (StringUtils.containedByAny(DEADLOCK_DETECTED_ERROR, ExceptionUtils.getRootCauseMessage(t), ExceptionUtils.getMessage(t))) { + if (!batchSortEnabled) { + log.warn("Deadlock was detected for method {} (tenant: {}). You might need to enable 'sql.batch_sort' option.", method, tenantId); + } else { + log.error("Deadlock was detected for method {} (tenant: {}). Arguments passed: \n{}\n The error: ", + method, tenantId, join(joinPoint.getArgs(), System.lineSeparator()), t); + } + } } + reportMethodExecution(tenantId, method, false, startTime); + } + + private void reportSuccessfulMethodExecution(TenantId tenantId, String method, long startTime) { + reportMethodExecution(tenantId, method, true, startTime); + } + + private void reportMethodExecution(TenantId tenantId, String method, boolean success, long startTime) { statsMap.computeIfAbsent(tenantId, DbCallStats::new) .onMethodCall(method, success, System.currentTimeMillis() - startTime); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/AttributeKvInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/AttributeKvInsertRepository.java index 472652e357..229d4f5740 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/AttributeKvInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/AttributeKvInsertRepository.java @@ -25,6 +25,7 @@ import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallbackWithoutResult; import org.springframework.transaction.support.TransactionTemplate; import org.thingsboard.server.dao.model.sql.AttributeKvEntity; +import org.thingsboard.server.dao.util.SqlDao; import java.sql.PreparedStatement; import java.sql.SQLException; @@ -35,6 +36,7 @@ import java.util.regex.Pattern; @Repository @Slf4j +@SqlDao public abstract class AttributeKvInsertRepository { private static final ThreadLocal PATTERN_THREAD_LOCAL = ThreadLocal.withInitial(() -> Pattern.compile(String.valueOf(Character.MIN_VALUE))); @@ -58,7 +60,7 @@ public abstract class AttributeKvInsertRepository { @Value("${sql.remove_null_chars:true}") private boolean removeNullChars; - protected void saveOrUpdate(List entities) { + public void saveOrUpdate(List entities) { transactionTemplate.execute(new TransactionCallbackWithoutResult() { @Override protected void doInTransactionWithoutResult(TransactionStatus status) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java index e040789f65..99ee1a9b00 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java @@ -78,7 +78,7 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl @Value("${sql.attributes.batch_threads:4}") private int batchThreads; - @Value("${sql.batch_sort:false}") + @Value("${sql.batch_sort:true}") private boolean batchSortEnabled; private TbSqlBlockingQueueWrapper queue; diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/SqlAttributesInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/SqlAttributesInsertRepository.java index e18f5178a8..adac4b4892 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/SqlAttributesInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/SqlAttributesInsertRepository.java @@ -17,9 +17,11 @@ package org.thingsboard.server.dao.sql.attributes; import org.springframework.stereotype.Repository; import org.springframework.transaction.annotation.Transactional; +import org.thingsboard.server.dao.util.SqlDao; @Repository @Transactional +@SqlDao public class SqlAttributesInsertRepository extends AttributeKvInsertRepository { } \ No newline at end of file diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java index 85220ddf9f..cb91a5e674 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java @@ -31,6 +31,7 @@ import org.thingsboard.server.common.data.event.LifecycleEvent; import org.thingsboard.server.common.data.event.RuleChainDebugEvent; import org.thingsboard.server.common.data.event.RuleNodeDebugEvent; import org.thingsboard.server.common.data.event.StatisticsEvent; +import org.thingsboard.server.dao.util.SqlDao; import javax.annotation.PostConstruct; import java.sql.PreparedStatement; @@ -45,6 +46,7 @@ import java.util.stream.Collectors; @Repository @Transactional +@SqlDao public class EventInsertRepository { private static final ThreadLocal PATTERN_THREAD_LOCAL = ThreadLocal.withInitial(() -> Pattern.compile(String.valueOf(Character.MIN_VALUE))); @@ -81,7 +83,7 @@ public class EventInsertRepository { "VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING;"); } - protected void save(List entities) { + public void save(List entities) { Map> eventsByType = entities.stream().collect(Collectors.groupingBy(Event::getType, Collectors.toList())); transactionTemplate.execute(new TransactionCallbackWithoutResult() { @Override diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java index 62ee11b828..4332319f0c 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java @@ -106,7 +106,7 @@ public class JpaBaseEventDao implements EventDao { @Value("${sql.events.batch_threads:3}") private int batchThreads; - @Value("${sql.batch_sort:false}") + @Value("${sql.batch_sort:true}") private boolean batchSortEnabled; private TbSqlBlockingQueueWrapper queue; diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java index 3de21e41bf..8439a4a063 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java @@ -63,7 +63,7 @@ public abstract class AbstractSqlTimeseriesDao extends BaseAbstractSqlTimeseries @Value("${sql.timescale.batch_threads:4}") protected int timescaleBatchThreads; - @Value("${sql.batch_sort:false}") + @Value("${sql.batch_sort:true}") protected boolean batchSortEnabled; @Value("${sql.ttl.ts.ts_key_value_ttl:0}") diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java index 00933140fb..214f7f0d92 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java @@ -94,7 +94,7 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme @Value("${sql.ts_latest.batch_threads:4}") private int tsLatestBatchThreads; - @Value("${sql.batch_sort:false}") + @Value("${sql.batch_sort:true}") protected boolean batchSortEnabled; @Autowired diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/latest/sql/SqlLatestInsertTsRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/latest/sql/SqlLatestInsertTsRepository.java index 451ba29987..087f5389da 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/latest/sql/SqlLatestInsertTsRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/latest/sql/SqlLatestInsertTsRepository.java @@ -24,6 +24,7 @@ import org.springframework.transaction.support.TransactionCallbackWithoutResult; import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity; import org.thingsboard.server.dao.sqlts.insert.AbstractInsertRepository; import org.thingsboard.server.dao.sqlts.insert.latest.InsertLatestTsRepository; +import org.thingsboard.server.dao.util.SqlDao; import org.thingsboard.server.dao.util.SqlTsLatestAnyDao; import java.sql.PreparedStatement; @@ -36,6 +37,7 @@ import java.util.List; @SqlTsLatestAnyDao @Repository @Transactional +@SqlDao public class SqlLatestInsertTsRepository extends AbstractInsertRepository implements InsertLatestTsRepository { @Value("${sql.ts_latest.update_by_latest_ts:true}") diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 116e59880e..2f87f0699b 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -31,10 +31,10 @@ import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.QueueId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.rule.RuleNode; @@ -63,6 +63,7 @@ import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.user.UserService; +import java.util.List; import java.util.Set; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -184,6 +185,10 @@ public interface TbContext { // TODO: Does this changes the message? TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action); + TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes); + + TbMsg attributesDeletedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List keys); + void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId); /* diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java index fd69303501..b9f4b7085e 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java @@ -21,6 +21,7 @@ import com.google.gson.JsonElement; import com.google.gson.JsonParser; import com.google.gson.JsonPrimitive; import lombok.extern.slf4j.Slf4j; +import org.jetbrains.annotations.NotNull; import org.thingsboard.common.util.CollectionsUtil; import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; @@ -74,6 +75,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { if (DataConstants.ATTRIBUTES_UPDATED.equals(msg.getType()) || DataConstants.ATTRIBUTES_DELETED.equals(msg.getType()) || DataConstants.ACTIVITY_EVENT.equals(msg.getType()) || + DataConstants.INACTIVITY_EVENT.equals(msg.getType()) || SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msg.getType())) { if (!msg.getMetaData().getData().isEmpty()) { long now = System.currentTimeMillis(); @@ -89,25 +91,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { long startTime = entityView.getStartTimeMs(); long endTime = entityView.getEndTimeMs(); if ((endTime != 0 && endTime > now && startTime < now) || (endTime == 0 && startTime < now)) { - if (DataConstants.ATTRIBUTES_UPDATED.equals(msg.getType()) || - DataConstants.ACTIVITY_EVENT.equals(msg.getType()) || - SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msg.getType())) { - Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); - List filteredAttributes = - attributes.stream().filter(attr -> attributeContainsInEntityView(scope, attr.getKey(), entityView)).collect(Collectors.toList()); - ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes, - new FutureCallback() { - @Override - public void onSuccess(@Nullable Void result) { - transformAndTellNext(ctx, msg, entityView); - } - - @Override - public void onFailure(Throwable t) { - ctx.tellFailure(msg, t); - } - }); - } else if (DataConstants.ATTRIBUTES_DELETED.equals(msg.getType())) { + if (DataConstants.ATTRIBUTES_DELETED.equals(msg.getType())) { List attributes = new ArrayList<>(); for (JsonElement element : new JsonParser().parse(msg.getData()).getAsJsonObject().get("attributes").getAsJsonArray()) { if (element.isJsonPrimitive()) { @@ -120,9 +104,15 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { List filteredAttributes = attributes.stream().filter(attr -> attributeContainsInEntityView(scope, attr, entityView)).collect(Collectors.toList()); if (!filteredAttributes.isEmpty()) { - ctx.getAttributesService().removeAll(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes); - transformAndTellNext(ctx, msg, entityView); + ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes, + getFutureCallback(ctx, msg, entityView)); } + } else { + Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); + List filteredAttributes = + attributes.stream().filter(attr -> attributeContainsInEntityView(scope, attr.getKey(), entityView)).collect(Collectors.toList()); + ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes, + getFutureCallback(ctx, msg, entityView)); } } } @@ -137,6 +127,21 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { } } + @NotNull + private FutureCallback getFutureCallback(TbContext ctx, TbMsg msg, EntityView entityView) { + return new FutureCallback() { + @Override + public void onSuccess(@Nullable Void result) { + transformAndTellNext(ctx, msg, entityView); + } + + @Override + public void onFailure(Throwable t) { + ctx.tellFailure(msg, t); + } + }; + } + private void transformAndTellNext(TbContext ctx, TbMsg msg, EntityView entityView) { ctx.enqueueForTellNext(ctx.newMsg(msg.getQueueName(), msg.getType(), entityView.getId(), msg.getCustomerId(), msg.getMetaData(), msg.getData()), SUCCESS); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java index 9b9974cbb9..d68b29734c 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java @@ -210,9 +210,13 @@ public class TbDeviceProfileNode implements TbNode { DeviceState deviceState = deviceStates.get(deviceId); if (deviceState != null) { DeviceProfileId currentProfileId = deviceState.getProfileId(); - Device device = JacksonUtil.fromString(deviceJson, Device.class); - if (!currentProfileId.equals(device.getDeviceProfileId())) { - removeDeviceState(deviceId); + try { + Device device = JacksonUtil.fromString(deviceJson, Device.class); + if (!currentProfileId.equals(device.getDeviceProfileId())) { + removeDeviceState(deviceId); + } + } catch (IllegalArgumentException e) { + log.debug("[{}] Received device update notification with non-device msg body: [{}][{}]", ctx.getSelfId(), deviceId, e); } } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesDeleteNodeCallback.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesDeleteNodeCallback.java new file mode 100644 index 0000000000..57bfc5d129 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesDeleteNodeCallback.java @@ -0,0 +1,45 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.telemetry; + +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.server.common.msg.TbMsg; + +import javax.annotation.Nullable; +import java.util.List; + +@Slf4j +public class AttributesDeleteNodeCallback extends TelemetryNodeCallback { + + private String scope; + private List keys; + + public AttributesDeleteNodeCallback(TbContext ctx, TbMsg msg, String scope, List keys) { + super(ctx, msg); + this.scope = scope; + this.keys = keys; + } + + @Override + public void onSuccess(@Nullable Void result) { + TbContext ctx = this.getCtx(); + TbMsg tbMsg = this.getMsg(); + ctx.enqueue(ctx.attributesDeletedActionMsg(tbMsg.getOriginator(), ctx.getSelfId(), scope, keys), + () -> ctx.tellSuccess(tbMsg), + throwable -> ctx.tellFailure(tbMsg, throwable)); + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesUpdateNodeCallback.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesUpdateNodeCallback.java new file mode 100644 index 0000000000..f8ee4a29b1 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesUpdateNodeCallback.java @@ -0,0 +1,44 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.telemetry; + +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.msg.TbMsg; + +import javax.annotation.Nullable; +import java.util.List; + +public class AttributesUpdateNodeCallback extends TelemetryNodeCallback { + + private String scope; + private List attributes; + + public AttributesUpdateNodeCallback(TbContext ctx, TbMsg msg, String scope, List attributes) { + super(ctx, msg); + this.scope = scope; + this.attributes = attributes; + } + + @Override + public void onSuccess(@Nullable Void result) { + TbContext ctx = this.getCtx(); + TbMsg tbMsg = this.getMsg(); + ctx.enqueue(ctx.attributesUpdatedActionMsg(tbMsg.getOriginator(), ctx.getSelfId(), scope, attributes), + () -> ctx.tellSuccess(tbMsg), + throwable -> ctx.tellFailure(tbMsg, throwable)); + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java index 9f1c0d4e20..b810774832 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java @@ -31,7 +31,7 @@ import org.thingsboard.server.common.msg.session.SessionMsgType; import org.thingsboard.server.common.transport.adaptor.JsonConverter; import java.util.ArrayList; -import java.util.Set; +import java.util.List; @Slf4j @RuleNode( @@ -39,7 +39,9 @@ import java.util.Set; name = "save attributes", configClazz = TbMsgAttributesNodeConfiguration.class, nodeDescription = "Saves attributes data", - nodeDetails = "Saves entity attributes based on configurable scope parameter. Expects messages with 'POST_ATTRIBUTES_REQUEST' message type", + nodeDetails = "Saves entity attributes based on configurable scope parameter. Expects messages with 'POST_ATTRIBUTES_REQUEST' message type. " + + "If upsert(update/insert) operation is completed successfully, rule node will send the \"Attributes Updated\" " + + "event to the root chain of the message originator and send the incoming message via Success chain, otherwise, Failure chain is used.", uiResources = {"static/rulenode/rulenode-core-config.js"}, configDirective = "tbActionNodeAttributesConfig", icon = "file_upload" @@ -63,15 +65,19 @@ public class TbMsgAttributesNode implements TbNode { return; } String src = msg.getData(); - Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(src)); + List attributes = new ArrayList<>(JsonConverter.convertToAttributes(new JsonParser().parse(src))); + if (attributes.isEmpty()) { + ctx.tellSuccess(msg); + return; + } String notifyDeviceStr = msg.getMetaData().getValue("notifyDevice"); ctx.getTelemetryService().saveAndNotify( ctx.getTenantId(), msg.getOriginator(), config.getScope(), - new ArrayList<>(attributes), + attributes, config.getNotifyDevice() || StringUtils.isEmpty(notifyDeviceStr) || Boolean.parseBoolean(notifyDeviceStr), - new TelemetryNodeCallback(ctx, msg) + new AttributesUpdateNodeCallback(ctx, msg, config.getScope(), attributes) ); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java new file mode 100644 index 0000000000..e8839291a0 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java @@ -0,0 +1,73 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.telemetry; + +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.rule.engine.api.RuleNode; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.api.TbNode; +import org.thingsboard.rule.engine.api.TbNodeConfiguration; +import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.server.common.data.StringUtils; +import org.thingsboard.server.common.data.plugin.ComponentType; +import org.thingsboard.server.common.msg.TbMsg; + +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.stream.Collectors; + +@Slf4j +@RuleNode( + type = ComponentType.ACTION, + name = "delete attributes", + configClazz = TbMsgDeleteAttributesConfiguration.class, + nodeDescription = "Delete attributes for Message Originator.", + nodeDetails = "Attempt to remove attributes by selected keys. If msg originator doesn't have an attribute with " + + " a key selected in the configuration, it will be ignored. If delete operation is completed successfully, " + + " rule node will send the \"Attributes Deleted\" event to the root chain of the message originator and " + + " send the incoming message via Success chain, otherwise, Failure chain is used.", + uiResources = {"static/rulenode/rulenode-core-config.js"}, + configDirective = "tbActionNodeDeleteAttributesConfig", + icon = "remove_circle" +) +public class TbMsgDeleteAttributes implements TbNode { + + private TbMsgDeleteAttributesConfiguration config; + private String scope; + private List keys; + + @Override + public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { + this.config = TbNodeUtils.convert(configuration, TbMsgDeleteAttributesConfiguration.class); + this.scope = config.getScope(); + this.keys = config.getKeys(); + } + + @Override + public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException { + List keysToDelete = keys.stream() + .map(keyPattern -> TbNodeUtils.processPattern(keyPattern, msg)) + .distinct() + .filter(StringUtils::isNotBlank) + .collect(Collectors.toList()); + if (keysToDelete.isEmpty()) { + ctx.tellSuccess(msg); + } else { + ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), scope, keysToDelete, new AttributesDeleteNodeCallback(ctx, msg, scope, keysToDelete)); + } + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java new file mode 100644 index 0000000000..03bef2a4e6 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java @@ -0,0 +1,38 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.telemetry; + +import lombok.Data; +import org.thingsboard.rule.engine.api.NodeConfiguration; +import org.thingsboard.server.common.data.DataConstants; + +import java.util.Collections; +import java.util.List; + +@Data +public class TbMsgDeleteAttributesConfiguration implements NodeConfiguration { + + private String scope; + private List keys; + + @Override + public TbMsgDeleteAttributesConfiguration defaultConfiguration() { + TbMsgDeleteAttributesConfiguration configuration = new TbMsgDeleteAttributesConfiguration(); + configuration.setScope(DataConstants.SERVER_SCOPE); + configuration.setKeys(Collections.emptyList()); + return configuration; + } +} diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java new file mode 100644 index 0000000000..8972a3aa1f --- /dev/null +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java @@ -0,0 +1,125 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.telemetry; + +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.junit.MockitoJUnitRunner; +import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.api.TbNodeConfiguration; +import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.queue.TbMsgCallback; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.function.Consumer; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.BDDMockito.willReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +@Slf4j +public class TbMsgDeleteAttributesTest { + final ObjectMapper mapper = new ObjectMapper(); + + DeviceId deviceId; + TbMsgDeleteAttributes node; + TbMsgDeleteAttributesConfiguration config; + TbNodeConfiguration nodeConfiguration; + TbContext ctx; + TbMsgCallback callback; + + RuleEngineTelemetryService telemetryService; + + @BeforeEach + void setUp() throws TbNodeException { + deviceId = new DeviceId(UUID.randomUUID()); + callback = mock(TbMsgCallback.class); + ctx = mock(TbContext.class); + config = new TbMsgDeleteAttributesConfiguration().defaultConfiguration(); + config.setKeys(List.of("${TestAttribute_1}", "$[TestAttribute_2]", "$[TestAttribute_3]", "TestAttribute_4")); + nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); + node = spy(new TbMsgDeleteAttributes()); + node.init(ctx, nodeConfiguration); + telemetryService = mock(RuleEngineTelemetryService.class); + + willReturn(telemetryService).given(ctx).getTelemetryService(); + willAnswer(invocation -> { + TelemetryNodeCallback callBack = invocation.getArgument(4); + callBack.onSuccess(null); + return null; + }).given(telemetryService).deleteAndNotify( + any(), any(), anyString(), anyList(), any()); + } + + @AfterEach + void tearDown() { + node.destroy(); + } + + @Test + void givenDefaultConfig_whenVerify_thenOK() { + TbMsgDeleteAttributesConfiguration defaultConfig = new TbMsgDeleteAttributesConfiguration().defaultConfiguration(); + assertThat(defaultConfig.getScope()).isEqualTo(DataConstants.SERVER_SCOPE); + assertThat(defaultConfig.getKeys()).isEqualTo(Collections.emptyList()); + } + + @Test + void givenMsg_whenOnMsg_thenVerifyOutput() throws Exception { + final Map mdMap = Map.of( + "TestAttribute_1", "temperature", + "city", "NY" + ); + final TbMsgMetaData metaData = new TbMsgMetaData(mdMap); + final String data = "{\"TestAttribute_2\": \"humidity\", \"TestAttribute_3\": \"voltage\"}"; + + TbMsg msg = TbMsg.newMsg("POST_ATTRIBUTES_REQUEST", deviceId, metaData, data, callback); + node.onMsg(ctx, msg); + + ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); + ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + + verify(ctx, times(1)).enqueue(any(), successCaptor.capture(), failureCaptor.capture()); + successCaptor.getValue().run(); + verify(ctx, times(1)).tellSuccess(newMsgCaptor.capture()); + + verify(ctx, times(1)).attributesDeletedActionMsg(any(), any(), anyString(), anyList()); + verify(ctx, never()).tellFailure(any(), any()); + verify(telemetryService, times(1)).deleteAndNotify(any(), any(), anyString(), anyList(), any()); + } +}