From 407511be52b1f7de51a74a62cf6d58d7c4ff12aa Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Wed, 5 Feb 2025 12:51:50 +0200 Subject: [PATCH] Cache implementation --- .../server/actors/ActorSystemContext.java | 6 + ...alculatedFieldManagerMessageProcessor.java | 44 ++---- .../service/cf/CalculatedFieldCache.java | 4 - .../cf/DefaultCalculatedFieldCache.java | 60 ++------- ...efaultCalculatedFieldExecutionService.java | 125 ------------------ .../cf/DefaultCalculatedFieldInitService.java | 36 ++--- .../CalculatedFieldEntityProfileCache.java | 35 +++++ ...aultCalculatedFieldEntityProfileCache.java | 86 ++++++++++++ .../cf/cache/TenantEntityProfileCache.java | 118 +++++++++++++++++ .../server/dao/asset/AssetService.java | 3 + .../server/dao/device/DeviceService.java | 3 + .../common/data/ProfileEntityIdInfo.java | 55 ++++++++ .../server/queue/util/AfterStartUp.java | 7 +- .../server/dao/asset/AssetDao.java | 3 + .../server/dao/asset/BaseAssetService.java | 8 ++ .../server/dao/device/DeviceDao.java | 4 + .../server/dao/device/DeviceServiceImpl.java | 8 ++ .../server/dao/sql/asset/JpaAssetDao.java | 12 ++ .../sql/device/AbstractNativeRepository.java | 54 ++++++++ .../device/DefaultNativeAssetRepository.java | 55 ++++++++ .../device/DefaultNativeDeviceRepository.java | 52 ++++---- .../server/dao/sql/device/JpaDeviceDao.java | 7 + .../dao/sql/device/NativeAssetRepository.java | 20 +++ .../sql/device/NativeDeviceRepository.java | 4 +- .../device/NativeProfileEntityRepository.java | 26 ++++ 25 files changed, 578 insertions(+), 257 deletions(-) create mode 100644 application/src/main/java/org/thingsboard/server/service/cf/cache/CalculatedFieldEntityProfileCache.java create mode 100644 application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java create mode 100644 application/src/main/java/org/thingsboard/server/service/cf/cache/TenantEntityProfileCache.java create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/ProfileEntityIdInfo.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/device/AbstractNativeRepository.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeAssetRepository.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeAssetRepository.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeProfileEntityRepository.java diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index daa6744fd9..f581d0b266 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -106,6 +106,7 @@ import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import org.thingsboard.server.service.cf.CalculatedFieldExecutionService; +import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.component.ComponentDiscoveryService; import org.thingsboard.server.service.edge.rpc.EdgeRpcService; @@ -516,6 +517,11 @@ public class ActorSystemContext { @Getter private CalculatedFieldExecutionService calculatedFieldExecutionService; + @Lazy + @Autowired(required = false) + @Getter + private CalculatedFieldEntityProfileCache calculatedFieldEntityProfileCache; + @Value("${actors.session.max_concurrent_sessions_per_device:1}") @Getter private long maxConcurrentSessionsPerDevice; 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 b2ff883236..cd2e5e876f 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 @@ -26,14 +26,11 @@ import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.id.AssetId; -import org.thingsboard.server.common.data.id.AssetProfileId; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.DeviceId; -import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.page.PageDataIterable; import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg; @@ -42,6 +39,7 @@ import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.dao.cf.CalculatedFieldService; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; import org.thingsboard.server.service.cf.CalculatedFieldExecutionService; +import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.profile.TbAssetProfileCache; @@ -50,7 +48,6 @@ import org.thingsboard.server.service.profile.TbDeviceProfileCache; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; -import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -69,19 +66,19 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private final Map calculatedFields = new HashMap<>(); private final Map> entityIdCalculatedFields = new ConcurrentHashMap<>(); private final ConcurrentMap> entityIdCalculatedFieldLinks = new ConcurrentHashMap<>(); - private final ConcurrentMap> profileEntities = new ConcurrentHashMap<>(); private final CalculatedFieldExecutionService cfExecService; + private final CalculatedFieldEntityProfileCache cfEntityCache; private final CalculatedFieldService cfDaoService; private final TbAssetProfileCache assetProfileCache; private final TbDeviceProfileCache deviceProfileCache; protected TbActorCtx ctx; final TenantId tenantId; - private final static int initFetchPackSize = 1024; CalculatedFieldManagerMessageProcessor(ActorSystemContext systemContext, TenantId tenantId) { super(systemContext); + this.cfEntityCache = systemContext.getCalculatedFieldEntityProfileCache(); this.cfExecService = systemContext.getCalculatedFieldExecutionService(); this.cfDaoService = systemContext.getCalculatedFieldService(); this.assetProfileCache = systemContext.getAssetProfileCache(); @@ -173,8 +170,10 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private void onEntityCreated(ComponentLifecycleMsg msg, TbCallback callback) { EntityId entityId = msg.getEntityId(); + EntityId profileId = getProfileId(tenantId, entityId); + cfEntityCache.add(tenantId, entityId, profileId); var entityIdFields = getCalculatedFieldsByEntityId(entityId); - var profileIdFields = getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId)); + var profileIdFields = getCalculatedFieldsByEntityId(profileId); var fieldsCount = entityIdFields.size() + profileIdFields.size(); if (fieldsCount > 0) { MultipleTbCallback multiCallback = new MultipleTbCallback(fieldsCount, callback); @@ -187,6 +186,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private void onEntityUpdated(ComponentLifecycleMsg msg, TbCallback callback) { if (msg.getOldProfileId() != null && msg.getOldProfileId() != msg.getProfileId()) { + cfEntityCache.update(tenantId, msg.getOldProfileId(), msg.getProfileId(), msg.getEntityId()); var oldProfileCfs = getCalculatedFieldsByEntityId(msg.getOldProfileId()); var newProfileCfs = getCalculatedFieldsByEntityId(msg.getProfileId()); var fieldsCount = oldProfileCfs.size() + newProfileCfs.size(); @@ -202,6 +202,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) { + cfEntityCache.evict(tenantId, msg.getEntityId()); getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); } @@ -288,7 +289,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware EntityId entityId = cfCtx.getEntityId(); EntityType entityType = cfCtx.getEntityId().getEntityType(); if (isProfileEntity(entityType)) { - var entityIds = getEntitiesByProfile(entityId); + var entityIds = cfEntityCache.getMyEntityIdsByProfileId(tenantId, entityId); if (!entityIds.isEmpty()) { //TODO: no need to do this if we cache all created actors and know which one belong to us; var multiCallback = new MultipleTbCallback(entityIds.size(), callback); @@ -335,7 +336,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware var cf = calculatedFields.get(link.cfId()); if (EntityType.DEVICE_PROFILE.equals(targetEntityType) || EntityType.ASSET_PROFILE.equals(targetEntityType)) { // iterate over all entities that belong to profile and push the message for corresponding CF - var entityIds = getEntitiesByProfile(targetEntityId); + var entityIds = cfEntityCache.getMyEntityIdsByProfileId(tenantId, targetEntityId); if (!entityIds.isEmpty()) { MultipleTbCallback callback = new MultipleTbCallback(entityIds.size(), msg.getCallback()); var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, callback); @@ -392,34 +393,11 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware return result; } - private Set getEntitiesByProfile(EntityId entityProfileId) { - Set entities = profileEntities.get(entityProfileId); - if (entities == null) { - entities = switch (entityProfileId.getEntityType()) { - case ASSET_PROFILE -> profileEntities.computeIfAbsent(entityProfileId, profileId -> { - Set assetIds = new HashSet<>(); - (new PageDataIterable<>(pageLink -> - systemContext.getAssetService().findAssetIdsByTenantIdAndAssetProfileId(tenantId, (AssetProfileId) profileId, pageLink), initFetchPackSize)).forEach(assetIds::add); - return assetIds; - }); - case DEVICE_PROFILE -> profileEntities.computeIfAbsent(entityProfileId, profileId -> { - Set deviceIds = new HashSet<>(); - (new PageDataIterable<>(pageLink -> - systemContext.getDeviceService().findDeviceIdsByTenantIdAndDeviceProfileId(tenantId, (DeviceProfileId) entityProfileId, pageLink), initFetchPackSize)).forEach(deviceIds::add); - return deviceIds; - }); - default -> throw new IllegalArgumentException("Entity type should be ASSET_PROFILE or DEVICE_PROFILE."); - }; - } - log.trace("[{}] Found entities by profile in cache: {}", entityProfileId, entities); - return entities; - } - private void initCf(CalculatedFieldCtx cfCtx, TbCallback callback, boolean forceStateReinit) { EntityId entityId = cfCtx.getEntityId(); EntityType entityType = cfCtx.getEntityId().getEntityType(); if (isProfileEntity(entityType)) { - var entityIds = getEntitiesByProfile(entityId); + var entityIds = cfEntityCache.getMyEntityIdsByProfileId(tenantId, entityId); if (!entityIds.isEmpty()) { var multiCallback = new MultipleTbCallback(entityIds.size(), callback); entityIds.forEach(id -> initCfForEntity(id, cfCtx, forceStateReinit, multiCallback)); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java index ff3bda5da5..96a3694543 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldCache.java @@ -31,16 +31,12 @@ public interface CalculatedFieldCache { List getCalculatedFieldsByEntityId(EntityId entityId); - List getCalculatedFieldLinks(CalculatedFieldId calculatedFieldId); - List getCalculatedFieldLinksByEntityId(EntityId entityId); CalculatedFieldCtx getCalculatedFieldCtx(CalculatedFieldId calculatedFieldId); List getCalculatedFieldCtxsByEntityId(EntityId entityId); - Set getEntitiesByProfile(TenantId tenantId, EntityId entityId); - void addCalculatedField(TenantId tenantId, CalculatedFieldId calculatedFieldId); void updateCalculatedField(TenantId tenantId, CalculatedFieldId calculatedFieldId); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java index 7e841a0cf8..ed5ae23a17 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java @@ -15,31 +15,28 @@ */ package org.thingsboard.server.service.cf; -import jakarta.annotation.PostConstruct; import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.thingsboard.script.api.tbel.TbelInvokeService; +import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration; -import org.thingsboard.server.common.data.id.AssetProfileId; import org.thingsboard.server.common.data.id.CalculatedFieldId; -import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageDataIterable; -import org.thingsboard.server.dao.asset.AssetService; +import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg; +import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg; import org.thingsboard.server.dao.cf.CalculatedFieldService; -import org.thingsboard.server.dao.device.DeviceService; +import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import java.util.Collections; -import java.util.HashSet; import java.util.List; -import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CopyOnWriteArrayList; @@ -51,32 +48,32 @@ import java.util.concurrent.locks.ReentrantLock; @RequiredArgsConstructor public class DefaultCalculatedFieldCache implements CalculatedFieldCache { + private static final Integer UNKNOWN_PARTITION = -1; + private final Lock calculatedFieldFetchLock = new ReentrantLock(); private final CalculatedFieldService calculatedFieldService; - private final AssetService assetService; - private final DeviceService deviceService; private final TbelInvokeService tbelInvokeService; + private final ActorSystemContext actorSystemContext; private final ConcurrentMap calculatedFields = new ConcurrentHashMap<>(); private final ConcurrentMap> entityIdCalculatedFields = new ConcurrentHashMap<>(); private final ConcurrentMap> calculatedFieldLinks = new ConcurrentHashMap<>(); private final ConcurrentMap> entityIdCalculatedFieldLinks = new ConcurrentHashMap<>(); private final ConcurrentMap calculatedFieldsCtx = new ConcurrentHashMap<>(); - private final ConcurrentMap> entityIdCalculatedFieldCtxs = new ConcurrentHashMap<>(); - private final ConcurrentMap> profileEntities = new ConcurrentHashMap<>(); @Value("${calculatedField.initFetchPackSize:50000}") @Getter private int initFetchPackSize; - @PostConstruct + @AfterStartUp(order = AfterStartUp.CF_READ_CF_SERVICE) public void init() { PageDataIterable cfs = new PageDataIterable<>(calculatedFieldService::findAllCalculatedFields, initFetchPackSize); cfs.forEach(cf -> calculatedFields.putIfAbsent(cf.getId(), cf)); calculatedFields.values().forEach(cf -> entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cf) ); + cfs.forEach(cf -> actorSystemContext.tell(new CalculatedFieldInitMsg(cf.getTenantId(), cf))); PageDataIterable cfls = new PageDataIterable<>(calculatedFieldService::findAllCalculatedFieldLinks, initFetchPackSize); cfls.forEach(link -> calculatedFieldLinks.computeIfAbsent(link.getCalculatedFieldId(), id -> new CopyOnWriteArrayList<>()).add(link)); calculatedFieldLinks.values().stream() @@ -84,6 +81,7 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { .forEach(link -> entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(link) ); + cfls.forEach(link -> actorSystemContext.tell(new CalculatedFieldLinkInitMsg(link.getTenantId(), link))); } @Override @@ -96,11 +94,6 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { return entityIdCalculatedFields.getOrDefault(entityId, new CopyOnWriteArrayList<>()); } - @Override - public List getCalculatedFieldLinks(CalculatedFieldId calculatedFieldId) { - return calculatedFieldLinks.getOrDefault(calculatedFieldId, new CopyOnWriteArrayList<>()); - } - @Override public List getCalculatedFieldLinksByEntityId(EntityId entityId) { return entityIdCalculatedFieldLinks.getOrDefault(entityId, new CopyOnWriteArrayList<>()); @@ -139,39 +132,6 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { .toList(); } - @Override - public Set getEntitiesByProfile(TenantId tenantId, EntityId entityProfileId) { - Set entities = profileEntities.get(entityProfileId); - if (entities == null) { - calculatedFieldFetchLock.lock(); - try { - entities = profileEntities.get(entityProfileId); - if (entities == null) { - entities = switch (entityProfileId.getEntityType()) { - case ASSET_PROFILE -> profileEntities.computeIfAbsent(entityProfileId, profileId -> { - Set assetIds = new HashSet<>(); - (new PageDataIterable<>(pageLink -> - assetService.findAssetIdsByTenantIdAndAssetProfileId(tenantId, (AssetProfileId) profileId, pageLink), initFetchPackSize)).forEach(assetIds::add); - return assetIds; - }); - case DEVICE_PROFILE -> profileEntities.computeIfAbsent(entityProfileId, profileId -> { - Set deviceIds = new HashSet<>(); - (new PageDataIterable<>(pageLink -> - deviceService.findDeviceIdsByTenantIdAndDeviceProfileId(tenantId, (DeviceProfileId) entityProfileId, pageLink), initFetchPackSize)).forEach(deviceIds::add); - return deviceIds; - }); - default -> - throw new IllegalArgumentException("Entity type should be ASSET_PROFILE or DEVICE_PROFILE."); - }; - } - } finally { - calculatedFieldFetchLock.unlock(); - } - } - log.trace("[{}] Found entities by profile in cache: {}", entityProfileId, entities); - return entities; - } - @Override public void addCalculatedField(TenantId tenantId, CalculatedFieldId calculatedFieldId) { calculatedFieldFetchLock.lock(); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java index d8e42c916d..5c4ab5ef6d 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java @@ -277,48 +277,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas @Override protected Map>> onAddedPartitions(Set addedPartitions) { var result = new HashMap>>(); -// PageDataIterable cfs = new PageDataIterable<>(calculatedFieldService::findAllCalculatedFields, initFetchPackSize); -// Map> tpiTargetEntityMap = new HashMap<>(); -// -// for (CalculatedField cf : cfs) { -// -// Consumer resolvePartition = entityId -> { -// TopicPartitionInfo tpi; -// try { -// tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, cf.getTenantId(), entityId); -// if (addedPartitions.contains(tpi) && states.keySet().stream().noneMatch(ctxId -> ctxId.cfId().equals(cf.getId()))) { -// tpiTargetEntityMap.computeIfAbsent(tpi, k -> new ArrayList<>()).add(new CalculatedFieldEntityCtxId(cf.getId(), entityId)); -// } -// } catch (Exception e) { -// log.warn("Failed to resolve partition for CalculatedFieldEntityCtxId: entityId=[{}], tenantId=[{}]. Reason: {}", -// entityId, cf.getTenantId(), e.getMessage()); -// } -// }; -// -// EntityId cfEntityId = cf.getEntityId(); -// if (isProfileEntity(cfEntityId)) { -// calculatedFieldCache.getEntitiesByProfile(cf.getTenantId(), cfEntityId).forEach(resolvePartition); -// } else { -// resolvePartition.accept(cfEntityId); -// } -// } -// -// for (var entry : tpiTargetEntityMap.entrySet()) { -// for (List partition : Lists.partition(entry.getValue(), 1000)) { -// log.info("[{}] Submit task for CalculatedFields: {}", entry.getKey(), partition.size()); -// var future = calculatedFieldExecutor.submit(() -> { -// try { -// for (CalculatedFieldEntityCtxId ctxId : partition) { -// restoreState(ctxId.cfId(), ctxId.entityId()); -// } -// } catch (Throwable t) { -// log.error("Unexpected exception while restoring CalculatedField states", t); -// throw t; -// } -// }); -// result.computeIfAbsent(entry.getKey(), k -> new ArrayList<>()).add(future); -// } -// } return result; } @@ -331,89 +289,6 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas states.keySet().removeIf(ctxId -> ctxId.cfId().equals(calculatedFieldId)); } -// @Override -// public void onEntityUpdateMsg(CalculatedFieldEntityUpdateMsgProto proto, TbCallback callback) { -// try { -// TenantId tenantId = TenantId.fromUUID(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())); -// EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())); -// -// TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId); -// if (tpi.isMyPartition()) { -// log.info("Received CalculatedFieldEntityUpdateMsgProto for processing: tenantId=[{}], entityId=[{}]", tenantId, entityId); -// if (proto.getDeleted()) { -// log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity deleted from profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); -// -// EntityId oldProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB())); -// calculatedFieldCache.getCalculatedFieldsByEntityId(entityId).forEach(cf -> clearState(cf.getId(), entityId)); -// calculatedFieldCache.getCalculatedFieldsByEntityId(oldProfileId).forEach(cf -> clearState(cf.getId(), entityId)); -// } -// if (proto.getAdded()) { -// log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity added to profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); -// -// EntityId newProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getNewProfileIdMSB(), proto.getNewProfileIdLSB())); -// initializeStateForEntityByProfile(entityId, newProfileId, callback); -// } -// if (proto.getUpdated()) { -// log.info("Executing CalculatedFieldEntityUpdateMsgProto msg: entity changed the profile, tenantId=[{}], entityId=[{}]", tenantId, entityId); -// -// EntityId oldProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getOldProfileIdMSB(), proto.getOldProfileIdLSB())); -// EntityId newProfileId = EntityIdFactory.getByTypeAndUuid(proto.getEntityProfileType(), new UUID(proto.getNewProfileIdMSB(), proto.getNewProfileIdLSB())); -// -// calculatedFieldCache.getCalculatedFieldsByEntityId(oldProfileId).forEach(cf -> clearState(cf.getId(), entityId)); -// initializeStateForEntityByProfile(entityId, newProfileId, callback); -// } -// } else { -// clusterService.pushNotificationToCalculatedFields(tenantId, entityId, ToCalculatedFieldNotificationMsg.newBuilder().setEntityUpdateMsg(proto).build(), null); -// } -// } catch (Exception e) { -// log.trace("Failed to process entity update msg: [{}]", proto, e); -// } -// } - - private void clearState(CalculatedFieldId calculatedFieldId, EntityId entityId) { - log.warn("Executing clearState, calculatedFieldId=[{}], entityId=[{}]", calculatedFieldId, entityId); - } - - private void initializeStateForEntityByProfile(EntityId entityId, EntityId profileId, TbCallback callback) { - calculatedFieldCache.getCalculatedFieldsByEntityId(profileId).stream() - .map(cf -> calculatedFieldCache.getCalculatedFieldCtx(cf.getId())) - .forEach(cfCtx -> initializeStateForEntity(cfCtx, entityId, callback)); - } - - private void initializeStateForEntity(CalculatedFieldCtx calculatedFieldCtx, EntityId entityId, TbCallback callback) { - initializeStateForEntity(calculatedFieldCtx, entityId, new HashMap<>(), callback); - } - - private void initializeStateForEntity(CalculatedFieldCtx calculatedFieldCtx, EntityId entityId, Map commonArguments, TbCallback callback) { - Map argumentValues = new HashMap<>(commonArguments); - List> futures = new ArrayList<>(); - - calculatedFieldCtx.getArguments().forEach((key, argument) -> { - if (!commonArguments.containsKey(key)) { - futures.add(Futures.transform(fetchArgumentValue(calculatedFieldCtx.getTenantId(), entityId, argument), - result -> { - argumentValues.put(key, result); - return result; - }, calculatedFieldCallbackExecutor)); - } - }); - - Futures.addCallback(Futures.allAsList(futures), new FutureCallback<>() { - @Override - public void onSuccess(List results) { -// updateOrInitializeState(calculatedFieldCtx, entityId, argumentValues, new ArrayList<>()); - callback.onSuccess(); - } - - @Override - public void onFailure(Throwable t) { - log.error("Failed to initialize state for entity: [{}]", entityId, t); - callback.onFailure(t); - } - }, calculatedFieldCallbackExecutor); - } - - @Override public void pushMsgToRuleEngine(TenantId tenantId, EntityId entityId, CalculatedFieldResult calculatedFieldResult, List cfIds, TbCallback callback) { try { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldInitService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldInitService.java index 87dda7013f..f71617e204 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldInitService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldInitService.java @@ -17,41 +17,49 @@ package org.thingsboard.server.service.cf; import lombok.Getter; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.calculatedField.CalculatedFieldStateRestoreMsg; -import org.thingsboard.server.common.data.cf.CalculatedField; -import org.thingsboard.server.common.data.cf.CalculatedFieldLink; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.page.PageDataIterable; -import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg; -import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg; -import org.thingsboard.server.dao.cf.CalculatedFieldService; +import org.thingsboard.server.dao.asset.AssetService; +import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbRuleEngineComponent; +import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService; +@Slf4j @Service @TbRuleEngineComponent @RequiredArgsConstructor public class DefaultCalculatedFieldInitService implements CalculatedFieldInitService { - private final CalculatedFieldService calculatedFieldService; + private final CalculatedFieldEntityProfileCache entityProfileCache; private final CalculatedFieldStateService stateService; + private final ActorSystemContext actorSystemContext; + private final AssetService assetService; + private final DeviceService deviceService; @Value("${calculated_fields.init_fetch_pack_size:50000}") @Getter private int initFetchPackSize; - @AfterStartUp(order = AfterStartUp.CF_INIT_SERVICE) + @AfterStartUp(order = AfterStartUp.CF_READ_PROFILE_ENTITIES_SERVICE) public void initCalculatedFieldDefinitions() { - PageDataIterable cfs = new PageDataIterable<>(calculatedFieldService::findAllCalculatedFields, initFetchPackSize); - cfs.forEach(cf -> actorSystemContext.tell(new CalculatedFieldInitMsg(cf.getTenantId(), cf))); - PageDataIterable cfls = new PageDataIterable<>(calculatedFieldService::findAllCalculatedFieldLinks, initFetchPackSize); - cfls.forEach(link -> actorSystemContext.tell(new CalculatedFieldLinkInitMsg(link.getTenantId(), link))); - //TODO: combine with the DefaultCalculatedFieldCache. - + PageDataIterable deviceIdInfos = new PageDataIterable<>(deviceService::findProfileEntityIdInfos, initFetchPackSize); + for (ProfileEntityIdInfo idInfo : deviceIdInfos) { + log.trace("Processing device record: {}", idInfo); + entityProfileCache.add(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId()); + } + PageDataIterable assetIdInfos = new PageDataIterable<>(assetService::findProfileEntityIdInfos, initFetchPackSize); + for (ProfileEntityIdInfo idInfo : assetIdInfos) { + log.trace("Processing asset record: {}", idInfo); + entityProfileCache.add(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId()); + } } @AfterStartUp(order = AfterStartUp.CF_STATE_RESTORE_SERVICE) @@ -59,6 +67,4 @@ public class DefaultCalculatedFieldInitService implements CalculatedFieldInitSer stateService.restoreStates().forEach((k, v) -> actorSystemContext.tell(new CalculatedFieldStateRestoreMsg(k, v))); } - - } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/cache/CalculatedFieldEntityProfileCache.java b/application/src/main/java/org/thingsboard/server/service/cf/cache/CalculatedFieldEntityProfileCache.java new file mode 100644 index 0000000000..6b0718eb84 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/cf/cache/CalculatedFieldEntityProfileCache.java @@ -0,0 +1,35 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.cf.cache; + +import org.springframework.context.ApplicationListener; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; + +import java.util.Collection; + +public interface CalculatedFieldEntityProfileCache extends ApplicationListener { + + void add(TenantId tenantId, EntityId profileId, EntityId entityId); + + void update(TenantId tenantId, EntityId oldProfileId, EntityId newProfileId, EntityId entityId); + + void evict(TenantId tenantId, EntityId entityId); + + Collection getMyEntityIdsByProfileId(TenantId tenantId, EntityId profileId); + +} diff --git a/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java b/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java new file mode 100644 index 0000000000..13f95c547d --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java @@ -0,0 +1,86 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.cf.cache; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.queue.discovery.HashPartitionService; +import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.TbApplicationEventListener; +import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; +import org.thingsboard.server.queue.util.TbRuleEngineComponent; + +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.stream.Collectors; + +@TbRuleEngineComponent +@Service +@Slf4j +@RequiredArgsConstructor +public class DefaultCalculatedFieldEntityProfileCache extends TbApplicationEventListener implements CalculatedFieldEntityProfileCache { + + private static final Integer UNKNOWN = -1; + private final ConcurrentMap tenantCache = new ConcurrentHashMap<>(); + private final PartitionService partitionService; + private volatile List myPartitions = Collections.emptyList(); + + @Override + protected void onTbApplicationEvent(PartitionChangeEvent event) { + myPartitions = event.getCalculatedFieldsPartitions().stream() + .filter(TopicPartitionInfo::isMyPartition) + .map(tpi -> tpi.getPartition().orElse(UNKNOWN)).collect(Collectors.toList()); + //Naive approach that need to be improved. + tenantCache.values().forEach(cache -> cache.setMyPartitions(myPartitions)); + } + + @Override + public void add(TenantId tenantId, EntityId profileId, EntityId entityId) { + var tpi = partitionService.resolve(HashPartitionService.CALCULATED_FIELD_QUEUE_KEY, entityId); + var partition = tpi.getPartition().orElse(UNKNOWN); + tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()) + .add(profileId, entityId, partition, tpi.isMyPartition()); + } + + @Override + public void update(TenantId tenantId, EntityId oldProfileId, EntityId newProfileId, EntityId entityId) { + var tpi = partitionService.resolve(HashPartitionService.CALCULATED_FIELD_QUEUE_KEY, entityId); + var partition = tpi.getPartition().orElse(UNKNOWN); + var cache = tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()); + //TODO: make this method atomic; + cache.remove(oldProfileId, entityId); + cache.add(newProfileId, entityId, partition, tpi.isMyPartition()); + } + + @Override + public void evict(TenantId tenantId, EntityId entityId) { + var cache = tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()); + cache.removeEntityId(entityId); + } + + @Override + public Collection getMyEntityIdsByProfileId(TenantId tenantId, EntityId profileId) { + return tenantCache.computeIfAbsent(tenantId, id -> new TenantEntityProfileCache()).getMyEntityIdsByProfileId(profileId); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/cf/cache/TenantEntityProfileCache.java b/application/src/main/java/org/thingsboard/server/service/cf/cache/TenantEntityProfileCache.java new file mode 100644 index 0000000000..b3499f3cc4 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/cf/cache/TenantEntityProfileCache.java @@ -0,0 +1,118 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.cf.cache; + +import org.thingsboard.server.common.data.id.EntityId; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +public class TenantEntityProfileCache { + + private final ReadWriteLock lock = new ReentrantReadWriteLock(); + private final Map>> allEntities = new HashMap<>(); + private final Map> myEntities = new HashMap<>(); + + public void setMyPartitions(List myPartitions) { + lock.writeLock().lock(); + try { + myEntities.clear(); + myPartitions.forEach(partitionId -> { + var map = allEntities.get(partitionId); + if (map != null) { + map.forEach((profileId, entityIds) -> myEntities.computeIfAbsent(profileId, k -> new HashSet<>()).addAll(entityIds)); + } + }); + } finally { + lock.writeLock().unlock(); + } + } + + public void removeProfileId(EntityId profileId) { + lock.writeLock().lock(); + try { + // Remove from allEntities + allEntities.values().forEach(map -> map.remove(profileId)); + // Remove from myEntities + myEntities.remove(profileId); + } finally { + lock.writeLock().unlock(); + } + } + + public void removeEntityId(EntityId entityId) { + lock.writeLock().lock(); + try { + // Remove from allEntities + allEntities.values().forEach(map -> map.values().forEach(set -> set.remove(entityId))); + // Remove from myEntities + myEntities.values().forEach(set -> set.remove(entityId)); + } finally { + lock.writeLock().unlock(); + } + } + + public void remove(EntityId profileId, EntityId entityId) { + lock.writeLock().lock(); + try { + // Remove from allEntities + allEntities.values().forEach(map -> removeSafely(map, profileId, entityId)); + // Remove from myEntities + removeSafely(myEntities, profileId, entityId); + } finally { + lock.writeLock().unlock(); + } + } + + public void add(EntityId profileId, EntityId entityId, Integer partition, boolean mine) { + lock.writeLock().lock(); + try { + if (mine) { + myEntities.computeIfAbsent(profileId, k -> new HashSet<>()).add(entityId); + } + allEntities.computeIfAbsent(partition, k -> new HashMap<>()).computeIfAbsent(profileId, p -> new HashSet<>()).add(entityId); + } finally { + lock.writeLock().unlock(); + } + } + + public Collection getMyEntityIdsByProfileId(EntityId profileId) { + lock.readLock().lock(); + try { + var entities = myEntities.getOrDefault(profileId, Collections.emptySet()); + List result = new ArrayList<>(entities.size()); + result.addAll(entities); + return result; + } finally { + lock.readLock().unlock(); + } + } + + private void removeSafely(Map> map, EntityId profileId, EntityId entityId) { + var set = map.get(profileId); + if (set != null) { + set.remove(entityId); + } + } +} diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/asset/AssetService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/asset/AssetService.java index 2a9fe08827..15ed81d4c2 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/asset/AssetService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/asset/AssetService.java @@ -17,6 +17,7 @@ package org.thingsboard.server.dao.asset; import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.EntitySubtype; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.asset.AssetInfo; import org.thingsboard.server.common.data.asset.AssetSearchQuery; @@ -63,6 +64,8 @@ public interface AssetService extends EntityDaoService { PageData findAssetInfosByTenantIdAndAssetProfileId(TenantId tenantId, AssetProfileId assetProfileId, PageLink pageLink); + PageData findProfileEntityIdInfos(PageLink pageLink); + PageData findAssetIdsByTenantIdAndAssetProfileId(TenantId tenantId, AssetProfileId assetProfileId, PageLink pageLink); ListenableFuture> findAssetsByTenantIdAndIdsAsync(TenantId tenantId, List assetIds); diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/device/DeviceService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/device/DeviceService.java index 0848e5a1cb..eb018aef1d 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/device/DeviceService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/device/DeviceService.java @@ -23,6 +23,7 @@ import org.thingsboard.server.common.data.DeviceInfoFilter; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.DeviceTransportType; import org.thingsboard.server.common.data.EntitySubtype; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.device.DeviceSearchQuery; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; @@ -73,6 +74,8 @@ public interface DeviceService extends EntityDaoService { PageData findDeviceIdInfos(PageLink pageLink); + PageData findProfileEntityIdInfos(PageLink pageLink); + PageData findDevicesByTenantIdAndType(TenantId tenantId, String type, PageLink pageLink); PageData findDeviceIdsByTenantIdAndDeviceProfileId(TenantId tenantId, DeviceProfileId deviceProfileId, PageLink pageLink); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/ProfileEntityIdInfo.java b/common/data/src/main/java/org/thingsboard/server/common/data/ProfileEntityIdInfo.java new file mode 100644 index 0000000000..83b56c38a7 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/ProfileEntityIdInfo.java @@ -0,0 +1,55 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data; + +import lombok.Data; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.data.id.AssetId; +import org.thingsboard.server.common.data.id.AssetProfileId; +import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.DeviceProfileId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; + +import java.io.Serializable; +import java.util.UUID; + +@Data +@Slf4j +public class ProfileEntityIdInfo implements Serializable, HasTenantId { + + private static final long serialVersionUID = 8532058281983868003L; + + private final TenantId tenantId; + private final EntityId profileId; + private final EntityId entityId; + + private ProfileEntityIdInfo(UUID tenantId, EntityId profileId, EntityId entityId) { + this.tenantId = TenantId.fromUUID(tenantId); + this.profileId = profileId; + this.entityId = entityId; + } + + public static ProfileEntityIdInfo create(UUID tenantId, DeviceProfileId profileId, DeviceId entityId) { + return new ProfileEntityIdInfo(tenantId, profileId, entityId); + } + + public static ProfileEntityIdInfo create(UUID tenantId, AssetProfileId profileId, AssetId entityId) { + return new ProfileEntityIdInfo(tenantId, profileId, entityId); + } + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/util/AfterStartUp.java b/common/queue/src/main/java/org/thingsboard/server/queue/util/AfterStartUp.java index 70ae853e1e..f97c3b0a1b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/util/AfterStartUp.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/util/AfterStartUp.java @@ -38,9 +38,10 @@ public @interface AfterStartUp { int ACTOR_SYSTEM = 9; int REGULAR_SERVICE = 10; - int CF_INIT_SERVICE = 10; - int CF_STATE_RESTORE_SERVICE = 11; - int CF_CONSUMER_SERVICE = 12; + int CF_READ_PROFILE_ENTITIES_SERVICE = 10; + int CF_READ_CF_SERVICE = 11; + int CF_STATE_RESTORE_SERVICE = 12; + int CF_CONSUMER_SERVICE = 13; int BEFORE_TRANSPORT_SERVICE = Integer.MAX_VALUE - 1001; int TRANSPORT_SERVICE = Integer.MAX_VALUE - 1000; diff --git a/dao/src/main/java/org/thingsboard/server/dao/asset/AssetDao.java b/dao/src/main/java/org/thingsboard/server/dao/asset/AssetDao.java index 8e40ec6d87..783a084ef3 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/asset/AssetDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/asset/AssetDao.java @@ -17,6 +17,7 @@ package org.thingsboard.server.dao.asset; import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.EntitySubtype; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.asset.AssetInfo; import org.thingsboard.server.common.data.id.AssetId; @@ -236,4 +237,6 @@ public interface AssetDao extends Dao, TenantEntityDao, ExportableEntityD PageData findAssetsByTenantIdAndEdgeIdAndType(UUID tenantId, UUID edgeId, String type, PageLink pageLink); PageData> getAllAssetTypes(PageLink pageLink); + + PageData findProfileEntityIdInfos(PageLink pageLink); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java b/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java index 564da127ab..607d09e23c 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/asset/BaseAssetService.java @@ -26,6 +26,7 @@ import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.event.TransactionalEventListener; import org.thingsboard.server.common.data.EntitySubtype; import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.asset.AssetInfo; @@ -285,6 +286,13 @@ public class BaseAssetService extends AbstractCachedEntityService findProfileEntityIdInfos(PageLink pageLink) { + log.trace("Executing findProfileEntityIdInfos, pageLink [{}]", pageLink); + validatePageLink(pageLink); + return assetDao.findProfileEntityIdInfos(pageLink); + } + @Override public PageData findAssetIdsByTenantIdAndAssetProfileId(TenantId tenantId, AssetProfileId assetProfileId, PageLink pageLink) { log.trace("Executing findAssetIdsByTenantIdAndAssetProfileId, tenantId [{}], assetProfileId [{}]", tenantId, assetProfileId); diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceDao.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceDao.java index 7c82df7703..b474e0b8e5 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceDao.java @@ -22,6 +22,7 @@ import org.thingsboard.server.common.data.DeviceInfo; import org.thingsboard.server.common.data.DeviceInfoFilter; import org.thingsboard.server.common.data.DeviceTransportType; import org.thingsboard.server.common.data.EntitySubtype; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; @@ -230,5 +231,8 @@ public interface DeviceDao extends Dao, TenantEntityDao, ExportableEntit PageData findDeviceIdInfos(PageLink pageLink); + PageData findProfileEntityIdInfos(PageLink pageLink); + PageData findDeviceInfosByFilter(DeviceInfoFilter filter, PageLink pageLink); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java index ffebd5f032..99ae252f79 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java @@ -39,6 +39,7 @@ import org.thingsboard.server.common.data.DeviceTransportType; import org.thingsboard.server.common.data.EntitySubtype; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityView; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.audit.ActionType; @@ -386,6 +387,13 @@ public class DeviceServiceImpl extends CachedVersionedEntityService findProfileEntityIdInfos(PageLink pageLink) { + log.trace("Executing findProfileEntityIdInfos, pageLink [{}]", pageLink); + validatePageLink(pageLink); + return deviceDao.findProfileEntityIdInfos(pageLink); + } + @Override public PageData findDevicesByTenantIdAndType(TenantId tenantId, String type, PageLink pageLink) { log.trace("Executing findDevicesByTenantIdAndType, tenantId [{}], type [{}], pageLink [{}]", tenantId, type, pageLink); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/asset/JpaAssetDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/asset/JpaAssetDao.java index db6472f825..650d991521 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/asset/JpaAssetDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/asset/JpaAssetDao.java @@ -22,6 +22,7 @@ import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.EntitySubtype; import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.asset.AssetInfo; import org.thingsboard.server.common.data.id.AssetId; @@ -35,6 +36,8 @@ import org.thingsboard.server.dao.asset.AssetDao; import org.thingsboard.server.dao.model.sql.AssetEntity; import org.thingsboard.server.dao.model.sql.AssetInfoEntity; import org.thingsboard.server.dao.sql.JpaAbstractDao; +import org.thingsboard.server.dao.sql.device.NativeAssetRepository; +import org.thingsboard.server.dao.sql.device.NativeDeviceRepository; import org.thingsboard.server.dao.util.SqlDao; import java.util.Arrays; @@ -56,6 +59,9 @@ public class JpaAssetDao extends JpaAbstractDao implements A @Autowired private AssetRepository assetRepository; + @Autowired + private NativeAssetRepository nativeAssetRepository; + @Autowired private AssetProfileRepository assetProfileRepository; @@ -252,6 +258,12 @@ public class JpaAssetDao extends JpaAbstractDao implements A DaoUtil.toPageable(pageLink, Arrays.asList(new SortOrder("tenantId"), new SortOrder("type"))))); } + @Override + public PageData findProfileEntityIdInfos(PageLink pageLink) { + log.debug("Find profile device id infos by pageLink [{}]", pageLink); + return nativeAssetRepository.findProfileEntityIdInfos(DaoUtil.toPageable(pageLink)); + } + @Override public Long countByTenantId(TenantId tenantId) { return assetRepository.countByTenantId(tenantId.getId()); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/device/AbstractNativeRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/device/AbstractNativeRepository.java new file mode 100644 index 0000000000..bde9278cd4 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/device/AbstractNativeRepository.java @@ -0,0 +1,54 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.sql.device; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.domain.Pageable; +import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; +import org.springframework.stereotype.Repository; +import org.springframework.transaction.support.TransactionTemplate; +import org.thingsboard.server.common.data.page.PageData; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import java.util.stream.Collectors; + +@RequiredArgsConstructor +@Repository +@Slf4j +public class AbstractNativeRepository { + + private final NamedParameterJdbcTemplate jdbcTemplate; + private final TransactionTemplate transactionTemplate; + + protected PageData find(String countQuery, String findQuery, Pageable pageable, Function, T> mapper) { + return transactionTemplate.execute(status -> { + long startTs = System.currentTimeMillis(); + int totalElements = jdbcTemplate.queryForObject(countQuery, Collections.emptyMap(), Integer.class); + log.debug("Count query took {} ms", System.currentTimeMillis() - startTs); + startTs = System.currentTimeMillis(); + List> rows = jdbcTemplate.queryForList(String.format(findQuery, pageable.getPageSize(), pageable.getOffset()), Collections.emptyMap()); + log.debug("Main query took {} ms", System.currentTimeMillis() - startTs); + int totalPages = pageable.getPageSize() > 0 ? (int) Math.ceil((float) totalElements / pageable.getPageSize()) : 1; + boolean hasNext = pageable.getPageSize() > 0 && totalElements > pageable.getOffset() + rows.size(); + var data = rows.stream().map(mapper).collect(Collectors.toList()); + return new PageData<>(data, totalPages, totalElements, hasNext); + }); + } +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeAssetRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeAssetRepository.java new file mode 100644 index 0000000000..43f80e0a86 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeAssetRepository.java @@ -0,0 +1,55 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.sql.device; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.domain.Pageable; +import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; +import org.springframework.stereotype.Repository; +import org.springframework.transaction.support.TransactionTemplate; +import org.thingsboard.server.common.data.DeviceIdInfo; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; +import org.thingsboard.server.common.data.id.AssetId; +import org.thingsboard.server.common.data.id.AssetProfileId; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.DeviceProfileId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.page.PageData; + +import java.util.UUID; + +@Repository +@Slf4j +public class DefaultNativeAssetRepository extends AbstractNativeRepository implements NativeAssetRepository { + + private final String COUNT_QUERY = "SELECT count(id) FROM asset;"; + + public DefaultNativeAssetRepository(NamedParameterJdbcTemplate jdbcTemplate, TransactionTemplate transactionTemplate) { + super(jdbcTemplate, transactionTemplate); + } + + @Override + public PageData findProfileEntityIdInfos(Pageable pageable) { + String PROFILE_DEVICE_ID_INFO_QUERY = "SELECT tenant_id as tenantId, asset_profile_id as profileId, id as id FROM asset ORDER BY created_time ASC LIMIT %s OFFSET %s"; + return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, row -> { + AssetId id = new AssetId((UUID) row.get("id")); + AssetProfileId profileId = new AssetProfileId((UUID) row.get("profileId")); + var tenantIdObj = row.get("tenantId"); + return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id); + }); + } + +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeDeviceRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeDeviceRepository.java index 002c796089..67b45ba70d 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeDeviceRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/device/DefaultNativeDeviceRepository.java @@ -15,50 +15,50 @@ */ package org.thingsboard.server.dao.sql.device; -import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.data.domain.Pageable; import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; import org.springframework.stereotype.Repository; import org.springframework.transaction.support.TransactionTemplate; import org.thingsboard.server.common.data.DeviceIdInfo; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; -import java.util.Collections; -import java.util.List; -import java.util.Map; import java.util.UUID; -import java.util.stream.Collectors; -@RequiredArgsConstructor @Repository @Slf4j -public class DefaultNativeDeviceRepository implements NativeDeviceRepository { +public class DefaultNativeDeviceRepository extends AbstractNativeRepository implements NativeDeviceRepository { private final String COUNT_QUERY = "SELECT count(id) FROM device;"; - private final String QUERY = "SELECT tenant_id as tenantId, customer_id as customerId, id as id FROM device ORDER BY created_time ASC LIMIT %s OFFSET %s"; - private final NamedParameterJdbcTemplate jdbcTemplate; - private final TransactionTemplate transactionTemplate; + + public DefaultNativeDeviceRepository(NamedParameterJdbcTemplate jdbcTemplate, TransactionTemplate transactionTemplate) { + super(jdbcTemplate, transactionTemplate); + } @Override public PageData findDeviceIdInfos(Pageable pageable) { - return transactionTemplate.execute(status -> { - long startTs = System.currentTimeMillis(); - int totalElements = jdbcTemplate.queryForObject(COUNT_QUERY, Collections.emptyMap(), Integer.class); - log.debug("Count query took {} ms", System.currentTimeMillis() - startTs); - startTs = System.currentTimeMillis(); - List> rows = jdbcTemplate.queryForList(String.format(QUERY, pageable.getPageSize(), pageable.getOffset()), Collections.emptyMap()); - log.debug("Main query took {} ms", System.currentTimeMillis() - startTs); - int totalPages = pageable.getPageSize() > 0 ? (int) Math.ceil((float) totalElements / pageable.getPageSize()) : 1; - boolean hasNext = pageable.getPageSize() > 0 && totalElements > pageable.getOffset() + rows.size(); - var data = rows.stream().map(row -> { - UUID id = (UUID) row.get("id"); - var tenantIdObj = row.get("tenantId"); - var customerIdObj = row.get("customerId"); - return new DeviceIdInfo(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), customerIdObj != null ? (UUID) customerIdObj : null, id); - }).collect(Collectors.toList()); - return new PageData<>(data, totalPages, totalElements, hasNext); + String DEVICE_ID_INFO_QUERY = "SELECT tenant_id as tenantId, customer_id as customerId, id as id FROM device ORDER BY created_time ASC LIMIT %s OFFSET %s"; + return find(COUNT_QUERY, DEVICE_ID_INFO_QUERY, pageable, row -> { + UUID id = (UUID) row.get("id"); + var tenantIdObj = row.get("tenantId"); + var customerIdObj = row.get("customerId"); + return new DeviceIdInfo(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), customerIdObj != null ? (UUID) customerIdObj : null, id); }); } + + @Override + public PageData findProfileEntityIdInfos(Pageable pageable) { + String PROFILE_DEVICE_ID_INFO_QUERY = "SELECT tenant_id as tenantId, device_profile_id as profileId, id as id FROM device ORDER BY created_time ASC LIMIT %s OFFSET %s"; + return find(COUNT_QUERY, PROFILE_DEVICE_ID_INFO_QUERY, pageable, row -> { + DeviceId id = new DeviceId((UUID) row.get("id")); + DeviceProfileId profileId = new DeviceProfileId((UUID) row.get("profileId")); + var tenantIdObj = row.get("tenantId"); + return ProfileEntityIdInfo.create(tenantIdObj != null ? (UUID) tenantIdObj : TenantId.SYS_TENANT_ID.getId(), profileId, id); + }); + } + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/device/JpaDeviceDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/device/JpaDeviceDao.java index bf0def9fca..74afdca880 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/device/JpaDeviceDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/device/JpaDeviceDao.java @@ -29,6 +29,7 @@ import org.thingsboard.server.common.data.DeviceInfoFilter; import org.thingsboard.server.common.data.DeviceTransportType; import org.thingsboard.server.common.data.EntitySubtype; import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; @@ -272,6 +273,12 @@ public class JpaDeviceDao extends JpaAbstractDao implement return nativeDeviceRepository.findDeviceIdInfos(DaoUtil.toPageable(pageLink)); } + @Override + public PageData findProfileEntityIdInfos(PageLink pageLink) { + log.debug("Find profile device id infos by pageLink [{}]", pageLink); + return nativeDeviceRepository.findProfileEntityIdInfos(DaoUtil.toPageable(pageLink)); + } + @Override public Device findByTenantIdAndExternalId(UUID tenantId, UUID externalId) { return DaoUtil.getData(deviceRepository.findByTenantIdAndExternalId(tenantId, externalId)); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeAssetRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeAssetRepository.java new file mode 100644 index 0000000000..6a0539ee9d --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeAssetRepository.java @@ -0,0 +1,20 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.sql.device; + +public interface NativeAssetRepository extends NativeProfileEntityRepository { + +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeDeviceRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeDeviceRepository.java index f968749461..925bf12618 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeDeviceRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeDeviceRepository.java @@ -17,10 +17,12 @@ package org.thingsboard.server.dao.sql.device; import org.springframework.data.domain.Pageable; import org.thingsboard.server.common.data.DeviceIdInfo; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.page.PageData; -public interface NativeDeviceRepository { +public interface NativeDeviceRepository extends NativeProfileEntityRepository { PageData findDeviceIdInfos(Pageable pageable); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeProfileEntityRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeProfileEntityRepository.java new file mode 100644 index 0000000000..1d3dbbec45 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeProfileEntityRepository.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.sql.device; + +import org.springframework.data.domain.Pageable; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; +import org.thingsboard.server.common.data.page.PageData; + +public interface NativeProfileEntityRepository { + + PageData findProfileEntityIdInfos(Pageable pageable); + +}