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

fixed processing telemetry batch

pull/14623/head
IrynaMatveieva 8 months ago
parent
4728af3641
commit
e0357aff50
3 changed files with 131 additions and 15 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. 34
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  2. 32
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java
  3. 80
      application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java

34
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
View File

@ -36,6 +36,7 @@ import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto;
@ -48,6 +49,7 @@ 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.SingleValueArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry;
import java.util.ArrayList;
import java.util.Collection;
@ -62,6 +64,8 @@ import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry.getValueForTsRecord;
/**
* @author Andrew Shvayka
@ -356,13 +360,39 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
ReferencedEntityKey key = new ReferencedEntityKey(item.getKv().getKey(), ArgumentType.TS_LATEST, null);
Set<String> argNames = args.get(key);
if (argNames != null) {
argNames.forEach(argName -> arguments.put(argName, new SingleValueArgumentEntry(item)));
SingleValueArgumentEntry incoming = new SingleValueArgumentEntry(item);
argNames.forEach(argName -> arguments.compute(argName, (name, existing) -> {
if (existing == null) {
return incoming;
}
existing.updateEntry(incoming);
return existing;
}));
}
key = new ReferencedEntityKey(item.getKv().getKey(), ArgumentType.TS_ROLLING, null);
argNames = args.get(key);
if (argNames != null) {
argNames.forEach(argName -> arguments.put(argName, new SingleValueArgumentEntry(item)));
Double recordValue = getValueForTsRecord(ProtoUtils.fromProto(item.getKv()));
argNames.forEach(argName -> arguments.compute(argName, (name, existing) -> {
if (existing instanceof TsRollingArgumentEntry rolling) {
if (recordValue != null) {
rolling.getTsRecords().put(item.getTs(), recordValue);
}
return rolling;
}
TsRollingArgumentEntry rolling = new TsRollingArgumentEntry();
if (recordValue != null) {
rolling.getTsRecords().put(item.getTs(), recordValue);
}
if (existing instanceof SingleValueArgumentEntry single) {
Double existingValue = getValueForTsRecord(single.getKvEntryValue());
if (existingValue != null) {
rolling.getTsRecords().put(single.getTs(), existingValue);
}
}
return rolling;
}));
}
}
return arguments;

32
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java
View File

@ -122,20 +122,11 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
}
private void addTsRecord(Long ts, KvEntry value) {
try {
switch (value.getDataType()) {
case LONG -> value.getLongValue().ifPresent(aLong -> tsRecords.put(ts, aLong.doubleValue()));
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)));
}
} catch (Exception e) {
tsRecords.put(ts, Double.NaN);
log.debug("Invalid value '{}' for time series rolling arguments. Only numeric values are supported.", value.getValue());
} finally {
cleanupExpiredRecords();
Double recordValue = getValueForTsRecord(value);
if (recordValue != null) {
tsRecords.put(ts, recordValue);
}
cleanupExpiredRecords();
}
private void addTsRecord(Long ts, double value) {
@ -150,4 +141,19 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
tsRecords.entrySet().removeIf(tsRecord -> tsRecord.getKey() < System.currentTimeMillis() - timeWindow);
}
public static Double getValueForTsRecord(KvEntry value) {
try {
return switch (value.getDataType()) {
case LONG -> value.getLongValue().map(Long::doubleValue).orElse(null);
case DOUBLE -> value.getDoubleValue().orElse(null);
case BOOLEAN -> value.getBooleanValue().map(b -> b ? 1.0 : 0.0).orElse(null);
case STRING -> value.getStrValue().map(Double::parseDouble).orElse(null);
case JSON -> value.getJsonValue().map(Double::parseDouble).orElse(null);
};
} catch (Exception e) {
log.debug("Invalid value '{}' for time series rolling arguments. Only numeric values are supported.", value.getValue());
return Double.NaN;
}
}
}

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

@ -796,6 +796,86 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
});
}
@Test
public void testCalculatedFieldWhenBatchOfTelemetrySent() throws Exception {
Device testDevice = createDevice("Test device", "1234567890");
doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"a\":5}"));
doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"b\":10}"));
doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"b\":20}"));
CalculatedField calculatedField = new CalculatedField();
calculatedField.setEntityId(testDevice.getId());
calculatedField.setType(CalculatedFieldType.SCRIPT);
calculatedField.setName("Script CF");
calculatedField.setDebugSettings(DebugSettings.all());
SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration();
ReferencedEntityKey refEntityKeyA = new ReferencedEntityKey("a", ArgumentType.TS_LATEST, null);
Argument argumentA = new Argument();
argumentA.setRefEntityKey(refEntityKeyA);
Argument argumentB = new Argument();
ReferencedEntityKey refEntityKeyB = new ReferencedEntityKey("b", ArgumentType.TS_ROLLING, null);
argumentB.setTimeWindow(TimeUnit.MINUTES.toMillis(10));
argumentB.setLimit(1000);
argumentB.setRefEntityKey(refEntityKeyB);
config.setArguments(Map.of("a", argumentA, "b", argumentB));
config.setExpression("""
return {
"latestA": a,
"avgB": b.avg
};
""");
Output output = new Output();
output.setType(OutputType.TIME_SERIES);
config.setOutput(output);
calculatedField.setConfiguration(config);
doPost("/api/calculatedField", calculatedField, CalculatedField.class);
await().alias("create CF -> perform initial calculation").atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
ObjectNode result = getLatestTelemetry(testDevice.getId(), "latestA", "avgB");
assertThat(result).isNotNull();
assertThat(result.get("latestA").get(0).get("value").asText()).isEqualTo("5");
assertThat(result.get("avgB").get(0).get("value").asText()).isEqualTo("15.0");
});
long now = System.currentTimeMillis();
doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("""
[{
"ts": %s,
"values": {
"a": 6,
"b": 100
}
}, {
"ts": %s,
"values": {
"a": 7,
"b": 200
}
}, {
"ts": %s,
"values": {
"a": 8,
"b": 300
}
}]""", now - TimeUnit.MINUTES.toMillis(2), now, now - TimeUnit.MINUTES.toMillis(5))));
await().alias("update telemetry -> recalculate state").atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
ObjectNode result = getLatestTelemetry(testDevice.getId(), "latestA", "avgB");
assertThat(result).isNotNull();
assertThat(result.get("latestA").get(0).get("value").asText()).isEqualTo("7");
assertThat(result.get("avgB").get(0).get("value").asText()).isEqualTo("126.0");
});
}
private ObjectNode getLatestTelemetry(EntityId entityId, String... keys) throws Exception {
return doGetAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() + "/values/timeseries?keys=" + String.join(",", keys), ObjectNode.class);
}

Write Preview
Loading…
Cancel
Save