Browse Source

Refactoring of the arguments

pull/12691/head
Andrii Shvaika 2 years ago
parent
commit
d7412edceb
  1. 5
      application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
  2. 3
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java
  3. 17
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java
  4. 28
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java
  5. 7
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
  6. 73
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java
  7. 30
      application/src/test/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldStateTest.java
  8. 18
      application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java
  9. 12
      common/proto/src/main/proto/queue.proto
  10. 4
      common/script/script-api/src/main/java/org/thingsboard/script/api/AbstractScriptInvokeService.java
  11. 6
      common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/DefaultTbelInvokeService.java
  12. 20
      common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfArg.java
  13. 2
      common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfObject.java
  14. 3
      common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfSingleValueArg.java
  15. 29
      common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfTsDoubleVal.java
  16. 73
      common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfTsRollingArg.java

5
application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@ -810,7 +810,8 @@ public class ActorSystemContext {
Futures.addCallback(future, RULE_CHAIN_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor()); Futures.addCallback(future, RULE_CHAIN_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor());
} }
public void persistCalculatedFieldDebugEvent(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId, Map<String, ArgumentEntry> arguments, UUID tbMsgId, TbMsgType tbMsgType, String result, Throwable error) { public void persistCalculatedFieldDebugEvent(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId, Map<String, ArgumentEntry> arguments,
UUID tbMsgId, TbMsgType tbMsgType, String result, Throwable error) {
if (cfDebugPerTenantEnabled) { if (cfDebugPerTenantEnabled) {
TbRateLimits rateLimits = cfDebugPerTenantLimits.computeIfAbsent(tenantId, id -> new TbRateLimits(cfDebugPerTenantLimitsConfiguration)); TbRateLimits rateLimits = cfDebugPerTenantLimits.computeIfAbsent(tenantId, id -> new TbRateLimits(cfDebugPerTenantLimitsConfiguration));
@ -831,7 +832,7 @@ public class ActorSystemContext {
if (arguments != null) { if (arguments != null) {
eventBuilder.arguments(JacksonUtil.toString( eventBuilder.arguments(JacksonUtil.toString(
arguments.entrySet().stream() arguments.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getValue())) .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().toTbelCfArg()))
)); ));
} }
if (result != null) { if (result != null) {

3
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java

@ -18,6 +18,7 @@ package org.thingsboard.server.service.cf.ctx.state;
import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonSubTypes; import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonTypeInfo; import com.fasterxml.jackson.annotation.JsonTypeInfo;
import org.thingsboard.script.api.tbel.TbelCfArg;
import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry;
@ -52,4 +53,6 @@ public interface ArgumentEntry {
@JsonIgnore @JsonIgnore
ArgumentEntry copy(); ArgumentEntry copy();
TbelCfArg toTbelCfArg();
} }

17
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java

@ -32,6 +32,8 @@ import org.thingsboard.server.common.util.KvProtoUtil;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto; import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto;
import org.thingsboard.server.gen.transport.TransportProtos.TsDoubleValProto;
import org.thingsboard.server.gen.transport.TransportProtos.TsRollingArgumentProto;
import org.thingsboard.server.gen.transport.TransportProtos.TsValueListProto; import org.thingsboard.server.gen.transport.TransportProtos.TsValueListProto;
import org.thingsboard.server.gen.transport.TransportProtos.TsValueProto; import org.thingsboard.server.gen.transport.TransportProtos.TsValueProto;
import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.AfterStartUp;
@ -130,11 +132,11 @@ public class RocksDBCalculatedFieldStateService implements CalculatedFieldStateS
return builder.build(); return builder.build();
} }
private TsValueListProto toRollingArgumentProto(String argName, TsRollingArgumentEntry entry) { private TsRollingArgumentProto toRollingArgumentProto(String argName, TsRollingArgumentEntry entry) {
TsValueListProto.Builder builder = TsValueListProto.newBuilder().setKey(argName); TsRollingArgumentProto.Builder builder = TsRollingArgumentProto.newBuilder().setKey(argName);
if (entry != TsRollingArgumentEntry.EMPTY) { if (entry != TsRollingArgumentEntry.EMPTY) {
entry.getTsRecords().forEach((ts, value) -> builder.addTsValue(KvProtoUtil.toTsValueProto(ts, value))); entry.getTsRecords().forEach((ts, value) -> builder.addTsValue(TsDoubleValProto.newBuilder().setTs(ts).setValue(value).build()));
} }
return builder.build(); return builder.build();
@ -173,15 +175,12 @@ public class RocksDBCalculatedFieldStateService implements CalculatedFieldStateS
return new SingleValueArgumentEntry(ts, kvEntry, proto.getVersion()); return new SingleValueArgumentEntry(ts, kvEntry, proto.getVersion());
} }
private TsRollingArgumentEntry fromRollingArgumentProto(TsValueListProto proto) { private TsRollingArgumentEntry fromRollingArgumentProto(TsRollingArgumentProto proto) {
if (proto.getTsValueCount() <= 0) { if (proto.getTsValueCount() <= 0) {
return (TsRollingArgumentEntry) TsRollingArgumentEntry.EMPTY; return (TsRollingArgumentEntry) TsRollingArgumentEntry.EMPTY;
} }
TreeMap<Long, BasicKvEntry> tsRecords = new TreeMap<>(); TreeMap<Long, Double> tsRecords = new TreeMap<>();
proto.getTsValueList().forEach(tsValueProto -> { proto.getTsValueList().forEach(tsValueProto -> tsRecords.put(tsValueProto.getTs(), tsValueProto.getValue()));
BasicKvEntry kvEntry = (BasicKvEntry) KvProtoUtil.fromTsValueProto(proto.getKey(), tsValueProto);
tsRecords.put(tsValueProto.getTs(), kvEntry);
});
return new TsRollingArgumentEntry(tsRecords); return new TsRollingArgumentEntry(tsRecords);
} }

