Browse Source

added check for supported entities

pull/14291/head
IrynaMatveieva 11 months ago
parent
commit
87e7a14bdb
  1. 26
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
  2. 9
      application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java
  3. 5
      common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java
  4. 36
      dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java

26
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) { 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()) { switch (msg.getEntityId().getEntityType()) {
case DEVICE, ASSET -> entityProfileCache.removeEntityId(msg.getEntityId()); case DEVICE, ASSET -> entityProfileCache.removeEntityId(msg.getEntityId());
case CUSTOMER -> ownerEntities.remove(msg.getEntityId()); case CUSTOMER -> ownerEntities.remove(msg.getEntityId());
} }
ownerEntities.values().forEach(entities -> entities.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()); 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<EntityRelation> entityRelations = relationService.findEntityRelations(tenantId, msg.getEntityId());
if (entityRelations.isEmpty()) {
callback.onSuccess();
}
entityRelations.forEach(entityRelation -> {
Function<EntityId, TriConsumer<EntityId, CalculatedFieldCtx, TbCallback>> deleteAction =
relatedId -> (entityId, ctx, cb) -> deleteRelatedEntity(entityId, relatedId, ctx, cb);
processRelationIfSupported(entityRelation, callback, deleteAction);
});
}
private void onRelationChangedEvent(ComponentLifecycleMsg msg, TbCallback callback) { private void onRelationChangedEvent(ComponentLifecycleMsg msg, TbCallback callback) {
Function<EntityId, TriConsumer<EntityId, CalculatedFieldCtx, TbCallback>> relationAction = switch (msg.getEvent()) { Function<EntityId, TriConsumer<EntityId, CalculatedFieldCtx, TbCallback>> relationAction = switch (msg.getEvent()) {
case RELATION_UPDATED -> relatedId -> (entityId, ctx, cb) -> initRelatedEntity(entityId, relatedId, ctx, cb); 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); EntityRelation entityRelation = JacksonUtil.treeToValue(msg.getInfo(), EntityRelation.class);
processRelationIfSupported(entityRelation, callback, relationAction);
}
private void processRelationIfSupported(EntityRelation entityRelation,
TbCallback callback,
Function<EntityId, TriConsumer<EntityId, CalculatedFieldCtx, TbCallback>> relationAction) {
EntityId toId = entityRelation.getTo(); EntityId toId = entityRelation.getTo();
EntityId fromId = entityRelation.getFrom(); EntityId fromId = entityRelation.getFrom();
String relationType = entityRelation.getType(); String relationType = entityRelation.getType();

9
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.Optional;
import java.util.Set; import java.util.Set;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.function.Predicate;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import static org.thingsboard.server.common.data.cf.CalculatedFieldType.PROPAGATION; import static org.thingsboard.server.common.data.cf.CalculatedFieldType.PROPAGATION;
@ -185,7 +186,8 @@ public abstract class AbstractCalculatedFieldProcessingService {
} }
private ListenableFuture<List<EntityId>> resolveRelatedEntities(TenantId tenantId, EntityId entityId, RelationPathLevel relation) { private ListenableFuture<List<EntityId>> resolveRelatedEntities(TenantId tenantId, EntityId entityId, RelationPathLevel relation) {
ListenableFuture<List<EntityRelation>> relationsFut = relationService.findByRelationPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation))); Predicate<EntityRelation> filter = entityRelation -> CalculatedField.isSupportedRefEntity(entityRelation.getFrom()) && CalculatedField.isSupportedRefEntity(entityRelation.getTo());
ListenableFuture<List<EntityRelation>> relationsFut = relationService.findFilteredRelationsByPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation)), filter);
return Futures.transform(relationsFut, relations -> { return Futures.transform(relationsFut, relations -> {
if (relations == null) { if (relations == null) {
@ -195,11 +197,9 @@ public abstract class AbstractCalculatedFieldProcessingService {
return switch (relation.direction()) { return switch (relation.direction()) {
case FROM -> relations.stream() case FROM -> relations.stream()
.map(EntityRelation::getTo) .map(EntityRelation::getTo)
.filter(CalculatedField::isSupportedRefEntity)
.toList(); .toList();
case TO -> relations.stream() case TO -> relations.stream()
.map(EntityRelation::getFrom) .map(EntityRelation::getFrom)
.filter(CalculatedField::isSupportedRefEntity)
.findFirst() .findFirst()
.map(List::of) .map(List::of)
.orElseGet(Collections::emptyList); .orElseGet(Collections::emptyList);
@ -224,7 +224,8 @@ public abstract class AbstractCalculatedFieldProcessingService {
case CURRENT_OWNER -> Futures.immediateFuture(List.of(resolveOwnerArgument(tenantId, entityId))); case CURRENT_OWNER -> Futures.immediateFuture(List.of(resolveOwnerArgument(tenantId, entityId)));
case RELATION_PATH_QUERY -> { case RELATION_PATH_QUERY -> {
var configuration = (RelationPathQueryDynamicSourceConfiguration) refDynamicSourceConfiguration; var configuration = (RelationPathQueryDynamicSourceConfiguration) refDynamicSourceConfiguration;
yield Futures.transform(relationService.findByRelationPathQueryAsync(tenantId, configuration.toRelationPathQuery(entityId)), Predicate<EntityRelation> filter = entityRelation -> CalculatedField.isSupportedRefEntity(entityRelation.getFrom()) && CalculatedField.isSupportedRefEntity(entityRelation.getTo());
yield Futures.transform(relationService.findFilteredRelationsByPathQueryAsync(tenantId, configuration.toRelationPathQuery(entityId), filter),
configuration::resolveEntityIds, calculatedFieldCallbackExecutor); configuration::resolveEntityIds, calculatedFieldCallbackExecutor);
} }
}; };

5
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 org.thingsboard.server.common.data.rule.RuleChainType;
import java.util.List; import java.util.List;
import java.util.function.Predicate;
/** /**
* Created by ashvayka on 27.04.17. * Created by ashvayka on 27.04.17.
@ -86,9 +87,9 @@ public interface RelationService {
ListenableFuture<List<EntityRelation>> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery); ListenableFuture<List<EntityRelation>> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery);
List<EntityRelation> findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery); ListenableFuture<List<EntityRelation>> findFilteredRelationsByPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery, Predicate<EntityRelation> relationFilter);
List<EntityRelation> findEntityRelations(TenantId tenantId, EntityId entityId); List<EntityRelation> findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery);
// TODO: This method may be useful for some validations in the future // TODO: This method may be useful for some validations in the future
// ListenableFuture<Boolean> checkRecursiveRelation(EntityId from, EntityId to); // ListenableFuture<Boolean> checkRecursiveRelation(EntityId from, EntityId to);

36
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.ScheduledExecutorService;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer; 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.validateId;
import static org.thingsboard.server.dao.service.Validator.validatePositiveNumber; import static org.thingsboard.server.dao.service.Validator.validatePositiveNumber;
@ -507,6 +508,11 @@ public class BaseRelationService implements RelationService {
@Override @Override
public ListenableFuture<List<EntityRelation>> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery) { public ListenableFuture<List<EntityRelation>> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery) {
return findFilteredRelationsByPathQueryAsync(tenantId, relationPathQuery, null);
}
@Override
public ListenableFuture<List<EntityRelation>> findFilteredRelationsByPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery, Predicate<EntityRelation> relationFilter) {
log.trace("Executing findByRelationPathQuery, tenantId [{}], relationPathQuery {}", tenantId, relationPathQuery); log.trace("Executing findByRelationPathQuery, tenantId [{}], relationPathQuery {}", tenantId, relationPathQuery);
validateId(tenantId, id -> "Invalid tenant id: " + id); validateId(tenantId, id -> "Invalid tenant id: " + id);
validate(relationPathQuery); validate(relationPathQuery);
@ -518,10 +524,24 @@ public class BaseRelationService implements RelationService {
case FROM -> findByFromAndTypeAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON); case FROM -> findByFromAndTypeAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON);
case TO -> findByToAndTypeAsync(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 ? return Futures.transform(relationsFuture, entityRelations -> {
entityRelations.subList(0, limit) : entityRelations, MoreExecutors.directExecutor()); if (entityRelations == null || entityRelations.isEmpty()) {
return Collections.emptyList();
}
List<EntityRelation> 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<EntityRelation> entityRelations = relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit);
return relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations;
});
}
private List<EntityRelation> filterRelations(List<EntityRelation> entityRelations, Predicate<EntityRelation> relationFilter) {
return entityRelations.stream()
.filter(relationFilter)
.toList();
} }
@Override @Override
@ -541,16 +561,6 @@ public class BaseRelationService implements RelationService {
return relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit); return relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit);
} }
@Override
public List<EntityRelation> findEntityRelations(TenantId tenantId, EntityId entityId) {
List<EntityRelation> 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) { private void validate(EntityRelationPathQuery relationPathQuery) {
validateId((UUIDBased) relationPathQuery.rootEntityId(), id -> "Invalid root entity id: " + id); validateId((UUIDBased) relationPathQuery.rootEntityId(), id -> "Invalid root entity id: " + id);
List<RelationPathLevel> levels = relationPathQuery.levels(); List<RelationPathLevel> levels = relationPathQuery.levels();

Loading…
Cancel
Save