diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 09488bfe4e..daa6744fd9 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/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 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 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 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 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); } } diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java index b8f357fd87..9392e3c02a 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java @@ -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 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 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 cfIdList, MultipleTbCallback callback, - Map newArgValues) { + Map 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 cfIdList, CalculatedFieldState state, TbCallback callback) { + private void processStateIfReady(CalculatedFieldCtx ctx, List 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; + } + } diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java index e82f407e3a..b2ff883236 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java +++ b/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); } } diff --git a/application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java b/application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java index efe25e3883..96a97aeeff 100644 --- a/application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java +++ b/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 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 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)", diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java index 9aca0d5b3f..d8e42c916d 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java +++ b/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 states = new ConcurrentHashMap<>(); - private static final int MAX_LAST_RECORDS_VALUE = 1024; - private static final Set 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 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> 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 entries = request.getEntries(); List 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 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 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 calculatedFieldIds) { + private CalculatedFieldTelemetryMsgProto.Builder buildTelemetryMsgProto(TenantId tenantId, EntityId entityId, List 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(); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java b/application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java index 3aed65eced..fe800f61ad 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/RocksDBService.java +++ b/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 getAll() { - Map map = new HashMap<>(); + public Map getAll() { + Map 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; } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java index d37e7ebc3c..c670e97580 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/CalculatedFieldStateService.java @@ -24,8 +24,6 @@ public interface CalculatedFieldStateService { Map restoreStates(); - CalculatedFieldState restoreState(CalculatedFieldEntityCtxId ctxId); - void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback); void removeState(CalculatedFieldEntityCtxId stateId, TbCallback callback); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java index b261840bfd..be471a8ad9 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java +++ b/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); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java index ca243e25b5..f5f681b505 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java @@ -15,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 requiredArguments; protected Map arguments; - public BaseCalculatedFieldState() { - arguments = new HashMap<>(); + public BaseCalculatedFieldState(List requiredArguments) { + this.requiredArguments = requiredArguments; + this.arguments = new HashMap<>(); } @Override public Map getArguments() { - return this.arguments; + return arguments; + } + + @Override + public List 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); + } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java index 6ca4369230..4600628838 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java @@ -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 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 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 updatedTelemetry = proto.getTsDataList().stream() + .map(ProtoUtils::fromProto) + .toList(); + return linkMatches(entityId, updatedTelemetry); + } else { + AttributeScope scope = AttributeScope.valueOf(proto.getScope().name()); + List 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; } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java index f261e586a4..173d299da6 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java @@ -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 getArguments(); + List getRequiredArguments(); + boolean updateState(Map argumentValues); ListenableFuture performCalculation(CalculatedFieldCtx ctx); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java index f5c882ea52..6487ce1a43 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java +++ b/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 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 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); + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java index 0421055fef..4a24d13c93 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java +++ b/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 requiredArguments) { + super(requiredArguments); + } + @Override public CalculatedFieldType getType() { return CalculatedFieldType.SCRIPT; } + @Override + protected void validateNewEntry(ArgumentEntry newEntry) { + } + @Override public ListenableFuture performCalculation(CalculatedFieldCtx ctx) { arguments.forEach((key, argumentEntry) -> { - if (argumentEntry instanceof TsRollingArgumentEntry) { + if (argumentEntry instanceof TsRollingArgumentEntry tsRollingEntry) { Argument argument = ctx.getArguments().get(key); - TreeMap tsRecords = ((TsRollingArgumentEntry) argumentEntry).getTsRecords(); + TreeMap tsRecords = tsRollingEntry.getTsRecords(); if (tsRecords.size() > argument.getLimit()) { tsRecords.pollFirstEntry(); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java index e16d310b3e..01cf9cff70 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java @@ -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 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 performCalculation(CalculatedFieldCtx ctx) { - String expression = ctx.getExpression(); - ThreadLocal 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 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 variables = new HashMap<>(); - this.arguments.forEach((k, v) -> variables.put(k, Double.parseDouble(v.getValue().toString()))); - expr.setVariables(variables); double expressionResult = expr.evaluate(); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java index d5f7b1aee6..0832e53e5e 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java +++ b/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; + } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java index 1118e3af13..ddae0c3513 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java @@ -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 tsRecords = new TreeMap<>(); + private TreeMap tsRecords = new TreeMap<>(); public TsRollingArgumentEntry(List 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 tsRecordEntry : tsRollingEntry.getTsRecords().entrySet()) { + updated |= addTsRecordIfAbsent(tsRecordEntry.getKey(), tsRecordEntry.getValue()); } + return updated; } - public void addAllTsRecords(Map newRecords) { - for (Map.Entry 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."); } } diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/cf/DefaultTbCalculatedFieldService.java b/application/src/main/java/org/thingsboard/server/service/entitiy/cf/DefaultTbCalculatedFieldService.java index 2e6e975636..1ccfe2375c 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/cf/DefaultTbCalculatedFieldService.java +++ b/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 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 & 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); diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/cf/TbCalculatedFieldService.java b/application/src/main/java/org/thingsboard/server/service/entitiy/cf/TbCalculatedFieldService.java index 89931b8541..cb7e809059 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/cf/TbCalculatedFieldService.java +++ b/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 findAllByTenantIdAndEntityId(EntityId entityId, SecurityUser user, PageLink pageLink); + void delete(CalculatedField calculatedField, SecurityUser user); } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java index b272694b16..944e24480e 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java +++ b/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); } diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldStateTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldStateTest.java new file mode 100644 index 0000000000..42ff828dbd --- /dev/null +++ b/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 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 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 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 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 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; + } + +} \ No newline at end of file diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldStateTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldStateTest.java new file mode 100644 index 0000000000..d6b384d85b --- /dev/null +++ b/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 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 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 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; + } + +} \ No newline at end of file diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java new file mode 100644 index 0000000000..203d7b3d71 --- /dev/null +++ b/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(); + } + +} \ No newline at end of file diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java new file mode 100644 index 0000000000..5a6f5e96e5 --- /dev/null +++ b/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 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 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."); + } + +} \ No newline at end of file diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldService.java index 3a508a5c08..f08d54190f 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldService.java @@ -44,6 +44,8 @@ public interface CalculatedFieldService extends EntityDaoService { PageData findAllCalculatedFields(PageLink pageLink); + PageData findAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink); + void deleteCalculatedField(TenantId tenantId, CalculatedFieldId calculatedFieldId); int deleteAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java index 6aa9075a79..0fd630fead 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java +++ b/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; }; } diff --git a/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java b/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java index f0a34de08e..b81af2bfb9 100644 --- a/common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java +++ b/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(); diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 18bcb7c6f4..6ee4108f9a 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/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. diff --git a/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java b/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java index 26ed4134cc..d37a4a53c1 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java +++ b/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 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); diff --git a/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldDao.java b/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldDao.java index 39663d0afc..3efb4011ed 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldDao.java @@ -37,8 +37,12 @@ public interface CalculatedFieldDao extends Dao { PageData findAll(PageLink pageLink); + PageData findAllByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink); + List removeAllByEntityId(TenantId tenantId, EntityId entityId); boolean existsByEntityId(TenantId tenantId, EntityId entityId); + long countCFByEntityId(TenantId tenantId, EntityId entityId); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java b/dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java index 80e421f350..db8997ceb8 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java +++ b/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 { @@ -29,13 +33,40 @@ public class CalculatedFieldDataValidator extends DataValidator @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!"); + } + } + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldRepository.java index bed6f2d3a2..2aeca659bc 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldRepository.java +++ b/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 findAllByTenantIdAndEntityId(UUID tenantId, UUID entityId); + Page findAllByTenantIdAndEntityId(UUID tenantId, UUID entityId, Pageable pageable); + List findAllByTenantId(UUID tenantId); List removeAllByTenantIdAndEntityId(UUID tenantId, UUID entityId); + long countByTenantIdAndEntityId(UUID tenantId, UUID entityId); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/DefaultNativeCalculatedFieldRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/DefaultNativeCalculatedFieldRepository.java index 677234dc20..daca6d5056 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/DefaultNativeCalculatedFieldRepository.java +++ b/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; diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldDao.java index cdcffdd440..703bdfbf6f 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldDao.java @@ -71,6 +71,12 @@ public class JpaCalculatedFieldDao extends JpaAbstractDao 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 removeAllByEntityId(TenantId tenantId, EntityId entityId) { @@ -82,6 +88,11 @@ public class JpaCalculatedFieldDao extends JpaAbstractDao getEntityClass() { return CalculatedFieldEntity.class; diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java index 406b2d0d5c..74f3468f49 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/AttributesSaveRequest.java +++ b/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 entries; private final boolean notifyDevice; private final List previousCalculatedFieldIds; + private final UUID tbMsgId; + private final TbMsgType tbMsgType; private final FutureCallback callback; public static Builder builder() { @@ -56,6 +60,8 @@ public class AttributesSaveRequest { private List entries; private boolean notifyDevice = true; private List previousCalculatedFieldIds; + private UUID tbMsgId; + private TbMsgType tbMsgType; private FutureCallback 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 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); } } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java index 95eb788e5f..957b0cf108 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesSaveRequest.java +++ b/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 previousCalculatedFieldIds; + private final UUID tbMsgId; + private final TbMsgType tbMsgType; private final FutureCallback callback; public static Builder builder() { @@ -59,6 +63,8 @@ public class TimeseriesSaveRequest { private boolean saveLatest = true; private boolean onlyLatest; private List 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 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); } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java index ae2fce6575..925d70daa1 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java +++ b/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()); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java index 89f7844313..3287cc825f 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java +++ b/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()); }