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 8f632647c5..a86ddde2a1 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 @@ -27,9 +27,11 @@ import org.thingsboard.server.actors.service.DefaultActorService; import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.asset.AssetProfile; import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; @@ -74,6 +76,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ScheduledFuture; @@ -292,17 +295,33 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) { + // 2 = 1 for entity processing + 1 for relation processing + MultipleTbCallback multiCallback = new MultipleTbCallback(2, callback); + deleteEntityRelations(msg, multiCallback); switch (msg.getEntityId().getEntityType()) { case DEVICE, ASSET -> entityProfileCache.removeEntityId(msg.getEntityId()); case CUSTOMER -> ownerEntities.remove(msg.getEntityId()); } ownerEntities.values().forEach(entities -> entities.remove(msg.getEntityId())); - if (isMyPartition(msg.getEntityId(), callback)) { + if (isMyPartition(msg.getEntityId(), multiCallback)) { log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId()); - getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); + getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), multiCallback)); } } + private void deleteEntityRelations(ComponentLifecycleMsg msg, TbCallback callback) { + List entityRelations = relationService.findEntityRelations(tenantId, msg.getEntityId()); + if (entityRelations.isEmpty()) { + callback.onSuccess(); + } + entityRelations.forEach(entityRelation -> { + Function> deleteAction = + relatedId -> (entityId, ctx, cb) -> deleteRelatedEntity(entityId, relatedId, ctx, cb); + + processRelationIfSupported(entityRelation, callback, deleteAction); + }); + } + private void onRelationChangedEvent(ComponentLifecycleMsg msg, TbCallback callback) { Function> relationAction = switch (msg.getEvent()) { case RELATION_UPDATED -> relatedId -> (entityId, ctx, cb) -> initRelatedEntity(entityId, relatedId, ctx, cb); @@ -316,16 +335,26 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } EntityRelation entityRelation = JacksonUtil.treeToValue(msg.getInfo(), EntityRelation.class); + processRelationIfSupported(entityRelation, callback, relationAction); + } + + private void processRelationIfSupported(EntityRelation entityRelation, + TbCallback callback, + Function> relationAction) { EntityId toId = entityRelation.getTo(); EntityId fromId = entityRelation.getFrom(); String relationType = entityRelation.getType(); + if (!(CalculatedField.isSupportedRefEntity(toId) || CalculatedField.isSupportedRefEntity(fromId))) { + callback.onSuccess(); + return; + } + MultipleTbCallback callbackForToAndFrom = new MultipleTbCallback(2, callback); processRelationByDirection(EntitySearchDirection.TO, relationType, toId, callbackForToAndFrom, relationAction.apply(fromId)); processRelationByDirection(EntitySearchDirection.FROM, relationType, fromId, callbackForToAndFrom, relationAction.apply(toId)); } - private void processRelationByDirection(EntitySearchDirection direction, String relationType, EntityId mainId, @@ -339,7 +368,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware List matchingCfs = cfsByEntityIdAndProfile.stream() .filter(cf -> { - if (cf.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config ) { + if (cf.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config) { RelationPathLevel relation = config.getRelation(); return direction.equals(relation.direction()) && relationType.equals(relation.relationType()); } else { @@ -717,8 +746,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private EntityId getProfileId(TenantId tenantId, EntityId entityId) { return switch (entityId.getEntityType()) { - case ASSET -> assetProfileCache.get(tenantId, (AssetId) entityId).getId(); - case DEVICE -> deviceProfileCache.get(tenantId, (DeviceId) entityId).getId(); + case ASSET -> Optional.ofNullable(assetProfileCache.get(tenantId, (AssetId) entityId)).map(AssetProfile::getId).orElse(null); + case DEVICE -> Optional.ofNullable(deviceProfileCache.get(tenantId, (DeviceId) entityId)).map(DeviceProfile::getId).orElse(null); default -> null; }; } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java index 945792ebcd..f20df4159d 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java @@ -24,6 +24,7 @@ import jakarta.annotation.PreDestroy; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.server.common.data.cf.CalculatedField; 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.RelationPathQueryDynamicSourceConfiguration; @@ -194,8 +195,14 @@ public abstract class AbstractCalculatedFieldProcessingService { return switch (relation.direction()) { case FROM -> relations.stream() .map(EntityRelation::getTo) + .filter(CalculatedField::isSupportedRefEntity) .toList(); - case TO -> relations.isEmpty() ? List.of() : List.of(relations.get(0).getFrom()); + case TO -> relations.stream() + .map(EntityRelation::getFrom) + .filter(CalculatedField::isSupportedRefEntity) + .findFirst() + .map(List::of) + .orElseGet(Collections::emptyList); }; }, calculatedFieldCallbackExecutor); } 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 c7e369a862..8f5d857d74 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 @@ -162,7 +162,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS } private boolean checkEntityForCalculatedFields(TenantId tenantId, EntityId entityId, Predicate filter, Predicate linkedEntityFilter, Predicate dynamicSourceFilter, Predicate relatedEntityFilter) { - if (!CalculatedField.SUPPORTED_REFERENCED_ENTITIES.contains(entityId.getEntityType())) { + if (!CalculatedField.isSupportedRefEntity(entityId)) { return false; } diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java b/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java index 48a318f981..9a6a96155f 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java +++ b/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java @@ -29,7 +29,6 @@ import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; -import org.thingsboard.server.common.data.ObjectType; import org.thingsboard.server.common.data.TbResource; import org.thingsboard.server.common.data.TbResourceInfo; import org.thingsboard.server.common.data.Tenant; @@ -49,6 +48,7 @@ import org.thingsboard.server.common.data.job.Job; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.notification.NotificationRequest; import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; +import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.rule.RuleChainType; import org.thingsboard.server.common.data.security.DeviceCredentials; @@ -274,10 +274,13 @@ public class EntityStateSourcingListener { @TransactionalEventListener(fallbackExecution = true) public void handleEvent(RelationActionEvent relationEvent) { - if (relationEvent.getActionType() == ActionType.RELATION_ADD_OR_UPDATE) { - tbClusterService.onRelationUpdated(relationEvent.getTenantId(), relationEvent.getRelation(), TbQueueCallback.EMPTY); - } else if (relationEvent.getActionType() == ActionType.RELATION_DELETED) { - tbClusterService.onRelationDeleted(relationEvent.getTenantId(), relationEvent.getRelation(), TbQueueCallback.EMPTY); + EntityRelation relation = relationEvent.getRelation(); + if (CalculatedField.isSupportedRefEntity(relation.getFrom()) && CalculatedField.isSupportedRefEntity(relation.getTo())) { + if (relationEvent.getActionType() == ActionType.RELATION_ADD_OR_UPDATE) { + tbClusterService.onRelationUpdated(relationEvent.getTenantId(), relation, TbQueueCallback.EMPTY); + } else if (relationEvent.getActionType() == ActionType.RELATION_DELETED) { + tbClusterService.onRelationDeleted(relationEvent.getTenantId(), relation, TbQueueCallback.EMPTY); + } } } diff --git a/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java b/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java index a7b67baf3f..f4eb2b5167 100644 --- a/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java @@ -51,6 +51,7 @@ import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.EntitySearchDirection; import org.thingsboard.server.common.data.relation.RelationPathLevel; import org.thingsboard.server.common.data.relation.RelationTypeGroup; +import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.controller.AbstractControllerTest; import org.thingsboard.server.dao.service.DaoSqlTest; @@ -299,6 +300,45 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr }); } + @Test + public void testCreateCfAndRelationToRuleChain_checkAggregation() throws Exception { + Asset asset2 = createAsset("Asset 2", assetProfile.getId()); + Device device3 = createDevice("Device 3", "1234567890333"); + postTelemetry(device3.getId(), "{\"occupied\":true}"); + + RuleChain ruleChain = new RuleChain(); + ruleChain.setName("RuleChain"); + ruleChain = doPost("/api/ruleChain", ruleChain, RuleChain.class); + postTelemetry(ruleChain.getId(), "{\"occupied\":true}"); + + createEntityRelation(asset2.getId(), device3.getId(), "Contains"); + createEntityRelation(asset2.getId(), ruleChain.getId(), "Contains"); + + createOccupancyCF(asset2.getId()); + + await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset2.getId(), Map.of( + "freeSpaces", "0", + "occupiedSpaces", "1", + "totalSpaces", "1" + )); + }); + + postTelemetry(ruleChain.getId(), "{\"occupied\":true}"); + + await().alias("update telemetry on rule chain and no aggregation performed").atMost(deduplicationInterval, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset2.getId(), Map.of( + "freeSpaces", "0", + "occupiedSpaces", "1", + "totalSpaces", "1" + )); + }); + } + @Test public void testDeleteCf_checkNoAggregation() throws Exception { CalculatedField cf = createOccupancyCF(asset.getId()); @@ -466,6 +506,24 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr }); } + @Test + public void testDeleteEntityByRelation_checkAggregation() throws Exception { + createOccupancyCF(asset.getId()); + checkInitialCalculation(); + + doDelete("/api/device/" + device1.getId()).andExpect(status().isOk()); + + await().alias("create relation and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of( + "freeSpaces", "1", + "occupiedSpaces", "0", + "totalSpaces", "1" + )); + }); + } + @Test public void testUpdateRelationPath_checkAggregation() throws Exception { CalculatedField cf = createOccupancyCF(asset.getId()); diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java index 214dc5247e..12b788435f 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java @@ -88,6 +88,8 @@ public interface RelationService { List findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery); + List findEntityRelations(TenantId tenantId, EntityId entityId); + // TODO: This method may be useful for some validations in the future // ListenableFuture checkRecursiveRelation(EntityId from, EntityId to); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java index 0d2543d4c2..8c5adefcf8 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java @@ -65,6 +65,10 @@ public class CalculatedField extends BaseData implements HasN EntityType.DEVICE, EntityType.ASSET, EntityType.CUSTOMER, EntityType.TENANT )); + public static boolean isSupportedRefEntity(EntityId entity) { + return SUPPORTED_REFERENCED_ENTITIES.contains(entity.getEntityType()); + } + private TenantId tenantId; private EntityId entityId; diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java index ff0d09ef2c..3fc5a9d5e7 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java @@ -541,6 +541,16 @@ public class BaseRelationService implements RelationService { return relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit); } + @Override + public List findEntityRelations(TenantId tenantId, EntityId entityId) { + List relations = new ArrayList<>(); + for (RelationTypeGroup relationTypeGroup : RelationTypeGroup.values()) { + relations.addAll(findByFrom(tenantId, entityId, relationTypeGroup)); + relations.addAll(findByTo(tenantId, entityId, relationTypeGroup)); + } + return relations; + } + private void validate(EntityRelationPathQuery relationPathQuery) { validateId((UUIDBased) relationPathQuery.rootEntityId(), id -> "Invalid root entity id: " + id); List levels = relationPathQuery.levels();