tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge pull request #14291 from irynamatveieva/fix/relation-update-event 

Fixed relation update event processing
pull/14303/head
Viacheslav Klimov 12 months ago
committed by GitHub
parent
243ba16cc3 da61794ec5
commit
3b5c240eea
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
8 changed files with 119 additions and 16 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 15
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
  2. 14
      application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java
  3. 2
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java
  4. 13
      application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java
  5. 58
      application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java
  6. 3
      common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java
  7. 4
      common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java
  8. 26
      dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java

15
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
View File

@ -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<CalculatedFieldCtx> 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;
};
}

14
application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java
View File

@ -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<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 -> {
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<EntityRelation> filter = entityRelation -> CalculatedField.isSupportedRefEntity(entityRelation.getFrom()) && CalculatedField.isSupportedRefEntity(entityRelation.getTo());
yield Futures.transform(relationService.findFilteredRelationsByPathQueryAsync(tenantId, configuration.toRelationPathQuery(entityId), filter),
configuration::resolveEntityIds, calculatedFieldCallbackExecutor);
}
};

2
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java
View File

@ -162,7 +162,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
}
private boolean checkEntityForCalculatedFields(TenantId tenantId, EntityId entityId, Predicate<CalculatedFieldCtx> filter, Predicate<CalculatedFieldCtx> linkedEntityFilter, Predicate<CalculatedFieldCtx> dynamicSourceFilter, Predicate<CalculatedFieldCtx> relatedEntityFilter) {
if (!CalculatedField.SUPPORTED_REFERENCED_ENTITIES.contains(entityId.getEntityType())) {
if (!CalculatedField.isSupportedRefEntity(entityId)) {
return false;
}

13
application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java
View File

@ -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);
}
}
}

58
application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java
View File

@ -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());

3
common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java
View File

@ -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<List<EntityRelation>> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery);
ListenableFuture<List<EntityRelation>> findFilteredRelationsByPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery, Predicate<EntityRelation> relationFilter);
List<EntityRelation> findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery);
// TODO: This method may be useful for some validations in the future

4
common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java
View File

@ -65,6 +65,10 @@ public class CalculatedField extends BaseData<CalculatedFieldId> 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;

26
dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java
View File

@ -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<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);
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<EntityRelation> 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<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

Write Preview
Loading…
Cancel
Save