fix TbGetTelemetryNode (#1353)

TbGetTelemetryNode - get start and end ts from metadata

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNode.java

@@ -21,8 +21,15 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.util.concurrent.ListenableFuture;
+import lombok.Data;
+import lombok.NoArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.thingsboard.rule.engine.api.*;
+import org.apache.commons.lang3.math.NumberUtils;
+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.DonAsynchron;
 import org.thingsboard.rule.engine.api.util.TbNodeUtils;
 import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
@@ -37,7 +44,9 @@ import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
-import static org.thingsboard.rule.engine.metadata.TbGetTelemetryNodeConfiguration.*;
+import static org.thingsboard.rule.engine.metadata.TbGetTelemetryNodeConfiguration.FETCH_MODE_ALL;
+import static org.thingsboard.rule.engine.metadata.TbGetTelemetryNodeConfiguration.FETCH_MODE_FIRST;
+import static org.thingsboard.rule.engine.metadata.TbGetTelemetryNodeConfiguration.MAX_FETCH_SIZE;
 import static org.thingsboard.server.common.data.kv.Aggregation.NONE;
 
 /**
@@ -57,8 +66,6 @@ public class TbGetTelemetryNode implements TbNode {
 
     private TbGetTelemetryNodeConfiguration config;
     private List<String> tsKeyNames;
-    private long startTsOffset;
-    private long endTsOffset;
     private int limit;
     private ObjectMapper mapper;
     private String fetchMode;
@@ -67,8 +74,6 @@ public class TbGetTelemetryNode implements TbNode {
     public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbGetTelemetryNodeConfiguration.class);
         tsKeyNames = config.getLatestTsKeyNames();
-        startTsOffset = TimeUnit.valueOf(config.getStartIntervalTimeUnit()).toMillis(config.getStartInterval());
-        endTsOffset = TimeUnit.valueOf(config.getEndIntervalTimeUnit()).toMillis(config.getEndInterval());
         limit = config.getFetchMode().equals(FETCH_MODE_ALL) ? MAX_FETCH_SIZE : 1;
         fetchMode = config.getFetchMode();
         mapper = new ObjectMapper();
@@ -82,8 +87,8 @@ public class TbGetTelemetryNode implements TbNode {
             ctx.tellFailure(msg, new IllegalStateException("Telemetry is not selected!"));
         } else {
             try {
-                List<ReadTsKvQuery> queries = buildQueries();
+                checkMetadataKeyPatterns(msg);
-                ListenableFuture<List<TsKvEntry>> list = ctx.getTimeseriesService().findAll(ctx.getTenantId(), msg.getOriginator(), queries);
+                ListenableFuture<List<TsKvEntry>> list = ctx.getTimeseriesService().findAll(ctx.getTenantId(), msg.getOriginator(), buildQueries(msg));
                 DonAsynchron.withCallback(list, data -> {
                     process(data, msg);
                     TbMsg newMsg = ctx.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), msg.getData());
@@ -95,10 +100,12 @@ public class TbGetTelemetryNode implements TbNode {
         }
     }
 
-    private List<ReadTsKvQuery> buildQueries() {
+    @Override
-        long ts = System.currentTimeMillis();
+    public void destroy() {
-        long startTs = ts - startTsOffset;
+
-        long endTs = ts - endTsOffset;
+    }
+
+    private List<ReadTsKvQuery> buildQueries(TbMsg msg) {
         String orderBy;
         if (fetchMode.equals(FETCH_MODE_FIRST) || fetchMode.equals(FETCH_MODE_ALL)) {
             orderBy = "ASC";
@@ -106,7 +113,7 @@ public class TbGetTelemetryNode implements TbNode {
             orderBy = "DESC";
         }
         return tsKeyNames.stream()
-                .map(key -> new BaseReadTsKvQuery(key, startTs, endTs, 1, limit, NONE, orderBy))
+                .map(key -> new BaseReadTsKvQuery(key, getInterval(msg).getStartTs(), getInterval(msg).getEndTs(), 1, limit, NONE, orderBy))
                 .collect(Collectors.toList());
     }
 
@@ -162,8 +169,79 @@ public class TbGetTelemetryNode implements TbNode {
         return obj;
     }
 
-    @Override
+    private Interval getInterval(TbMsg msg) {
-    public void destroy() {
+        Interval interval = new Interval();
+        if (config.isUseMetadataIntervalPatterns()) {
+            if (isParsable(msg, config.getStartIntervalPattern())) {
+                interval.setStartTs(Long.parseLong(TbNodeUtils.processPattern(config.getStartIntervalPattern(), msg.getMetaData())));
+            }
+            if (isParsable(msg, config.getEndIntervalPattern())) {
+                interval.setEndTs(Long.parseLong(TbNodeUtils.processPattern(config.getEndIntervalPattern(), msg.getMetaData())));
+            }
+        } else {
+            long ts = System.currentTimeMillis();
+            interval.setStartTs(ts - TimeUnit.valueOf(config.getStartIntervalTimeUnit()).toMillis(config.getStartInterval()));
+            interval.setEndTs(ts - TimeUnit.valueOf(config.getEndIntervalTimeUnit()).toMillis(config.getEndInterval()));
+        }
+        return interval;
+    }
+
+    private boolean isParsable(TbMsg msg, String pattern) {
+        return NumberUtils.isParsable(TbNodeUtils.processPattern(pattern, msg.getMetaData()));
+    }
 
+    private void checkMetadataKeyPatterns(TbMsg msg) {
+        isUndefined(msg, config.getStartIntervalPattern(), config.getEndIntervalPattern());
+        isInvalid(msg, config.getStartIntervalPattern(), config.getEndIntervalPattern());
     }
+
+    private void isUndefined(TbMsg msg, String startIntervalPattern, String endIntervalPattern) {
+        if (getMetadataValue(msg, startIntervalPattern) == null && getMetadataValue(msg, endIntervalPattern) == null) {
+            throw new IllegalArgumentException("Message metadata values: '" +
+                    replaceRegex(startIntervalPattern) + "' and '" +
+                    replaceRegex(endIntervalPattern) + "' are undefined");
+        } else {
+            if (getMetadataValue(msg, startIntervalPattern) == null) {
+                throw new IllegalArgumentException("Message metadata value: '" +
+                        replaceRegex(startIntervalPattern) + "' is undefined");
+            }
+            if (getMetadataValue(msg, endIntervalPattern) == null) {
+                throw new IllegalArgumentException("Message metadata value: '" +
+                        replaceRegex(endIntervalPattern) + "' is undefined");
+            }
+        }
+    }
+
+    private void isInvalid(TbMsg msg, String startIntervalPattern, String endIntervalPattern) {
+        if (getInterval(msg).getStartTs() == null && getInterval(msg).getEndTs() == null) {
+            throw new IllegalArgumentException("Message metadata values: '" +
+                    replaceRegex(startIntervalPattern) + "' and '" +
+                    replaceRegex(endIntervalPattern) + "' have invalid format");
+        } else {
+            if (getInterval(msg).getStartTs() == null) {
+                throw new IllegalArgumentException("Message metadata value: '" +
+                        replaceRegex(startIntervalPattern) + "' has invalid format");
+            }
+            if (getInterval(msg).getEndTs() == null) {
+                throw new IllegalArgumentException("Message metadata value: '" +
+                        replaceRegex(endIntervalPattern) + "' has invalid format");
+            }
+        }
+    }
+
+    private String getMetadataValue(TbMsg msg, String pattern) {
+        return msg.getMetaData().getValue(replaceRegex(pattern));
+    }
+
+    private String replaceRegex(String pattern) {
+        return pattern.replaceAll("[${}]", "");
+    }
+
+    @Data
+    @NoArgsConstructor
+    private static class Interval {
+        private Long startTs;
+        private Long endTs;
+    }
+
 }

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNodeConfiguration.java

@@ -35,6 +35,12 @@ public class TbGetTelemetryNodeConfiguration implements NodeConfiguration<TbGetT
 
     private int startInterval;
     private int endInterval;
+
+    private String startIntervalPattern;
+    private String endIntervalPattern;
+
+    private boolean useMetadataIntervalPatterns;
+
     private String startIntervalTimeUnit;
     private String endIntervalTimeUnit;
     private String fetchMode; //FIRST, LAST, LATEST
@@ -52,6 +58,9 @@ public class TbGetTelemetryNodeConfiguration implements NodeConfiguration<TbGetT
         configuration.setStartInterval(2);
         configuration.setEndIntervalTimeUnit(TimeUnit.MINUTES.name());
         configuration.setEndInterval(1);
+        configuration.setUseMetadataIntervalPatterns(false);
+        configuration.setStartIntervalPattern("");
+        configuration.setEndIntervalPattern("");
         return configuration;
     }
 }