From cc0e5417887266847e4e5ac2cf9b41a8023bdba5 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Wed, 29 Jan 2020 16:58:52 +0200 Subject: [PATCH] fixed tell-failure for attributes fetch node (#2365) * fixed tell-failure for attributes fetch node --- .../server/common/data/DataConstants.java | 1 + .../metadata/TbAbstractGetAttributesNode.java | 81 +++++++++++++------ 2 files changed, 59 insertions(+), 23 deletions(-) diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java index 6e606e2ad5..afb0dbeba6 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java @@ -27,6 +27,7 @@ public class DataConstants { public static final String CLIENT_SCOPE = "CLIENT_SCOPE"; public static final String SERVER_SCOPE = "SERVER_SCOPE"; public static final String SHARED_SCOPE = "SHARED_SCOPE"; + public static final String LATEST_TS = "LATEST_TS"; public static final String[] allScopes() { return new String[]{CLIENT_SCOPE, SHARED_SCOPE, SERVER_SCOPE}; diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java index 9cbd433173..4db628a7f4 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java @@ -29,16 +29,20 @@ import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.msg.TbMsg; +import java.util.ArrayList; import java.util.List; - +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; import static org.thingsboard.common.util.DonAsynchron.withCallback; import static org.thingsboard.rule.engine.api.TbRelationTypes.FAILURE; import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS; import static org.thingsboard.server.common.data.DataConstants.CLIENT_SCOPE; +import static org.thingsboard.server.common.data.DataConstants.LATEST_TS; import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE; import static org.thingsboard.server.common.data.DataConstants.SHARED_SCOPE; @@ -82,40 +86,43 @@ public abstract class TbAbstractGetAttributesNode> failuresMap = new ConcurrentHashMap<>(); ListenableFuture> allFutures = Futures.allAsList( - putLatestTelemetry(ctx, entityId, msg, config.getLatestTsKeyNames()), - putAttrAsync(ctx, entityId, msg, CLIENT_SCOPE, config.getClientAttributeNames(), "cs_"), - putAttrAsync(ctx, entityId, msg, SHARED_SCOPE, config.getSharedAttributeNames(), "shared_"), - putAttrAsync(ctx, entityId, msg, SERVER_SCOPE, config.getServerAttributeNames(), "ss_") + putLatestTelemetry(ctx, entityId, msg, LATEST_TS, config.getLatestTsKeyNames(), failuresMap), + putAttrAsync(ctx, entityId, msg, CLIENT_SCOPE, config.getClientAttributeNames(), failuresMap, "cs_"), + putAttrAsync(ctx, entityId, msg, SHARED_SCOPE, config.getSharedAttributeNames(), failuresMap, "shared_"), + putAttrAsync(ctx, entityId, msg, SERVER_SCOPE, config.getServerAttributeNames(), failuresMap, "ss_") ); - withCallback(allFutures, i -> ctx.tellNext(msg, SUCCESS), t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); + withCallback(allFutures, i -> { + if (!failuresMap.isEmpty()) { + throw reportFailures(failuresMap); + } + ctx.tellNext(msg, SUCCESS); + }, t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); } - private ListenableFuture putAttrAsync(TbContext ctx, EntityId entityId, TbMsg msg, String scope, List keys, String prefix) { + private ListenableFuture putAttrAsync(TbContext ctx, EntityId entityId, TbMsg msg, String scope, List keys, ConcurrentHashMap> failuresMap, String prefix) { if (CollectionUtils.isEmpty(keys)) { return Futures.immediateFuture(null); } - ListenableFuture> latest = ctx.getAttributesService().find(ctx.getTenantId(), entityId, scope, keys); - return Futures.transform(latest, l -> { - l.forEach(r -> { + ListenableFuture> attributeKvEntryListFuture = ctx.getAttributesService().find(ctx.getTenantId(), entityId, scope, keys); + return Futures.transform(attributeKvEntryListFuture, attributeKvEntryList -> { + if (!CollectionUtils.isEmpty(attributeKvEntryList)) { + List existingAttributesKvEntry = attributeKvEntryList.stream().filter(attributeKvEntry -> keys.contains(attributeKvEntry.getKey())).collect(Collectors.toList()); + existingAttributesKvEntry.forEach(kvEntry -> msg.getMetaData().putValue(prefix + kvEntry.getKey(), kvEntry.getValueAsString())); + if (existingAttributesKvEntry.size() != keys.size() && BooleanUtils.toBooleanDefaultIfNull(this.config.isTellFailureIfAbsent(), true)) { + getNotExistingKeys(existingAttributesKvEntry, keys).forEach(key -> computeFailuresMap(scope, failuresMap, key)); + } + } else { if (BooleanUtils.toBooleanDefaultIfNull(this.config.isTellFailureIfAbsent(), true)) { - if (r.getValue() != null) { - msg.getMetaData().putValue(prefix + r.getKey(), r.getValueAsString()); - } else { - throw new RuntimeException("[" + scope + "][" + r.getKey() + "] attribute value is not present in the DB!"); - } - } else { - if (r.getValue() != null) { - msg.getMetaData().putValue(prefix + r.getKey(), r.getValueAsString()); - } + keys.forEach(key -> computeFailuresMap(scope, failuresMap, key)); } - - }); + } return null; }); } - private ListenableFuture putLatestTelemetry(TbContext ctx, EntityId entityId, TbMsg msg, List keys) { + private ListenableFuture putLatestTelemetry(TbContext ctx, EntityId entityId, TbMsg msg, String scope, List keys, ConcurrentHashMap> failuresMap) { if (CollectionUtils.isEmpty(keys)) { return Futures.immediateFuture(null); } @@ -125,7 +132,7 @@ public abstract class TbAbstractGetAttributesNode getNotExistingKeys(List existingAttributesKvEntry, List allKeys) { + List existingKeys = existingAttributesKvEntry.stream().map(KvEntry::getKey).collect(Collectors.toList()); + return allKeys.stream().filter(key -> !existingKeys.contains(key)).collect(Collectors.toList()); + } + + private void computeFailuresMap(String scope, ConcurrentHashMap> failuresMap, String key) { + List failures = failuresMap.computeIfAbsent(scope, k -> new ArrayList<>()); + failures.add(key); + } + + private RuntimeException reportFailures(ConcurrentHashMap> failuresMap) { + StringBuilder errorMessage = new StringBuilder("The following attribute/telemetry keys is not present in the DB: ").append("\n"); + if (failuresMap.containsKey(CLIENT_SCOPE)) { + errorMessage.append("\t").append("[" + CLIENT_SCOPE + "]:").append(failuresMap.get(CLIENT_SCOPE).toString()).append("\n"); + } + if (failuresMap.containsKey(SERVER_SCOPE)) { + errorMessage.append("\t").append("[" + SERVER_SCOPE + "]:").append(failuresMap.get(SERVER_SCOPE).toString()).append("\n"); + } + if (failuresMap.containsKey(SHARED_SCOPE)) { + errorMessage.append("\t").append("[" + SHARED_SCOPE + "]:").append(failuresMap.get(SHARED_SCOPE).toString()).append("\n"); + } + if (failuresMap.containsKey(LATEST_TS)) { + errorMessage.append("\t").append("[" + LATEST_TS + "]:").append(failuresMap.get(LATEST_TS).toString()).append("\n"); + } + failuresMap.clear(); + return new RuntimeException(errorMessage.toString()); + } }