Browse Source

Merge pull request #12547 from irynamatveieva/calculated-fields

Calculated fields
pull/12586/head
Andrew Shvayka 2 years ago
committed by GitHub
parent
commit
5fcc3a6aec
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
  1. 52
      application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
  2. 42
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  3. 24
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
  4. 35
      application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java
  5. 54
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java
  6. 32
      application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java
  7. 2
      application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java
  8. 2
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java
  9. 45
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
  10. 61
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
  11. 3
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java
  12. 151
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java
  13. 16
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java
  14. 39
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java
  15. 42
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
  16. 68
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java
  17. 37
      application/src/main/java/org/thingsboard/server/service/entitiy/cf/DefaultTbCalculatedFieldService.java
  18. 5
      application/src/main/java/org/thingsboard/server/service/entitiy/cf/TbCalculatedFieldService.java
  19. 6
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
  20. 241
      application/src/test/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldStateTest.java
  21. 216
      application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldStateTest.java
  22. 72
      application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java
  23. 96
      application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java
  24. 2
      common/dao-api/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldService.java
  25. 7
      common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java
  26. 39
      common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java
  27. 28
      common/proto/src/main/proto/queue.proto
  28. 8
      dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java
  29. 4
      dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldDao.java
  30. 31
      dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java
  31. 6
      dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldRepository.java
  32. 5
      dao/src/main/java/org/thingsboard/server/dao/sql/cf/DefaultNativeCalculatedFieldRepository.java
  33. 11
      dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldDao.java
  34. 18
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java
  35. 18
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java
  36. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java
  37. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java

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

