diff --git a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java index 232476c15a..f5ec782992 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java @@ -88,7 +88,7 @@ public abstract class AbstractCalculatedFieldProcessingService { protected abstract String getExecutorNamePrefix(); protected ListenableFuture> fetchArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) { - Map> argFutures = switch (ctx.getCalculatedField().getType()) { + Map> argFutures = switch (ctx.getCfType()) { case GEOFENCING -> fetchGeofencingCalculatedFieldArguments(ctx, entityId, false, ts); case SIMPLE, SCRIPT, ALARM, PROPAGATION -> getBaseCalculatedFieldArguments(ctx, entityId, ts); }; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java index d027a2f9fe..c6af49bf4d 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java @@ -15,7 +15,9 @@ */ package org.thingsboard.server.service.cf.ctx.state; +import com.fasterxml.jackson.databind.node.ObjectNode; import lombok.Getter; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.utils.CalculatedFieldUtils; @@ -105,6 +107,20 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { protected void validateNewEntry(String key, ArgumentEntry newEntry) {} + protected ObjectNode toSimpleResult(boolean useLatestTs, ObjectNode valuesNode) { + if (!useLatestTs) { + return valuesNode; + } + long latestTs = getLatestTimestamp(); + if (latestTs == -1) { + return valuesNode; + } + ObjectNode resultNode = JacksonUtil.newObjectNode(); + resultNode.put("ts", latestTs); + resultNode.set("values", valuesNode); + return resultNode; + } + private void updateLastUpdateTimestamp(ArgumentEntry entry) { long newTs = this.latestTimestamp; if (entry instanceof SingleValueArgumentEntry singleValueArgumentEntry) { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java index ece459350f..339ba40838 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java @@ -103,6 +103,7 @@ public class CalculatedFieldCtx { private long scheduledUpdateIntervalMillis; private Argument propagationArgument; + private boolean applyExpressionForResolvedArguments; public CalculatedFieldCtx(CalculatedField calculatedField, ActorSystemContext systemContext) { @@ -159,6 +160,7 @@ public class CalculatedFieldCtx { } if (calculatedField.getConfiguration() instanceof PropagationCalculatedFieldConfiguration propagationConfig) { propagationArgument = propagationConfig.toPropagationArgument(); + applyExpressionForResolvedArguments = propagationConfig.isApplyExpressionToResolvedArguments(); relationQueryDynamicArguments = true; } } @@ -177,13 +179,13 @@ public class CalculatedFieldCtx { public void init() { switch (cfType) { - case SCRIPT, PROPAGATION -> { - try { - initTbelExpression(expression); - initialized = true; - } catch (Exception e) { - throw new RuntimeException("Failed to init calculated field ctx. Invalid expression syntax.", e); + case SCRIPT -> initTbelExpression(); + case PROPAGATION -> { + if (applyExpressionForResolvedArguments) { + initTbelExpression(); + return; } + initialized = true; } case GEOFENCING -> initialized = true; case SIMPLE -> { @@ -206,6 +208,15 @@ public class CalculatedFieldCtx { } } + private void initTbelExpression() { + try { + initTbelExpression(expression); + initialized = true; + } catch (Exception e) { + throw new RuntimeException("Failed to init calculated field ctx. Invalid expression syntax.", e); + } + } + public double evaluateSimpleExpression(String expressionStr, CalculatedFieldState state) { Expression expression = simpleExpressions.get(expressionStr).get(); for (Map.Entry entry : state.getArguments().entrySet()) { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java index ad8005af83..ecd8b0d0f7 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java @@ -26,6 +26,7 @@ import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState; +import org.thingsboard.server.service.cf.ctx.state.propagation.PropagationCalculatedFieldState; import java.util.Map; @@ -36,7 +37,8 @@ import static org.thingsboard.server.utils.CalculatedFieldUtils.toSingleValueArg @Type(value = SimpleCalculatedFieldState.class, name = "SIMPLE"), @Type(value = ScriptCalculatedFieldState.class, name = "SCRIPT"), @Type(value = GeofencingCalculatedFieldState.class, name = "GEOFENCING"), - @Type(value = AlarmCalculatedFieldState.class, name = "ALARM") + @Type(value = AlarmCalculatedFieldState.class, name = "ALARM"), + @Type(value = PropagationCalculatedFieldState.class, name = "PROPAGATION") }) public interface CalculatedFieldState { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java index 3a98fee361..bd5ec720df 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java @@ -84,16 +84,7 @@ public class SimpleCalculatedFieldState extends BaseCalculatedFieldState { } else { valuesNode.set(outputName, JacksonUtil.valueToTree(result)); } - - long latestTs = getLatestTimestamp(); - if (useLatestTs && latestTs != -1) { - ObjectNode resultNode = JacksonUtil.newObjectNode(); - resultNode.put("ts", latestTs); - resultNode.set("values", valuesNode); - return resultNode; - } else { - return valuesNode; - } + return toSimpleResult(useLatestTs, valuesNode); } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/geofencing/GeofencingCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/geofencing/GeofencingCalculatedFieldState.java index f3bf8750cf..da1acc1905 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/geofencing/GeofencingCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/geofencing/GeofencingCalculatedFieldState.java @@ -172,13 +172,7 @@ public class GeofencingCalculatedFieldState extends BaseCalculatedFieldState { } private JsonNode toResultNode(OutputType outputType, ObjectNode valuesNode) { - if (OutputType.ATTRIBUTES.equals(outputType) || latestTimestamp == -1) { - return valuesNode; - } - ObjectNode resultNode = JacksonUtil.newObjectNode(); - resultNode.put("ts", latestTimestamp); - resultNode.set("values", valuesNode); - return resultNode; + return toSimpleResult(outputType == OutputType.TIME_SERIES, valuesNode); } private GeofencingEvalResult aggregateZoneGroup(List zoneResults) { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java index c7d49a4d40..f09ae93999 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java @@ -32,6 +32,7 @@ public class PropagationArgumentEntry implements ArgumentEntry { private boolean forceResetPrevious; + // TODO: do we need to persist this? public PropagationArgumentEntry(List propagationEntityIds) { this.propagationEntityIds = propagationEntityIds; } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationCalculatedFieldState.java index 21a4493c91..9554b28024 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationCalculatedFieldState.java @@ -15,10 +15,14 @@ */ package org.thingsboard.server.service.cf.ctx.state.propagation; +import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.cf.CalculatedFieldType; +import org.thingsboard.server.common.data.cf.configuration.Output; +import org.thingsboard.server.common.data.cf.configuration.OutputType; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.service.cf.CalculatedFieldResult; import org.thingsboard.server.service.cf.PropagationCalculatedFieldResult; @@ -26,6 +30,7 @@ import org.thingsboard.server.service.cf.TelemetryCalculatedFieldResult; 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.ScriptCalculatedFieldState; +import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; import java.util.Map; @@ -37,6 +42,12 @@ public class PropagationCalculatedFieldState extends ScriptCalculatedFieldState super(entityId); } + @Override + public void init(CalculatedFieldCtx ctx) { + super.init(ctx); + requiredArguments.add(PROPAGATION_CONFIG_ARGUMENT); + } + @Override public CalculatedFieldType getType() { return CalculatedFieldType.PROPAGATION; @@ -48,12 +59,42 @@ public class PropagationCalculatedFieldState extends ScriptCalculatedFieldState if (!(argumentEntry instanceof PropagationArgumentEntry propagationArgumentEntry) || propagationArgumentEntry.isEmpty()) { return Futures.immediateFuture(PropagationCalculatedFieldResult.builder().build()); } - return Futures.transform(super.performCalculation(updatedArgs, ctx), telemetryCfResult -> - PropagationCalculatedFieldResult.builder() - .propagationEntityIds(propagationArgumentEntry.getPropagationEntityIds()) - .result((TelemetryCalculatedFieldResult) telemetryCfResult) - .build(), - MoreExecutors.directExecutor()); + if (ctx.isApplyExpressionForResolvedArguments()) { + return Futures.transform(super.performCalculation(updatedArgs, ctx), telemetryCfResult -> + PropagationCalculatedFieldResult.builder() + .propagationEntityIds(propagationArgumentEntry.getPropagationEntityIds()) + .result((TelemetryCalculatedFieldResult) telemetryCfResult) + .build(), + MoreExecutors.directExecutor()); + } + return Futures.immediateFuture(PropagationCalculatedFieldResult.builder() + .propagationEntityIds(propagationArgumentEntry.getPropagationEntityIds()) + .result(toTelemetryResult(ctx)) + .build()); + } + + private TelemetryCalculatedFieldResult toTelemetryResult(CalculatedFieldCtx ctx) { + Output output = ctx.getOutput(); + TelemetryCalculatedFieldResult.TelemetryCalculatedFieldResultBuilder telemetryCfBuilder = + TelemetryCalculatedFieldResult.builder() + .type(output.getType()) + .scope(output.getScope()); + ObjectNode valuesNode = JacksonUtil.newObjectNode(); + arguments.forEach((argumentName, argumentEntry) -> { + if (argumentEntry instanceof PropagationArgumentEntry) { + return; + } + if (argumentEntry instanceof SingleValueArgumentEntry singleArgumentEntry) { + // TODO: use argumentName as a key or no? + JacksonUtil.addKvEntry(valuesNode, singleArgumentEntry.getKvEntryValue(), argumentName); + return; + } + throw new IllegalArgumentException("Unsupported argument type: " + argumentEntry.getType() + " detected for argument: " + argumentName + ". " + + "Only Latest telemetry or Attribute arguments supported for 'Arguments Only' propagation mode!"); + }); + ObjectNode result = toSimpleResult(output.getType() == OutputType.TIME_SERIES, valuesNode); + telemetryCfBuilder.result(result); + return telemetryCfBuilder.build(); } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/PropagationCalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/PropagationCalculatedFieldConfiguration.java index 7585c30438..b592264d6c 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/PropagationCalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/PropagationCalculatedFieldConfiguration.java @@ -33,6 +33,8 @@ public class PropagationCalculatedFieldConfiguration extends BaseCalculatedField private EntitySearchDirection direction; private String relationType; + private boolean applyExpressionToResolvedArguments; + @Override public CalculatedFieldType getType() { return CalculatedFieldType.PROPAGATION; @@ -48,6 +50,19 @@ public class PropagationCalculatedFieldConfiguration extends BaseCalculatedField if (StringUtils.isBlank(relationType)) { throw new IllegalArgumentException("Propagation calculated field relation type must be specified!"); } + if (!applyExpressionToResolvedArguments) { + arguments.forEach((name, argument) -> { + if (argument.getRefEntityKey() == null) { + throw new IllegalArgumentException("Argument: '" + name + "' doesn't have reference entity key configured!"); + } + if (argument.getRefEntityKey().getType() == ArgumentType.TS_ROLLING) { + throw new IllegalArgumentException("Argument type: 'Time series rolling' detected for argument: '" + name + "'! " + + "Only 'Attribute' or 'Latest telemetry' arguments are allowed for in 'Arguments only' propagation mode!"); + } + }); + } else if (StringUtils.isBlank(expression)) { + throw new IllegalArgumentException("Expression must be specified for 'Expression result' propagation mode!"); + } } public Argument toPropagationArgument() {