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

set ts -1 when default value used

pull/14526/head
IrynaMatveieva 9 months ago
parent
784f1efcde
commit
958e44f7d6
7 changed files with 112 additions and 66 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 68
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java
  2. 5
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
  3. 5
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java
  4. 6
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
  5. 75
      application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java
  6. 17
      application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java
  7. 2
      common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfCtx.java

68
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java
View File

@ -23,7 +23,6 @@ import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.math.NumberUtils;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg;
@ -31,21 +30,14 @@ import org.thingsboard.server.actors.calculatedField.MultipleTbCallback;
import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.cf.configuration.Argument;
import org.thingsboard.server.common.data.cf.configuration.OutputType;
import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.Aggregation;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.common.data.kv.DoubleDataEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
import org.thingsboard.server.common.data.kv.StringDataEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
@ -70,9 +62,6 @@ import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.ScriptCalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.SimpleCalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry;
import java.util.ArrayList;
@ -80,12 +69,16 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import static org.thingsboard.server.common.data.DataConstants.SCOPE;
import static org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry.DEFAULT_TS;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultAttributeEntry;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultTsKvEntry;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createStateByType;
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformSingleValueArgument;
import static org.thingsboard.server.utils.CalculatedFieldUtils.toProto;
@TbRuleEngineComponent
@ -244,30 +237,17 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP
private ListenableFuture<ArgumentEntry> fetchKvEntry(TenantId tenantId, EntityId entityId, Argument argument) {
return switch (argument.getRefEntityKey().getType()) {
case TS_ROLLING -> fetchTsRolling(tenantId, entityId, argument);
case ATTRIBUTE -> transformSingleValueArgument(
Futures.transform(
attributesService.find(tenantId, entityId, argument.getRefEntityKey().getScope(), argument.getRefEntityKey().getKey()),
result -> result.or(() -> Optional.of(new BaseAttributeKvEntry(createDefaultKvEntry(argument), System.currentTimeMillis(), 0L))),
calculatedFieldCallbackExecutor)
);
case TS_LATEST -> transformSingleValueArgument(
Futures.transform(
timeseriesService.findLatest(tenantId, entityId, argument.getRefEntityKey().getKey()),
result -> result.or(() -> Optional.of(new BasicTsKvEntry(System.currentTimeMillis(), createDefaultKvEntry(argument), 0L))),
calculatedFieldCallbackExecutor));
case ATTRIBUTE -> Futures.transform(
attributesService.find(tenantId, entityId, argument.getRefEntityKey().getScope(), argument.getRefEntityKey().getKey()),
result -> transformSingleValueArgument(result.orElseGet(() -> createDefaultAttributeEntry(argument, DEFAULT_TS))),
calculatedFieldCallbackExecutor);
case TS_LATEST -> Futures.transform(
timeseriesService.findLatest(tenantId, entityId, argument.getRefEntityKey().getKey()),
result -> transformSingleValueArgument(result.orElseGet(() -> createDefaultTsKvEntry(argument, DEFAULT_TS))),
calculatedFieldCallbackExecutor);
};
}
private ListenableFuture<ArgumentEntry> transformSingleValueArgument(ListenableFuture<Optional<? extends KvEntry>> kvEntryFuture) {
return Futures.transform(kvEntryFuture, kvEntry -> {
if (kvEntry.isPresent() && kvEntry.get().getValue() != null) {
return ArgumentEntry.createSingleValueArgument(kvEntry.get());
} else {
return new SingleValueArgumentEntry();
}
}, calculatedFieldCallbackExecutor);
}
private ListenableFuture<ArgumentEntry> fetchTsRolling(TenantId tenantId, EntityId entityId, Argument argument) {
long currentTime = System.currentTimeMillis();
long timeWindow = argument.getTimeWindow() == 0 ? System.currentTimeMillis() : argument.getTimeWindow();
@ -282,28 +262,6 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP
return Futures.transform(tsRollingFuture, tsRolling -> tsRolling == null ? new TsRollingArgumentEntry(limit, timeWindow) : ArgumentEntry.createTsRollingArgument(tsRolling, limit, timeWindow), calculatedFieldCallbackExecutor);
}
private KvEntry createDefaultKvEntry(Argument argument) {
String key = argument.getRefEntityKey().getKey();
String defaultValue = argument.getDefaultValue();
if (StringUtils.isBlank(defaultValue)) {
return new StringDataEntry(key, null);
}
if (NumberUtils.isParsable(defaultValue)) {
return new DoubleDataEntry(key, Double.parseDouble(defaultValue));
}
if ("true".equalsIgnoreCase(defaultValue) || "false".equalsIgnoreCase(defaultValue)) {
return new BooleanDataEntry(key, Boolean.parseBoolean(defaultValue));
}
return new StringDataEntry(key, defaultValue);
}
private CalculatedFieldState createStateByType(CalculatedFieldCtx ctx) {
return switch (ctx.getCfType()) {
case SIMPLE -> new SimpleCalculatedFieldState(ctx.getArgNames());
case SCRIPT -> new ScriptCalculatedFieldState(ctx.getArgNames());
};
}
private static class TbCallbackWrapper implements TbQueueCallback {
private final TbCallback callback;

5
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
View File

@ -48,6 +48,11 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
this(new ArrayList<>(), new HashMap<>(), false, DEFAULT_LAST_UPDATE_TS);
}
public long getLatestTimestamp() {
return latestTimestamp == DEFAULT_LAST_UPDATE_TS ? System.currentTimeMillis() : latestTimestamp;
}
@Override
public boolean updateState(CalculatedFieldCtx ctx, Map<String, ArgumentEntry> argumentValues) {
if (arguments == null) {

5
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java
View File

@ -99,10 +99,9 @@ public class SimpleCalculatedFieldState extends BaseCalculatedFieldState {
valuesNode.set(outputName, JacksonUtil.valueToTree(result));
}
long latestTs = getLatestTimestamp();
if (useLatestTs && latestTs != DEFAULT_LAST_UPDATE_TS) {
if (useLatestTs) {
ObjectNode resultNode = JacksonUtil.newObjectNode();
resultNode.put("ts", latestTs);
resultNode.put("ts", getLatestTimestamp());
resultNode.set("values", valuesNode);
return resultNode;
} else {

6
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
View File

@ -19,7 +19,6 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.core.type.TypeReference;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.script.api.tbel.TbelCfArg;
import org.thingsboard.script.api.tbel.TbelCfSingleValueArg;
@ -32,13 +31,12 @@ import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto;
import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
import static org.thingsboard.server.service.cf.ctx.state.BaseCalculatedFieldState.DEFAULT_LAST_UPDATE_TS;
@Data
@AllArgsConstructor
public class SingleValueArgumentEntry implements ArgumentEntry {
public static final Long DEFAULT_VERSION = -1L;
public static final Long DEFAULT_TS = -1L;
private long ts;
private BasicKvEntry kvEntryValue;
@ -47,7 +45,7 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
private boolean forceResetPrevious;
public SingleValueArgumentEntry() {
this.ts = DEFAULT_LAST_UPDATE_TS;
this.ts = DEFAULT_TS;
this.version = DEFAULT_VERSION;
}

75
application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java
View File

@ -0,0 +1,75 @@
/**
* Copyright © 2016-2025 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.utils;
import lombok.NonNull;
import org.apache.commons.lang3.math.NumberUtils;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.cf.configuration.Argument;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.common.data.kv.DoubleDataEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.StringDataEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.ScriptCalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.SimpleCalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry;
import static org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry.DEFAULT_VERSION;
public class CalculatedFieldArgumentUtils {
public static ArgumentEntry transformSingleValueArgument(@NonNull KvEntry kvEntry) {
return kvEntry.getValue() != null ? ArgumentEntry.createSingleValueArgument(kvEntry) : new SingleValueArgumentEntry();
}
public static TsKvEntry createDefaultTsKvEntry(Argument argument, long ts) {
return new BasicTsKvEntry(ts, createDefaultKvEntry(argument), DEFAULT_VERSION);
}
public static AttributeKvEntry createDefaultAttributeEntry(Argument argument, long ts) {
return new BaseAttributeKvEntry(createDefaultKvEntry(argument), ts, DEFAULT_VERSION);
}
private static KvEntry createDefaultKvEntry(Argument argument) {
String key = argument.getRefEntityKey().getKey();
String defaultValue = argument.getDefaultValue();
if (StringUtils.isBlank(defaultValue)) {
return new StringDataEntry(key, null);
}
if (NumberUtils.isParsable(defaultValue)) {
return new DoubleDataEntry(key, Double.parseDouble(defaultValue));
}
if ("true".equalsIgnoreCase(defaultValue) || "false".equalsIgnoreCase(defaultValue)) {
return new BooleanDataEntry(key, Boolean.parseBoolean(defaultValue));
}
return new StringDataEntry(key, defaultValue);
}
public static CalculatedFieldState createStateByType(CalculatedFieldCtx ctx) {
return switch (ctx.getCfType()) {
case SIMPLE -> new SimpleCalculatedFieldState(ctx.getArgNames());
case SCRIPT -> new ScriptCalculatedFieldState(ctx.getArgNames());
};
}
}

17
application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java
View File

@ -570,8 +570,6 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
@Test
public void testScriptCalculatedFieldWhenUsedLatestTsInScript() throws Exception {
Device testDevice = createDevice("Test device", "1234567890");
long ts = System.currentTimeMillis() - 300000L;
doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"temperature\":30}}", ts)));
CalculatedField calculatedField = new CalculatedField();
calculatedField.setEntityId(testDevice.getId());
@ -585,6 +583,7 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
Argument argument = new Argument();
ReferencedEntityKey refEntityKey = new ReferencedEntityKey("temperature", ArgumentType.TS_LATEST, null);
argument.setRefEntityKey(refEntityKey);
argument.setDefaultValue("20");
config.setArguments(Map.of("T", argument));
config.setExpression("return {\"ts\": ctx.latestTs, \"values\": {\"fahrenheitTemp\": (T * 1.8) + 32}};");
@ -596,7 +595,19 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
CalculatedField savedCalculatedField = doPost("/api/calculatedField", calculatedField, CalculatedField.class);
await().alias("create CF -> perform initial calculation").atMost(TIMEOUT, TimeUnit.SECONDS)
await().alias("create CF -> perform initial calculation with default value").atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
ObjectNode fahrenheitTemp = getLatestTelemetry(testDevice.getId(), "fahrenheitTemp");
assertThat(fahrenheitTemp).isNotNull();
assertThat(fahrenheitTemp.get("fahrenheitTemp").get(0).get("ts").asText()).isNotEqualTo("-1");
assertThat(fahrenheitTemp.get("fahrenheitTemp").get(0).get("value").asText()).isEqualTo("68.0");
});
long ts = System.currentTimeMillis() - 10L;
doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"temperature\":30}}", ts)));
await().alias("update telemetry -> perform calculation").atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
ObjectNode fahrenheitTemp = getLatestTelemetry(testDevice.getId(), "fahrenheitTemp");

2
common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfCtx.java
View File

@ -29,7 +29,7 @@ public class TbelCfCtx implements TbelCfObject {
public TbelCfCtx(Map<String, TbelCfArg> args, long latestTs) {
this.args = Collections.unmodifiableMap(args);
this.latestTs = latestTs != -1 ? latestTs : System.currentTimeMillis();
this.latestTs = latestTs;
}
@Override

Write Preview
Loading…
Cancel
Save