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..ec4598947a 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; @@ -320,12 +323,16 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware 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 +346,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 +724,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..c7bbf83066 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; @@ -56,6 +57,7 @@ import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutionException; +import java.util.function.Predicate; import java.util.stream.Collectors; import static org.thingsboard.server.common.data.cf.CalculatedFieldType.PROPAGATION; @@ -184,7 +186,8 @@ public abstract class AbstractCalculatedFieldProcessingService { } private ListenableFuture> resolveRelatedEntities(TenantId tenantId, EntityId entityId, RelationPathLevel relation) { - ListenableFuture> relationsFut = relationService.findByRelationPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation))); + Predicate filter = entityRelation -> CalculatedField.isSupportedRefEntity(entityRelation.getFrom()) && CalculatedField.isSupportedRefEntity(entityRelation.getTo()); + ListenableFuture> relationsFut = relationService.findFilteredRelationsByPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation)), filter); return Futures.transform(relationsFut, relations -> { if (relations == null) { @@ -195,7 +198,11 @@ public abstract class AbstractCalculatedFieldProcessingService { case FROM -> relations.stream() .map(EntityRelation::getTo) .toList(); - case TO -> relations.isEmpty() ? List.of() : List.of(relations.get(0).getFrom()); + case TO -> relations.stream() + .map(EntityRelation::getFrom) + .findFirst() + .map(List::of) + .orElseGet(Collections::emptyList); }; }, calculatedFieldCallbackExecutor); } @@ -217,7 +224,8 @@ public abstract class AbstractCalculatedFieldProcessingService { case CURRENT_OWNER -> Futures.immediateFuture(List.of(resolveOwnerArgument(tenantId, entityId))); case RELATION_PATH_QUERY -> { var configuration = (RelationPathQueryDynamicSourceConfiguration) refDynamicSourceConfiguration; - yield Futures.transform(relationService.findByRelationPathQueryAsync(tenantId, configuration.toRelationPathQuery(entityId)), + Predicate filter = entityRelation -> CalculatedField.isSupportedRefEntity(entityRelation.getFrom()) && CalculatedField.isSupportedRefEntity(entityRelation.getTo()); + yield Futures.transform(relationService.findFilteredRelationsByPathQueryAsync(tenantId, configuration.toRelationPathQuery(entityId), filter), configuration::resolveEntityIds, 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..aaa52f99a7 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 @@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.relation.RelationTypeGroup; import org.thingsboard.server.common.data.rule.RuleChainType; import java.util.List; +import java.util.function.Predicate; /** * Created by ashvayka on 27.04.17. @@ -86,6 +87,8 @@ public interface RelationService { ListenableFuture> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery); + ListenableFuture> findFilteredRelationsByPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery, Predicate relationFilter); + List findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery); // TODO: This method may be useful for some validations in the future 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..ffb800c7f7 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 @@ -71,6 +71,7 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; +import java.util.function.Predicate; import static org.thingsboard.server.dao.service.Validator.validateId; import static org.thingsboard.server.dao.service.Validator.validatePositiveNumber; @@ -507,6 +508,11 @@ public class BaseRelationService implements RelationService { @Override public ListenableFuture> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery) { + return findFilteredRelationsByPathQueryAsync(tenantId, relationPathQuery, null); + } + + @Override + public ListenableFuture> findFilteredRelationsByPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery, Predicate relationFilter) { log.trace("Executing findByRelationPathQuery, tenantId [{}], relationPathQuery {}", tenantId, relationPathQuery); validateId(tenantId, id -> "Invalid tenant id: " + id); validate(relationPathQuery); @@ -518,10 +524,24 @@ public class BaseRelationService implements RelationService { case FROM -> findByFromAndTypeAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON); case TO -> findByToAndTypeAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON); }; - return Futures.transform(relationsFuture, entityRelations -> entityRelations.size() > limit ? - entityRelations.subList(0, limit) : entityRelations, MoreExecutors.directExecutor()); + return Futures.transform(relationsFuture, entityRelations -> { + if (entityRelations == null || entityRelations.isEmpty()) { + return Collections.emptyList(); + } + List relations = relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations; + return relations.size() > limit ? relations.subList(0, limit) : relations; + }, MoreExecutors.directExecutor()); } - return executor.submit(() -> relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit)); + return executor.submit(() -> { + List entityRelations = relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit); + return relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations; + }); + } + + private List filterRelations(List entityRelations, Predicate relationFilter) { + return entityRelations.stream() + .filter(relationFilter) + .toList(); } @Override