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 a86ddde2a1..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 @@ -295,33 +295,17 @@ 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(), multiCallback)) { + if (isMyPartition(msg.getEntityId(), callback)) { log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId()); - getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), multiCallback)); + getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); } } - 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); @@ -335,12 +319,6 @@ 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(); 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 f20df4159d..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 @@ -57,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; @@ -185,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,11 +197,9 @@ public abstract class AbstractCalculatedFieldProcessingService { return switch (relation.direction()) { case FROM -> relations.stream() .map(EntityRelation::getTo) - .filter(CalculatedField::isSupportedRefEntity) .toList(); case TO -> relations.stream() .map(EntityRelation::getFrom) - .filter(CalculatedField::isSupportedRefEntity) .findFirst() .map(List::of) .orElseGet(Collections::emptyList); @@ -224,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/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 12b788435f..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,9 +87,9 @@ public interface RelationService { ListenableFuture> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery); - List findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery); + ListenableFuture> findFilteredRelationsByPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery, Predicate relationFilter); - List findEntityRelations(TenantId tenantId, EntityId entityId); + List findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery); // TODO: This method may be useful for some validations in the future // ListenableFuture checkRecursiveRelation(EntityId from, EntityId to); 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 3fc5a9d5e7..dbde6efd96 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 ? entityRelations.subList(0, limit) : entityRelations; + }, 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 @@ -541,16 +561,6 @@ 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();