@ -50,6 +50,7 @@ import org.thingsboard.server.common.data.event.RuleNodeDebugEvent;
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.msg.TbMsgType;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.TbMsg;
@ -105,6 +106,7 @@ import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
import org.thingsboard.server.service.cf.CalculatedFieldExecutionService;
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
import org.thingsboard.server.service.component.ComponentDiscoveryService;
import org.thingsboard.server.service.edge.rpc.EdgeRpcService;
import org.thingsboard.server.service.entitiy.entityview.TbEntityViewService;
@ -129,6 +131,7 @@ import org.thingsboard.server.service.transport.TbCoreToTransportService;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
@ -744,29 +747,34 @@ public class ActorSystemContext {
}
}
public void persistCalculatedFieldDebugEvent(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId, Map<String, String> arguments, TbMsg tbMsg, Throwable error) {
if (checkLimits(tenantId, tbMsg, error)) {
try {
CalculatedFieldDebugEvent.CalculatedFieldDebugEventBuilder event = CalculatedFieldDebugEvent.builder()
.tenantId(tenantId)
.entityId(entityId.getId())
.serviceId(getServiceId())
.calculatedFieldId(calculatedFieldId)
.eventEntity(tbMsg.getOriginator())
.msgId(tbMsg.getId())
.msgType(tbMsg.getType())
.arguments(JacksonUtil.toString(arguments))
.result(tbMsg.getData());
if (error != null) {
event.error(toString(error));
}
ListenableFuture<Void> future = eventService.saveAsync(event.build());
Futures.addCallback(future, CALCULATED_FIELD_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor());
} catch (IllegalArgumentException ex) {
log.warn("Failed to persist calculated field debug message", ex);
public void persistCalculatedFieldDebugEvent(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId, Map<String, ArgumentEntry> arguments, UUID tbMsgId, TbMsgType tbMsgType, String result, Throwable error) {
try {
CalculatedFieldDebugEvent.CalculatedFieldDebugEventBuilder eventBuilder = CalculatedFieldDebugEvent.builder()
.tenantId(tenantId)
.entityId(entityId.getId())
.serviceId(getServiceId())
.calculatedFieldId(calculatedFieldId)
.eventEntity(entityId);
if (tbMsgId != null) {
eventBuilder.msgId(tbMsgId);
}
if (tbMsgType != null) {
eventBuilder.msgType(tbMsgType.name());
}
if (arguments != null) {
eventBuilder.arguments(JacksonUtil.toString(arguments));
}
if (result != null) {
eventBuilder.result(result);
}
if (error != null) {
eventBuilder.error(toString(error));
}
ListenableFuture<Void> future = eventService.saveAsync(eventBuilder.build());
Futures.addCallback(future, CALCULATED_FIELD_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor());
} catch (IllegalArgumentException ex) {
log.warn("Failed to persist calculated field debug message", ex);
}
}

42
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java

@ -18,19 +18,18 @@ package org.thingsboard.server.actors.calculatedField;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.thingsboard.common.util.DebugModeUtil;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.TbActorCtx;
import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.cf.configuration.ArgumentType;
import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey;
import org.thingsboard.server.common.data.id.AssetProfileId;
import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageDataIterable;
import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto;
@ -54,9 +53,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/**
@ -95,7 +92,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
states.remove(cfCtx.getCfId());
}
var cfState = getOrInitState(cfCtx);
processStateIfReady(cfCtx, Collections.singletonList(cfCtx.getCfId()), cfState, msg.getCallback());
processStateIfReady(cfCtx, Collections.singletonList(cfCtx.getCfId()), cfState, null, null, msg.getCallback());
}
public void process(CalculatedFieldEntityDeleteMsg msg) {
@ -135,9 +132,9 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
callback.onSuccess(CALLBACKS_PER_CF);
} else {
if (proto.getTsDataCount() > 0) {
processArgumentValuesUpdate(ctx, cfIds, callback, mapToArguments(ctx, msg.getEntityId(), proto.getTsDataList()));
processArgumentValuesUpdate(ctx, cfIds, callback, mapToArguments(ctx, msg.getEntityId(), proto.getTsDataList()), toTbMsgId(proto), toTbMsgType(proto));
} else if (proto.getAttrDataCount() > 0) {
processArgumentValuesUpdate(ctx, cfIds, callback, mapToArguments(ctx, msg.getEntityId(), proto.getScope(), proto.getAttrDataList()));
processArgumentValuesUpdate(ctx, cfIds, callback, mapToArguments(ctx, msg.getEntityId(), proto.getScope(), proto.getAttrDataList()), toTbMsgId(proto), toTbMsgType(proto));
} else {
callback.onSuccess(CALLBACKS_PER_CF);
}
@ -160,17 +157,17 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
@SneakyThrows
private void processTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback) {
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getTsDataList()));
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getTsDataList()), toTbMsgId(proto), toTbMsgType(proto));
}
@SneakyThrows
private void processAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback) {
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getScope(), proto.getAttrDataList()));
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getScope(), proto.getAttrDataList()), toTbMsgId(proto), toTbMsgType(proto));
}
@SneakyThrows
private void processArgumentValuesUpdate(CalculatedFieldCtx ctx, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback,
Map<String, ArgumentEntry> newArgValues) {
Map<String, ArgumentEntry> newArgValues, UUID tbMsgId, TbMsgType tbMsgType) {
if (newArgValues.isEmpty()) {
callback.onSuccess(CALLBACKS_PER_CF);
}
@ -178,7 +175,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
if (state.updateState(newArgValues)) {
cfIdList = new ArrayList<>(cfIdList);
cfIdList.add(ctx.getCfId());
processStateIfReady(ctx, cfIdList, state, callback);
processStateIfReady(ctx, cfIdList, state, tbMsgId, tbMsgType, callback);
} else {
callback.onSuccess(CALLBACKS_PER_CF);
}
@ -202,10 +199,13 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
}
@SneakyThrows
private void processStateIfReady(CalculatedFieldCtx ctx, List<CalculatedFieldId> cfIdList, CalculatedFieldState state, TbCallback callback) {
private void processStateIfReady(CalculatedFieldCtx ctx, List<CalculatedFieldId> cfIdList, CalculatedFieldState state, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) {
if (state.isReady()) {
CalculatedFieldResult calculationResult = state.performCalculation(ctx).get(5, TimeUnit.SECONDS);
cfService.pushMsgToRuleEngine(tenantId, entityId, calculationResult, cfIdList, callback);
if (DebugModeUtil.isDebugAllAvailable(ctx.getCalculatedField())) {
systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArguments(), tbMsgId, tbMsgType, JacksonUtil.writeValueAsString(calculationResult.getResultMap()), null);
}
} else {
callback.onSuccess(); // State was updated but no calculation performed;
}
@ -276,4 +276,18 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
return cfIds;
}
private UUID toTbMsgId(CalculatedFieldTelemetryMsgProto proto) {
if (proto.getTbMsgIdMSB() != 0 && proto.getTbMsgIdLSB() != 0) {
return new UUID(proto.getTbMsgIdMSB(), proto.getTbMsgIdLSB());
}
return null;
}
private TbMsgType toTbMsgType(CalculatedFieldTelemetryMsgProto proto) {
if (!proto.getTbMsgType().isEmpty()) {
return TbMsgType.valueOf(proto.getTbMsgType());
}
return null;
}
}

24
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java

@ -16,6 +16,7 @@
package org.thingsboard.server.actors.calculatedField;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.DebugModeUtil;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.TbActorCtx;
import org.thingsboard.server.actors.TbActorRef;
@ -95,6 +96,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
public void onFieldInitMsg(CalculatedFieldInitMsg msg) {
var cf = msg.getCf();
var cfCtx = new CalculatedFieldCtx(cf, systemContext.getTbelInvokeService());
try {
cfCtx.init();
} catch (Exception e) {
if (DebugModeUtil.isDebugAllAvailable(cf)) {
systemContext.persistCalculatedFieldDebugEvent(cf.getTenantId(), cf.getId(), cf.getEntityId(), null, null, null, null, e);
}
}
calculatedFields.put(cf.getId(), cfCtx);
// We use copy on write lists to safely pass the reference to another actor for the iteration.
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
@ -114,7 +122,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
if (calculatedFields.containsKey(msg.getId().cfId())) {
getOrCreateActor(msg.getId().entityId()).tell(msg);
} else {
// TODO: remove state from storage
cfExecService.deleteStateFromStorage(msg.getId(), msg.getCallback());
}
}
@ -209,6 +217,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
callback.onSuccess();
} else {
var cfCtx = new CalculatedFieldCtx(cf, systemContext.getTbelInvokeService());
try {
cfCtx.init();
} catch (Exception e) {
if (DebugModeUtil.isDebugAllAvailable(cf)) {
systemContext.persistCalculatedFieldDebugEvent(cf.getTenantId(), cf.getId(), cf.getEntityId(), null, null, null, null, e);
}
}
calculatedFields.put(cf.getId(), cfCtx);
// We use copy on write lists to safely pass the reference to another actor for the iteration.
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
@ -249,6 +264,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
// We use copy on write lists to safely pass the reference to another actor for the iteration.
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
if (newCfCtx.hasSignificantChanges(oldCfCtx)) {
try {
newCfCtx.init();
} catch (Exception e) {
if (DebugModeUtil.isDebugAllAvailable(newCf)) {
systemContext.persistCalculatedFieldDebugEvent(newCf.getTenantId(), newCf.getId(), newCf.getEntityId(), null, null, null, null, e);
}
}
initCf(newCfCtx, callback, true);
}
}

35
application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java

@ -32,6 +32,8 @@ import org.springframework.web.bind.annotation.RestController;
import org.thingsboard.server.common.data.cf.CalculatedField;
import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityIdFactory;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.config.annotations.ApiOperation;
@ -40,7 +42,8 @@ import org.thingsboard.server.service.entitiy.cf.TbCalculatedFieldService;
import org.thingsboard.server.service.security.permission.Operation;
import static org.thingsboard.server.controller.ControllerConstants.CF_TEXT_SEARCH_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.PAGE_DATA_PARAMETERS;
import static org.thingsboard.server.controller.ControllerConstants.ENTITY_ID_PARAM_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.ENTITY_TYPE_PARAM_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.PAGE_NUMBER_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.PAGE_SIZE_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.SORT_ORDER_DESCRIPTION;
@ -91,25 +94,25 @@ public class CalculatedFieldController extends BaseController {
return calculatedField;
}
@ApiOperation(value = "Get Calculated Fields (getCalculatedFields)",
notes = "Returns a page of calculated fields. " + PAGE_DATA_PARAMETERS
@ApiOperation(value = "Get Calculated Fields by Entity Id (getCalculatedFieldsByEntityId)",
notes = "Fetch the Calculated Fields based on the provided Entity Id."
)
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')")
@RequestMapping(value = "/calculatedFields", params = {"pageSize", "page"}, method = RequestMethod.GET)
@RequestMapping(value = "/{entityType}/{entityId}/calculatedFields", params = {"pageSize", "page"}, method = RequestMethod.GET)
@ResponseBody
public PageData<CalculatedField> getCalculatedFields(
@Parameter(description = PAGE_SIZE_DESCRIPTION, required = true)
@RequestParam int pageSize,
@Parameter(description = PAGE_NUMBER_DESCRIPTION, required = true)
@RequestParam int page,
@Parameter(description = CF_TEXT_SEARCH_DESCRIPTION)
@RequestParam(required = false) String textSearch,
@Parameter(description = SORT_PROPERTY_DESCRIPTION, schema = @Schema(allowableValues = {"createdTime", "name"}))
@RequestParam(required = false) String sortProperty,
@Parameter(description = SORT_ORDER_DESCRIPTION, schema = @Schema(allowableValues = {"ASC", "DESC"}))
@RequestParam(required = false) String sortOrder) throws ThingsboardException {
public PageData<CalculatedField> getCalculatedFieldsByEntityId(
@Parameter(description = ENTITY_TYPE_PARAM_DESCRIPTION, required = true, schema = @Schema(defaultValue = "DEVICE")) @PathVariable("entityType") String entityType,
@Parameter(description = ENTITY_ID_PARAM_DESCRIPTION, required = true) @PathVariable("entityId") String entityIdStr,
@Parameter(description = PAGE_SIZE_DESCRIPTION, required = true) @RequestParam int pageSize,
@Parameter(description = PAGE_NUMBER_DESCRIPTION, required = true) @RequestParam int page,
@Parameter(description = CF_TEXT_SEARCH_DESCRIPTION) @RequestParam(required = false) String textSearch,
@Parameter(description = SORT_PROPERTY_DESCRIPTION, schema = @Schema(allowableValues = {"createdTime", "name"})) @RequestParam(required = false) String sortProperty,
@Parameter(description = SORT_ORDER_DESCRIPTION, schema = @Schema(allowableValues = {"ASC", "DESC"})) @RequestParam(required = false) String sortOrder) throws ThingsboardException {
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder);
return checkNotNull(calculatedFieldService.findAllCalculatedFields(pageLink));
checkParameter("entityId", entityIdStr);
EntityId entityId = EntityIdFactory.getByTypeAndUuid(entityType, entityIdStr);
checkEntityId(entityId, Operation.READ_CALCULATED_FIELD);
return checkNotNull(tbCalculatedFieldService.findAllByTenantIdAndEntityId(entityId, getCurrentUser(), pageLink));
}
@ApiOperation(value = "Delete Calculated Field (deleteCalculatedField)",

54
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java

@ -40,7 +40,6 @@ import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
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.OutputType;
import org.thingsboard.server.common.data.id.AssetId;
@ -62,6 +61,7 @@ import org.thingsboard.server.common.data.kv.StringDataEntry;
import org.thingsboard.server.common.data.kv.TimeseriesSaveResult;
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;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.common.msg.queue.TbCallback;
@ -70,17 +70,19 @@ import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.dao.attributes.AttributesService;
import org.thingsboard.server.dao.cf.CalculatedFieldService;
import org.thingsboard.server.dao.timeseries.TimeseriesService;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.dao.usagerecord.ApiLimitService;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdProto;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto.Builder;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto;
import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg;
import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
import org.thingsboard.server.queue.TbQueueCallback;
import org.thingsboard.server.queue.TbQueueMsgMetadata;
import org.thingsboard.server.queue.discovery.HashPartitionService;
import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtx;
import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
@ -117,6 +119,7 @@ import java.util.stream.Collectors;
import static org.thingsboard.server.common.data.DataConstants.SCOPE;
import static org.thingsboard.server.common.util.ProtoUtils.toTsKvProto;
import static org.thingsboard.server.queue.discovery.HashPartitionService.CALCULATED_FIELD_QUEUE_KEY;
@Service
@Slf4j
@ -141,14 +144,13 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
private final TimeseriesService timeseriesService;
private final CalculatedFieldStateService stateService;
private final TbClusterService clusterService;
private final ApiLimitService apiLimitService;
private ListeningExecutorService calculatedFieldExecutor;
private ListeningExecutorService calculatedFieldCallbackExecutor;
private final ConcurrentMap<CalculatedFieldEntityCtxId, CalculatedFieldEntityCtx> states = new ConcurrentHashMap<>();
private static final int MAX_LAST_RECORDS_VALUE = 1024;
private static final Set<EntityType> supportedReferencedEntities = EnumSet.of(
EntityType.DEVICE, EntityType.ASSET, EntityType.CUSTOMER, EntityType.TENANT
);
@ -245,7 +247,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
argFutures.put(entry.getKey(), argValueFuture);
}
return Futures.whenAllComplete(argFutures.values()).call(() -> {
var result = createStateByType(ctx.getCfType());
var result = createStateByType(ctx);
result.updateState(argFutures.entrySet().stream()
.collect(Collectors.toMap(
Entry::getKey, // Keep the key as is
@ -450,7 +452,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
if (broadcast) {
broadcasts.add(link);
} else {
TopicPartitionInfo tpi = partitionService.resolve(HashPartitionService.CALCULATED_FIELD_QUEUE_KEY, link.entityId());
TopicPartitionInfo tpi = partitionService.resolve(CALCULATED_FIELD_QUEUE_KEY, link.entityId());
unicasts.computeIfAbsent(tpi, k -> new ArrayList<>()).add(link);
}
}
@ -483,7 +485,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
}
private CalculatedFieldLinkedTelemetryMsgProto buildLinkedTelemetryMsgProto(CalculatedFieldTelemetryMsgProto telemetryProto, List<CalculatedFieldEntityCtxId> links) {
TransportProtos.CalculatedFieldLinkedTelemetryMsgProto.Builder builder = TransportProtos.CalculatedFieldLinkedTelemetryMsgProto.newBuilder();
Builder builder = CalculatedFieldLinkedTelemetryMsgProto.newBuilder();
builder.setMsg(telemetryProto);
for (CalculatedFieldEntityCtxId link : links) {
builder.addLinks(toProto(link));
@ -492,8 +494,10 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
}
//TODO: IM: move to utils;
private TransportProtos.CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) {
return TransportProtos.CalculatedFieldEntityCtxIdProto.newBuilder()
private CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) {
return CalculatedFieldEntityCtxIdProto.newBuilder()
.setTenantIdMSB(ctxId.tenantId().getId().getMostSignificantBits())
.setTenantIdLSB(ctxId.tenantId().getId().getLeastSignificantBits())
.setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits())
.setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits())
.setEntityType(ctxId.entityId().getEntityType().name())
@ -557,7 +561,8 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
long currentTime = System.currentTimeMillis();
long timeWindow = argument.getTimeWindow() == 0 ? System.currentTimeMillis() : argument.getTimeWindow();
long startTs = currentTime - timeWindow;
int limit = argument.getLimit() == 0 ? MAX_LAST_RECORDS_VALUE : argument.getLimit();
long maxDataPoints = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxDataPointsPerRollingArg);
int limit = argument.getLimit() == 0 ? (int) maxDataPoints : argument.getLimit();
ReadTsKvQuery query = new BaseReadTsKvQuery(argument.getRefEntityKey().getKey(), startTs, currentTime, 0, limit, Aggregation.NONE);
ListenableFuture<List<TsKvEntry>> tsRollingFuture = timeseriesService.findAll(tenantId, entityId, List.of(query));
@ -584,10 +589,10 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
return payload;
}
private CalculatedFieldState createStateByType(CalculatedFieldType calculatedFieldType) {
return switch (calculatedFieldType) {
case SIMPLE -> new SimpleCalculatedFieldState();
case SCRIPT -> new ScriptCalculatedFieldState();
private CalculatedFieldState createStateByType(CalculatedFieldCtx ctx) {
return switch (ctx.getCfType()) {
case SIMPLE -> new SimpleCalculatedFieldState(ctx.getArgNames());
case SCRIPT -> new ScriptCalculatedFieldState(ctx.getArgNames());
};
}
@ -606,7 +611,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
private ToCalculatedFieldMsg toCalculatedFieldTelemetryMsgProto(TimeseriesSaveRequest request, TimeseriesSaveResult result) {
ToCalculatedFieldMsg.Builder msg = ToCalculatedFieldMsg.newBuilder();
CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = buildTelemetryMsgProto(request.getTenantId(), request.getEntityId(), request.getPreviousCalculatedFieldIds());
CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = buildTelemetryMsgProto(request.getTenantId(), request.getEntityId(), request.getPreviousCalculatedFieldIds(), request.getTbMsgId(), request.getTbMsgType());
List<TsKvEntry> entries = request.getEntries();
List<Long> versions = result.getVersions();
for (int i = 0; i < entries.size(); i++) {
@ -622,7 +627,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
private ToCalculatedFieldMsg toCalculatedFieldTelemetryMsgProto(AttributesSaveRequest request, List<Long> versions) {
ToCalculatedFieldMsg.Builder msg = ToCalculatedFieldMsg.newBuilder();
CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = buildTelemetryMsgProto(request.getTenantId(), request.getEntityId(), request.getPreviousCalculatedFieldIds());
CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = buildTelemetryMsgProto(request.getTenantId(), request.getEntityId(), request.getPreviousCalculatedFieldIds(), request.getTbMsgId(), request.getTbMsgType());
telemetryMsg.setScope(AttributeScopeProto.valueOf(request.getScope().name()));
List<AttributeKvEntry> entries = request.getEntries();
for (int i = 0; i < entries.size(); i++) {
@ -635,7 +640,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
return msg.build();
}
private CalculatedFieldTelemetryMsgProto.Builder buildTelemetryMsgProto(TenantId tenantId, EntityId entityId, List<CalculatedFieldId> calculatedFieldIds) {
private CalculatedFieldTelemetryMsgProto.Builder buildTelemetryMsgProto(TenantId tenantId, EntityId entityId, List<CalculatedFieldId> calculatedFieldIds, UUID tbMsgId, TbMsgType tbMsgType) {
CalculatedFieldTelemetryMsgProto.Builder telemetryMsg = CalculatedFieldTelemetryMsgProto.newBuilder();
telemetryMsg.setTenantIdMSB(tenantId.getId().getMostSignificantBits());
@ -651,11 +656,20 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
}
}
if (tbMsgId != null) {
telemetryMsg.setTbMsgIdMSB(tbMsgId.getMostSignificantBits());
telemetryMsg.setTbMsgIdLSB(tbMsgId.getLeastSignificantBits());
}
if (tbMsgType != null) {
telemetryMsg.setTbMsgType(tbMsgType.name());
}
return telemetryMsg;
}
private TransportProtos.CalculatedFieldIdProto toProto(CalculatedFieldId cfId) {
return TransportProtos.CalculatedFieldIdProto.newBuilder()
private CalculatedFieldIdProto toProto(CalculatedFieldId cfId) {
return CalculatedFieldIdProto.newBuilder()
.setCalculatedFieldIdMSB(cfId.getId().getMostSignificantBits())
.setCalculatedFieldIdLSB(cfId.getId().getLeastSignificantBits())
.build();

32
application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java

@ -22,6 +22,8 @@ import org.rocksdb.RocksIterator;
import org.rocksdb.WriteOptions;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
import org.thingsboard.server.utils.RocksDBConfig;
import java.nio.charset.StandardCharsets;
@ -49,9 +51,17 @@ public class RocksDBService {
}
}
public void delete(String key) {
public void put(CalculatedFieldEntityCtxIdProto key, CalculatedFieldStateProto value) {
try {
db.delete(writeOptions, key.getBytes(StandardCharsets.UTF_8));
db.put(writeOptions, key.toByteArray(), value.toByteArray());
} catch (RocksDBException e) {
log.error("Failed to store data to RocksDB", e);
}
}
public void delete(CalculatedFieldEntityCtxIdProto key) {
try {
db.delete(writeOptions, key.toByteArray());
} catch (RocksDBException e) {
log.error("Failed to delete data from RocksDB", e);
}
@ -67,18 +77,20 @@ public class RocksDBService {
}
}
public Map<String, String> getAll() {
Map<String, String> map = new HashMap<>();
public Map<CalculatedFieldEntityCtxIdProto, CalculatedFieldStateProto> getAll() {
Map<CalculatedFieldEntityCtxIdProto, CalculatedFieldStateProto> results = new HashMap<>();
try (RocksIterator iterator = db.newIterator()) {
for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) {
String key = new String(iterator.key(), StandardCharsets.UTF_8);
String value = new String(iterator.value(), StandardCharsets.UTF_8);
map.put(key, value);
try {
CalculatedFieldEntityCtxIdProto key = CalculatedFieldEntityCtxIdProto.parseFrom(iterator.key());
CalculatedFieldStateProto value = CalculatedFieldStateProto.parseFrom(iterator.value());
results.put(key, value);
} catch (Exception e) {
log.error("Failed to retrieve data from RocksDB", e);
}
}
} catch (Exception e) {
log.error("Failed to retrieve data from RocksDB", e);
}
return map;
return results;
}
}

2
application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java

@ -24,8 +24,6 @@ public interface CalculatedFieldStateService {
Map<CalculatedFieldEntityCtxId, CalculatedFieldState> restoreStates();
CalculatedFieldState restoreState(CalculatedFieldEntityCtxId ctxId);
void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback);
void removeState(CalculatedFieldEntityCtxId stateId, TbCallback callback);

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

@ -39,7 +39,7 @@ public interface ArgumentEntry {
Object getValue();
boolean hasUpdatedValue(ArgumentEntry entry);
boolean updateEntry(ArgumentEntry entry);
static ArgumentEntry createSingleValueArgument(KvEntry kvEntry) {
return new SingleValueArgumentEntry(kvEntry);

45
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java

@ -15,20 +15,31 @@
*/
package org.thingsboard.server.service.cf.ctx.state;
import lombok.NoArgsConstructor;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@NoArgsConstructor
public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
protected List<String> requiredArguments;
protected Map<String, ArgumentEntry> arguments;
public BaseCalculatedFieldState() {
arguments = new HashMap<>();
public BaseCalculatedFieldState(List<String> requiredArguments) {
this.requiredArguments = requiredArguments;
this.arguments = new HashMap<>();
}
@Override
public Map<String, ArgumentEntry> getArguments() {
return this.arguments;
return arguments;
}
@Override
public List<String> getRequiredArguments() {
return requiredArguments;
}
@Override
@ -44,22 +55,12 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
ArgumentEntry newEntry = entry.getValue();
ArgumentEntry existingEntry = arguments.get(key);
if (existingEntry == null || existingEntry.hasUpdatedValue(newEntry)) {
if (existingEntry instanceof TsRollingArgumentEntry existingTsRollingEntry && newEntry instanceof TsRollingArgumentEntry newTsRollingEntry) {
existingTsRollingEntry.addAllTsRecords(newTsRollingEntry.getTsRecords());
} else if (existingEntry instanceof TsRollingArgumentEntry existingTsRollingEntry && newEntry instanceof SingleValueArgumentEntry singleValueEntry) {
existingTsRollingEntry.addTsRecord(singleValueEntry.getTs(), singleValueEntry.getValue());
} else if (existingEntry instanceof SingleValueArgumentEntry existingSingleValueEntry && newEntry instanceof SingleValueArgumentEntry singleValueEntry) {
// Long existingVersion = existingSingleValueEntry.getVersion();
// Long newVersion = singleValueEntry.getVersion();
// if (newVersion != null && (existingVersion == null || newVersion > existingVersion)) {
// arguments.put(key, newEntry.copy());
// }
arguments.put(key, newEntry.copy());
} else {
arguments.put(key, newEntry.copy());
}
if (existingEntry == null) {
validateNewEntry(newEntry);
arguments.put(key, newEntry);
stateUpdated = true;
} else {
stateUpdated = existingEntry.updateEntry(newEntry);
}
}
@ -68,7 +69,11 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
@Override
public boolean isReady() {
//TODO: IM
return true;
return arguments.keySet().containsAll(requiredArguments) &&
!arguments.containsValue(SingleValueArgumentEntry.EMPTY) &&
!arguments.containsValue(TsRollingArgumentEntry.EMPTY);
}
protected abstract void validateNewEntry(ArgumentEntry newEntry);
}

61
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java

@ -16,6 +16,9 @@
package org.thingsboard.server.service.cf.ctx.state;
import lombok.Data;
import net.objecthunter.exp4j.Expression;
import net.objecthunter.exp4j.ExpressionBuilder;
import org.mvel2.MVEL;
import org.thingsboard.script.api.tbel.TbelInvokeService;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.cf.CalculatedField;
@ -31,6 +34,7 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.util.TbPair;
import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto;
import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
@ -43,6 +47,8 @@ import java.util.stream.Collectors;
@Data
public class CalculatedFieldCtx {
private CalculatedField calculatedField;
private CalculatedFieldId cfId;
private TenantId tenantId;
private EntityId entityId;
@ -57,8 +63,13 @@ public class CalculatedFieldCtx {
private String expression;
private TbelInvokeService tbelInvokeService;
private CalculatedFieldScriptEngine calculatedFieldScriptEngine;
private ThreadLocal<Expression> customExpression;
private boolean initialized;
public CalculatedFieldCtx(CalculatedField calculatedField, TbelInvokeService tbelInvokeService) {
this.calculatedField = calculatedField;
this.cfId = calculatedField.getId();
this.tenantId = calculatedField.getTenantId();
this.entityId = calculatedField.getEntityId();
@ -85,8 +96,28 @@ public class CalculatedFieldCtx {
this.output = configuration.getOutput();
this.expression = configuration.getExpression();
this.tbelInvokeService = tbelInvokeService;
if (CalculatedFieldType.SCRIPT.equals(calculatedField.getType())) {
this.calculatedFieldScriptEngine = initEngine(tenantId, expression, tbelInvokeService);
}
public void init() {
if (CalculatedFieldType.SCRIPT.equals(cfType)) {
try {
this.calculatedFieldScriptEngine = initEngine(tenantId, expression, tbelInvokeService);
initialized = true;
} catch (Exception e) {
throw new RuntimeException("Failed to init calculated field ctx. Invalid expression syntax.", e);
}
} else {
if (isValidExpression(expression)) {
this.customExpression = ThreadLocal.withInitial(() ->
new ExpressionBuilder(expression)
.implicitMultiplication(true)
.variables(this.arguments.keySet())
.build()
);
initialized = true;
} else {
throw new RuntimeException("Failed to init calculated field ctx. Invalid expression syntax.");
}
}
}
@ -103,6 +134,15 @@ public class CalculatedFieldCtx {
);
}
private boolean isValidExpression(String expression) {
try {
MVEL.compileExpression(expression);
return true;
} catch (Exception e) {
return false;
}
}
public boolean matches(List<AttributeKvEntry> values, AttributeScope scope) {
return matchesAttributes(mainEntityArguments, values, scope);
}
@ -146,8 +186,18 @@ public class CalculatedFieldCtx {
}
public boolean linkMatches(EntityId entityId, CalculatedFieldTelemetryMsgProto proto) {
//TODO: IM - implement
return true;
if (!proto.getTsDataList().isEmpty()) {
List<TsKvEntry> updatedTelemetry = proto.getTsDataList().stream()
.map(ProtoUtils::fromProto)
.toList();
return linkMatches(entityId, updatedTelemetry);
} else {
AttributeScope scope = AttributeScope.valueOf(proto.getScope().name());
List<AttributeKvEntry> updatedTelemetry = proto.getAttrDataList().stream()
.map(ProtoUtils::fromProto)
.toList();
return linkMatches(entityId, updatedTelemetry, scope);
}
}
public CalculatedFieldEntityCtxId toCalculatedFieldEntityCtxId() {
@ -158,7 +208,8 @@ public class CalculatedFieldCtx {
boolean entityIdChanged = !entityId.equals(other.entityId);
boolean typeChanged = !cfType.equals(other.cfType);
boolean argumentsChanged = !arguments.equals(other.arguments);
return entityIdChanged || typeChanged || argumentsChanged;
boolean expressionChanged = !expression.equals(other.expression);
return entityIdChanged || typeChanged || argumentsChanged || expressionChanged;
}
}

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

@ -22,6 +22,7 @@ import com.google.common.util.concurrent.ListenableFuture;
import org.thingsboard.server.common.data.cf.CalculatedFieldType;
import org.thingsboard.server.service.cf.CalculatedFieldResult;
import java.util.List;
import java.util.Map;
@JsonTypeInfo(
@ -40,6 +41,8 @@ public interface CalculatedFieldState {
Map<String, ArgumentEntry> getArguments();
List<String> getRequiredArguments();
boolean updateState(Map<String, ArgumentEntry> argumentValues);
ListenableFuture<CalculatedFieldResult> performCalculation(CalculatedFieldCtx ctx);

151
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java

@ -18,14 +18,28 @@ package org.thingsboard.server.service.cf.ctx.state;
import lombok.RequiredArgsConstructor;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.cf.CalculatedFieldType;
import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityIdFactory;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.BasicKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
import org.thingsboard.server.gen.transport.TransportProtos.RollingArgumentProto;
import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto;
import org.thingsboard.server.gen.transport.TransportProtos.SingleValueProto;
import org.thingsboard.server.service.cf.RocksDBService;
import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
import java.util.Map;
import java.util.Optional;
import java.util.TreeMap;
import java.util.UUID;
import java.util.stream.Collectors;
@Service
@ -39,28 +53,137 @@ public class RocksDBStateService implements CalculatedFieldStateService {
public Map<CalculatedFieldEntityCtxId, CalculatedFieldState> restoreStates() {
return rocksDBService.getAll().entrySet().stream()
.collect(Collectors.toMap(
entry -> JacksonUtil.fromString(entry.getKey(), CalculatedFieldEntityCtxId.class),
entry -> JacksonUtil.fromString(entry.getValue(), CalculatedFieldState.class)
entry -> fromProto(entry.getKey()),
entry -> fromProto(entry.getValue())
));
}
@Override
public CalculatedFieldState restoreState(CalculatedFieldEntityCtxId ctxId) {
return Optional.ofNullable(rocksDBService.get(JacksonUtil.writeValueAsString(ctxId)))
.map(storedState -> JacksonUtil.fromString(storedState, CalculatedFieldState.class))
.orElse(null);
}
@Override
public void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback){
rocksDBService.put(JacksonUtil.writeValueAsString(stateId), JacksonUtil.writeValueAsString(state));
public void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) {
rocksDBService.put(toProto(stateId), toProto(stateId, state));
callback.onSuccess();
}
@Override
public void removeState(CalculatedFieldEntityCtxId ctxId, TbCallback callback) {
rocksDBService.delete(JacksonUtil.writeValueAsString(ctxId));
rocksDBService.delete(toProto(ctxId));
callback.onSuccess();
}
private CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) {
return CalculatedFieldEntityCtxIdProto.newBuilder()
.setTenantIdMSB(ctxId.tenantId().getId().getMostSignificantBits())
.setTenantIdLSB(ctxId.tenantId().getId().getLeastSignificantBits())
.setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits())
.setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits())
.setEntityType(ctxId.entityId().getEntityType().name())
.setEntityIdMSB(ctxId.entityId().getId().getMostSignificantBits())
.setEntityIdLSB(ctxId.entityId().getId().getLeastSignificantBits())
.build();
}
private CalculatedFieldEntityCtxId fromProto(CalculatedFieldEntityCtxIdProto ctxIdProto) {
TenantId tenantId = TenantId.fromUUID(new UUID(ctxIdProto.getTenantIdMSB(), ctxIdProto.getTenantIdLSB()));
EntityId entityId = EntityIdFactory.getByTypeAndUuid(ctxIdProto.getEntityType(), new UUID(ctxIdProto.getEntityIdMSB(), ctxIdProto.getEntityIdLSB()));
CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(ctxIdProto.getCalculatedFieldIdMSB(), ctxIdProto.getCalculatedFieldIdLSB()));
return new CalculatedFieldEntityCtxId(tenantId, calculatedFieldId, entityId);
}
private CalculatedFieldStateProto toProto(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state) {
CalculatedFieldStateProto.Builder builder = CalculatedFieldStateProto.newBuilder()
.setId(toProto(stateId))
.setType(state.getType().name())
.addAllRequiredArguments(state.getRequiredArguments());
state.getArguments().forEach((argName, argEntry) -> {
if (argEntry instanceof SingleValueArgumentEntry singleValueArgumentEntry) {
builder.addSingleValueArguments(toSingleValueArgumentProto(argName, singleValueArgumentEntry));
} else if (argEntry instanceof TsRollingArgumentEntry rollingArgumentEntry) {
builder.addRollingValueArguments(toRollingArgumentProto(argName, rollingArgumentEntry));
}
});
return builder.build();
}
private SingleValueArgumentProto toSingleValueArgumentProto(String argName, SingleValueArgumentEntry entry) {
SingleValueProto.Builder singleValueProtoBuilder = SingleValueProto.newBuilder()
.setTs(entry.getTs());
if (entry.getVersion() != null) {
singleValueProtoBuilder.setVersion(entry.getVersion());
}
KvEntry value = entry.getKvEntryValue();
if (value != null) {
singleValueProtoBuilder.setHasV(true)
.setValue(ProtoUtils.toKeyValueProto(value));
}
return SingleValueArgumentProto.newBuilder()
.setArgName(argName)
.setValue(singleValueProtoBuilder.build())
.build();
}
private RollingArgumentProto toRollingArgumentProto(String argName, TsRollingArgumentEntry entry) {
RollingArgumentProto.Builder rollingArgumentProtoBuilder = RollingArgumentProto.newBuilder()
.setArgName(argName);
entry.getTsRecords().forEach((ts, value) -> {
SingleValueProto.Builder singleValueProtoBuilder = SingleValueProto.newBuilder()
.setTs(ts);
if (value != null) {
singleValueProtoBuilder.setHasV(true)
.setValue(ProtoUtils.toKeyValueProto(value));
}
rollingArgumentProtoBuilder.addValues(singleValueProtoBuilder.build());
});
return rollingArgumentProtoBuilder.build();
}
private CalculatedFieldState fromProto(CalculatedFieldStateProto proto) {
if (StringUtils.isEmpty(proto.getType())) {
return null;
}
CalculatedFieldType type = CalculatedFieldType.valueOf(proto.getType());
CalculatedFieldState state = switch (type) {
case SIMPLE -> new SimpleCalculatedFieldState(proto.getRequiredArgumentsList());
case SCRIPT -> new ScriptCalculatedFieldState(proto.getRequiredArgumentsList());
};
proto.getSingleValueArgumentsList().forEach(argProto ->
state.getArguments().put(argProto.getArgName(), fromSingleValueArgumentProto(argProto)));
if (CalculatedFieldType.SCRIPT.equals(type)) {
proto.getRollingValueArgumentsList().forEach(argProto ->
state.getArguments().put(argProto.getArgName(), fromRollingArgumentProto(argProto)));
}
return state;
}
private SingleValueArgumentEntry fromSingleValueArgumentProto(SingleValueArgumentProto proto) {
SingleValueProto valueProto = proto.getValue();
BasicKvEntry value = valueProto.getHasV() ? ProtoUtils.fromProto(valueProto.getValue()) : null;
return new SingleValueArgumentEntry(valueProto.getTs(), value, valueProto.getVersion());
}
private TsRollingArgumentEntry fromRollingArgumentProto(RollingArgumentProto proto) {
TreeMap<Long, BasicKvEntry> tsRecords = new TreeMap<>();
proto.getValuesList().forEach(singleValueProto -> {
BasicKvEntry value = singleValueProto.getHasV() ? ProtoUtils.fromProto(singleValueProto.getValue()) : null;
tsRecords.put(singleValueProto.getTs(), value);
});
return new TsRollingArgumentEntry(tsRecords);
}
}

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

@ -19,30 +19,42 @@ import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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.Output;
import org.thingsboard.server.common.data.kv.BasicKvEntry;
import org.thingsboard.server.service.cf.CalculatedFieldResult;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
@Data
@Slf4j
@NoArgsConstructor
public class ScriptCalculatedFieldState extends BaseCalculatedFieldState {
public ScriptCalculatedFieldState(List<String> requiredArguments) {
super(requiredArguments);
}
@Override
public CalculatedFieldType getType() {
return CalculatedFieldType.SCRIPT;
}
@Override
protected void validateNewEntry(ArgumentEntry newEntry) {
}
@Override
public ListenableFuture<CalculatedFieldResult> performCalculation(CalculatedFieldCtx ctx) {
arguments.forEach((key, argumentEntry) -> {
if (argumentEntry instanceof TsRollingArgumentEntry) {
if (argumentEntry instanceof TsRollingArgumentEntry tsRollingEntry) {
Argument argument = ctx.getArguments().get(key);
TreeMap<Long, Object> tsRecords = ((TsRollingArgumentEntry) argumentEntry).getTsRecords();
TreeMap<Long, BasicKvEntry> tsRecords = tsRollingEntry.getTsRecords();
if (tsRecords.size() > argument.getLimit()) {
tsRecords.pollFirstEntry();
}

39
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java

@ -18,38 +18,47 @@ package org.thingsboard.server.service.cf.ctx.state;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.Data;
import net.objecthunter.exp4j.Expression;
import net.objecthunter.exp4j.ExpressionBuilder;
import lombok.NoArgsConstructor;
import org.thingsboard.server.common.data.cf.CalculatedFieldType;
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 java.util.HashMap;
import java.util.List;
import java.util.Map;
@Data
@NoArgsConstructor
public class SimpleCalculatedFieldState extends BaseCalculatedFieldState {
public SimpleCalculatedFieldState(List<String> requiredArguments) {
super(requiredArguments);
}
@Override
public CalculatedFieldType getType() {
return CalculatedFieldType.SIMPLE;
}
@Override
protected void validateNewEntry(ArgumentEntry newEntry) {
if (newEntry instanceof TsRollingArgumentEntry) {
throw new IllegalArgumentException("Rolling argument entry is not supported for simple calculated fields.");
}
}
@Override
public ListenableFuture<CalculatedFieldResult> performCalculation(CalculatedFieldCtx ctx) {
String expression = ctx.getExpression();
ThreadLocal<Expression> customExpression = new ThreadLocal<>();
var expr = customExpression.get();
if (expr == null) {
expr = new ExpressionBuilder(expression)
.implicitMultiplication(true)
.variables(this.arguments.keySet())
.build();
customExpression.set(expr);
var expr = ctx.getCustomExpression().get();
for (Map.Entry<String, ArgumentEntry> entry : this.arguments.entrySet()) {
try {
BasicKvEntry kvEntry = ((SingleValueArgumentEntry) entry.getValue()).getKvEntryValue();
expr.setVariable(entry.getKey(), Double.parseDouble(kvEntry.getValueAsString()));
} catch (NumberFormatException e) {
throw new IllegalArgumentException("Argument '" + entry.getKey() + "' is not a number.");
}
}
Map<String, Double> variables = new HashMap<>();
this.arguments.forEach((k, v) -> variables.put(k, Double.parseDouble(v.getValue().toString())));
expr.setVariables(variables);
double expressionResult = expr.evaluate();

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

@ -15,13 +15,14 @@
*/
package org.thingsboard.server.service.cf.ctx.state;
import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BasicKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.util.KvProtoUtil;
import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto;
import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
@ -34,20 +35,19 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
public static final ArgumentEntry EMPTY = new SingleValueArgumentEntry(0);
private long ts;
private Object value;
private BasicKvEntry kvEntryValue;
private Long version;
public SingleValueArgumentEntry(TsKvProto entry) {
this.ts = entry.getTs();
this.version = entry.getVersion();
this.value = ProtoUtils.fromProto(entry).getValue();
this.kvEntryValue = ProtoUtils.fromProto(entry.getKv());
}
public SingleValueArgumentEntry(AttributeValueProto entry) {
this.ts = entry.getLastUpdateTs();
this.version = entry.getVersion();
this.value = ProtoUtils.fromProto(entry).getValue();
this.kvEntryValue = ProtoUtils.basicKvEntryFromProto(entry);
}
public SingleValueArgumentEntry(KvEntry entry) {
@ -58,7 +58,7 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
this.ts = attributeKvEntry.getLastUpdateTs();
this.version = attributeKvEntry.getVersion();
}
this.value = entry.getValue();
this.kvEntryValue = ProtoUtils.basicKvEntryFromKvEntry(entry);
}
/**
@ -66,7 +66,7 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
* */
private SingleValueArgumentEntry(int ignored) {
this.ts = System.currentTimeMillis();
this.value = null;
this.kvEntryValue = null;
}
@Override
@ -74,19 +74,33 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
return ArgumentEntryType.SINGLE_VALUE;
}
@Override
@JsonIgnore
public Object getValue() {
return value;
return kvEntryValue.getValue();
}
@Override
public boolean hasUpdatedValue(ArgumentEntry entry) {
return this.ts != ((SingleValueArgumentEntry) entry).getTs();
public ArgumentEntry copy() {
return new SingleValueArgumentEntry(this.ts, this.kvEntryValue, this.version);
}
@Override
public ArgumentEntry copy() {
return new SingleValueArgumentEntry(this.ts, this.value, this.version);
}
public boolean updateEntry(ArgumentEntry entry) {
if (entry instanceof SingleValueArgumentEntry singleValueEntry) {
if (singleValueEntry.getTs() == this.ts) {
return false;
}
Long newVersion = singleValueEntry.getVersion();
if (newVersion == null || this.version == null || newVersion > this.version) {
this.ts = singleValueEntry.getTs();
this.kvEntryValue = singleValueEntry.getKvEntryValue();
this.version = newVersion;
return true;
}
} else {
throw new IllegalArgumentException("Unsupported argument entry type for single value argument entry: " + entry.getType());
}
return false;
}
}

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

@ -21,11 +21,15 @@ import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.math.NumberUtils;
import org.thingsboard.server.common.data.kv.BasicKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.util.ProtoUtils;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.stream.Collectors;
@Data
@NoArgsConstructor
@ -37,10 +41,10 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
private static final int MAX_ROLLING_ARGUMENT_ENTRY_SIZE = 1000;
private TreeMap<Long, Object> tsRecords = new TreeMap<>();
private TreeMap<Long, BasicKvEntry> tsRecords = new TreeMap<>();
public TsRollingArgumentEntry(List<TsKvEntry> kvEntries) {
kvEntries.forEach(tsKvEntry -> addTsRecord(tsKvEntry.getTs(), tsKvEntry.getValue()));
kvEntries.forEach(tsKvEntry -> addTsRecord(tsKvEntry.getTs(), tsKvEntry));
}
/**
@ -58,14 +62,15 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
@JsonIgnore
@Override
public Object getValue() {
return tsRecords;
}
return tsRecords.entrySet()
.stream()
.collect(Collectors.toMap(
Map.Entry::getKey,
entry -> entry.getValue().getValue(),
(oldValue, newValue) -> oldValue,
TreeMap::new
));
@Override
public boolean hasUpdatedValue(ArgumentEntry entry) {
return entry instanceof SingleValueArgumentEntry ?
!tsRecords.containsKey(((SingleValueArgumentEntry) entry).getTs()) :
!tsRecords.keySet().containsAll(((TsRollingArgumentEntry) entry).getTsRecords().keySet());
}
@Override
@ -73,20 +78,45 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
return new TsRollingArgumentEntry(new TreeMap<>(tsRecords));
}
public void addTsRecord(Long key, Object value) {
if (NumberUtils.isParsable(value.toString())) {
tsRecords.put(key, value);
if (tsRecords.size() > MAX_ROLLING_ARGUMENT_ENTRY_SIZE) {
tsRecords.pollFirstEntry();
}
@Override
public boolean updateEntry(ArgumentEntry entry) {
if (entry instanceof TsRollingArgumentEntry tsRollingEntry) {
return updateTsRollingEntry(tsRollingEntry);
} else if (entry instanceof SingleValueArgumentEntry singleValueEntry) {
return updateSingleValueEntry(singleValueEntry);
} else {
log.warn("Argument type 'TS_ROLLING' only supports numeric values.");
throw new IllegalArgumentException("Unsupported argument entry type for rolling argument entry: " + entry.getType());
}
}
private boolean updateTsRollingEntry(TsRollingArgumentEntry tsRollingEntry) {
boolean updated = false;
for (Map.Entry<Long, BasicKvEntry> tsRecordEntry : tsRollingEntry.getTsRecords().entrySet()) {
updated |= addTsRecordIfAbsent(tsRecordEntry.getKey(), tsRecordEntry.getValue());
}
return updated;
}
public void addAllTsRecords(Map<Long, Object> newRecords) {
for (Map.Entry<Long, Object> entry : newRecords.entrySet()) {
addTsRecord(entry.getKey(), entry.getValue());
private boolean updateSingleValueEntry(SingleValueArgumentEntry singleValueEntry) {
return addTsRecordIfAbsent(singleValueEntry.getTs(), singleValueEntry.getKvEntryValue());
}
private boolean addTsRecordIfAbsent(Long ts, KvEntry value) {
if (!tsRecords.containsKey(ts)) {
addTsRecord(ts, value);
return true;
}
return false;
}
private void addTsRecord(Long ts, KvEntry value) {
if (NumberUtils.isParsable(value.getValue().toString())) {
tsRecords.put(ts, ProtoUtils.basicKvEntryFromKvEntry(value));
if (tsRecords.size() > MAX_ROLLING_ARGUMENT_ENTRY_SIZE) {
tsRecords.pollFirstEntry();
}
} else {
throw new IllegalArgumentException("Argument type " + getType() + " only supports numeric values.");
}
}

37
application/src/main/java/org/thingsboard/server/service/entitiy/cf/DefaultTbCalculatedFieldService.java

@ -29,6 +29,8 @@ import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.HasId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.dao.cf.CalculatedFieldService;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.entitiy.AbstractTbEntityService;
@ -46,9 +48,6 @@ import static org.thingsboard.server.dao.service.Validator.validateEntityId;
@RequiredArgsConstructor
public class DefaultTbCalculatedFieldService extends AbstractTbEntityService implements TbCalculatedFieldService {
private static final int MAX_ARGUMENT_SIZE = 10;
private static final int MAX_CALCULATED_FIELD_NUMBER = 10;
private final CalculatedFieldService calculatedFieldService;
@Override
@ -56,9 +55,11 @@ public class DefaultTbCalculatedFieldService extends AbstractTbEntityService imp
ActionType actionType = calculatedField.getId() == null ? ActionType.ADDED : ActionType.UPDATED;
TenantId tenantId = calculatedField.getTenantId();
try {
checkCalculatedFieldNumber(tenantId, calculatedField.getEntityId());
if (ActionType.UPDATED.equals(actionType)) {
CalculatedField existingCf = calculatedFieldService.findById(tenantId, calculatedField.getId());
checkForEntityChange(existingCf, calculatedField);
}
checkEntityExistence(tenantId, calculatedField.getEntityId());
checkArgumentSize(calculatedField.getConfiguration());
checkReferencedEntities(calculatedField.getConfiguration(), user);
CalculatedField savedCalculatedField = checkNotNull(calculatedFieldService.save(calculatedField));
logEntityActionService.logEntityAction(tenantId, savedCalculatedField.getId(), savedCalculatedField, actionType, user);
@ -74,6 +75,13 @@ public class DefaultTbCalculatedFieldService extends AbstractTbEntityService imp
return calculatedFieldService.findById(user.getTenantId(), calculatedFieldId);
}
@Override
public PageData<CalculatedField> findAllByTenantIdAndEntityId(EntityId entityId, SecurityUser user, PageLink pageLink) {
TenantId tenantId = user.getTenantId();
checkEntityExistence(tenantId, entityId);
return calculatedFieldService.findAllCalculatedFieldsByEntityId(tenantId, entityId, pageLink);
}
@Override
@Transactional
public void delete(CalculatedField calculatedField, SecurityUser user) {
@ -89,6 +97,12 @@ public class DefaultTbCalculatedFieldService extends AbstractTbEntityService imp
}
}
private void checkForEntityChange(CalculatedField oldCalculatedField, CalculatedField newCalculatedField) {
if (!oldCalculatedField.getEntityId().equals(newCalculatedField.getEntityId())) {
throw new IllegalArgumentException("Changing the calculated field target entity after initialization is prohibited.");
}
}
private void checkEntityExistence(TenantId tenantId, EntityId entityId) {
switch (entityId.getEntityType()) {
case ASSET, DEVICE, ASSET_PROFILE, DEVICE_PROFILE ->
@ -110,19 +124,6 @@ public class DefaultTbCalculatedFieldService extends AbstractTbEntityService imp
}
private void checkArgumentSize(CalculatedFieldConfiguration calculatedFieldConfig) {
if (calculatedFieldConfig.getArguments().size() > MAX_ARGUMENT_SIZE) {
throw new IllegalArgumentException("Too many arguments: " + calculatedFieldConfig.getArguments().size() + ". Max number of argument is " + MAX_ARGUMENT_SIZE);
}
}
private void checkCalculatedFieldNumber(TenantId tenantId, EntityId entityId) {
int numberOfCalculatedFieldsByEntityId = calculatedFieldService.findCalculatedFieldIdsByEntityId(tenantId, entityId).size();
if (numberOfCalculatedFieldsByEntityId >= MAX_CALCULATED_FIELD_NUMBER) {
throw new IllegalArgumentException("Max number of calculated fields for entity is " + MAX_CALCULATED_FIELD_NUMBER);
}
}
private <E extends HasId<I> & HasTenantId, I extends EntityId> E findEntity(TenantId tenantId, EntityId entityId) {
return switch (entityId.getEntityType()) {
case TENANT, CUSTOMER, ASSET, DEVICE -> (E) entityService.fetchEntity(tenantId, entityId).orElse(null);

5
application/src/main/java/org/thingsboard/server/service/entitiy/cf/TbCalculatedFieldService.java

@ -18,6 +18,9 @@ package org.thingsboard.server.service.entitiy.cf;
import org.thingsboard.server.common.data.cf.CalculatedField;
import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.service.security.model.SecurityUser;
public interface TbCalculatedFieldService {
@ -26,6 +29,8 @@ public interface TbCalculatedFieldService {
CalculatedField findById(CalculatedFieldId calculatedFieldId, SecurityUser user);
PageData<CalculatedField> findAllByTenantIdAndEntityId(EntityId entityId, SecurityUser user, PageLink pageLink);
void delete(CalculatedField calculatedField, SecurityUser user);
}

6
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java

@ -43,7 +43,6 @@ import org.thingsboard.server.common.data.edge.EdgeEventActionType;
import org.thingsboard.server.common.data.edge.EdgeEventType;
import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.AssetProfileId;
import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.EdgeId;
@ -103,7 +102,6 @@ import org.thingsboard.server.service.ota.OtaPackageStateService;
import org.thingsboard.server.service.profile.TbAssetProfileCache;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
@ -729,13 +727,13 @@ public class DefaultTbClusterService implements TbClusterService {
@Override
public void onCalculatedFieldUpdated(CalculatedField calculatedField, CalculatedField oldCalculatedField, TbQueueCallback callback) {
var msg = new ComponentLifecycleMsg(calculatedField.getTenantId(), calculatedField.getEntityId(), oldCalculatedField == null ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED);
var msg = new ComponentLifecycleMsg(calculatedField.getTenantId(), calculatedField.getId(), oldCalculatedField == null ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED);
broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), callback);
}
@Override
public void onCalculatedFieldDeleted(CalculatedField calculatedField, TbQueueCallback callback) {
var msg = new ComponentLifecycleMsg(calculatedField.getTenantId(), calculatedField.getEntityId(), ComponentLifecycleEvent.DELETED);
var msg = new ComponentLifecycleMsg(calculatedField.getTenantId(), calculatedField.getId(), ComponentLifecycleEvent.DELETED);
broadcastToCalculatedFields(ToCalculatedFieldNotificationMsg.newBuilder().setComponentLifecycleMsg(toProto(msg)).build(), callback);
}

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

@ -0,0 +1,241 @@
/**
* 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.server.service.cf.ctx.state;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.thingsboard.script.api.tbel.DefaultTbelInvokeService;
import org.thingsboard.script.api.tbel.TbelInvokeService;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.cf.CalculatedField;
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.ArgumentType;
import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration;
import org.thingsboard.server.common.data.cf.configuration.Output;
import org.thingsboard.server.common.data.cf.configuration.OutputType;
import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey;
import org.thingsboard.server.common.data.cf.configuration.SimpleCalculatedFieldConfiguration;
import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.BasicKvEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.service.cf.CalculatedFieldResult;
import java.util.HashMap;
import java.util.Map;
import java.util.TreeMap;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import static org.assertj.core.api.Assertions.assertThat;
@SpringBootTest(classes = DefaultTbelInvokeService.class)
public class ScriptCalculatedFieldStateTest {
private final TenantId TENANT_ID = TenantId.fromUUID(UUID.fromString("5b18e321-3327-4290-b996-d72a65e90382"));
private final DeviceId DEVICE_ID = new DeviceId(UUID.fromString("5512071d-5abc-411d-a907-4cdb6539c2eb"));
private final AssetId ASSET_ID = new AssetId(UUID.fromString("5bc010ae-bcfd-46c8-98b9-8ee8c8955a76"));
private final SingleValueArgumentEntry assetHumidityArgEntry = new SingleValueArgumentEntry(System.currentTimeMillis() - 10, new LongDataEntry("assetHumidity", 43L), 122L);
private final TsRollingArgumentEntry deviceTemperatureArgEntry = createRollingArgEntry();
private final long ts = System.currentTimeMillis();
private ScriptCalculatedFieldState state;
private CalculatedFieldCtx ctx;
@Autowired
private TbelInvokeService tbelInvokeService;
@BeforeEach
void setUp() {
ctx = new CalculatedFieldCtx(getCalculatedField(), tbelInvokeService);
ctx.init();
state = new ScriptCalculatedFieldState(ctx.getArgNames());
}
@Test
void testType() {
assertThat(state.getType()).isEqualTo(CalculatedFieldType.SCRIPT);
}
@Test
void testUpdateState() {
state.arguments = new HashMap<>(Map.of("assetHumidity", assetHumidityArgEntry));
Map<String, ArgumentEntry> newArgs = Map.of("deviceTemperature", deviceTemperatureArgEntry);
boolean stateUpdated = state.updateState(newArgs);
assertThat(stateUpdated).isTrue();
assertThat(state.getArguments()).containsExactlyInAnyOrderEntriesOf(
Map.of(
"assetHumidity", assetHumidityArgEntry,
"deviceTemperature", deviceTemperatureArgEntry
)
);
}
@Test
void testUpdateStateWhenUpdateExistingEntry() {
state.arguments = new HashMap<>(Map.of("deviceTemperature", deviceTemperatureArgEntry, "assetHumidity", assetHumidityArgEntry));
SingleValueArgumentEntry newArgEntry = new SingleValueArgumentEntry(ts, new LongDataEntry("assetHumidity", 41L), 349L);
Map<String, ArgumentEntry> newArgs = Map.of("assetHumidity", newArgEntry);
boolean stateUpdated = state.updateState(newArgs);
assertThat(stateUpdated).isTrue();
assertThat(state.getArguments()).containsExactlyInAnyOrderEntriesOf(
Map.of(
"assetHumidity", newArgEntry,
"deviceTemperature", deviceTemperatureArgEntry
)
);
}
@Test
void testPerformCalculation() throws ExecutionException, InterruptedException {
state.arguments = new HashMap<>(Map.of("deviceTemperature", deviceTemperatureArgEntry, "assetHumidity", assetHumidityArgEntry));
CalculatedFieldResult result = state.performCalculation(ctx).get();
assertThat(result).isNotNull();
Output output = getCalculatedFieldConfig().getOutput();
assertThat(result.getType()).isEqualTo(output.getType());
assertThat(result.getScope()).isEqualTo(output.getScope());
assertThat(result.getResultMap()).isEqualTo(Map.of("averageDeviceTemperature", 13.0, "assetHumidity", 43L));
}
@Test
void testPerformCalculationWhenOldTelemetry() throws ExecutionException, InterruptedException {
TsRollingArgumentEntry argumentEntry = new TsRollingArgumentEntry();
TreeMap<Long, BasicKvEntry> values = new TreeMap<>();
values.put(ts - 40000, new LongDataEntry("deviceTemperature", 4L));// will not be used for calculation
values.put(ts - 45000, new LongDataEntry("deviceTemperature", 2L));// will not be used for calculation
values.put(ts - 20, new LongDataEntry("deviceTemperature", 0L));
argumentEntry.setTsRecords(values);
state.arguments = new HashMap<>(Map.of("deviceTemperature", argumentEntry, "assetHumidity", assetHumidityArgEntry));
CalculatedFieldResult result = state.performCalculation(ctx).get();
assertThat(result).isNotNull();
Output output = getCalculatedFieldConfig().getOutput();
assertThat(result.getType()).isEqualTo(output.getType());
assertThat(result.getScope()).isEqualTo(output.getScope());
assertThat(result.getResultMap()).isEqualTo(Map.of("averageDeviceTemperature", 0.0, "assetHumidity", 43L));
}
@Test
void testPerformCalculationWhenArgumentsMoreThanLimit() throws ExecutionException, InterruptedException {
TsRollingArgumentEntry argumentEntry = new TsRollingArgumentEntry();
TreeMap<Long, BasicKvEntry> values = new TreeMap<>();
values.put(ts - 20, new LongDataEntry("deviceTemperature", 1000L));// will not be used
values.put(ts - 18, new LongDataEntry("deviceTemperature", 0L));
values.put(ts - 16, new LongDataEntry("deviceTemperature", 0L));
values.put(ts - 14, new LongDataEntry("deviceTemperature", 0L));
values.put(ts - 12, new LongDataEntry("deviceTemperature", 0L));
values.put(ts - 10, new LongDataEntry("deviceTemperature", 0L));
argumentEntry.setTsRecords(values);
state.arguments = new HashMap<>(Map.of("deviceTemperature", argumentEntry, "assetHumidity", assetHumidityArgEntry));
CalculatedFieldResult result = state.performCalculation(ctx).get();
assertThat(result).isNotNull();
Output output = getCalculatedFieldConfig().getOutput();
assertThat(result.getType()).isEqualTo(output.getType());
assertThat(result.getScope()).isEqualTo(output.getScope());
assertThat(result.getResultMap()).isEqualTo(Map.of("averageDeviceTemperature", 0.0, "assetHumidity", 43L));
}
@Test
void testIsReadyWhenNotAllArgPresent() {
assertThat(state.isReady()).isFalse();
}
@Test
void testIsReadyWhenAllArgPresent() {
state.arguments = new HashMap<>(Map.of("deviceTemperature", deviceTemperatureArgEntry, "assetHumidity", assetHumidityArgEntry));
assertThat(state.isReady()).isTrue();
}
@Test
void testIsReadyWhenEmptyEntryPresents() {
state.arguments = new HashMap<>(Map.of("deviceTemperature", TsRollingArgumentEntry.EMPTY, "assetHumidity", assetHumidityArgEntry));
assertThat(state.isReady()).isFalse();
}
private TsRollingArgumentEntry createRollingArgEntry() {
TsRollingArgumentEntry argumentEntry = new TsRollingArgumentEntry();
long ts = System.currentTimeMillis();
TreeMap<Long, BasicKvEntry> values = new TreeMap<>();
values.put(ts - 40, new LongDataEntry("deviceTemperature", 10L));
values.put(ts - 30, new LongDataEntry("deviceTemperature", 12L));
values.put(ts - 20, new LongDataEntry("deviceTemperature", 17L));
argumentEntry.setTsRecords(values);
return argumentEntry;
}
private CalculatedField getCalculatedField() {
CalculatedField calculatedField = new CalculatedField();
calculatedField.setTenantId(TENANT_ID);
calculatedField.setEntityId(ASSET_ID);
calculatedField.setType(CalculatedFieldType.SCRIPT);
calculatedField.setName("Test Calculated Field");
calculatedField.setConfigurationVersion(1);
calculatedField.setConfiguration(getCalculatedFieldConfig());
calculatedField.setVersion(1L);
return calculatedField;
}
private CalculatedFieldConfiguration getCalculatedFieldConfig() {
SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration();
Argument argument1 = new Argument();
argument1.setRefEntityId(DEVICE_ID);
ReferencedEntityKey refEntityKey1 = new ReferencedEntityKey("temperature", ArgumentType.TS_ROLLING, null);
argument1.setRefEntityKey(refEntityKey1);
argument1.setLimit(5);
argument1.setTimeWindow(30000);
Argument argument2 = new Argument();
ReferencedEntityKey refEntityKey2 = new ReferencedEntityKey("humidity", ArgumentType.TS_LATEST, null);
argument1.setRefEntityKey(refEntityKey2);
config.setArguments(Map.of("deviceTemperature", argument1, "assetHumidity", argument2));
config.setExpression("var result = 0; foreach(element : deviceTemperature.entrySet()) { result += element.getValue(); } var map = {}; map.put(\"averageDeviceTemperature\", result / deviceTemperature.size()); map.put(\"assetHumidity\", assetHumidity); return map;");
Output output = new Output();
output.setType(OutputType.ATTRIBUTES);
output.setScope(AttributeScope.SERVER_SCOPE);
config.setOutput(output);
return config;
}
}

216
application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldStateTest.java

@ -0,0 +1,216 @@
/**
* 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.server.service.cf.ctx.state;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.cf.CalculatedField;
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.ArgumentType;
import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration;
import org.thingsboard.server.common.data.cf.configuration.Output;
import org.thingsboard.server.common.data.cf.configuration.OutputType;
import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey;
import org.thingsboard.server.common.data.cf.configuration.SimpleCalculatedFieldConfiguration;
import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.common.data.kv.StringDataEntry;
import org.thingsboard.server.service.cf.CalculatedFieldResult;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
public class SimpleCalculatedFieldStateTest {
private final TenantId TENANT_ID = TenantId.fromUUID(UUID.fromString("5b18e321-3327-4290-b996-d72a65e90382"));
private final DeviceId DEVICE_ID = new DeviceId(UUID.fromString("5512071d-5abc-411d-a907-4cdb6539c2eb"));
private final AssetId ASSET_ID = new AssetId(UUID.fromString("5bc010ae-bcfd-46c8-98b9-8ee8c8955a76"));
private final SingleValueArgumentEntry key1ArgEntry = new SingleValueArgumentEntry(System.currentTimeMillis() - 10, new LongDataEntry("key1", 11L), 145L);
private final SingleValueArgumentEntry key2ArgEntry = new SingleValueArgumentEntry(System.currentTimeMillis() - 6, new LongDataEntry("key2", 15L), 165L);
private final SingleValueArgumentEntry key3ArgEntry = new SingleValueArgumentEntry(System.currentTimeMillis() - 3, new LongDataEntry("key3", 23L), 184L);
private SimpleCalculatedFieldState state;
private CalculatedFieldCtx ctx;
@BeforeEach
void setUp() {
ctx = new CalculatedFieldCtx(getCalculatedField(), null);
ctx.init();
state = new SimpleCalculatedFieldState(ctx.getArgNames());
}
@Test
void testType() {
assertThat(state.getType()).isEqualTo(CalculatedFieldType.SIMPLE);
}
@Test
void testUpdateState() {
state.arguments = new HashMap<>(Map.of(
"key1", key1ArgEntry,
"key2", key2ArgEntry
));
Map<String, ArgumentEntry> newArgs = Map.of("key3", key3ArgEntry);
boolean stateUpdated = state.updateState(newArgs);
assertThat(stateUpdated).isTrue();
assertThat(state.getArguments()).containsExactlyInAnyOrderEntriesOf(
Map.of(
"key1", key1ArgEntry,
"key2", key2ArgEntry,
"key3", key3ArgEntry
)
);
}
@Test
void testUpdateStateWhenUpdateExistingEntry() {
state.arguments = new HashMap<>(Map.of("key1", key1ArgEntry));
SingleValueArgumentEntry newArgEntry = new SingleValueArgumentEntry(System.currentTimeMillis(), new LongDataEntry("key1", 18L), 190L);
Map<String, ArgumentEntry> newArgs = Map.of("key1", newArgEntry);
boolean stateUpdated = state.updateState(newArgs);
assertThat(stateUpdated).isTrue();
assertThat(state.getArguments()).containsExactlyInAnyOrderEntriesOf(Map.of("key1", newArgEntry));
}
@Test
void testUpdateStateWhenRollingEntryPassed() {
state.arguments = new HashMap<>(Map.of(
"key1", key1ArgEntry,
"key2", key2ArgEntry
));
Map<String, ArgumentEntry> newArgs = Map.of("key3", TsRollingArgumentEntry.EMPTY);
assertThatThrownBy(() -> state.updateState(newArgs))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Rolling argument entry is not supported for simple calculated fields.");
}
@Test
void testPerformCalculation() throws ExecutionException, InterruptedException {
state.arguments = new HashMap<>(Map.of(
"key1", key1ArgEntry,
"key2", key2ArgEntry,
"key3", key3ArgEntry
));
CalculatedFieldResult result = state.performCalculation(ctx).get();
assertThat(result).isNotNull();
Output output = getCalculatedFieldConfig().getOutput();
assertThat(result.getType()).isEqualTo(output.getType());
assertThat(result.getScope()).isEqualTo(output.getScope());
assertThat(result.getResultMap()).isEqualTo(Map.of("output", 49.0));
}
@Test
void testPerformCalculationWhenPassedNotNumber() {
state.arguments = new HashMap<>(Map.of(
"key1", key1ArgEntry,
"key2", new SingleValueArgumentEntry(System.currentTimeMillis() - 9, new StringDataEntry("key2", "string"), 124L),
"key3", key3ArgEntry
));
assertThatThrownBy(() -> state.performCalculation(ctx))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Argument 'key2' is not a number.");
}
@Test
void testIsReadyWhenNotAllArgPresent() {
assertThat(state.isReady()).isFalse();
}
@Test
void testIsReadyWhenAllArgPresent() {
state.arguments = new HashMap<>(Map.of(
"key1", key1ArgEntry,
"key2", key2ArgEntry,
"key3", key3ArgEntry
));
assertThat(state.isReady()).isTrue();
}
@Test
void testIsReadyWhenEmptyEntryPresents() {
state.arguments = new HashMap<>(Map.of(
"key1", key1ArgEntry,
"key2", key2ArgEntry
));
state.getArguments().put("key3", SingleValueArgumentEntry.EMPTY);
assertThat(state.isReady()).isFalse();
}
private CalculatedField getCalculatedField() {
CalculatedField calculatedField = new CalculatedField();
calculatedField.setTenantId(TENANT_ID);
calculatedField.setEntityId(DEVICE_ID);
calculatedField.setType(CalculatedFieldType.SIMPLE);
calculatedField.setName("Test Calculated Field");
calculatedField.setConfigurationVersion(1);
calculatedField.setConfiguration(getCalculatedFieldConfig());
calculatedField.setVersion(1L);
return calculatedField;
}
private CalculatedFieldConfiguration getCalculatedFieldConfig() {
SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration();
Argument argument1 = new Argument();
argument1.setRefEntityId(ASSET_ID);
ReferencedEntityKey refEntityKey1 = new ReferencedEntityKey("temp1", ArgumentType.TS_LATEST, null);
argument1.setRefEntityKey(refEntityKey1);
Argument argument2 = new Argument();
argument2.setRefEntityId(ASSET_ID);
ReferencedEntityKey refEntityKey2 = new ReferencedEntityKey("temp2", ArgumentType.ATTRIBUTE, null);
argument2.setRefEntityKey(refEntityKey2);
Argument argument3 = new Argument();
argument3.setRefEntityId(ASSET_ID);
ReferencedEntityKey refEntityKey3 = new ReferencedEntityKey("temp3", ArgumentType.TS_LATEST, null);
argument3.setRefEntityKey(refEntityKey3);
config.setArguments(Map.of("key1", argument1, "key2", argument2, "key3", argument3));
config.setExpression("key1 + key2 + key3");
Output output = new Output();
output.setName("output");
output.setType(OutputType.ATTRIBUTES);
output.setScope(AttributeScope.SERVER_SCOPE);
config.setOutput(output);
return config;
}
}

72
application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java

@ -0,0 +1,72 @@
/**
* 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.server.service.cf.ctx.state;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.thingsboard.server.common.data.kv.LongDataEntry;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
public class SingleValueArgumentEntryTest {
private SingleValueArgumentEntry entry;
private final long ts = System.currentTimeMillis();
@BeforeEach
void setUp() {
entry = new SingleValueArgumentEntry(ts, new LongDataEntry("key", 11L), 363L);
}
@Test
void testArgumentEntryType() {
assertThat(entry.getType()).isEqualTo(ArgumentEntryType.SINGLE_VALUE);
}
@Test
void testUpdateEntryWhenRollingEntryPassed() {
assertThatThrownBy(() -> entry.updateEntry(TsRollingArgumentEntry.EMPTY))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Unsupported argument entry type for single value argument entry: " + ArgumentEntryType.TS_ROLLING);
}
@Test
void testUpdateEntryWithThaSameTs() {
assertThat(entry.updateEntry(new SingleValueArgumentEntry(ts, new LongDataEntry("key", 13L), 363L))).isFalse();
}
@Test
void testUpdateEntryWhenNewVersionIsNull() {
assertThat(entry.updateEntry(new SingleValueArgumentEntry(ts + 16, new LongDataEntry("key", 13L), null))).isTrue();
assertThat(entry.getValue()).isEqualTo(13L);
assertThat(entry.getVersion()).isNull();
}
@Test
void testUpdateEntryWhenNewVersionIsGreaterThanCurrent() {
assertThat(entry.updateEntry(new SingleValueArgumentEntry(ts + 18, new LongDataEntry("key", 18L), 369L))).isTrue();
assertThat(entry.getValue()).isEqualTo(18L);
assertThat(entry.getVersion()).isEqualTo(369L);
}
@Test
void testUpdateEntryWhenNewVersionIsLessThanCurrent() {
assertThat(entry.updateEntry(new SingleValueArgumentEntry(ts + 18, new LongDataEntry("key", 18L), 234L))).isFalse();
}
}

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

@ -0,0 +1,96 @@
/**
* 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.server.service.cf.ctx.state;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.thingsboard.server.common.data.kv.BasicKvEntry;
import org.thingsboard.server.common.data.kv.DoubleDataEntry;
import org.thingsboard.server.common.data.kv.StringDataEntry;
import java.util.Map;
import java.util.TreeMap;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
public class TsRollingArgumentEntryTest {
private TsRollingArgumentEntry entry;
private final long ts = System.currentTimeMillis();
@BeforeEach
void setUp() {
TreeMap<Long, BasicKvEntry> values = new TreeMap<>();
values.put(ts - 40, new DoubleDataEntry("key", 10.0));
values.put(ts - 30, new DoubleDataEntry("key", 12.0));
values.put(ts - 20, new DoubleDataEntry("key", 17.0));
entry = new TsRollingArgumentEntry(values);
}
@Test
void testArgumentEntryType() {
assertThat(entry.getType()).isEqualTo(ArgumentEntryType.TS_ROLLING);
}
@Test
void testUpdateEntryWhenSingleValueEntryPassed() {
SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 10, new DoubleDataEntry("key", 23.0), 123L);
assertThat(entry.updateEntry(newEntry)).isTrue();
assertThat(entry.getTsRecords()).hasSize(4);
assertThat(entry.getTsRecords().get(ts - 10).getValue()).isEqualTo(23.0);
}
@Test
void testUpdateEntryWhenSingleValueEntryWithTheSameTsPassed() {
SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 20, new DoubleDataEntry("key", 23.0), 123L);
assertThat(entry.updateEntry(newEntry)).isFalse();
}
@Test
void testUpdateEntryWhenRollingEntryPassed() {
TsRollingArgumentEntry newEntry = new TsRollingArgumentEntry();
TreeMap<Long, BasicKvEntry> values = new TreeMap<>();
values.put(ts - 20, new DoubleDataEntry("key", 16.0));
values.put(ts - 10, new DoubleDataEntry("key", 7.0));
values.put(ts - 5, new DoubleDataEntry("key", 1.0));
newEntry.setTsRecords(values);
assertThat(entry.updateEntry(newEntry)).isTrue();
assertThat(entry.getTsRecords()).hasSize(5);
assertThat(entry.getTsRecords()).isEqualTo(Map.of(
ts - 40, new DoubleDataEntry("key", 10.0),
ts - 30, new DoubleDataEntry("key", 12.0),
ts - 20, new DoubleDataEntry("key", 17.0),
ts - 10, new DoubleDataEntry("key", 7.0),
ts - 5, new DoubleDataEntry("key", 1.0)
));
}
@Test
void testUpdateEntryWhenValueIsNotNumber() {
SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 10, new StringDataEntry("key", "string"), 123L);
assertThatThrownBy(() -> entry.updateEntry(newEntry))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Argument type " + ArgumentEntryType.TS_ROLLING + " only supports numeric values.");
}
}

2
common/dao-api/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldService.java

@ -44,6 +44,8 @@ public interface CalculatedFieldService extends EntityDaoService {
PageData<CalculatedField> findAllCalculatedFields(PageLink pageLink);
PageData<CalculatedField> findAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink);
void deleteCalculatedField(TenantId tenantId, CalculatedFieldId calculatedFieldId);
int deleteAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId);

7
common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java

@ -135,6 +135,12 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura
private double warnThreshold;
private long maxCalculatedFieldsPerTenant;
private long maxCalculatedFieldsPerEntity;
private long maxArgumentsPerCF;
private long maxDataPointsPerRollingArg;
private long maxStateSizeInKBytes;
@Override
public long getProfileThreshold(ApiUsageRecordKey key) {
return switch (key) {
@ -175,6 +181,7 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura
case DASHBOARD -> maxDashboards;
case RULE_CHAIN -> maxRuleChains;
case EDGE -> maxEdges;
case CALCULATED_FIELD -> maxCalculatedFieldsPerTenant;
default -> 0;
};
}

39
common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java

@ -58,6 +58,7 @@ import org.thingsboard.server.common.data.id.TenantProfileId;
import org.thingsboard.server.common.data.kv.AttributeKey;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BasicKvEntry;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.common.data.kv.DoubleDataEntry;
@ -90,7 +91,7 @@ import org.thingsboard.server.common.msg.rule.engine.DeviceDeleteMsg;
import org.thingsboard.server.common.msg.rule.engine.DeviceEdgeUpdateMsg;
import org.thingsboard.server.common.msg.rule.engine.DeviceNameOrTypeUpdateMsg;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto;
import java.util.ArrayList;
import java.util.Arrays;
@ -655,6 +656,42 @@ public class ProtoUtils {
return new BaseAttributeKvEntry(entry, proto.getLastUpdateTs(), proto.hasVersion() ? proto.getVersion() : null);
}
public static BasicKvEntry basicKvEntryFromProto(TransportProtos.AttributeValueProto proto) {
boolean hasValue = proto.getHasV();
String key = proto.getKey();
return switch (proto.getType()) {
case BOOLEAN_V -> new BooleanDataEntry(key, hasValue ? proto.getBoolV() : null);
case LONG_V -> new LongDataEntry(key, hasValue ? proto.getLongV() : null);
case DOUBLE_V -> new DoubleDataEntry(key, hasValue ? proto.getDoubleV() : null);
case STRING_V -> new StringDataEntry(key, hasValue ? proto.getStringV() : null);
case JSON_V -> new JsonDataEntry(key, hasValue ? proto.getJsonV() : null);
default -> null;
};
}
public static BasicKvEntry fromProto(KeyValueProto proto) {
String key = proto.getKey();
return switch (proto.getType()) {
case BOOLEAN_V -> new BooleanDataEntry(key, proto.getBoolV());
case LONG_V -> new LongDataEntry(key, proto.getLongV());
case DOUBLE_V -> new DoubleDataEntry(key, proto.getDoubleV());
case STRING_V -> new StringDataEntry(key, proto.getStringV());
case JSON_V -> new JsonDataEntry(key, proto.getJsonV());
default -> null;
};
}
public static BasicKvEntry basicKvEntryFromKvEntry(KvEntry kvEntry) {
String key = kvEntry.getKey();
return switch (kvEntry.getDataType()) {
case BOOLEAN -> new BooleanDataEntry(key, kvEntry.getBooleanValue().orElse(null));
case LONG -> new LongDataEntry(key, kvEntry.getLongValue().orElse(null));
case DOUBLE -> new DoubleDataEntry(key, kvEntry.getDoubleValue().orElse(null));
case STRING -> new StringDataEntry(key, kvEntry.getStrValue().orElse(null));
case JSON -> new JsonDataEntry(key, kvEntry.getJsonValue().orElse(null));
};
}
public static TsKvEntry fromProto(TransportProtos.TsKvProto proto) {
TransportProtos.KeyValueProto kvProto = proto.getKv();
String key = kvProto.getKey();

28
common/proto/src/main/proto/queue.proto

@ -783,6 +783,9 @@ message CalculatedFieldTelemetryMsgProto {
repeated TsKvProto tsData = 9;
AttributeScopeProto scope = 10;
repeated AttributeValueProto attrData = 11;
int64 tbMsgIdMSB = 12;
int64 tbMsgIdLSB = 13;
string tbMsgType = 14;
}
message CalculatedFieldLinkedTelemetryMsgProto {
@ -791,11 +794,13 @@ message CalculatedFieldLinkedTelemetryMsgProto {
}
message CalculatedFieldEntityCtxIdProto {
int64 calculatedFieldIdMSB = 1;
int64 calculatedFieldIdLSB = 2;
string entityType = 3;
int64 entityIdMSB = 4;
int64 entityIdLSB = 5;
int64 tenantIdMSB = 1;
int64 tenantIdLSB = 2;
int64 calculatedFieldIdMSB = 3;
int64 calculatedFieldIdLSB = 4;
string entityType = 5;
int64 entityIdMSB = 6;
int64 entityIdLSB = 7;
}
message CalculatedFieldIdProto {
@ -806,13 +811,8 @@ message CalculatedFieldIdProto {
message SingleValueProto {
int64 ts = 1;
int64 version = 2;
KeyValueType type = 3;
bool has_v = 4;
bool bool_v = 5;
int64 long_v = 6;
double double_v = 7;
string string_v = 8;
string json_v = 9;
KeyValueProto value = 5;
}
message SingleValueArgumentProto {
@ -828,8 +828,10 @@ message RollingArgumentProto {
message CalculatedFieldStateProto {
CalculatedFieldEntityCtxIdProto id = 1;
// int32 version = 2;
repeated SingleValueArgumentProto singleValueArguments = 3;
repeated RollingArgumentProto rollingValueArguments = 4;
string type = 3;
repeated string requiredArguments = 4;
repeated SingleValueArgumentProto singleValueArguments = 5;
repeated RollingArgumentProto rollingValueArguments = 6;
}
//Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.

8
dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java

@ -118,6 +118,14 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements
return calculatedFieldDao.findAll(pageLink);
}
@Override
public PageData<CalculatedField> findAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink) {
log.trace("Executing findAllByEntityId, entityId [{}], pageLink [{}]", entityId, pageLink);
validateId(entityId.getId(), id -> INCORRECT_ENTITY_ID + id);
validatePageLink(pageLink);
return calculatedFieldDao.findAllByEntityId(tenantId, entityId, pageLink);
}
@Override
public void deleteCalculatedField(TenantId tenantId, CalculatedFieldId calculatedFieldId) {
validateId(tenantId, id -> INCORRECT_TENANT_ID + id);

4
dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldDao.java

@ -37,8 +37,12 @@ public interface CalculatedFieldDao extends Dao<CalculatedField> {
PageData<CalculatedField> findAll(PageLink pageLink);
PageData<CalculatedField> findAllByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink);
List<CalculatedField> removeAllByEntityId(TenantId tenantId, EntityId entityId);
boolean existsByEntityId(TenantId tenantId, EntityId entityId);
long countCFByEntityId(TenantId tenantId, EntityId entityId);
}

31
dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java

@ -17,11 +17,15 @@ package org.thingsboard.server.dao.service.validator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.cf.CalculatedField;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
import org.thingsboard.server.dao.cf.CalculatedFieldDao;
import org.thingsboard.server.dao.exception.DataValidationException;
import org.thingsboard.server.dao.service.DataValidator;
import org.thingsboard.server.dao.usagerecord.ApiLimitService;
@Component
public class CalculatedFieldDataValidator extends DataValidator<CalculatedField> {
@ -29,13 +33,40 @@ public class CalculatedFieldDataValidator extends DataValidator<CalculatedField>
@Autowired
private CalculatedFieldDao calculatedFieldDao;
@Autowired
private ApiLimitService apiLimitService;
@Override
protected void validateCreate(TenantId tenantId, CalculatedField calculatedField) {
validateNumberOfEntitiesPerTenant(tenantId, EntityType.CALCULATED_FIELD);
validateNumberOfCFsPerEntity(tenantId, calculatedField.getEntityId());
validateNumberOfArgumentsPerCF(tenantId, calculatedField);
}
@Override
protected CalculatedField validateUpdate(TenantId tenantId, CalculatedField calculatedField) {
CalculatedField old = calculatedFieldDao.findById(calculatedField.getTenantId(), calculatedField.getId().getId());
if (old == null) {
throw new DataValidationException("Can't update non existing calculated field!");
}
validateNumberOfArgumentsPerCF(tenantId, calculatedField);
return old;
}
private void validateNumberOfCFsPerEntity(TenantId tenantId, EntityId entityId) {
long maxCFsPerEntity = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxCalculatedFieldsPerEntity);
long countCFByEntityId = calculatedFieldDao.countCFByEntityId(tenantId, entityId);
if (countCFByEntityId == maxCFsPerEntity) {
throw new DataValidationException("Calculated fields per entity limit reached!");
}
}
private void validateNumberOfArgumentsPerCF(TenantId tenantId, CalculatedField calculatedField) {
long maxArgumentsPerCF = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxArgumentsPerCF);
if (calculatedField.getConfiguration().getArguments().size() > maxArgumentsPerCF) {
throw new DataValidationException("Calculated field arguments limit reached!");
}
}
}

6
dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldRepository.java

@ -15,6 +15,8 @@
*/
package org.thingsboard.server.dao.sql.cf;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.repository.JpaRepository;
import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.dao.model.sql.CalculatedFieldEntity;
@ -30,8 +32,12 @@ public interface CalculatedFieldRepository extends JpaRepository<CalculatedField
List<CalculatedFieldEntity> findAllByTenantIdAndEntityId(UUID tenantId, UUID entityId);
Page<CalculatedFieldEntity> findAllByTenantIdAndEntityId(UUID tenantId, UUID entityId, Pageable pageable);
List<CalculatedFieldEntity> findAllByTenantId(UUID tenantId);
List<CalculatedFieldEntity> removeAllByTenantIdAndEntityId(UUID tenantId, UUID entityId);
long countByTenantIdAndEntityId(UUID tenantId, UUID entityId);
}

5
dao/src/main/java/org/thingsboard/server/dao/sql/cf/DefaultNativeCalculatedFieldRepository.java

@ -29,6 +29,7 @@ import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
import org.thingsboard.server.common.data.cf.CalculatedFieldLinkConfiguration;
import org.thingsboard.server.common.data.cf.CalculatedFieldType;
import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration;
import org.thingsboard.server.common.data.debug.DebugSettings;
import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.CalculatedFieldLinkId;
import org.thingsboard.server.common.data.id.EntityIdFactory;
@ -78,18 +79,20 @@ public class DefaultNativeCalculatedFieldRepository implements NativeCalculatedF
int configurationVersion = (int) row.get("configuration_version");
JsonNode configuration = JacksonUtil.toJsonNode((String) row.get("configuration"));
long version = row.get("version") != null ? (long) row.get("version") : 0;
String debugSettings = (String) row.get("debug_settings");
Object externalIdObj = row.get("external_id");
CalculatedField calculatedField = new CalculatedField();
calculatedField.setId(new CalculatedFieldId(id));
calculatedField.setCreatedTime(createdTime);
calculatedField.setTenantId(new TenantId(tenantId));
calculatedField.setTenantId(TenantId.fromUUID(tenantId));
calculatedField.setEntityId(EntityIdFactory.getByTypeAndUuid(entityType, entityId));
calculatedField.setType(type);
calculatedField.setName(name);
calculatedField.setConfigurationVersion(configurationVersion);
calculatedField.setConfiguration(JacksonUtil.treeToValue(configuration, CalculatedFieldConfiguration.class));
calculatedField.setVersion(version);
calculatedField.setDebugSettings(JacksonUtil.fromString(debugSettings, DebugSettings.class));
calculatedField.setExternalId(externalIdObj != null ? new CalculatedFieldId(UUID.fromString((String) externalIdObj)) : null);
return calculatedField;

11
dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldDao.java

@ -71,6 +71,12 @@ public class JpaCalculatedFieldDao extends JpaAbstractDao<CalculatedFieldEntity,
return nativeCalculatedFieldRepository.findCalculatedFields(DaoUtil.toPageable(pageLink));
}
@Override
public PageData<CalculatedField> findAllByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink) {
log.debug("Try to find calculated fields by entityId[{}] and pageLink [{}]", entityId, pageLink);
return DaoUtil.toPageData(calculatedFieldRepository.findAllByTenantIdAndEntityId(tenantId.getId(), entityId.getId(), DaoUtil.toPageable(pageLink)));
}
@Override
@Transactional
public List<CalculatedField> removeAllByEntityId(TenantId tenantId, EntityId entityId) {
@ -82,6 +88,11 @@ public class JpaCalculatedFieldDao extends JpaAbstractDao<CalculatedFieldEntity,
return calculatedFieldRepository.existsByTenantIdAndEntityId(tenantId.getId(), entityId.getId());
}
@Override
public long countCFByEntityId(TenantId tenantId, EntityId entityId) {
return calculatedFieldRepository.countByTenantIdAndEntityId(tenantId.getId(), entityId.getId());
}
@Override
protected Class<CalculatedFieldEntity> getEntityClass() {
return CalculatedFieldEntity.class;

18
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java

@ -28,8 +28,10 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.msg.TbMsgType;
import java.util.List;
import java.util.UUID;
@Getter
@ToString
@ -42,6 +44,8 @@ public class AttributesSaveRequest {
private final List<AttributeKvEntry> entries;
private final boolean notifyDevice;
private final List<CalculatedFieldId> previousCalculatedFieldIds;
private final UUID tbMsgId;
private final TbMsgType tbMsgType;
private final FutureCallback<Void> callback;
public static Builder builder() {
@ -56,6 +60,8 @@ public class AttributesSaveRequest {
private List<AttributeKvEntry> entries;
private boolean notifyDevice = true;
private List<CalculatedFieldId> previousCalculatedFieldIds;
private UUID tbMsgId;
private TbMsgType tbMsgType;
private FutureCallback<Void> callback;
Builder() {}
@ -108,6 +114,16 @@ public class AttributesSaveRequest {
return this;
}
public Builder tbMsgId(UUID tbMsgId) {
this.tbMsgId = tbMsgId;
return this;
}
public Builder tbMsgType(TbMsgType tbMsgType) {
this.tbMsgType = tbMsgType;
return this;
}
public Builder callback(FutureCallback<Void> callback) {
this.callback = callback;
return this;
@ -128,7 +144,7 @@ public class AttributesSaveRequest {
}
public AttributesSaveRequest build() {
return new AttributesSaveRequest(tenantId, entityId, scope, entries, notifyDevice, previousCalculatedFieldIds, callback);
return new AttributesSaveRequest(tenantId, entityId, scope, entries, notifyDevice, previousCalculatedFieldIds, tbMsgId, tbMsgType, callback);
}
}

18
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java

@ -27,8 +27,10 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.msg.TbMsgType;
import java.util.List;
import java.util.UUID;
@Getter
@AllArgsConstructor(access = AccessLevel.PRIVATE)
@ -42,6 +44,8 @@ public class TimeseriesSaveRequest {
private final boolean saveLatest;
private final boolean onlyLatest;
private final List<CalculatedFieldId> previousCalculatedFieldIds;
private final UUID tbMsgId;
private final TbMsgType tbMsgType;
private final FutureCallback<Void> callback;
public static Builder builder() {
@ -59,6 +63,8 @@ public class TimeseriesSaveRequest {
private boolean saveLatest = true;
private boolean onlyLatest;
private List<CalculatedFieldId> previousCalculatedFieldIds;
private UUID tbMsgId;
private TbMsgType tbMsgType;
Builder() {}
@ -111,6 +117,16 @@ public class TimeseriesSaveRequest {
return this;
}
public Builder tbMsgId(UUID tbMsgId) {
this.tbMsgId = tbMsgId;
return this;
}
public Builder tbMsgType(TbMsgType tbMsgType) {
this.tbMsgType = tbMsgType;
return this;
}
public Builder callback(FutureCallback<Void> callback) {
this.callback = callback;
return this;
@ -131,7 +147,7 @@ public class TimeseriesSaveRequest {
}
public TimeseriesSaveRequest build() {
return new TimeseriesSaveRequest(tenantId, customerId, entityId, entries, ttl, saveLatest, onlyLatest, previousCalculatedFieldIds, callback);
return new TimeseriesSaveRequest(tenantId, customerId, entityId, entries, ttl, saveLatest, onlyLatest, previousCalculatedFieldIds, tbMsgId, tbMsgType, callback);
}
}

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java

@ -126,6 +126,8 @@ public class TbMsgAttributesNode implements TbNode {
.entries(attributes)
.notifyDevice(config.isNotifyDevice() || checkNotifyDeviceMdValue(msg.getMetaData().getValue(NOTIFY_DEVICE_METADATA_KEY)))
.previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds())
.tbMsgId(msg.getId())
.tbMsgType(msg.getInternalType())
.callback(callback)
.build());
}

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java

@ -113,6 +113,8 @@ public class TbMsgTimeseriesNode implements TbNode {
.ttl(ttl)
.saveLatest(!config.isSkipLatestPersistence())
.previousCalculatedFieldIds(msg.getPreviousCalculatedFieldIds())
.tbMsgId(msg.getId())
.tbMsgType(msg.getInternalType())
.callback(new TelemetryNodeCallback(ctx, msg))
.build());
}

Loading…
Cancel
Save