diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java index 49d1af2b0f..257e9d8565 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java @@ -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 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; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java index e01d8b7369..9f9ad5db73 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java @@ -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; + } + } + } diff --git a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java index c83e8c5ee3..1da3673695 100644 --- a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java @@ -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); }