From 62da129d6bb08e67243084af811f4dacee083bf6 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Thu, 27 Feb 2025 16:46:55 +0200 Subject: [PATCH] Review comments fixed --- ...alculatedFieldManagerMessageProcessor.java | 14 ++-- .../EntityCalculatedFieldTelemetryMsg.java | 1 + .../controller/CalculatedFieldController.java | 5 +- .../service/cf/CalculatedFieldCache.java | 1 - ...faultCalculatedFieldProcessingService.java | 6 -- .../DefaultCalculatedFieldQueueService.java | 18 +---- ...aultCalculatedFieldEntityProfileCache.java | 2 +- .../service/cf/ctx/state/ArgumentEntry.java | 12 +-- .../KafkaCalculatedFieldStateService.java | 5 +- .../RocksDBCalculatedFieldStateService.java | 7 +- .../ctx/state/SingleValueArgumentEntry.java | 3 - ...CalculatedFieldAttributeUpdateRequest.java | 72 ------------------ ...CalculatedFieldTelemetryUpdateRequest.java | 39 ---------- ...alculatedFieldTimeSeriesUpdateRequest.java | 75 ------------------- .../entitiy/AbstractTbEntityService.java | 10 --- ...faultTbCalculatedFieldConsumerService.java | 13 +--- .../queue/DefaultTbClusterService.java | 4 +- .../server/utils/CalculatedFieldUtils.java | 8 ++ .../server/actors/TbActorMailbox.java | 5 +- .../server/dao/cf/CalculatedFieldService.java | 15 ---- .../common/data/cf/CalculatedFieldLink.java | 6 +- .../cf/CalculatedFieldLinkConfiguration.java | 31 -------- .../BaseCalculatedFieldConfiguration.java | 31 -------- .../CalculatedFieldConfiguration.java | 4 - .../common/data/id/CalculatedFieldLinkId.java | 2 +- .../thingsboard/server/common/msg/TbMsg.java | 2 +- common/proto/src/main/proto/queue.proto | 2 +- .../DefaultTbServiceInfoProvider.java | 7 +- .../provider/TbQueueProducerProvider.java | 2 +- .../DefaultTbQueueRequestTemplateTest.java | 18 +---- .../discovery/ZkDiscoveryServiceTest.java | 6 +- .../script/api/tbel/TbelCfSingleValueArg.java | 8 +- .../dao/cf/BaseCalculatedFieldService.java | 46 ------------ .../server/dao/cf/CalculatedFieldDao.java | 2 - .../server/dao/cf/CalculatedFieldLinkDao.java | 3 - .../server/dao/model/ModelConstants.java | 1 - .../model/sql/CalculatedFieldLinkEntity.java | 20 ----- ...efaultNativeCalculatedFieldRepository.java | 2 - .../dao/sql/cf/JpaCalculatedFieldDao.java | 5 -- .../dao/sql/cf/JpaCalculatedFieldLinkDao.java | 5 -- .../main/resources/sql/schema-entities.sql | 1 - pom.xml | 2 +- 42 files changed, 57 insertions(+), 464 deletions(-) delete mode 100644 application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldAttributeUpdateRequest.java delete mode 100644 application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTelemetryUpdateRequest.java delete mode 100644 application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTimeSeriesUpdateRequest.java delete mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldLinkConfiguration.java 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 3395f279a1..d26d89626e 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 @@ -65,8 +65,8 @@ import static org.thingsboard.server.utils.CalculatedFieldUtils.fromProto; public class CalculatedFieldManagerMessageProcessor extends AbstractContextAwareMsgProcessor { private final Map calculatedFields = new HashMap<>(); - private final Map> entityIdCalculatedFields = new ConcurrentHashMap<>(); - private final ConcurrentMap> entityIdCalculatedFieldLinks = new ConcurrentHashMap<>(); + private final Map> entityIdCalculatedFields = new HashMap<>(); + private final Map> entityIdCalculatedFieldLinks = new HashMap<>(); private final CalculatedFieldProcessingService cfExecService; private final CalculatedFieldStateService cfStateService; @@ -105,7 +105,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware 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) - entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new ArrayList<>()).add(cfCtx); + entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx); msg.getCallback().onSuccess(); } @@ -114,7 +114,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware var link = msg.getLink(); // 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) - entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new ArrayList<>()).add(link); + entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(link); msg.getCallback().onSuccess(); } @@ -262,7 +262,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } calculatedFields.put(newCf.getId(), newCfCtx); List oldCfList = entityIdCalculatedFields.get(newCf.getEntityId()); - List newCfList = new ArrayList<>(oldCfList.size()); + List newCfList = new CopyOnWriteArrayList<>(); boolean found = false; for (CalculatedFieldCtx oldCtx : oldCfList) { if (oldCtx.getCfId().equals(newCf.getId())) { @@ -456,13 +456,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private void addLinks(CalculatedField newCf) { var newLinks = newCf.getConfiguration().buildCalculatedFieldLinks(tenantId, newCf.getEntityId(), newCf.getId()); - newLinks.forEach(link -> entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new ArrayList<>()).add(link)); + newLinks.forEach(link -> entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(link)); } private void deleteLinks(CalculatedFieldCtx cfCtx) { var oldCf = cfCtx.getCalculatedField(); var oldLinks = oldCf.getConfiguration().buildCalculatedFieldLinks(tenantId, oldCf.getEntityId(), oldCf.getId()); - oldLinks.forEach(link -> entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new ArrayList<>()).remove(link)); + oldLinks.forEach(link -> entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).remove(link)); } public void onPartitionChange(CalculatedFieldPartitionChangeMsg msg) { diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldTelemetryMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldTelemetryMsg.java index 8aab7d2743..8ded4b6028 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldTelemetryMsg.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/EntityCalculatedFieldTelemetryMsg.java @@ -32,6 +32,7 @@ public class EntityCalculatedFieldTelemetryMsg implements ToCalculatedFieldSyste private final TenantId tenantId; private final EntityId entityId; private final CalculatedFieldTelemetryMsgProto proto; + // Both lists are effectively immutable in CalculatedFieldManagerMessageProcessor and must stay so. private final List entityIdFields; private final List profileIdFields; private final TbCallback callback; 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 b1e9dc9214..fbeade5819 100644 --- a/application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java +++ b/application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java @@ -35,6 +35,7 @@ import org.springframework.web.bind.annotation.RestController; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.script.api.tbel.TbelCfArg; import org.thingsboard.script.api.tbel.TbelInvokeService; +import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.HasTenantId; import org.thingsboard.server.common.data.cf.CalculatedField; @@ -53,6 +54,7 @@ import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldTbelScriptEngi import org.thingsboard.server.service.entitiy.cf.TbCalculatedFieldService; import org.thingsboard.server.service.security.model.SecurityUser; import org.thingsboard.server.service.security.permission.Operation; +import org.thingsboard.server.service.security.permission.Resource; import java.util.ArrayList; import java.util.Collections; @@ -127,6 +129,7 @@ public class CalculatedFieldController extends BaseController { public CalculatedField saveCalculatedField(@io.swagger.v3.oas.annotations.parameters.RequestBody(description = "A JSON value representing the calculated field.") @RequestBody CalculatedField calculatedField) throws Exception { calculatedField.setTenantId(getTenantId()); + checkEntity(calculatedField.getId(), calculatedField, Resource.CALCULATED_FIELD); checkEntityId(calculatedField.getEntityId(), Operation.WRITE_CALCULATED_FIELD); checkReferencedEntities(calculatedField.getConfiguration(), getCurrentUser()); return tbCalculatedFieldService.save(calculatedField, getCurrentUser()); @@ -176,7 +179,7 @@ public class CalculatedFieldController extends BaseController { public void deleteCalculatedField(@PathVariable(CALCULATED_FIELD_ID) String strCalculatedField) throws Exception { checkParameter(CALCULATED_FIELD_ID, strCalculatedField); CalculatedFieldId calculatedFieldId = new CalculatedFieldId(toUUID(strCalculatedField)); - CalculatedField calculatedField = tbCalculatedFieldService.findById(calculatedFieldId, getCurrentUser()); + CalculatedField calculatedField = checkCalculatedFieldId(calculatedFieldId, Operation.DELETE); checkEntityId(calculatedField.getEntityId(), Operation.WRITE_CALCULATED_FIELD); tbCalculatedFieldService.delete(calculatedField, getCurrentUser()); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java index 3fbfe26dc0..fb63432fed 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java @@ -23,7 +23,6 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import java.util.List; -import java.util.Set; public interface CalculatedFieldCache { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java index 5ca4993d3f..f69f5cd50b 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java @@ -21,11 +21,9 @@ import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; -import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.math.NumberUtils; -import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ThingsBoardExecutors; @@ -104,10 +102,6 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP private ListeningExecutorService calculatedFieldCallbackExecutor; - @Value("${calculatedField.initFetchPackSize:50000}") - @Getter - private int initFetchPackSize; - @PostConstruct public void init() { calculatedFieldCallbackExecutor = MoreExecutors.listeningDecorator(ThingsBoardExecutors.newWorkStealingPool( diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java index 7421dfe187..1b52032bb5 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java @@ -16,10 +16,8 @@ package org.thingsboard.server.service.cf; import com.google.common.util.concurrent.FutureCallback; -import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.thingsboard.rule.engine.api.AttributesDeleteRequest; import org.thingsboard.rule.engine.api.AttributesSaveRequest; @@ -40,7 +38,6 @@ import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto; import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto; -import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; @@ -59,6 +56,7 @@ import java.util.function.Predicate; import java.util.function.Supplier; import static org.thingsboard.server.common.util.ProtoUtils.toTsKvProto; +import static org.thingsboard.server.utils.CalculatedFieldUtils.toProto; @Service @Slf4j @@ -84,17 +82,10 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS EntityType.DEVICE, EntityType.ASSET, EntityType.CUSTOMER, EntityType.TENANT ); - @Value("${calculatedField.initFetchPackSize:50000}") - @Getter - private int initFetchPackSize; - @Override public void pushRequestToQueue(TimeseriesSaveRequest request, TimeseriesSaveResult result, FutureCallback callback) { var tenantId = request.getTenantId(); var entityId = request.getEntityId(); - //TODO: 1. check that request entity has calculated fields for entity or profile. If yes - push to corresponding partitions; - //TODO: 2. check that request entity has calculated field links. If yes - push to corresponding partitions; - //TODO: in 1 and 2 we should do the check as quick as possible. Should we also check the field/link keys?; checkEntityAndPushToQueue(tenantId, entityId, cf -> cf.matches(request.getEntries()), cf -> cf.linkMatches(entityId, request.getEntries()), () -> toCalculatedFieldTelemetryMsgProto(request, result), callback); } @@ -239,13 +230,6 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS return telemetryMsg; } - private CalculatedFieldIdProto toProto(CalculatedFieldId cfId) { - return CalculatedFieldIdProto.newBuilder() - .setCalculatedFieldIdMSB(cfId.getId().getMostSignificantBits()) - .setCalculatedFieldIdLSB(cfId.getId().getLeastSignificantBits()) - .build(); - } - private static TbQueueCallback wrap(FutureCallback callback) { if (callback != null) { return new FutureCallbackWrapper(callback); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java b/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java index 4377ee58aa..4cf62c01b3 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java @@ -38,7 +38,7 @@ import java.util.stream.Collectors; @Service @Slf4j @RequiredArgsConstructor -//TODO: remove and use TenantEntityProfileCache in each CalculatedFieldManagerMessageProcessor; +//TODO ashvayka: remove and use TenantEntityProfileCache in each CalculatedFieldManagerMessageProcessor; public class DefaultCalculatedFieldEntityProfileCache extends TbApplicationEventListener implements CalculatedFieldEntityProfileCache { private static final Integer UNKNOWN = 0; 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 6f827065af..83e10b8194 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 @@ -44,6 +44,12 @@ public interface ArgumentEntry { boolean isEmpty(); + TbelCfArg toTbelCfArg(); + + boolean isForceResetPrevious(); + + void setForceResetPrevious(boolean forceResetPrevious); + static ArgumentEntry createSingleValueArgument(KvEntry kvEntry) { return new SingleValueArgumentEntry(kvEntry); } @@ -52,10 +58,4 @@ public interface ArgumentEntry { return new TsRollingArgumentEntry(kvEntries, limit, timeWindow); } - TbelCfArg toTbelCfArg(); - - boolean isForceResetPrevious(); - - void setForceResetPrevious(boolean forceResetPrevious); - } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java index 0c2b5ac07e..edb47dd7fa 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java @@ -44,10 +44,7 @@ import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; -import static org.thingsboard.server.queue.common.AbstractTbQueueTemplate.bytesToString; -import static org.thingsboard.server.queue.common.AbstractTbQueueTemplate.bytesToUuid; -import static org.thingsboard.server.queue.common.AbstractTbQueueTemplate.stringToBytes; -import static org.thingsboard.server.queue.common.AbstractTbQueueTemplate.uuidToBytes; +import static org.thingsboard.server.queue.common.AbstractTbQueueTemplate.*; @Service @RequiredArgsConstructor diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java index 32ad6df551..a508eecada 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java @@ -37,7 +37,7 @@ public class RocksDBCalculatedFieldStateService extends AbstractCalculatedFieldS private final CfRocksDb cfRocksDb; - private Set partitions; + private boolean initialized; @Override protected void doPersist(CalculatedFieldEntityCtxId stateId, CalculatedFieldStateProto stateMsgProto, TbCallback callback) { @@ -53,7 +53,7 @@ public class RocksDBCalculatedFieldStateService extends AbstractCalculatedFieldS @Override public void restore(Set partitions) { - if (this.partitions == null) { + if (!this.initialized) { cfRocksDb.forEach((key, value) -> { try { processRestoredState(CalculatedFieldStateProto.parseFrom(value)); @@ -61,10 +61,9 @@ public class RocksDBCalculatedFieldStateService extends AbstractCalculatedFieldS log.error("[{}] Failed to process restored state", key, e); } }); + this.initialized = true; } - eventConsumer.update(partitions); - this.partitions = partitions; } @Override 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 2e8bb63f5f..a064a99935 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 @@ -100,8 +100,6 @@ public class SingleValueArgumentEntry implements ArgumentEntry { if (newVersion == null || this.version == null || newVersion > this.version) { this.ts = singleValueEntry.getTs(); this.version = newVersion; - - // TODO: should we persist updated ts and version values? BasicKvEntry newValue = singleValueEntry.getKvEntryValue(); if (this.kvEntryValue != null && this.kvEntryValue.getValue().equals(newValue.getValue())) { return false; @@ -109,7 +107,6 @@ public class SingleValueArgumentEntry implements ArgumentEntry { this.kvEntryValue = singleValueEntry.getKvEntryValue(); return true; } - } else { throw new IllegalArgumentException("Unsupported argument entry type for single value argument entry: " + entry.getType()); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldAttributeUpdateRequest.java b/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldAttributeUpdateRequest.java deleted file mode 100644 index f95b9041a3..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldAttributeUpdateRequest.java +++ /dev/null @@ -1,72 +0,0 @@ -/** - * Copyright © 2016-2025 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.telemetry; - -import lombok.AllArgsConstructor; -import lombok.Data; -import org.thingsboard.rule.engine.api.AttributesSaveRequest; -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.CalculatedFieldId; -import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.KvEntry; -import org.thingsboard.server.common.data.util.TbPair; -import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; - -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -@Data -@AllArgsConstructor -public class CalculatedFieldAttributeUpdateRequest implements CalculatedFieldTelemetryUpdateRequest { - - private TenantId tenantId; - private EntityId entityId; - private AttributeScope scope; - private List kvEntries; - private List previousCalculatedFieldIds; - - public CalculatedFieldAttributeUpdateRequest(AttributesSaveRequest request) { - this.tenantId = request.getTenantId(); - this.entityId = request.getEntityId(); - this.scope = request.getScope(); - this.kvEntries = request.getEntries(); - this.previousCalculatedFieldIds = request.getPreviousCalculatedFieldIds(); - } - - @Override - public Map getMappedTelemetry(CalculatedFieldCtx ctx, EntityId referencedEntityId) { - Map mappedKvEntries = new HashMap<>(); - Map, String> referencedKeys = ctx.getReferencedEntityKeys(); - - kvEntries.forEach(entry -> { - String key = entry.getKey(); - - ReferencedEntityKey referencedEntityKey = new ReferencedEntityKey(key, ArgumentType.ATTRIBUTE, scope); - - String argName = referencedKeys.get(new TbPair<>(referencedEntityId, referencedEntityKey)); - - if (argName != null) { - mappedKvEntries.put(argName, entry); - } - }); - - return mappedKvEntries; - } -} diff --git a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTelemetryUpdateRequest.java b/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTelemetryUpdateRequest.java deleted file mode 100644 index 9f1043c4d3..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTelemetryUpdateRequest.java +++ /dev/null @@ -1,39 +0,0 @@ -/** - * Copyright © 2016-2025 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.telemetry; - -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.kv.KvEntry; -import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; - -import java.util.List; -import java.util.Map; - -public interface CalculatedFieldTelemetryUpdateRequest { - - TenantId getTenantId(); - - EntityId getEntityId(); - - List getKvEntries(); - - List getPreviousCalculatedFieldIds(); - - Map getMappedTelemetry(CalculatedFieldCtx ctx, EntityId referencedEntityId); - -} diff --git a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTimeSeriesUpdateRequest.java b/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTimeSeriesUpdateRequest.java deleted file mode 100644 index 3f85108e0c..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/cf/telemetry/CalculatedFieldTimeSeriesUpdateRequest.java +++ /dev/null @@ -1,75 +0,0 @@ -/** - * Copyright © 2016-2025 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.telemetry; - -import lombok.AllArgsConstructor; -import lombok.Data; -import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; -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.CalculatedFieldId; -import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.KvEntry; -import org.thingsboard.server.common.data.util.TbPair; -import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; - -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -@Data -@AllArgsConstructor -public class CalculatedFieldTimeSeriesUpdateRequest implements CalculatedFieldTelemetryUpdateRequest { - - private TenantId tenantId; - private EntityId entityId; - private List kvEntries; - private List previousCalculatedFieldIds; - - public CalculatedFieldTimeSeriesUpdateRequest(TimeseriesSaveRequest request) { - this.tenantId = request.getTenantId(); - this.entityId = request.getEntityId(); - this.kvEntries = request.getEntries(); - this.previousCalculatedFieldIds = request.getPreviousCalculatedFieldIds(); - } - - @Override - public Map getMappedTelemetry(CalculatedFieldCtx ctx, EntityId referencedEntityId) { - Map mappedKvEntries = new HashMap<>(); - Map, String> referencedKeys = ctx.getReferencedEntityKeys(); - - kvEntries.forEach(entry -> { - String key = entry.getKey(); - - ReferencedEntityKey tsLatestKey = new ReferencedEntityKey(key, ArgumentType.TS_LATEST, null); - String argTsLatestName = referencedKeys.get(new TbPair<>(referencedEntityId, tsLatestKey)); - - if (argTsLatestName != null) { - mappedKvEntries.put(argTsLatestName, entry); - } else { - ReferencedEntityKey tsRollingKey = new ReferencedEntityKey(key, ArgumentType.TS_ROLLING, null); - String argTsRollingName = referencedKeys.get(new TbPair<>(referencedEntityId, tsRollingKey)); - - if (argTsRollingName != null) { - mappedKvEntries.put(argTsRollingName, entry); - } - } - }); - - return mappedKvEntries; - } -} diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/AbstractTbEntityService.java b/application/src/main/java/org/thingsboard/server/service/entitiy/AbstractTbEntityService.java index 99e62d9186..e9ee7202a7 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/AbstractTbEntityService.java +++ b/application/src/main/java/org/thingsboard/server/service/entitiy/AbstractTbEntityService.java @@ -78,16 +78,6 @@ public abstract class AbstractTbEntityService { @Lazy private EntitiesVersionControlService vcService; @Autowired - protected TenantService tenantService; - @Autowired - protected AssetService assetService; - @Autowired - protected DeviceService deviceService; - @Autowired - protected AssetProfileService assetProfileService; - @Autowired - protected DeviceProfileService deviceProfileService; - @Autowired protected EntityService entityService; protected boolean isTestProfile() { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java index bc15a64770..e76a2be9be 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java @@ -87,8 +87,6 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer private PartitionedQueueConsumerManager> eventConsumer; - private ListeningExecutorService calculatedFieldsExecutor; - public DefaultTbCalculatedFieldConsumerService(TbRuleEngineQueueFactory tbQueueFactory, ActorSystemContext actorContext, TbDeviceProfileCache deviceProfileCache, @@ -109,7 +107,6 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer @PostConstruct public void init() { super.init("tb-cf"); - this.calculatedFieldsExecutor = MoreExecutors.listeningDecorator(ThingsBoardExecutors.newWorkStealingPool(poolSize, "tb-cf-executor")); // TODO: multiple threads. this.eventConsumer = PartitionedQueueConsumerManager.>create() .queueKey(QueueKey.CF) @@ -127,9 +124,6 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer @PreDestroy public void destroy() { super.destroy(); - if (calculatedFieldsExecutor != null) { - calculatedFieldsExecutor.shutdownNow(); - } } @Override @@ -197,10 +191,9 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer packSubmitFuture.cancel(true); log.info("Timeout to process message: {}", pendingMsgHolder.getMsg()); } -// if (log.isDebugEnabled()) { -// ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue())); -// } - ctx.getAckMap().forEach((id, msg) -> log.warn("[{}] Timeout to process message: {}", id, msg.getValue())); // TODO: replace with commented above after testing + if (log.isDebugEnabled()) { + ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue())); + } ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue())); } consumer.commit(); 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 0ccfb764e2..800578ace4 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 @@ -196,12 +196,12 @@ public class DefaultTbClusterService implements TbClusterService { UUID msgId = UUID.randomUUID(); TbQueueProducer> toCfProducer = producerProvider.getCalculatedFieldsNotificationsMsgProducer(); Set tbReServices = partitionService.getAllServiceIds(ServiceType.TB_RULE_ENGINE); + MultipleTbQueueCallbackWrapper callbackWrapper = new MultipleTbQueueCallbackWrapper(tbReServices.size(), callback); for (String serviceId : tbReServices) { TopicPartitionInfo tpi = topicService.getCalculatedFieldNotificationsTopic(serviceId); - toCfProducer.send(tpi, new TbProtoQueueMsg<>(msgId, toCfMsg), null); + toCfProducer.send(tpi, new TbProtoQueueMsg<>(msgId, toCfMsg), callbackWrapper); toRuleEngineNfs.incrementAndGet(); } - callback.onSuccess(null); // TODO: refactor to be fair, similar to multi-value callback; } @Override diff --git a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java index 6f10c63492..77080c28c8 100644 --- a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java +++ b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java @@ -24,6 +24,7 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.BasicKvEntry; import org.thingsboard.server.common.util.KvProtoUtil; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto; import org.thingsboard.server.gen.transport.TransportProtos.TsDoubleValProto; @@ -42,6 +43,13 @@ import java.util.UUID; public class CalculatedFieldUtils { + public static CalculatedFieldIdProto toProto(CalculatedFieldId cfId) { + return CalculatedFieldIdProto.newBuilder() + .setCalculatedFieldIdMSB(cfId.getId().getMostSignificantBits()) + .setCalculatedFieldIdLSB(cfId.getId().getLeastSignificantBits()) + .build(); + } + public static CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) { return CalculatedFieldEntityCtxIdProto.newBuilder() .setTenantIdMSB(ctxId.tenantId().getId().getMostSignificantBits()) diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java index af6031fb83..4b822e0030 100644 --- a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java @@ -153,14 +153,13 @@ public final class TbActorMailbox implements TbActorCtx { } if (msg != null) { try { - log.debug("[{}] Going to process message: {}", selfId, msg); + log.trace("[{}] Going to process message: {}", selfId, msg); actor.process(msg); } catch (TbRuleNodeUpdateException updateException) { stopReason = TbActorStopReason.INIT_FAILED; destroy(updateException.getCause()); } catch (Throwable t) { - //TODO: revert; - log.error("[{}] Failed to process message: {}", selfId, msg, t); + log.debug("[{}] Failed to process message: {}", selfId, msg, t); ProcessFailureStrategy strategy = actor.onProcessFailure(msg, t); if (strategy.isStop()) { system.stop(selfId); 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 adc794e2c1..3d91790790 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 @@ -15,7 +15,6 @@ */ package org.thingsboard.server.dao.cf; -import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.id.CalculatedFieldId; @@ -34,14 +33,10 @@ public interface CalculatedFieldService extends EntityDaoService { CalculatedField findById(TenantId tenantId, CalculatedFieldId calculatedFieldId); - ListenableFuture findCalculatedFieldByIdAsync(TenantId tenantId, CalculatedFieldId calculatedFieldId); - List findCalculatedFieldIdsByEntityId(TenantId tenantId, EntityId entityId); List findCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId); - List findAllCalculatedFields(); - PageData findAllCalculatedFields(PageLink pageLink); PageData findAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink); @@ -54,22 +49,12 @@ public interface CalculatedFieldService extends EntityDaoService { CalculatedFieldLink findCalculatedFieldLinkById(TenantId tenantId, CalculatedFieldLinkId calculatedFieldLinkId); - ListenableFuture findCalculatedFieldLinkByIdAsync(TenantId tenantId, CalculatedFieldLinkId calculatedFieldLinkId); - - List findAllCalculatedFieldLinks(); - List findAllCalculatedFieldLinksById(TenantId tenantId, CalculatedFieldId calculatedFieldId); List findAllCalculatedFieldLinksByEntityId(TenantId tenantId, EntityId entityId); - ListenableFuture> findAllCalculatedFieldLinksByIdAsync(TenantId tenantId, CalculatedFieldId calculatedFieldId); - PageData findAllCalculatedFieldLinks(PageLink pageLink); boolean referencedInAnyCalculatedField(TenantId tenantId, EntityId referencedEntityId); - boolean referencedInAnyCalculatedFieldIncludingEntityId(TenantId tenantId, EntityId referencedEntityId); - - boolean existsCalculatedFieldByEntityId(TenantId tenantId, EntityId entityId); - } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldLink.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldLink.java index ce979214d2..3f048815da 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldLink.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldLink.java @@ -36,8 +36,6 @@ public class CalculatedFieldLink extends BaseData { @Schema(description = "JSON object with the Calculated Field Id. ", accessMode = Schema.AccessMode.READ_ONLY) private CalculatedFieldId calculatedFieldId; - @Schema - private transient CalculatedFieldLinkConfiguration configuration; public CalculatedFieldLink() { super(); @@ -47,11 +45,10 @@ public class CalculatedFieldLink extends BaseData { super(id); } - public CalculatedFieldLink(TenantId tenantId, EntityId entityId, CalculatedFieldId calculatedFieldId, CalculatedFieldLinkConfiguration configuration) { + public CalculatedFieldLink(TenantId tenantId, EntityId entityId, CalculatedFieldId calculatedFieldId) { this.tenantId = tenantId; this.entityId = entityId; this.calculatedFieldId = calculatedFieldId; - this.configuration = configuration; } @Override @@ -61,7 +58,6 @@ public class CalculatedFieldLink extends BaseData { .append("tenantId=").append(tenantId) .append(", entityId=").append(entityId) .append(", calculatedFieldId=").append(calculatedFieldId) - .append(", configuration=").append(configuration) .append(", createdTime=").append(createdTime) .append(", id=").append(id).append(']') .toString(); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldLinkConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldLinkConfiguration.java deleted file mode 100644 index b3aaa46b86..0000000000 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldLinkConfiguration.java +++ /dev/null @@ -1,31 +0,0 @@ -/** - * Copyright © 2016-2025 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.common.data.cf; - -import lombok.Data; - -import java.util.HashMap; -import java.util.Map; - -@Data -public class CalculatedFieldLinkConfiguration { - - private Map clientAttributes = new HashMap<>(); - private Map serverAttributes = new HashMap<>(); - private Map sharedAttributes = new HashMap<>(); - private Map timeSeries = new HashMap<>(); - -} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/BaseCalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/BaseCalculatedFieldConfiguration.java index 8ff4fab94a..8227ff4603 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/BaseCalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/BaseCalculatedFieldConfiguration.java @@ -17,7 +17,6 @@ package org.thingsboard.server.common.data.cf.configuration; import lombok.Data; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; -import org.thingsboard.server.common.data.cf.CalculatedFieldLinkConfiguration; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; @@ -42,35 +41,6 @@ public abstract class BaseCalculatedFieldConfiguration implements CalculatedFiel .collect(Collectors.toList()); } - @Override - public CalculatedFieldLinkConfiguration getReferencedEntityConfig(EntityId entityId) { - CalculatedFieldLinkConfiguration linkConfiguration = new CalculatedFieldLinkConfiguration(); - - arguments.entrySet().stream() - .filter(entry -> entry.getValue().getRefEntityId() != null && entry.getValue().getRefEntityId().equals(entityId)) - .forEach(entry -> { - ReferencedEntityKey refEntityKey = entry.getValue().getRefEntityKey(); - String argumentName = entry.getKey(); - - switch (refEntityKey.getType()) { - case ATTRIBUTE -> { - switch (refEntityKey.getScope()) { - case CLIENT_SCOPE -> - linkConfiguration.getClientAttributes().put(refEntityKey.getKey(), argumentName); - case SERVER_SCOPE -> - linkConfiguration.getServerAttributes().put(refEntityKey.getKey(), argumentName); - case SHARED_SCOPE -> - linkConfiguration.getSharedAttributes().put(refEntityKey.getKey(), argumentName); - } - } - case TS_LATEST, TS_ROLLING -> - linkConfiguration.getTimeSeries().put(refEntityKey.getKey(), argumentName); - } - }); - - return linkConfiguration; - } - @Override public List buildCalculatedFieldLinks(TenantId tenantId, EntityId cfEntityId, CalculatedFieldId calculatedFieldId) { return getReferencedEntities().stream() @@ -85,7 +55,6 @@ public abstract class BaseCalculatedFieldConfiguration implements CalculatedFiel link.setTenantId(tenantId); link.setEntityId(referencedEntityId); link.setCalculatedFieldId(calculatedFieldId); - link.setConfiguration(getReferencedEntityConfig(referencedEntityId)); return link; } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java index b97024c47f..c53f1fe5f1 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java @@ -19,7 +19,6 @@ import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.annotation.JsonSubTypes; import com.fasterxml.jackson.annotation.JsonTypeInfo; 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.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; @@ -53,9 +52,6 @@ public interface CalculatedFieldConfiguration { @JsonIgnore List getReferencedEntities(); - @JsonIgnore - CalculatedFieldLinkConfiguration getReferencedEntityConfig(EntityId entityId); - List buildCalculatedFieldLinks(TenantId tenantId, EntityId cfEntityId, CalculatedFieldId calculatedFieldId); CalculatedFieldLink buildCalculatedFieldLink(TenantId tenantId, EntityId referencedEntityId, CalculatedFieldId calculatedFieldId); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/id/CalculatedFieldLinkId.java b/common/data/src/main/java/org/thingsboard/server/common/data/id/CalculatedFieldLinkId.java index 75a0dd8b90..6a0c680bb6 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/id/CalculatedFieldLinkId.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/id/CalculatedFieldLinkId.java @@ -23,7 +23,7 @@ import org.thingsboard.server.common.data.EntityType; import java.util.UUID; @Schema -public class CalculatedFieldLinkId extends UUIDBased implements EntityId { +public class CalculatedFieldLinkId extends UUIDBased implements EntityId { private static final long serialVersionUID = 1L; diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java index c0ec06191d..f1c2236080 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java @@ -485,7 +485,7 @@ public final class TbMsg implements Serializable { } public TbMsgBuilder previousCalculatedFieldIds(List previousCalculatedFieldIds) { - this.previousCalculatedFieldIds = previousCalculatedFieldIds; + this.previousCalculatedFieldIds = new CopyOnWriteArrayList<>(previousCalculatedFieldIds); return this; } diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index fe412dbc0c..f41bf92592 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -1190,7 +1190,7 @@ message ComponentLifecycleMsgProto { int64 entityIdMSB = 4; int64 entityIdLSB = 5; ComponentLifecycleEvent event = 6; - //Since 4.0. To replace the + //Since 4.0. TODO: replace the DeviceNameOrTypeUpdateMsgProto string oldName = 7; string name = 8; int64 oldProfileIdMSB = 9; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java index a2753b8832..4bcbb5152c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java @@ -39,12 +39,7 @@ import java.util.Set; import java.util.UUID; import java.util.stream.Collectors; -import static org.thingsboard.common.util.SystemUtil.getCpuCount; -import static org.thingsboard.common.util.SystemUtil.getCpuUsage; -import static org.thingsboard.common.util.SystemUtil.getDiscSpaceUsage; -import static org.thingsboard.common.util.SystemUtil.getMemoryUsage; -import static org.thingsboard.common.util.SystemUtil.getTotalDiscSpace; -import static org.thingsboard.common.util.SystemUtil.getTotalMemory; +import static org.thingsboard.common.util.SystemUtil.*; @Component diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java index de2e372cbf..865637b2ff 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java @@ -15,8 +15,8 @@ */ package org.thingsboard.server.queue.provider; -import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplateTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplateTest.java index 1760555e07..d3975c78bd 100644 --- a/common/queue/src/test/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplateTest.java +++ b/common/queue/src/test/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplateTest.java @@ -37,23 +37,11 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.equalTo; -import static org.hamcrest.Matchers.greaterThanOrEqualTo; -import static org.hamcrest.Matchers.is; -import static org.hamcrest.Matchers.lessThan; +import static org.hamcrest.Matchers.*; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; -import static org.mockito.BDDMockito.willAnswer; -import static org.mockito.BDDMockito.willDoNothing; -import static org.mockito.BDDMockito.willReturn; -import static org.mockito.Mockito.RETURNS_DEEP_STUBS; -import static org.mockito.Mockito.atLeastOnce; -import static org.mockito.Mockito.lenient; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.spy; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; +import static org.mockito.BDDMockito.*; +import static org.mockito.Mockito.*; import static org.mockito.hamcrest.MockitoHamcrest.longThat; @Slf4j diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/discovery/ZkDiscoveryServiceTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/discovery/ZkDiscoveryServiceTest.java index c8d91b0c0f..19643f4ad4 100644 --- a/common/queue/src/test/java/org/thingsboard/server/queue/discovery/ZkDiscoveryServiceTest.java +++ b/common/queue/src/test/java/org/thingsboard/server/queue/discovery/ZkDiscoveryServiceTest.java @@ -42,11 +42,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.reset; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; +import static org.mockito.Mockito.*; @ExtendWith(MockitoExtension.class) public class ZkDiscoveryServiceTest { diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfSingleValueArg.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfSingleValueArg.java index 4b84f1a642..193b6ea1ae 100644 --- a/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfSingleValueArg.java +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfSingleValueArg.java @@ -22,6 +22,8 @@ import lombok.Data; @Data public class TbelCfSingleValueArg implements TbelCfArg { + public static final long OBJ_SIZE = 32L; // Approximate calculation; + private final long ts; private final Object value; @@ -36,7 +38,11 @@ public class TbelCfSingleValueArg implements TbelCfArg { @Override public long memorySize() { - return 8L; // TODO; + if (value instanceof String strValue) { + return OBJ_SIZE + strValue.length(); + } else { + return OBJ_SIZE; + } } @Override 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 9024ae1a70..416bff6ea3 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 @@ -84,13 +84,6 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements return calculatedFieldDao.findById(tenantId, calculatedFieldId.getId()); } - @Override - public ListenableFuture findCalculatedFieldByIdAsync(TenantId tenantId, CalculatedFieldId calculatedFieldId) { - log.trace("Executing findCalculatedFieldByIdAsync [{}]", calculatedFieldId); - validateId(calculatedFieldId, id -> INCORRECT_CALCULATED_FIELD_ID + id); - return calculatedFieldDao.findByIdAsync(tenantId, calculatedFieldId.getId()); - } - @Override public List findCalculatedFieldIdsByEntityId(TenantId tenantId, EntityId entityId) { log.trace("Executing findCalculatedFieldIdsByEntityId [{}]", entityId); @@ -105,12 +98,6 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements return calculatedFieldDao.findCalculatedFieldsByEntityId(tenantId, entityId); } - @Override - public List findAllCalculatedFields() { - log.trace("Executing findAll"); - return calculatedFieldDao.findAll(); - } - @Override public PageData findAllCalculatedFields(PageLink pageLink) { log.trace("Executing findAll, pageLink [{}]", pageLink); @@ -176,20 +163,6 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements return calculatedFieldLinkDao.findById(tenantId, calculatedFieldLinkId.getId()); } - @Override - public ListenableFuture findCalculatedFieldLinkByIdAsync(TenantId tenantId, CalculatedFieldLinkId calculatedFieldLinkId) { - log.trace("Executing findCalculatedFieldLinkByIdAsync [{}]", calculatedFieldLinkId); - validateId(tenantId, id -> INCORRECT_TENANT_ID + id); - validateId(calculatedFieldLinkId, id -> "Incorrect calculatedFieldLinkId " + id); - return calculatedFieldLinkDao.findByIdAsync(tenantId, calculatedFieldLinkId.getId()); - } - - @Override - public List findAllCalculatedFieldLinks() { - log.trace("Executing findAllCalculatedFieldLinks"); - return calculatedFieldLinkDao.findAll(); - } - @Override public List findAllCalculatedFieldLinksById(TenantId tenantId, CalculatedFieldId calculatedFieldId) { log.trace("Executing findAllCalculatedFieldLinksById, calculatedFieldId [{}]", calculatedFieldId); @@ -202,12 +175,6 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements return calculatedFieldLinkDao.findCalculatedFieldLinksByEntityId(tenantId, entityId); } - @Override - public ListenableFuture> findAllCalculatedFieldLinksByIdAsync(TenantId tenantId, CalculatedFieldId calculatedFieldId) { - log.trace("Executing findAllCalculatedFieldLinksByIdAsync, calculatedFieldId [{}]", calculatedFieldId); - return calculatedFieldLinkDao.findCalculatedFieldLinksByCalculatedFieldIdAsync(tenantId, calculatedFieldId); - } - @Override public PageData findAllCalculatedFieldLinks(PageLink pageLink) { log.trace("Executing findAllCalculatedFieldLinks, pageLink [{}]", pageLink); @@ -224,19 +191,6 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements .anyMatch(referencedEntities -> referencedEntities.contains(referencedEntityId)); } - @Override - public boolean referencedInAnyCalculatedFieldIncludingEntityId(TenantId tenantId, EntityId referencedEntityId) { - return calculatedFieldDao.findAllByTenantId(tenantId).stream() - .map(CalculatedField::getConfiguration) - .map(CalculatedFieldConfiguration::getReferencedEntities) - .anyMatch(referencedEntities -> referencedEntities.contains(referencedEntityId)); - } - - @Override - public boolean existsCalculatedFieldByEntityId(TenantId tenantId, EntityId entityId) { - return calculatedFieldDao.existsByEntityId(tenantId, entityId); - } - @Override public Optional> findEntity(TenantId tenantId, EntityId entityId) { return Optional.ofNullable(findById(tenantId, new CalculatedFieldId(entityId.getId()))); 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 025f3dc18c..a966977968 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 @@ -41,8 +41,6 @@ public interface CalculatedFieldDao extends Dao { 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/cf/CalculatedFieldLinkDao.java b/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldLinkDao.java index 904cb37e9c..8b4a5e7086 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldLinkDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldLinkDao.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.dao.cf; -import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; @@ -32,8 +31,6 @@ public interface CalculatedFieldLinkDao extends Dao { List findCalculatedFieldLinksByEntityId(TenantId tenantId, EntityId entityId); - ListenableFuture> findCalculatedFieldLinksByCalculatedFieldIdAsync(TenantId tenantId, CalculatedFieldId calculatedFieldId); - List findAll(); PageData findAll(PageLink pageLink); diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java index fdb46d3451..e2b21ed59b 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java @@ -739,7 +739,6 @@ public class ModelConstants { public static final String CALCULATED_FIELD_LINK_ENTITY_TYPE = ENTITY_TYPE_COLUMN; public static final String CALCULATED_FIELD_LINK_ENTITY_ID = ENTITY_ID_COLUMN; public static final String CALCULATED_FIELD_LINK_CALCULATED_FIELD_ID = "calculated_field_id"; - public static final String CALCULATED_FIELD_LINK_CONFIGURATION = "configuration"; protected static final String[] NONE_AGGREGATION_COLUMNS = new String[]{LONG_VALUE_COLUMN, DOUBLE_VALUE_COLUMN, BOOLEAN_VALUE_COLUMN, STRING_VALUE_COLUMN, JSON_VALUE_COLUMN, KEY_COLUMN, TS_COLUMN}; diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/CalculatedFieldLinkEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/CalculatedFieldLinkEntity.java index 34c1933065..2668a1a0d8 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/CalculatedFieldLinkEntity.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/CalculatedFieldLinkEntity.java @@ -22,22 +22,17 @@ import jakarta.persistence.Entity; import jakarta.persistence.Table; import lombok.Data; import lombok.EqualsAndHashCode; -import org.thingsboard.common.util.JacksonUtil; -import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; -import org.thingsboard.server.common.data.cf.CalculatedFieldLinkConfiguration; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.CalculatedFieldLinkId; import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.dao.model.BaseEntity; import org.thingsboard.server.dao.model.BaseSqlEntity; -import org.thingsboard.server.dao.util.mapping.JsonConverter; import java.util.UUID; import static org.thingsboard.server.dao.model.ModelConstants.CALCULATED_FIELD_LINK_CALCULATED_FIELD_ID; -import static org.thingsboard.server.dao.model.ModelConstants.CALCULATED_FIELD_LINK_CONFIGURATION; import static org.thingsboard.server.dao.model.ModelConstants.CALCULATED_FIELD_LINK_ENTITY_ID; import static org.thingsboard.server.dao.model.ModelConstants.CALCULATED_FIELD_LINK_ENTITY_TYPE; import static org.thingsboard.server.dao.model.ModelConstants.CALCULATED_FIELD_LINK_TABLE_NAME; @@ -61,24 +56,10 @@ public class CalculatedFieldLinkEntity extends BaseSqlEntity> findCalculatedFieldLinksByCalculatedFieldIdAsync(TenantId tenantId, CalculatedFieldId calculatedFieldId) { - return service.submit(() -> findCalculatedFieldLinksByCalculatedFieldId(tenantId, calculatedFieldId)); - } - @Override public List findAll() { return DaoUtil.convertDataList(calculatedFieldLinkRepository.findAll()); diff --git a/dao/src/main/resources/sql/schema-entities.sql b/dao/src/main/resources/sql/schema-entities.sql index 7fe57afd58..19551a786a 100644 --- a/dao/src/main/resources/sql/schema-entities.sql +++ b/dao/src/main/resources/sql/schema-entities.sql @@ -947,7 +947,6 @@ CREATE TABLE IF NOT EXISTS calculated_field_link ( entity_type VARCHAR(32), entity_id uuid NOT NULL, calculated_field_id uuid NOT NULL, - configuration varchar(10000), CONSTRAINT fk_calculated_field_id FOREIGN KEY (calculated_field_id) REFERENCES calculated_field(id) ON DELETE CASCADE ); diff --git a/pom.xml b/pom.xml index 052a2e41e7..0f26f4f05c 100755 --- a/pom.xml +++ b/pom.xml @@ -166,7 +166,7 @@ 2.19.0 9.2.0 - 9.4.0 + 9.10.0