28
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java

@ -21,15 +21,13 @@ import com.google.common.util.concurrent.MoreExecutors;
import lombok.Data; import lombok.Data;
import lombok.NoArgsConstructor; import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull; import org.thingsboard.script.api.tbel.TbelCfArg;
import org.mvel2.execution.ExecutionArrayList; import org.thingsboard.script.api.tbel.TbelCfSingleValueArg;
import org.thingsboard.script.api.tbel.TbCfArg; import org.thingsboard.script.api.tbel.TbelCfTsDoubleVal;
import org.thingsboard.script.api.tbel.TbCfSingleValueArg; import org.thingsboard.script.api.tbel.TbelCfTsRollingArg;
import org.thingsboard.script.api.tbel.TbCfTsRollingArg;
import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.cf.CalculatedFieldType;
import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.Argument;
import org.thingsboard.server.common.data.cf.configuration.Output; import org.thingsboard.server.common.data.cf.configuration.Output;
import org.thingsboard.server.common.data.kv.BasicKvEntry;
import org.thingsboard.server.service.cf.CalculatedFieldResult; import org.thingsboard.server.service.cf.CalculatedFieldResult;
import java.util.ArrayList; import java.util.ArrayList;
@ -60,7 +58,7 @@ public class ScriptCalculatedFieldState extends BaseCalculatedFieldState {
arguments.forEach((key, argumentEntry) -> { arguments.forEach((key, argumentEntry) -> {
if (argumentEntry instanceof TsRollingArgumentEntry tsRollingEntry) { if (argumentEntry instanceof TsRollingArgumentEntry tsRollingEntry) {
Argument argument = ctx.getArguments().get(key); Argument argument = ctx.getArguments().get(key);
TreeMap<Long, BasicKvEntry> tsRecords = tsRollingEntry.getTsRecords(); TreeMap<Long, Double> tsRecords = tsRollingEntry.getTsRecords();
if (tsRecords.size() > argument.getLimit()) { if (tsRecords.size() > argument.getLimit()) {
tsRecords.pollFirstEntry(); tsRecords.pollFirstEntry();
} }
@ -79,20 +77,8 @@ public class ScriptCalculatedFieldState extends BaseCalculatedFieldState {
); );
} }
private TbCfArg toTbelArgument(String key) { private TbelCfArg toTbelArgument(String key) {
ArgumentEntry argEntry = arguments.get(key); return arguments.get(key).toTbelCfArg();
if (argEntry instanceof SingleValueArgumentEntry svArg) {
return new TbCfSingleValueArg(svArg.getTs(), argEntry.getValue());
} else if (argEntry instanceof TsRollingArgumentEntry rollingArg) {
var tsRecords = rollingArg.getTsRecords();
List<TbCfSingleValueArg> values = new ArrayList<>(tsRecords.size());
for(var e : tsRecords.entrySet()){
values.add(new TbCfSingleValueArg(e.getKey(), e.getValue().getValue()));
}
return new TbCfTsRollingArg(values);
} else {
throw new RuntimeException("Argument is not supported for TBEL execution!");
}
} }
} }

7
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java

@ -19,6 +19,8 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.Data; import lombok.Data;
import lombok.NoArgsConstructor; import lombok.NoArgsConstructor;
import org.thingsboard.script.api.tbel.TbelCfArg;
import org.thingsboard.script.api.tbel.TbelCfSingleValueArg;
import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BasicKvEntry; import org.thingsboard.server.common.data.kv.BasicKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.KvEntry;
@ -84,6 +86,11 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
return new SingleValueArgumentEntry(this.ts, this.kvEntryValue, this.version); return new SingleValueArgumentEntry(this.ts, this.kvEntryValue, this.version);
} }
@Override
public TbelCfArg toTbelCfArg() {
return new TbelCfSingleValueArg(ts, kvEntryValue.getValue());
}
@Override @Override
public boolean updateEntry(ArgumentEntry entry) { public boolean updateEntry(ArgumentEntry entry) {
if (entry instanceof SingleValueArgumentEntry singleValueEntry) { if (entry instanceof SingleValueArgumentEntry singleValueEntry) {

73
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java

@ -21,11 +21,16 @@ import lombok.Data;
import lombok.NoArgsConstructor; import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.math.NumberUtils; import org.apache.commons.lang3.math.NumberUtils;
import org.thingsboard.script.api.tbel.TbelCfArg;
import org.thingsboard.script.api.tbel.TbelCfTsDoubleVal;
import org.thingsboard.script.api.tbel.TbelCfTsRollingArg;
import org.thingsboard.server.common.data.kv.BasicKvEntry; import org.thingsboard.server.common.data.kv.BasicKvEntry;
import org.thingsboard.server.common.data.kv.DataType;
import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.common.util.ProtoUtils;
import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.TreeMap; import java.util.TreeMap;
@ -41,7 +46,7 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
private static final int MAX_ROLLING_ARGUMENT_ENTRY_SIZE = 1000; private static final int MAX_ROLLING_ARGUMENT_ENTRY_SIZE = 1000;
private TreeMap<Long, BasicKvEntry> tsRecords = new TreeMap<>(); private TreeMap<Long, Double> tsRecords = new TreeMap<>();
public TsRollingArgumentEntry(List<TsKvEntry> kvEntries) { public TsRollingArgumentEntry(List<TsKvEntry> kvEntries) {
kvEntries.forEach(tsKvEntry -> addTsRecord(tsKvEntry.getTs(), tsKvEntry)); kvEntries.forEach(tsKvEntry -> addTsRecord(tsKvEntry.getTs(), tsKvEntry));
@ -62,15 +67,7 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
@JsonIgnore @JsonIgnore
@Override @Override
public Object getValue() { public Object getValue() {
return tsRecords.entrySet() return tsRecords;
.stream()
.collect(Collectors.toMap(
Map.Entry::getKey,
entry -> entry.getValue().getValue(),
(oldValue, newValue) -> oldValue,
TreeMap::new
));
} }
@Override @Override
@ -78,45 +75,57 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
return new TsRollingArgumentEntry(new TreeMap<>(tsRecords)); return new TsRollingArgumentEntry(new TreeMap<>(tsRecords));
} }
@Override
public TbelCfArg toTbelCfArg() {
List<TbelCfTsDoubleVal> values = new ArrayList<>(tsRecords.size());
for (var e : tsRecords.entrySet()) {
values.add(new TbelCfTsDoubleVal(e.getKey(), e.getValue()));
}
return new TbelCfTsRollingArg(values);
}
@Override @Override
public boolean updateEntry(ArgumentEntry entry) { public boolean updateEntry(ArgumentEntry entry) {
if (entry instanceof TsRollingArgumentEntry tsRollingEntry) { if (entry instanceof TsRollingArgumentEntry tsRollingEntry) {
return updateTsRollingEntry(tsRollingEntry); updateTsRollingEntry(tsRollingEntry);
} else if (entry instanceof SingleValueArgumentEntry singleValueEntry) { } else if (entry instanceof SingleValueArgumentEntry singleValueEntry) {
return updateSingleValueEntry(singleValueEntry); updateSingleValueEntry(singleValueEntry);
} else { } else {
throw new IllegalArgumentException("Unsupported argument entry type for rolling argument entry: " + entry.getType()); throw new IllegalArgumentException("Unsupported argument entry type for rolling argument entry: " + entry.getType());
} }
return true;
} }
private boolean updateTsRollingEntry(TsRollingArgumentEntry tsRollingEntry) { private void updateTsRollingEntry(TsRollingArgumentEntry tsRollingEntry) {
boolean updated = false; for (Map.Entry<Long, Double> tsRecordEntry : tsRollingEntry.getTsRecords().entrySet()) {
for (Map.Entry<Long, BasicKvEntry> tsRecordEntry : tsRollingEntry.getTsRecords().entrySet()) { addTsRecord(tsRecordEntry.getKey(), tsRecordEntry.getValue());
updated |= addTsRecordIfAbsent(tsRecordEntry.getKey(), tsRecordEntry.getValue());
} }
return updated;
} }
private boolean updateSingleValueEntry(SingleValueArgumentEntry singleValueEntry) { private void updateSingleValueEntry(SingleValueArgumentEntry singleValueEntry) {
return addTsRecordIfAbsent(singleValueEntry.getTs(), singleValueEntry.getKvEntryValue()); addTsRecord(singleValueEntry.getTs(), singleValueEntry.getKvEntryValue());
} }
private boolean addTsRecordIfAbsent(Long ts, KvEntry value) { private void addTsRecord(Long ts, KvEntry value) {
if (!tsRecords.containsKey(ts)) { switch (value.getDataType()) {
addTsRecord(ts, value); case LONG -> value.getLongValue().ifPresent(aLong -> tsRecords.put(ts, aLong.doubleValue()));
return true; case DOUBLE -> value.getDoubleValue().ifPresent(aDouble -> tsRecords.put(ts, aDouble));
case BOOLEAN -> value.getBooleanValue().ifPresent(aBoolean -> tsRecords.put(ts, aBoolean ? 1.0 : 0.0));
case STRING -> value.getStrValue().ifPresent(aString -> tsRecords.put(ts, Double.parseDouble(aString)));
case JSON -> value.getJsonValue().ifPresent(aString -> tsRecords.put(ts, Double.parseDouble(aString)));
//TODO: try catch
} }
return false; pollFirstEntryIfNeeded();
} }
private void addTsRecord(Long ts, KvEntry value) { private void addTsRecord(Long ts, double value) {
if (NumberUtils.isParsable(value.getValue().toString())) { tsRecords.put(ts, value);
tsRecords.put(ts, ProtoUtils.basicKvEntryFromKvEntry(value)); pollFirstEntryIfNeeded();
if (tsRecords.size() > MAX_ROLLING_ARGUMENT_ENTRY_SIZE) { }
tsRecords.pollFirstEntry();
} private void pollFirstEntryIfNeeded() {
} else { if (tsRecords.size() > MAX_ROLLING_ARGUMENT_ENTRY_SIZE) {
throw new IllegalArgumentException("Argument type " + getType() + " only supports numeric values."); tsRecords.pollFirstEntry();
} }
} }

30
application/src/test/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldStateTest.java

@ -134,10 +134,10 @@ public class ScriptCalculatedFieldStateTest {
void testPerformCalculationWhenOldTelemetry() throws ExecutionException, InterruptedException { void testPerformCalculationWhenOldTelemetry() throws ExecutionException, InterruptedException {
TsRollingArgumentEntry argumentEntry = new TsRollingArgumentEntry(); TsRollingArgumentEntry argumentEntry = new TsRollingArgumentEntry();
TreeMap<Long, BasicKvEntry> values = new TreeMap<>(); TreeMap<Long, Double> values = new TreeMap<>();
values.put(ts - 40000, new LongDataEntry("deviceTemperature", 4L));// will not be used for calculation values.put(ts - 40000, 4.0);// will not be used for calculation
values.put(ts - 45000, new LongDataEntry("deviceTemperature", 2L));// will not be used for calculation values.put(ts - 45000, 2.0);// will not be used for calculation
values.put(ts - 20, new LongDataEntry("deviceTemperature", 0L)); values.put(ts - 20, 0.0);
argumentEntry.setTsRecords(values); argumentEntry.setTsRecords(values);
@ -155,13 +155,13 @@ public class ScriptCalculatedFieldStateTest {
@Test @Test
void testPerformCalculationWhenArgumentsMoreThanLimit() throws ExecutionException, InterruptedException { void testPerformCalculationWhenArgumentsMoreThanLimit() throws ExecutionException, InterruptedException {
TsRollingArgumentEntry argumentEntry = new TsRollingArgumentEntry(); TsRollingArgumentEntry argumentEntry = new TsRollingArgumentEntry();
TreeMap<Long, BasicKvEntry> values = new TreeMap<>(); TreeMap<Long, Double> values = new TreeMap<>();
values.put(ts - 20, new LongDataEntry("deviceTemperature", 1000L));// will not be used values.put(ts - 20, 1000.0);// will not be used
values.put(ts - 18, new LongDataEntry("deviceTemperature", 0L)); values.put(ts - 18, 0.0);
values.put(ts - 16, new LongDataEntry("deviceTemperature", 0L)); values.put(ts - 16, 0.0);
values.put(ts - 14, new LongDataEntry("deviceTemperature", 0L)); values.put(ts - 14, 0.0);
values.put(ts - 12, new LongDataEntry("deviceTemperature", 0L)); values.put(ts - 12, 0.0);
values.put(ts - 10, new LongDataEntry("deviceTemperature", 0L)); values.put(ts - 10, 0.0);
argumentEntry.setTsRecords(values); argumentEntry.setTsRecords(values);
state.arguments = new HashMap<>(Map.of("deviceTemperature", argumentEntry, "assetHumidity", assetHumidityArgEntry)); state.arguments = new HashMap<>(Map.of("deviceTemperature", argumentEntry, "assetHumidity", assetHumidityArgEntry));
@ -198,10 +198,10 @@ public class ScriptCalculatedFieldStateTest {
TsRollingArgumentEntry argumentEntry = new TsRollingArgumentEntry(); TsRollingArgumentEntry argumentEntry = new TsRollingArgumentEntry();
long ts = System.currentTimeMillis(); long ts = System.currentTimeMillis();
TreeMap<Long, BasicKvEntry> values = new TreeMap<>(); TreeMap<Long, Double> values = new TreeMap<>();
values.put(ts - 40, new LongDataEntry("deviceTemperature", 10L)); values.put(ts - 40, 10.0);
values.put(ts - 30, new LongDataEntry("deviceTemperature", 12L)); values.put(ts - 30, 12.0);
values.put(ts - 20, new LongDataEntry("deviceTemperature", 17L)); values.put(ts - 20, 17.0);
argumentEntry.setTsRecords(values); argumentEntry.setTsRecords(values);
return argumentEntry; return argumentEntry;

18
application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java

@ -35,10 +35,10 @@ public class TsRollingArgumentEntryTest {
@BeforeEach @BeforeEach
void setUp() { void setUp() {
TreeMap<Long, BasicKvEntry> values = new TreeMap<>(); TreeMap<Long, Double> values = new TreeMap<>();
values.put(ts - 40, new DoubleDataEntry("key", 10.0)); values.put(ts - 40, 10.0);
values.put(ts - 30, new DoubleDataEntry("key", 12.0)); values.put(ts - 30, 12.0);
values.put(ts - 20, new DoubleDataEntry("key", 17.0)); values.put(ts - 20, 17.0);
entry = new TsRollingArgumentEntry(values); entry = new TsRollingArgumentEntry(values);
} }
@ -54,7 +54,7 @@ public class TsRollingArgumentEntryTest {
assertThat(entry.updateEntry(newEntry)).isTrue(); assertThat(entry.updateEntry(newEntry)).isTrue();
assertThat(entry.getTsRecords()).hasSize(4); assertThat(entry.getTsRecords()).hasSize(4);
assertThat(entry.getTsRecords().get(ts - 10).getValue()).isEqualTo(23.0); assertThat(entry.getTsRecords().get(ts - 10)).isEqualTo(23.0);
} }
@Test @Test
@ -67,10 +67,10 @@ public class TsRollingArgumentEntryTest {
@Test @Test
void testUpdateEntryWhenRollingEntryPassed() { void testUpdateEntryWhenRollingEntryPassed() {
TsRollingArgumentEntry newEntry = new TsRollingArgumentEntry(); TsRollingArgumentEntry newEntry = new TsRollingArgumentEntry();
TreeMap<Long, BasicKvEntry> values = new TreeMap<>(); TreeMap<Long, Double> values = new TreeMap<>();
values.put(ts - 20, new DoubleDataEntry("key", 16.0)); values.put(ts - 20, 16.0);
values.put(ts - 10, new DoubleDataEntry("key", 7.0)); values.put(ts - 10, 7.0);
values.put(ts - 5, new DoubleDataEntry("key", 1.0)); values.put(ts - 5, 1.0);
newEntry.setTsRecords(values); newEntry.setTsRecords(values);
assertThat(entry.updateEntry(newEntry)).isTrue(); assertThat(entry.updateEntry(newEntry)).isTrue();

12
common/proto/src/main/proto/queue.proto

@ -814,12 +814,22 @@ message SingleValueArgumentProto {
int64 version = 3; int64 version = 3;
} }
message TsDoubleValProto {
int64 ts = 1;
double value = 2;
}
message TsRollingArgumentProto {
string key = 1;
repeated TsDoubleValProto tsValue = 2;
}
message CalculatedFieldStateProto { message CalculatedFieldStateProto {
CalculatedFieldEntityCtxIdProto id = 1; CalculatedFieldEntityCtxIdProto id = 1;
// int32 version = 2; // int32 version = 2;
string type = 3; string type = 3;
repeated SingleValueArgumentProto singleValueArguments = 4; repeated SingleValueArgumentProto singleValueArguments = 4;
repeated TsValueListProto rollingValueArguments = 5; repeated TsRollingArgumentProto rollingValueArguments = 5;
} }
//Used to report session state to tb-Service and persist this state in the cache on the tb-Service level. //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.

4
common/script/script-api/src/main/java/org/thingsboard/script/api/AbstractScriptInvokeService.java

@ -22,6 +22,8 @@ import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.script.api.tbel.TbelCfArg;
import org.thingsboard.script.api.tbel.TbelCfObject;
import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
@ -256,6 +258,8 @@ public abstract class AbstractScriptInvokeService implements ScriptInvokeService
for (Object arg : args) { for (Object arg : args) {
if (arg instanceof CharSequence) { if (arg instanceof CharSequence) {
totalArgsSize += ((CharSequence) arg).length(); totalArgsSize += ((CharSequence) arg).length();
} else if (arg instanceof TbelCfObject tbelCfObj) {
totalArgsSize += tbelCfObj.memorySize();
} else { } else {
var str = JacksonUtil.toString(arg); var str = JacksonUtil.toString(arg);
if (str != null) { if (str != null) {

6
common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/DefaultTbelInvokeService.java

@ -32,7 +32,6 @@ import org.mvel2.MVEL;
import org.mvel2.ParserContext; import org.mvel2.ParserContext;
import org.mvel2.SandboxedParserConfiguration; import org.mvel2.SandboxedParserConfiguration;
import org.mvel2.ScriptMemoryOverflowException; import org.mvel2.ScriptMemoryOverflowException;
import org.mvel2.integration.PropertyHandlerFactory;
import org.mvel2.optimizers.OptimizerFactory; import org.mvel2.optimizers.OptimizerFactory;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@ -134,8 +133,9 @@ public class DefaultTbelInvokeService extends AbstractScriptInvokeService implem
parserConfig.registerDataType("Date", TbDate.class, val -> 8L); parserConfig.registerDataType("Date", TbDate.class, val -> 8L);
parserConfig.registerDataType("Random", Random.class, val -> 8L); parserConfig.registerDataType("Random", Random.class, val -> 8L);
parserConfig.registerDataType("Calendar", Calendar.class, val -> 8L); parserConfig.registerDataType("Calendar", Calendar.class, val -> 8L);
parserConfig.registerDataType("TbCfSingleValueArg", TbCfSingleValueArg.class, TbCfSingleValueArg::memorySize); parserConfig.registerDataType("TbelCfSingleValueArg", TbelCfSingleValueArg.class, TbelCfSingleValueArg::memorySize);
parserConfig.registerDataType("TbCfTsRollingArg", TbCfTsRollingArg.class, TbCfTsRollingArg::memorySize); parserConfig.registerDataType("TbelCfTsRollingArg", TbelCfTsRollingArg.class, TbelCfTsRollingArg::memorySize);
parserConfig.registerDataType("TbelCfTsDoubleVal", TbelCfTsDoubleVal.class, TbelCfTsDoubleVal::memorySize);
TbUtils.register(parserConfig); TbUtils.register(parserConfig);
executor = MoreExecutors.listeningDecorator(ThingsBoardExecutors.newWorkStealingPool(threadPoolSize, "tbel-executor")); executor = MoreExecutors.listeningDecorator(ThingsBoardExecutors.newWorkStealingPool(threadPoolSize, "tbel-executor"));
try { try {

20
common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfArg.java

@ -0,0 +1,20 @@
/**
* Copyright © 2016-2024 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.tbel;
public interface TbelCfArg extends TbelCfObject {
}

2
common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbCfArg.java → common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfObject.java

@ -15,7 +15,7 @@
*/ */
package org.thingsboard.script.api.tbel; package org.thingsboard.script.api.tbel;
public interface TbCfArg { public interface TbelCfObject {
long memorySize(); long memorySize();

3
common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbCfSingleValueArg.java → common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfSingleValueArg.java

@ -18,7 +18,7 @@ package org.thingsboard.script.api.tbel;
import lombok.Data; import lombok.Data;
@Data @Data
public class TbCfSingleValueArg implements TbCfArg { public class TbelCfSingleValueArg implements TbelCfArg {
private final long ts; private final long ts;
private final Object value; private final Object value;
@ -27,4 +27,5 @@ public class TbCfSingleValueArg implements TbCfArg {
public long memorySize() { public long memorySize() {
return 8L; // TODO; return 8L; // TODO;
} }
} }

29
common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbCfTsRollingArg.java → common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfTsDoubleVal.java

@ -15,33 +15,18 @@
*/ */
package org.thingsboard.script.api.tbel; package org.thingsboard.script.api.tbel;
import lombok.Getter; import lombok.Data;
import java.util.List; @Data
public class TbelCfTsDoubleVal implements TbelCfObject {
public class TbCfTsRollingArg implements TbCfArg { public static final long OBJ_SIZE = 32L; // Approximate calculation;
@Getter private final long ts;
private final List<TbCfSingleValueArg> values; private final double value;
public TbCfTsRollingArg(List<TbCfSingleValueArg> values) {
this.values = values;
}
@Override @Override
public long memorySize() { public long memorySize() {
return values.size() * 8L; //TODO; return OBJ_SIZE;
} }
public double max() {
double max = Double.MIN_VALUE;
for (TbCfSingleValueArg arg : values) {
double val = Double.valueOf(arg.getValue().toString());
if (max < val) {
max = val;
}
}
return max;
}
} }

73
common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfTsRollingArg.java

@ -0,0 +1,73 @@
/**
* Copyright © 2016-2024 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.tbel;
import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.Getter;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.function.Consumer;
import static org.thingsboard.script.api.tbel.TbelCfTsDoubleVal.OBJ_SIZE;
public class TbelCfTsRollingArg implements TbelCfArg, Iterable<TbelCfTsDoubleVal> {
@Getter
private final List<TbelCfTsDoubleVal> values;
public TbelCfTsRollingArg(List<TbelCfTsDoubleVal> values) {
this.values = Collections.unmodifiableList(values);
}
@Override
public long memorySize() {
return 12 + values.size() * OBJ_SIZE;
}
@JsonIgnore
public List<TbelCfTsDoubleVal> getValue() {
return values;
}
public double max() {
double max = Double.MIN_VALUE;
for (TbelCfTsDoubleVal value : values) {
double val = value.getValue();
if (max < val) {
max = val;
}
}
return max;
}
@JsonIgnore
public int getSize() {
return values.size();
}
@Override
public Iterator<TbelCfTsDoubleVal> iterator() {
return values.iterator();
}
@Override
public void forEach(Consumer<? super TbelCfTsDoubleVal> action) {
values.forEach(action);
}
}
Loading…
Cancel
Save