fixed tell-failure for attributes fetch node (#2365)

* fixed tell-failure for attributes fetch node

fixed tell-failure for attributes fetch node (#2365)
* fixed tell-failure for attributes fetch node
ShvaykaD · GitHub
1 parent 0dd313dd
Showing 2 changed files with 59 additions and 23 deletions
common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java
--- a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
View file @cc0e541
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
View file @cc0e541
@@ -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};
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java
View file @cc0e541
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java
View file @cc0e541
@@ -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<C extends TbGetAttributesNodeC
             ctx.tellNext(msg, FAILURE);
             return;
         }
+        ConcurrentHashMap<String, List<String>> failuresMap = new ConcurrentHashMap<>();
         ListenableFuture<List<Void>> 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<Void> putAttrAsync(TbContext ctx, EntityId entityId, TbMsg msg, String scope, List<String> keys, String prefix) {
+    private ListenableFuture<Void> putAttrAsync(TbContext ctx, EntityId entityId, TbMsg msg, String scope, List<String> keys, ConcurrentHashMap<String, List<String>> failuresMap, String prefix) {
         if (CollectionUtils.isEmpty(keys)) {
             return Futures.immediateFuture(null);
         }
-        ListenableFuture<List<AttributeKvEntry>> latest = ctx.getAttributesService().find(ctx.getTenantId(), entityId, scope, keys);
-        return Futures.transform(latest, l -> {
-            l.forEach(r -> {
+        ListenableFuture<List<AttributeKvEntry>> attributeKvEntryListFuture = ctx.getAttributesService().find(ctx.getTenantId(), entityId, scope, keys);
+        return Futures.transform(attributeKvEntryListFuture, attributeKvEntryList -> {
+            if (!CollectionUtils.isEmpty(attributeKvEntryList)) {
+                List<AttributeKvEntry> 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<Void> putLatestTelemetry(TbContext ctx, EntityId entityId, TbMsg msg, List<String> keys) {
+    private ListenableFuture<Void> putLatestTelemetry(TbContext ctx, EntityId entityId, TbMsg msg, String scope, List<String> keys, ConcurrentHashMap<String, List<String>> failuresMap) {
         if (CollectionUtils.isEmpty(keys)) {
             return Futures.immediateFuture(null);
         }
@@ -125,7 +132,7 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
                 boolean getLatestValueWithTs = BooleanUtils.toBooleanDefaultIfNull(this.config.isGetLatestValueWithTs(), false);
                 if (BooleanUtils.toBooleanDefaultIfNull(this.config.isTellFailureIfAbsent(), true)) {
                     if (r.getValue() == null) {
-                        throw new RuntimeException("[" + r.getKey() + "] telemetry value is not present in the DB!");
+                        computeFailuresMap(scope, failuresMap, r.getKey());
                     } else if (getLatestValueWithTs) {
                         putValueWithTs(msg, r);
                     } else {
@@ -164,4 +171,32 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
         }
         msg.getMetaData().putValue(r.getKey(), value.toString());
     }
+
+    private List<String> getNotExistingKeys(List<AttributeKvEntry> existingAttributesKvEntry, List<String> allKeys) {
+        List<String> 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<String, List<String>> failuresMap, String key) {
+        List<String> failures = failuresMap.computeIfAbsent(scope, k -> new ArrayList<>());
+        failures.add(key);
+    }
+
+    private RuntimeException reportFailures(ConcurrentHashMap<String, List<String>> 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());
+    }
 }