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

refactoring: moved cf cache cleanup logic from AbstractConsumerService to DefaultCalculatedFieldCache

pull/15277/head
dashevchenko 7 months ago
parent
7e9a970597
commit
c410b9317f
3 changed files with 96 additions and 93 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java
  2. 154
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java
  3. 33
      application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java

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

@ -61,7 +61,7 @@ public interface CalculatedFieldCache {
void addOwnerEntity(TenantId tenantId, EntityId entityId); void addOwnerEntity(TenantId tenantId, EntityId entityId);
void evictEntity(EntityId entityId); void evictOwnerEntity(EntityId entityId);
void evictOwner(EntityId owner); void evictOwner(EntityId owner);

154
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java
View File

@ -29,8 +29,6 @@ import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedField;
import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.common.data.cf.CalculatedFieldType;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration;
import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.CalculatedFieldId;
@ -40,6 +38,8 @@ import org.thingsboard.server.common.data.id.HasId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.TenantProfileId; import org.thingsboard.server.common.data.id.TenantProfileId;
import org.thingsboard.server.common.data.page.PageDataIterable; import org.thingsboard.server.common.data.page.PageDataIterable;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.dao.cf.CalculatedFieldService; import org.thingsboard.server.dao.cf.CalculatedFieldService;
import org.thingsboard.server.dao.tenant.TbTenantProfileCache; import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.AfterStartUp;
@ -50,8 +50,8 @@ import org.thingsboard.server.service.profile.TbDeviceProfileCache;
import java.util.Collections; import java.util.Collections;
import java.util.HashSet; import java.util.HashSet;
import java.util.List; import java.util.List;
import java.util.Set;
import java.util.Map; import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArrayList;
@ -273,12 +273,12 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
@Override @Override
public void updateOwnerEntity(TenantId tenantId, EntityId entityId) { public void updateOwnerEntity(TenantId tenantId, EntityId entityId) {
evictEntity(entityId); evictOwnerEntity(entityId);
addOwnerEntity(tenantId, entityId); addOwnerEntity(tenantId, entityId);
} }
@Override @Override
public void evictEntity(EntityId entityId) { public void evictOwnerEntity(EntityId entityId) {
ownerEntities.values().forEach(entities -> entities.remove(entityId)); ownerEntities.values().forEach(entities -> entities.remove(entityId));
} }
@ -297,71 +297,105 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
@EventListener(ComponentLifecycleMsg.class) @EventListener(ComponentLifecycleMsg.class)
public void onComponentLifecycleEvent(ComponentLifecycleMsg event) { public void onComponentLifecycleEvent(ComponentLifecycleMsg event) {
if (event.getEvent() != ComponentLifecycleEvent.DELETED) {
return;
}
switch (event.getEntityId().getEntityType()) { switch (event.getEntityId().getEntityType()) {
case TENANT_PROFILE:
if (event.getEvent() == ComponentLifecycleEvent.UPDATED) {
TenantProfileId tenantProfileId = new TenantProfileId(event.getEntityId().getId());
handleTenantProfileUpdate(tenantProfileId);
}
break;
case TENANT: case TENANT:
TenantId tenantId = event.getTenantId(); if (event.getEvent() == ComponentLifecycleEvent.DELETED) {
var removedCfIds = new HashSet<CalculatedFieldId>(); TenantId tenantId = event.getTenantId();
var removedCfEntityIds = new HashSet<EntityId>(); var removedCfIds = new HashSet<CalculatedFieldId>();
var removedLinkEntityIds = new HashSet<EntityId>(); var removedCfEntityIds = new HashSet<EntityId>();
for (Map.Entry<CalculatedFieldId, CalculatedField> entry : calculatedFields.entrySet()) { var removedLinkEntityIds = new HashSet<EntityId>();
CalculatedFieldId cfId = entry.getKey(); for (Map.Entry<CalculatedFieldId, CalculatedField> entry : calculatedFields.entrySet()) {
CalculatedField cf = entry.getValue(); CalculatedFieldId cfId = entry.getKey();
if (cf.getTenantId().equals(tenantId)) { CalculatedField cf = entry.getValue();
calculatedFields.remove(cfId); if (cf.getTenantId().equals(tenantId)) {
List<CalculatedFieldLink> links = calculatedFieldLinks.remove(cfId); calculatedFields.remove(cfId);
if (links != null) { List<CalculatedFieldLink> links = calculatedFieldLinks.remove(cfId);
links.forEach(link -> removedLinkEntityIds.add(link.entityId())); if (links != null) {
links.forEach(link -> removedLinkEntityIds.add(link.entityId()));
}
calculatedFieldsCtx.remove(cfId);
removedCfIds.add(cfId);
removedCfEntityIds.add(cf.getEntityId());
log.debug("[{}] evict calculated field from cache on tenant deletion: {}", cfId, cf);
} }
calculatedFieldsCtx.remove(cfId);
removedCfIds.add(cfId);
removedCfEntityIds.add(cf.getEntityId());
log.debug("[{}] evict calculated field from cache on tenant deletion: {}", cfId, cf);
} }
} removedCfEntityIds.forEach(entityId -> {
removedCfEntityIds.forEach(entityId -> { List<CalculatedField> cfs = entityIdCalculatedFields.get(entityId);
List<CalculatedField> cfs = entityIdCalculatedFields.get(entityId); if (cfs != null) {
if (cfs != null) { cfs.removeIf(cf -> removedCfIds.contains(cf.getId()));
cfs.removeIf(cf -> removedCfIds.contains(cf.getId())); if (cfs.isEmpty()) {
if (cfs.isEmpty()) { entityIdCalculatedFields.remove(entityId);
entityIdCalculatedFields.remove(entityId); }
} }
} });
}); removedLinkEntityIds.forEach(entityId -> {
removedLinkEntityIds.forEach(entityId -> { List<CalculatedFieldLink> entityLinks = entityIdCalculatedFieldLinks.get(entityId);
List<CalculatedFieldLink> entityLinks = entityIdCalculatedFieldLinks.get(entityId); if (entityLinks != null) {
if (entityLinks != null) { entityLinks.removeIf(link -> removedCfIds.contains(link.calculatedFieldId()));
entityLinks.removeIf(link -> removedCfIds.contains(link.calculatedFieldId())); if (entityLinks.isEmpty()) {
if (entityLinks.isEmpty()) { entityIdCalculatedFieldLinks.remove(entityId);
entityIdCalculatedFieldLinks.remove(entityId); }
} }
}
});
removedCfIds.forEach(calculatedFieldFetchLocks::remove);
break;
case DEVICE:
case ASSET:
case DEVICE_PROFILE:
case ASSET_PROFILE:
EntityId entityId = event.getEntityId();
List<CalculatedField> cfs = entityIdCalculatedFields.remove(entityId);
if (cfs != null) {
var cfIds = new HashSet<CalculatedFieldId>();
cfs.forEach(cf -> {
calculatedFields.remove(cf.getId());
calculatedFieldLinks.remove(cf.getId());
calculatedFieldsCtx.remove(cf.getId());
cfIds.add(cf.getId());
log.debug("[{}] evict calculated field from cache on entity deletion: {}", cf.getId(), cf);
}); });
entityIdCalculatedFieldLinks.values().forEach(list -> list.removeIf(link -> cfIds.contains(link.calculatedFieldId()))); removedCfIds.forEach(calculatedFieldFetchLocks::remove);
cfIds.forEach(calculatedFieldFetchLocks::remove); evictOwner(tenantId);
}
break;
case CUSTOMER:
if (event.getEvent().equals(ComponentLifecycleEvent.CREATED)) {
addOwnerEntity(event.getTenantId(), event.getEntityId());
} else if (event.getEvent().equals(ComponentLifecycleEvent.UPDATED) && event.isOwnerChanged()) {
updateOwnerEntity(event.getTenantId(), event.getEntityId());
} else if (event.getEvent() == ComponentLifecycleEvent.DELETED) {
evictOwner(event.getEntityId());
evictOwnerEntity(event.getEntityId());
}
break;
case DEVICE, ASSET:
if (event.getEvent().equals(ComponentLifecycleEvent.CREATED)) {
addOwnerEntity(event.getTenantId(), event.getEntityId());
} else if (event.getEvent().equals(ComponentLifecycleEvent.UPDATED) && event.isOwnerChanged()) {
updateOwnerEntity(event.getTenantId(), event.getEntityId());
} else if (event.getEvent().equals(ComponentLifecycleEvent.DELETED)) {
evictOwnerEntity(event.getEntityId());
evictEntity(event.getEntityId());
} }
entityIdCalculatedFieldLinks.remove(entityId);
break; break;
case DEVICE_PROFILE, ASSET_PROFILE:
evictEntity(event.getEntityId());
break;
case CALCULATED_FIELD:
if (event.getEvent() == ComponentLifecycleEvent.CREATED) {
addCalculatedField(event.getTenantId(), (CalculatedFieldId) event.getEntityId());
} else if (event.getEvent() == ComponentLifecycleEvent.UPDATED) {
updateCalculatedField(event.getTenantId(), (CalculatedFieldId) event.getEntityId());
} else {
evict((CalculatedFieldId) event.getEntityId());
}
}
}
private void evictEntity(EntityId entityId) {
List<CalculatedField> cfs = entityIdCalculatedFields.remove(entityId);
if (cfs != null) {
var cfIds = new HashSet<CalculatedFieldId>();
cfs.forEach(cf -> {
calculatedFields.remove(cf.getId());
calculatedFieldLinks.remove(cf.getId());
calculatedFieldsCtx.remove(cf.getId());
cfIds.add(cf.getId());
log.debug("[{}] evict calculated field from cache on entity deletion: {}", cf.getId(), cf);
});
entityIdCalculatedFieldLinks.values().forEach(list -> list.removeIf(link -> cfIds.contains(link.calculatedFieldId())));
cfIds.forEach(calculatedFieldFetchLocks::remove);
} }
entityIdCalculatedFieldLinks.remove(entityId);
} }
private Lock getFetchLock(CalculatedFieldId id) { private Lock getFetchLock(CalculatedFieldId id) {

33
application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java
View File

@ -26,7 +26,6 @@ import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.AssetProfileId; import org.thingsboard.server.common.data.id.AssetProfileId;
import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.id.DeviceProfileId;
@ -166,7 +165,6 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
tenantProfileCache.evict(tenantProfileId); tenantProfileCache.evict(tenantProfileId);
if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.UPDATED)) { if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.UPDATED)) {
apiUsageStateService.onTenantProfileUpdate(tenantProfileId); apiUsageStateService.onTenantProfileUpdate(tenantProfileId);
calculatedFieldCache.handleTenantProfileUpdate(tenantProfileId);
} }
} else if (EntityType.TENANT.equals(componentLifecycleMsg.getEntityId().getEntityType())) { } else if (EntityType.TENANT.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
if (TenantId.SYS_TENANT_ID.equals(tenantId)) { if (TenantId.SYS_TENANT_ID.equals(tenantId)) {
@ -179,7 +177,6 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
apiUsageStateService.onTenantUpdate(tenantId); apiUsageStateService.onTenantUpdate(tenantId);
} else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.DELETED)) { } else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.DELETED)) {
apiUsageStateService.onTenantDelete(tenantId); apiUsageStateService.onTenantDelete(tenantId);
calculatedFieldCache.evictOwner(tenantId);
partitionService.removeTenant(tenantId); partitionService.removeTenant(tenantId);
} }
} }
@ -187,45 +184,17 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
deviceProfileCache.evict(tenantId, new DeviceProfileId(componentLifecycleMsg.getEntityId().getId())); deviceProfileCache.evict(tenantId, new DeviceProfileId(componentLifecycleMsg.getEntityId().getId()));
} else if (EntityType.DEVICE.equals(componentLifecycleMsg.getEntityId().getEntityType())) { } else if (EntityType.DEVICE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
deviceProfileCache.evict(tenantId, new DeviceId(componentLifecycleMsg.getEntityId().getId())); deviceProfileCache.evict(tenantId, new DeviceId(componentLifecycleMsg.getEntityId().getId()));
if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.CREATED)) {
calculatedFieldCache.addOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
} else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.UPDATED) && componentLifecycleMsg.isOwnerChanged()) {
calculatedFieldCache.updateOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
} else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.DELETED)) {
calculatedFieldCache.evictEntity(componentLifecycleMsg.getEntityId());
}
} else if (EntityType.ASSET_PROFILE.equals(componentLifecycleMsg.getEntityId().getEntityType())) { } else if (EntityType.ASSET_PROFILE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
assetProfileCache.evict(tenantId, new AssetProfileId(componentLifecycleMsg.getEntityId().getId())); assetProfileCache.evict(tenantId, new AssetProfileId(componentLifecycleMsg.getEntityId().getId()));
} else if (EntityType.ASSET.equals(componentLifecycleMsg.getEntityId().getEntityType())) { } else if (EntityType.ASSET.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
assetProfileCache.evict(tenantId, new AssetId(componentLifecycleMsg.getEntityId().getId())); assetProfileCache.evict(tenantId, new AssetId(componentLifecycleMsg.getEntityId().getId()));
if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.CREATED)) {
calculatedFieldCache.addOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
} else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.UPDATED) && componentLifecycleMsg.isOwnerChanged()) {
calculatedFieldCache.updateOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
} else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.DELETED)) {
calculatedFieldCache.evictEntity(componentLifecycleMsg.getEntityId());
}
} else if (EntityType.ENTITY_VIEW.equals(componentLifecycleMsg.getEntityId().getEntityType())) { } else if (EntityType.ENTITY_VIEW.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
actorContext.getTbEntityViewService().onComponentLifecycleMsg(componentLifecycleMsg); actorContext.getTbEntityViewService().onComponentLifecycleMsg(componentLifecycleMsg);
} else if (EntityType.API_USAGE_STATE.equals(componentLifecycleMsg.getEntityId().getEntityType())) { } else if (EntityType.API_USAGE_STATE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
apiUsageStateService.onApiUsageStateUpdate(tenantId); apiUsageStateService.onApiUsageStateUpdate(tenantId);
} else if (EntityType.CUSTOMER.equals(componentLifecycleMsg.getEntityId().getEntityType())) { } else if (EntityType.CUSTOMER.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.CREATED)) { if (componentLifecycleMsg.getEvent() == ComponentLifecycleEvent.DELETED) {
calculatedFieldCache.addOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
} else if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.UPDATED) && componentLifecycleMsg.isOwnerChanged()) {
calculatedFieldCache.updateOwnerEntity(tenantId, componentLifecycleMsg.getEntityId());
} else if (componentLifecycleMsg.getEvent() == ComponentLifecycleEvent.DELETED) {
apiUsageStateService.onCustomerDelete((CustomerId) componentLifecycleMsg.getEntityId()); apiUsageStateService.onCustomerDelete((CustomerId) componentLifecycleMsg.getEntityId());
calculatedFieldCache.evictOwner(componentLifecycleMsg.getEntityId());
calculatedFieldCache.evictEntity(componentLifecycleMsg.getEntityId());
}
} else if (EntityType.CALCULATED_FIELD.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
if (componentLifecycleMsg.getEvent() == ComponentLifecycleEvent.CREATED) {
calculatedFieldCache.addCalculatedField(tenantId, (CalculatedFieldId) componentLifecycleMsg.getEntityId());
} else if (componentLifecycleMsg.getEvent() == ComponentLifecycleEvent.UPDATED) {
calculatedFieldCache.updateCalculatedField(tenantId, (CalculatedFieldId) componentLifecycleMsg.getEntityId());
} else {
calculatedFieldCache.evict((CalculatedFieldId) componentLifecycleMsg.getEntityId());
} }
} else if (EntityType.TB_RESOURCE.equals(componentLifecycleMsg.getEntityId().getEntityType())) { } else if (EntityType.TB_RESOURCE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
tbResourceDataCache.evictResourceData(tenantId, new TbResourceId(componentLifecycleMsg.getEntityId().getId())); tbResourceDataCache.evictResourceData(tenantId, new TbResourceId(componentLifecycleMsg.getEntityId().getId()));

Write Preview
Loading…
Cancel
Save