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 1ed919e922..b5fa1e6fbd 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -30,9 +30,9 @@ import org.springframework.context.annotation.Lazy; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Component; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.rule.engine.api.DeviceStateManager; import org.thingsboard.rule.engine.api.MailService; import org.thingsboard.rule.engine.api.NotificationCenter; -import org.thingsboard.rule.engine.api.DeviceStateManager; import org.thingsboard.rule.engine.api.SmsService; import org.thingsboard.rule.engine.api.notification.SlackService; import org.thingsboard.rule.engine.api.sms.SmsSenderFactory; @@ -110,7 +110,6 @@ import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; import org.thingsboard.server.service.cf.CalculatedFieldQueueService; import org.thingsboard.server.service.cf.CalculatedFieldStateService; -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; @@ -546,11 +545,6 @@ public class ActorSystemContext { @Getter private CalculatedFieldQueueService calculatedFieldQueueService; - @Lazy - @Autowired(required = false) - @Getter - private CalculatedFieldEntityProfileCache calculatedFieldEntityProfileCache; - @Value("${actors.session.max_concurrent_sessions_per_device:1}") @Getter private int maxConcurrentSessionsPerDevice; diff --git a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java index 904547a2b6..a79a182fa1 100644 --- a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java @@ -113,6 +113,8 @@ public class AppActor extends ContextAwareActor { case SESSION_TIMEOUT_MSG: ctx.broadcastToChildrenByType(msg, EntityType.TENANT); break; + case CF_CACHE_INIT_MSG: + case CF_INIT_PROFILE_ENTITY_MSG: case CF_INIT_MSG: case CF_LINK_INIT_MSG: case CF_STATE_RESTORE_MSG: diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java index 70ed2849e8..9f59a80e67 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java @@ -22,8 +22,10 @@ import org.thingsboard.server.actors.TbActorException; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.TbActorStopReason; import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; +import org.thingsboard.server.common.msg.cf.CalculatedFieldCacheInitMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg; +import org.thingsboard.server.common.msg.cf.CalculatedFieldInitProfileEntityMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg; @@ -65,6 +67,12 @@ public class CalculatedFieldManagerActor extends AbstractCalculatedFieldActor { case CF_PARTITIONS_CHANGE_MSG: processor.onPartitionChange((CalculatedFieldPartitionChangeMsg) msg); break; + case CF_CACHE_INIT_MSG: + processor.onCacheInitMsg((CalculatedFieldCacheInitMsg) msg); + break; + case CF_INIT_PROFILE_ENTITY_MSG: + processor.onProfileEntityMsg((CalculatedFieldInitProfileEntityMsg) msg); + break; case CF_INIT_MSG: processor.onFieldInitMsg((CalculatedFieldInitMsg) msg); break; 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 31152c7c98..a896bbd74b 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 @@ -15,7 +15,9 @@ */ package org.thingsboard.server.actors.calculatedField; +import lombok.Getter; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.TbActorCtx; import org.thingsboard.server.actors.TbActorRef; @@ -24,6 +26,7 @@ import org.thingsboard.server.actors.service.DefaultActorService; import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.id.AssetId; @@ -31,17 +34,22 @@ import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.DeviceId; 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.common.msg.cf.CalculatedFieldCacheInitMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg; +import org.thingsboard.server.common.msg.cf.CalculatedFieldInitProfileEntityMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg; import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.cf.CalculatedFieldService; +import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; import org.thingsboard.server.service.cf.CalculatedFieldStateService; -import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; +import org.thingsboard.server.service.cf.cache.TenantEntityProfileCache; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.profile.TbAssetProfileCache; @@ -68,22 +76,30 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private final CalculatedFieldProcessingService cfExecService; private final CalculatedFieldStateService cfStateService; - private final CalculatedFieldEntityProfileCache cfEntityCache; private final CalculatedFieldService cfDaoService; + private final DeviceService deviceService; + private final AssetService assetService; private final TbAssetProfileCache assetProfileCache; private final TbDeviceProfileCache deviceProfileCache; + private final TenantEntityProfileCache entityProfileCache; protected final TenantId tenantId; protected TbActorCtx ctx; + @Value("${queue.calculated_fields.init_tenant_fetch_pack_size:1000}") + @Getter + private int initFetchPackSize; + CalculatedFieldManagerMessageProcessor(ActorSystemContext systemContext, TenantId tenantId) { super(systemContext); - this.cfEntityCache = systemContext.getCalculatedFieldEntityProfileCache(); this.cfExecService = systemContext.getCalculatedFieldProcessingService(); this.cfStateService = systemContext.getCalculatedFieldStateService(); this.cfDaoService = systemContext.getCalculatedFieldService(); + this.deviceService = systemContext.getDeviceService(); + this.assetService = systemContext.getAssetService(); this.assetProfileCache = systemContext.getAssetProfileCache(); this.deviceProfileCache = systemContext.getDeviceProfileCache(); + this.entityProfileCache = new TenantEntityProfileCache(); this.tenantId = tenantId; } @@ -100,6 +116,19 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware ctx.stop(ctx.getSelf()); } + public void onCacheInitMsg(CalculatedFieldCacheInitMsg msg) { + log.debug("[{}] Processing CF actor init message.", msg.getTenantId().getId()); + initEntityProfileCache(); + initCalculatedFields(); + msg.getCallback().onSuccess(); + } + + public void onProfileEntityMsg(CalculatedFieldInitProfileEntityMsg msg) { + log.debug("[{}] Processing profile entity message.", msg.getTenantId().getId()); + entityProfileCache.add(msg.getProfileEntityId(), msg.getEntityId()); + msg.getCallback().onSuccess(); + } + public void onFieldInitMsg(CalculatedFieldInitMsg msg) throws CalculatedFieldException { log.debug("[{}] Processing CF init message.", msg.getCf().getId()); var cf = msg.getCf(); @@ -180,16 +209,35 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } break; } + case DEVICE_PROFILE: + case ASSET_PROFILE: { + switch (event) { + case DELETED: + onProfileDeleted(msg.getData(), msg.getCallback()); + break; + default: + msg.getCallback().onSuccess(); + break; + } + break; + } default: { msg.getCallback().onSuccess(); } } } + private void onProfileDeleted(ComponentLifecycleMsg msg, TbCallback callback) { + entityProfileCache.removeProfileId(msg.getEntityId()); + callback.onSuccess(); + } + private void onEntityCreated(ComponentLifecycleMsg msg, TbCallback callback) { EntityId entityId = msg.getEntityId(); EntityId profileId = getProfileId(tenantId, entityId); - cfEntityCache.add(tenantId, profileId, entityId); + if (profileId != null) { + entityProfileCache.add(profileId, entityId); + } if (!isMyPartition(entityId, callback)) { return; } @@ -207,7 +255,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private void onEntityUpdated(ComponentLifecycleMsg msg, TbCallback callback) { if (msg.getOldProfileId() != null && !msg.getOldProfileId().equals(msg.getProfileId())) { - cfEntityCache.update(tenantId, msg.getOldProfileId(), msg.getProfileId(), msg.getEntityId()); + entityProfileCache.update(msg.getOldProfileId(), msg.getProfileId(), msg.getEntityId()); if (!isMyPartition(msg.getEntityId(), callback)) { return; } @@ -226,7 +274,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) { - cfEntityCache.evict(tenantId, msg.getEntityId()); + entityProfileCache.removeEntityId(msg.getEntityId()); if (isMyPartition(msg.getEntityId(), callback)) { log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId()); getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); @@ -322,11 +370,15 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware EntityId entityId = cfCtx.getEntityId(); EntityType entityType = cfCtx.getEntityId().getEntityType(); if (isProfileEntity(entityType)) { - var entityIds = cfEntityCache.getMyEntityIdsByProfileId(tenantId, entityId); + var entityIds = entityProfileCache.getEntityIdsByProfileId(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); - entityIds.forEach(id -> deleteCfForEntity(id, cfId, multiCallback)); + entityIds.forEach(id -> { + if (isMyPartition(id, multiCallback)) { + deleteCfForEntity(id, cfId, multiCallback); + } + }); } else { callback.onSuccess(); } @@ -366,10 +418,11 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware EntityId sourceEntityId = msg.getEntityId(); log.debug("Received linked telemetry msg from entity [{}]", sourceEntityId); var proto = msg.getProto(); + var callback = msg.getCallback(); var linksList = proto.getLinksList(); if (linksList.isEmpty()) { log.debug("[{}] No CF links to process new telemetry.", msg.getTenantId()); - msg.getCallback().onSuccess(); + callback.onSuccess(); } for (var linkProto : linksList) { var link = fromProto(linkProto); @@ -378,21 +431,23 @@ 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 = cfEntityCache.getMyEntityIdsByProfileId(tenantId, targetEntityId); + var entityIds = entityProfileCache.getEntityIdsByProfileId(targetEntityId); if (!entityIds.isEmpty()) { - MultipleTbCallback callback = new MultipleTbCallback(entityIds.size(), msg.getCallback()); - var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, callback); + MultipleTbCallback multipleCallback = new MultipleTbCallback(entityIds.size(), callback); + var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, multipleCallback); entityIds.forEach(entityId -> { - log.debug("Pushing linked telemetry msg to specific actor [{}]", entityId); - getOrCreateActor(entityId).tell(newMsg); + if (isMyPartition(entityId, multipleCallback)) { + log.debug("Pushing linked telemetry msg to specific actor [{}]", entityId); + getOrCreateActor(entityId).tell(newMsg); + } }); } else { - msg.getCallback().onSuccess(); + callback.onSuccess(); } } else { - if (isMyPartition(targetEntityId, msg.getCallback())) { + if (isMyPartition(targetEntityId, callback)) { log.debug("Pushing linked telemetry msg to specific actor [{}]", targetEntityId); - var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, msg.getCallback()); + var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, callback); getOrCreateActor(targetEntityId).tell(newMsg); } } @@ -438,10 +493,14 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware EntityId entityId = cfCtx.getEntityId(); EntityType entityType = cfCtx.getEntityId().getEntityType(); if (isProfileEntity(entityType)) { - var entityIds = cfEntityCache.getMyEntityIdsByProfileId(tenantId, entityId); + var entityIds = entityProfileCache.getEntityIdsByProfileId(entityId); if (!entityIds.isEmpty()) { var multiCallback = new MultipleTbCallback(entityIds.size(), callback); - entityIds.forEach(id -> initCfForEntity(id, cfCtx, forceStateReinit, multiCallback)); + entityIds.forEach(id -> { + if (isMyPartition(id, multiCallback)) { + initCfForEntity(id, cfCtx, forceStateReinit, multiCallback); + } + }); } else { callback.onSuccess(); } @@ -479,7 +538,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware return switch (entityId.getEntityType()) { case ASSET -> assetProfileCache.get(tenantId, (AssetId) entityId).getId(); case DEVICE -> deviceProfileCache.get(tenantId, (DeviceId) entityId).getId(); - default -> null; + default -> + throw new IllegalArgumentException("'" + entityId.getEntityType() + "' is not profile entity." + entityId); }; } @@ -504,4 +564,45 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware public void onPartitionChange(CalculatedFieldPartitionChangeMsg msg) { ctx.broadcastToChildren(msg, true); } + + public void initCalculatedFields() { + PageDataIterable cfs = new PageDataIterable<>(pageLink -> cfDaoService.findCalculatedFieldsByTenantId(tenantId, pageLink), initFetchPackSize); + cfs.forEach(cf -> { + log.trace("Processing calculated field record: {}", cf); + try { + onFieldInitMsg(new CalculatedFieldInitMsg(cf.getTenantId(), cf)); + } catch (CalculatedFieldException e) { + log.error("Failed to process calculated field record: {}", cf, e); + } + }); + calculatedFields.values().forEach(cf -> { + entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cf); + }); + PageDataIterable cfls = new PageDataIterable<>(pageLink -> cfDaoService.findAllCalculatedFieldLinksByTenantId(tenantId, pageLink), initFetchPackSize); + cfls.forEach(link -> { + onLinkInitMsg(new CalculatedFieldLinkInitMsg(link.getTenantId(), link)); + }); + } + + private void initEntityProfileCache() { + PageDataIterable deviceIdInfos = new PageDataIterable<>(pageLink -> deviceService.findProfileEntityIdInfosByTenantId(tenantId, pageLink), initFetchPackSize); + for (ProfileEntityIdInfo idInfo : deviceIdInfos) { + log.trace("Processing device record: {}", idInfo); + try { + entityProfileCache.add(idInfo.getProfileId(), idInfo.getEntityId()); + } catch (Exception e) { + log.error("Failed to process device record: {}", idInfo, e); + } + } + PageDataIterable assetIdInfos = new PageDataIterable<>(pageLink -> assetService.findProfileEntityIdInfosByTenantId(tenantId, pageLink), initFetchPackSize); + for (ProfileEntityIdInfo idInfo : assetIdInfos) { + log.trace("Processing asset record: {}", idInfo); + try { + entityProfileCache.add(idInfo.getProfileId(), idInfo.getEntityId()); + } catch (Exception e) { + log.error("Failed to process asset record: {}", idInfo, e); + } + } + } + } diff --git a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java index 728f715af4..2a039327f2 100644 --- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java @@ -50,6 +50,7 @@ import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; import org.thingsboard.server.common.msg.aware.DeviceAwareMsg; import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg; +import org.thingsboard.server.common.msg.cf.CalculatedFieldCacheInitMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg; import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; @@ -176,6 +177,8 @@ public class TenantActor extends RuleChainManagerActor { case RULE_CHAIN_TO_RULE_CHAIN_MSG: onRuleChainMsg((RuleChainAwareMsg) msg); break; + case CF_CACHE_INIT_MSG: + case CF_INIT_PROFILE_ENTITY_MSG: case CF_INIT_MSG: case CF_LINK_INIT_MSG: case CF_STATE_RESTORE_MSG: @@ -199,6 +202,7 @@ public class TenantActor extends RuleChainManagerActor { } else { log.debug("[{}] CF Actor is not initialized. ToCalculatedFieldSystemMsg: [{}]", tenantId, msg); } + msg.getCallback().onSuccess(); return; } if (priority) { @@ -274,10 +278,12 @@ public class TenantActor extends RuleChainManagerActor { () -> DefaultActorService.CF_MANAGER_DISPATCHER_NAME, () -> new CalculatedFieldManagerActorCreator(systemContext, tenantId), () -> true); + cfActor.tellWithHighPriority(new CalculatedFieldCacheInitMsg(tenantId)); } catch (Exception e) { log.info("[{}] Failed to init CF Actor.", tenantId, e); } } + cfActor.tellWithHighPriority(msg); if (!ruleChainsInitialized) { log.info("Tenant {} is now managed by this service, initializing rule chains", tenantId); initRuleChains(); 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 219a261183..ed35d96cb7 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 @@ -64,7 +64,7 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { private final ConcurrentMap> entityIdCalculatedFieldLinks = new ConcurrentHashMap<>(); private final ConcurrentMap calculatedFieldsCtx = new ConcurrentHashMap<>(); - @Value("${calculatedField.initFetchPackSize:50000}") + @Value("${queue.calculated_fields.init_fetch_pack_size:50000}") @Getter private int initFetchPackSize; 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 cc3022fa29..79950f2e3f 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 @@ -20,13 +20,14 @@ 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.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.page.PageDataIterable; +import org.thingsboard.server.common.msg.cf.CalculatedFieldInitProfileEntityMsg; 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; @Slf4j @Service @@ -34,11 +35,11 @@ import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache @RequiredArgsConstructor public class DefaultCalculatedFieldInitService implements CalculatedFieldInitService { - private final CalculatedFieldEntityProfileCache entityProfileCache; private final AssetService assetService; private final DeviceService deviceService; + private final ActorSystemContext actorSystemContext; - @Value("${calculated_fields.init_fetch_pack_size:50000}") + @Value("${queue.calculated_fields.init_fetch_pack_size:50000}") @Getter private int initFetchPackSize; @@ -48,7 +49,7 @@ public class DefaultCalculatedFieldInitService implements CalculatedFieldInitSer for (ProfileEntityIdInfo idInfo : deviceIdInfos) { log.trace("Processing device record: {}", idInfo); try { - entityProfileCache.add(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId()); + actorSystemContext.tell(new CalculatedFieldInitProfileEntityMsg(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId())); } catch (Exception e) { log.error("Failed to process device record: {}", idInfo, e); } @@ -57,7 +58,7 @@ public class DefaultCalculatedFieldInitService implements CalculatedFieldInitSer for (ProfileEntityIdInfo idInfo : assetIdInfos) { log.trace("Processing asset record: {}", idInfo); try { - entityProfileCache.add(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId()); + actorSystemContext.tell(new CalculatedFieldInitProfileEntityMsg(idInfo.getTenantId(), idInfo.getProfileId(), idInfo.getEntityId())); } catch (Exception e) { log.error("Failed to process asset record: {}", idInfo, e); } 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 deleted file mode 100644 index 2f5772ae50..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/cf/cache/DefaultCalculatedFieldEntityProfileCache.java +++ /dev/null @@ -1,95 +0,0 @@ -/** - * Copyright © 2016-2025 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.DataConstants; -import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.msg.queue.ServiceType; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; -import org.thingsboard.server.queue.discovery.PartitionService; -import org.thingsboard.server.queue.discovery.QueueKey; -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.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.stream.Collectors; - -@TbRuleEngineComponent -@Service -@Slf4j -@RequiredArgsConstructor -//TODO ashvayka: remove and use TenantEntityProfileCache in each CalculatedFieldManagerMessageProcessor; -public class DefaultCalculatedFieldEntityProfileCache extends TbApplicationEventListener implements CalculatedFieldEntityProfileCache { - - private static final Integer UNKNOWN = 0; - private final ConcurrentMap tenantCache = new ConcurrentHashMap<>(); - private final PartitionService partitionService; - private volatile List myPartitions = Collections.emptyList(); - - @Override - protected void onTbApplicationEvent(PartitionChangeEvent event) { - myPartitions = event.getCfPartitions().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(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, tenantId, 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(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, tenantId, 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); - } - - @Override - public int getEntityIdPartition(TenantId tenantId, EntityId entityId) { - var tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, tenantId, entityId); - return tpi.getPartition().orElse(UNKNOWN); - } - -} 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 index 1a17b9b8be..6435ffb14c 100644 --- 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 @@ -32,31 +32,13 @@ 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(); - } - } + private final Map> allEntities = new HashMap<>(); 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); + allEntities.remove(profileId); } finally { lock.writeLock().unlock(); } @@ -66,9 +48,7 @@ public class TenantEntityProfileCache { 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)); + allEntities.values().forEach(set -> set.remove(entityId)); } finally { lock.writeLock().unlock(); } @@ -78,33 +58,33 @@ public class TenantEntityProfileCache { lock.writeLock().lock(); try { // Remove from allEntities - allEntities.values().forEach(map -> removeSafely(map, profileId, entityId)); - // Remove from myEntities - removeSafely(myEntities, profileId, entityId); + removeSafely(allEntities, profileId, entityId); } finally { lock.writeLock().unlock(); } } - public void add(EntityId profileId, EntityId entityId, Integer partition, boolean mine) { + public void add(EntityId profileId, EntityId entityId) { lock.writeLock().lock(); try { - if(EntityType.DEVICE.equals(profileId.getEntityType())){ - throw new RuntimeException("WTF?"); - } - if (mine) { - myEntities.computeIfAbsent(profileId, k -> new HashSet<>()).add(entityId); + if (EntityType.DEVICE.equals(profileId.getEntityType()) || EntityType.ASSET.equals(profileId.getEntityType())) { + throw new RuntimeException("Entity type '" + profileId.getEntityType() + "' is not a profileId."); } - allEntities.computeIfAbsent(partition, k -> new HashMap<>()).computeIfAbsent(profileId, p -> new HashSet<>()).add(entityId); + allEntities.computeIfAbsent(profileId, k -> new HashSet<>()).add(entityId); } finally { lock.writeLock().unlock(); } } - public Collection getMyEntityIdsByProfileId(EntityId profileId) { + public void update(EntityId oldProfileId, EntityId newProfileId, EntityId entityId) { + remove(oldProfileId, entityId); + add(newProfileId, entityId); + } + + public Collection getEntityIdsByProfileId(EntityId profileId) { lock.readLock().lock(); try { - var entities = myEntities.getOrDefault(profileId, Collections.emptySet()); + var entities = allEntities.getOrDefault(profileId, Collections.emptySet()); List result = new ArrayList<>(entities.size()); result.addAll(entities); return result; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java index 3620ad3639..533b487d38 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java @@ -91,7 +91,7 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta } } }) - .consumerCreator((config, partitionId) -> queueFactory.createCalculatedFieldStateConsumer()) + .consumerCreator((queueConfig, tpi) -> queueFactory.createCalculatedFieldStateConsumer()) .queueAdmin(queueFactory.getCalculatedFieldQueueAdmin()) .consumerExecutor(eventConsumer.getConsumerExecutor()) .scheduler(eventConsumer.getScheduler()) diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java index 0eaa506dfd..9dc6139ca5 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java @@ -22,11 +22,11 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Service; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; -import org.thingsboard.server.queue.common.state.DefaultQueueStateService; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.common.consumer.PartitionedQueueConsumerManager; +import org.thingsboard.server.queue.common.state.DefaultQueueStateService; import org.thingsboard.server.queue.discovery.QueueKey; import org.thingsboard.server.service.cf.AbstractCalculatedFieldStateService; import org.thingsboard.server.service.cf.CfRocksDb; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java index e3bc3be90d..f8c398b016 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java @@ -99,12 +99,12 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractPartitionBa @Override protected void onStartUp() { var queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME); - PartitionedQueueConsumerManager> eventConsumer = PartitionedQueueConsumerManager.>create() + var eventConsumer = PartitionedQueueConsumerManager.>create() .queueKey(queueKey) .topic(partitionService.getTopic(queueKey)) .pollInterval(pollInterval) .msgPackProcessor(this::processMsgs) - .consumerCreator((config, partitionId) -> queueFactory.createToCalculatedFieldMsgConsumer()) + .consumerCreator((queueConfig, tpi) -> queueFactory.createToCalculatedFieldMsgConsumer(tpi)) .queueAdmin(queueFactory.getCalculatedFieldQueueAdmin()) .consumerExecutor(consumersExecutor) .scheduler(scheduler) @@ -127,7 +127,7 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractPartitionBa protected void onPartitionChangeEvent(PartitionChangeEvent event) { try { event.getNewPartitions().forEach((queueKey, partitions) -> { - if (queueKey.getQueueName().equals(DataConstants.CF_QUEUE_NAME)) { + if (DataConstants.CF_QUEUE_NAME.equals(queueKey.getQueueName())) { stateService.restore(queueKey, partitions); } }); @@ -175,9 +175,7 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractPartitionBa packSubmitFuture.cancel(true); log.info("Timeout to process message: {}", pendingMsgHolder.getMsg()); } - if (log.isDebugEnabled()) { - ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue())); - } + ctx.getAckMap().forEach((id, msg) -> log.warn("[{}] Timeout to process message: {}", id, msg.getValue())); ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue())); } consumer.commit(); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index fe2d291765..47a0c473e9 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -206,7 +206,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService queueFactory.createToCoreMsgConsumer()) + .consumerCreator((config, tpi) -> queueFactory.createToCoreMsgConsumer()) .consumerExecutor(consumersExecutor) .scheduler(scheduler) .taskExecutor(mgmtExecutor) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java index d3dc2932f0..a9ae18a11f 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbEdgeConsumerService.java @@ -102,7 +102,7 @@ public class DefaultTbEdgeConsumerService extends AbstractConsumerService queueFactory.createEdgeMsgConsumer()) + .consumerCreator((config, tpi) -> queueFactory.createEdgeMsgConsumer()) .consumerExecutor(consumersExecutor) .scheduler(scheduler) .taskExecutor(mgmtExecutor) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java index c7f7f600a7..879da5afc6 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java @@ -72,7 +72,12 @@ public class TbRuleEngineQueueConsumerManager extends MainQueueConsumerManager { + Integer partitionId = tpi != null ? tpi.getPartition().orElse(-1) : null; + return ctx.getQueueFactory().createToRuleEngineMsgConsumer(queueConfig, partitionId); + }, + consumerExecutor, scheduler, taskExecutor, null); this.ctx = ctx; this.stats = new TbRuleEngineConsumerStats(queueKey, ctx.getStatsFactory()); } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 51a9a5f907..abafd81e7c 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1848,6 +1848,10 @@ queue: pool_size: "${TB_QUEUE_CF_POOL_SIZE:8}" # RocksDB path for storing CF states rocks_db_path: "${TB_QUEUE_CF_ROCKS_DB_PATH:${user.home}/.rocksdb/cf_states}" + # The fetch size specifies how many rows will be fetched from the database per request for initial fetching + init_fetch_pack_size: "${TB_QUEUE_CF_FETCH_PACK_SIZE:50000}" + # The fetch size specifies how many rows will be fetched from the database per request for per-tenant fetching + init_tenant_fetch_pack_size: "${TB_QUEUE_CF_TENANT_FETCH_PACK_SIZE:1000}" transport: # For high-priority notifications that require minimum latency and processing time notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" 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 09bc8f1f93..c22c9c4140 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 @@ -66,6 +66,8 @@ public interface AssetService extends EntityDaoService { PageData findProfileEntityIdInfos(PageLink pageLink); + PageData findProfileEntityIdInfosByTenantId(TenantId tenantId, 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/cf/CalculatedFieldService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldService.java index 3d91790790..5101d6d57e 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldService.java @@ -39,6 +39,8 @@ public interface CalculatedFieldService extends EntityDaoService { PageData findAllCalculatedFields(PageLink pageLink); + PageData findCalculatedFieldsByTenantId(TenantId tenantId, PageLink pageLink); + PageData findAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink); void deleteCalculatedField(TenantId tenantId, CalculatedFieldId calculatedFieldId); @@ -53,6 +55,8 @@ public interface CalculatedFieldService extends EntityDaoService { List findAllCalculatedFieldLinksByEntityId(TenantId tenantId, EntityId entityId); + PageData findAllCalculatedFieldLinksByTenantId(TenantId tenantId, PageLink pageLink); + PageData findAllCalculatedFieldLinks(PageLink pageLink); boolean referencedInAnyCalculatedField(TenantId tenantId, EntityId referencedEntityId); 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 4ef653855d..9eb258f182 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 @@ -76,6 +76,8 @@ public interface DeviceService extends EntityDaoService { PageData findProfileEntityIdInfos(PageLink pageLink); + PageData findProfileEntityIdInfosByTenantId(TenantId tenantId, PageLink pageLink); + PageData findDevicesByTenantIdAndType(TenantId tenantId, String type, PageLink pageLink); PageData findDeviceIdsByTenantIdAndDeviceProfileId(TenantId tenantId, DeviceProfileId deviceProfileId, PageLink pageLink); diff --git a/common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java b/common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java index 07575220eb..78d17ef368 100644 --- a/common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java +++ b/common/edqs/src/main/java/org/thingsboard/server/edqs/processor/EdqsProcessor.java @@ -139,7 +139,7 @@ public class EdqsProcessor implements TbQueueHandler, } consumer.commit(); }) - .consumerCreator((config, partitionId) -> queueFactory.createEdqsEventsConsumer()) + .consumerCreator((config, tpi) -> queueFactory.createEdqsEventsConsumer()) .queueAdmin(queueFactory.getEdqsQueueAdmin()) .consumerExecutor(consumersExecutor) .taskExecutor(taskExecutor) diff --git a/common/edqs/src/main/java/org/thingsboard/server/edqs/state/KafkaEdqsStateService.java b/common/edqs/src/main/java/org/thingsboard/server/edqs/state/KafkaEdqsStateService.java index efdb1ead1c..0efe6e7d3b 100644 --- a/common/edqs/src/main/java/org/thingsboard/server/edqs/state/KafkaEdqsStateService.java +++ b/common/edqs/src/main/java/org/thingsboard/server/edqs/state/KafkaEdqsStateService.java @@ -59,7 +59,8 @@ public class KafkaEdqsStateService implements EdqsStateService { private final EdqsConfig config; private final EdqsPartitionService partitionService; private final KafkaEdqsQueueFactory queueFactory; - @Autowired @Lazy + @Autowired + @Lazy private EdqsProcessor edqsProcessor; private PartitionedQueueConsumerManager> stateConsumer; @@ -93,7 +94,7 @@ public class KafkaEdqsStateService implements EdqsStateService { } consumer.commit(); }) - .consumerCreator((config, partitionId) -> queueFactory.createEdqsStateConsumer()) + .consumerCreator((config, tpi) -> queueFactory.createEdqsStateConsumer()) .queueAdmin(queueAdmin) .consumerExecutor(eventConsumer.getConsumerExecutor()) .taskExecutor(eventConsumer.getTaskExecutor()) diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java index 178caf7961..f1c404ce16 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java @@ -136,6 +136,8 @@ public enum MsgType { EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG, + CF_CACHE_INIT_MSG, // Sent to init caches for CF actor; + CF_INIT_PROFILE_ENTITY_MSG, // Sent to init profile entities cache; CF_INIT_MSG, // Sent to init particular calculated field; CF_LINK_INIT_MSG, // Sent to init particular calculated field; CF_STATE_RESTORE_MSG, // Sent to restore particular calculated field entity state; diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldCacheInitMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldCacheInitMsg.java new file mode 100644 index 0000000000..bf2054dfcf --- /dev/null +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldCacheInitMsg.java @@ -0,0 +1,33 @@ +/** + * Copyright © 2016-2025 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.msg.cf; + +import lombok.Data; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.MsgType; +import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; + +@Data +public class CalculatedFieldCacheInitMsg implements ToCalculatedFieldSystemMsg { + + private final TenantId tenantId; + + @Override + public MsgType getMsgType() { + return MsgType.CF_CACHE_INIT_MSG; + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/cf/cache/CalculatedFieldEntityProfileCache.java b/common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldInitProfileEntityMsg.java similarity index 51% rename from application/src/main/java/org/thingsboard/server/service/cf/cache/CalculatedFieldEntityProfileCache.java rename to common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldInitProfileEntityMsg.java index bb5ef91974..66cd2ac441 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/cache/CalculatedFieldEntityProfileCache.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldInitProfileEntityMsg.java @@ -13,24 +13,24 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.service.cf.cache; +package org.thingsboard.server.common.msg.cf; -import org.springframework.context.ApplicationListener; +import lombok.Data; 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 org.thingsboard.server.common.msg.MsgType; +import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; -import java.util.Collection; +@Data +public class CalculatedFieldInitProfileEntityMsg implements ToCalculatedFieldSystemMsg { -public interface CalculatedFieldEntityProfileCache extends ApplicationListener { + private final TenantId tenantId; + private final EntityId profileEntityId; + private final EntityId entityId; - void add(TenantId tenantId, EntityId profileId, EntityId entityId); + @Override + public MsgType getMsgType() { + return MsgType.CF_INIT_PROFILE_ENTITY_MSG; + } - void update(TenantId tenantId, EntityId oldProfileId, EntityId newProfileId, EntityId entityId); - - void evict(TenantId tenantId, EntityId entityId); - - Collection getMyEntityIdsByProfileId(TenantId tenantId, EntityId profileId); - - int getEntityIdPartition(TenantId tenantId, EntityId entityId); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/MainQueueConsumerManager.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/MainQueueConsumerManager.java index ef9728344c..2233855a37 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/MainQueueConsumerManager.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/MainQueueConsumerManager.java @@ -54,7 +54,7 @@ public class MainQueueConsumerManager msgPackProcessor; - protected final BiFunction> consumerCreator; + protected final BiFunction> consumerCreator; @Getter protected final ExecutorService consumerExecutor; @Getter @@ -74,7 +74,7 @@ public class MainQueueConsumerManager msgPackProcessor, - BiFunction> consumerCreator, + BiFunction> consumerCreator, ExecutorService consumerExecutor, ScheduledExecutorService scheduler, ExecutorService taskExecutor, @@ -313,7 +313,7 @@ public class MainQueueConsumerManager onStop.accept(tpi) : null; TbQueueConsumerTask consumer = new TbQueueConsumerTask<>(key, () -> { - TbQueueConsumer queueConsumer = consumerCreator.apply(config, partitionId); + TbQueueConsumer queueConsumer = consumerCreator.apply(config, tpi); if (startOffsetProvider != null && queueConsumer instanceof TbKafkaConsumerTemplate kafkaConsumer) { kafkaConsumer.setStartOffsetProvider(startOffsetProvider); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/PartitionedQueueConsumerManager.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/PartitionedQueueConsumerManager.java index 0de1e53753..1b19fcab0e 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/PartitionedQueueConsumerManager.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/PartitionedQueueConsumerManager.java @@ -45,7 +45,7 @@ public class PartitionedQueueConsumerManager extends MainQ @Builder(builderMethodName = "create") // not to conflict with super.builder() public PartitionedQueueConsumerManager(QueueKey queueKey, String topic, long pollInterval, MsgPackProcessor msgPackProcessor, - BiFunction> consumerCreator, TbQueueAdmin queueAdmin, + BiFunction> consumerCreator, TbQueueAdmin queueAdmin, ExecutorService consumerExecutor, ScheduledExecutorService scheduler, ExecutorService taskExecutor, Consumer uncaughtErrorHandler) { super(queueKey, QueueConfig.of(true, pollInterval), msgPackProcessor, consumerCreator, consumerExecutor, scheduler, taskExecutor, uncaughtErrorHandler); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java index 89d83af826..085d04f28c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java @@ -22,6 +22,7 @@ import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; @@ -133,7 +134,7 @@ public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE } @Override - public TbQueueConsumer> createToCalculatedFieldMsgConsumer() { + public TbQueueConsumer> createToCalculatedFieldMsgConsumer(TopicPartitionInfo tpi) { return new InMemoryTbQueueConsumer<>(storage, topicService.buildTopicName(calculatedFieldSettings.getEventTopic())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java index dcebe085b3..0738f660df 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java @@ -20,10 +20,12 @@ import jakarta.annotation.PreDestroy; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; import org.thingsboard.server.gen.transport.TransportProtos.FromEdqsMsg; @@ -513,15 +515,24 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi } @Override - public TbQueueConsumer> createToCalculatedFieldMsgConsumer() { + public TbQueueConsumer> createToCalculatedFieldMsgConsumer(TopicPartitionInfo tpi) { + String queueName = DataConstants.CF_QUEUE_NAME; + if (tpi == null) { + throw new IllegalArgumentException("TopicPartitionInfo is required."); + } + TenantId tenantId = tpi.getTenantId().orElse(TenantId.SYS_TENANT_ID); + Integer partitionId = tpi.getPartition().orElseThrow(() -> new IllegalArgumentException("PartitionId is required.")); + String groupId = topicService.buildConsumerGroupId("cf-", tenantId, queueName, partitionId); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic())); - consumerBuilder.clientId("monolith-calculated-field-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); - consumerBuilder.groupId(topicService.buildTopicName("monolith-calculated-field-consumer")); + consumerBuilder.clientId("cf-" + queueName + "-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); + consumerBuilder.groupId(groupId); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCalculatedFieldMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(cfAdmin); consumerBuilder.statsService(consumerStatsService); + return consumerBuilder.build(); } @@ -571,7 +582,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi .readFromBeginning(true) .stopWhenRead(true) .clientId("monolith-calculated-field-state-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()) - .groupId(topicService.buildTopicName("monolith-calculated-field-state-consumer")) + .groupId(null) // not using consumer group management .decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), msg.getData() != null ? CalculatedFieldStateProto.parseFrom(msg.getData()) : null, msg.getHeaders())) .admin(cfStateAdmin) .statsService(consumerStatsService) diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java index d43ef5c9ac..b4884ae72c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java @@ -20,8 +20,11 @@ import jakarta.annotation.PreDestroy; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; import org.thingsboard.server.gen.transport.TransportProtos.FromEdqsMsg; @@ -313,12 +316,20 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { } @Override - public TbQueueConsumer> createToCalculatedFieldMsgConsumer() { + public TbQueueConsumer> createToCalculatedFieldMsgConsumer(TopicPartitionInfo tpi) { + String queueName = DataConstants.CF_QUEUE_NAME; + if (tpi == null) { + throw new IllegalArgumentException("TopicPartitionInfo is required."); + } + TenantId tenantId = tpi.getTenantId().orElse(TenantId.SYS_TENANT_ID); + Integer partitionId = tpi.getPartition().orElseThrow(() -> new IllegalArgumentException("PartitionId is required.")); + String groupId = topicService.buildConsumerGroupId("cf-", tenantId, queueName, partitionId); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(topicService.buildTopicName(calculatedFieldSettings.getEventTopic())); - consumerBuilder.clientId("tb-rule-engine-calculated-field-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); - consumerBuilder.groupId(topicService.buildTopicName("tb-rule-engine-calculated-field-consumer")); + consumerBuilder.clientId("cf-" + queueName + "-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()); + consumerBuilder.groupId(groupId); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCalculatedFieldMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.admin(cfAdmin); consumerBuilder.statsService(consumerStatsService); @@ -371,7 +382,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { .readFromBeginning(true) .stopWhenRead(true) .clientId("tb-rule-engine-calculated-field-state-consumer-" + serviceInfoProvider.getServiceId() + "-" + consumerCount.incrementAndGet()) - .groupId(topicService.buildTopicName("tb-rule-engine-calculated-field-state-consumer")) + .groupId(null) // not using consumer group management .decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), msg.getData() != null ? CalculatedFieldStateProto.parseFrom(msg.getData()) : null, msg.getHeaders())) .admin(cfStateAdmin) .statsService(consumerStatsService) diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java index 329e3e346a..18bb6db14a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java @@ -16,6 +16,7 @@ package org.thingsboard.server.queue.provider; import org.thingsboard.server.common.data.queue.Queue; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldMsg; @@ -95,8 +96,8 @@ public interface TbRuleEngineQueueFactory extends TbUsageStatsClientQueueFactory /** * Used to consume messages by TB Rule Engine Service * - * @return * @param configuration + * @return */ TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration); @@ -104,9 +105,9 @@ public interface TbRuleEngineQueueFactory extends TbUsageStatsClientQueueFactory * Used to consume messages by TB Rule Engine Service * Intended usage for consumer per partition strategy * - * @return TbQueueConsumer * @param configuration - * @param partitionId as a suffix for consumer name + * @param partitionId as a suffix for consumer name + * @return TbQueueConsumer */ default TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration, Integer partitionId) { return createToRuleEngineMsgConsumer(configuration); @@ -121,7 +122,7 @@ public interface TbRuleEngineQueueFactory extends TbUsageStatsClientQueueFactory TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate(); - TbQueueConsumer> createToCalculatedFieldMsgConsumer(); + TbQueueConsumer> createToCalculatedFieldMsgConsumer(TopicPartitionInfo tpi); TbQueueAdmin getCalculatedFieldQueueAdmin(); 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 46c29b867b..6a3aaa4a7f 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 @@ -36,10 +36,11 @@ public @interface AfterStartUp { int STARTUP_SERVICE = 8; int ACTOR_SYSTEM = 9; - int REGULAR_SERVICE = 10; int CF_READ_PROFILE_ENTITIES_SERVICE = 10; - int CF_READ_CF_SERVICE = 11; + int CF_READ_CF_SERVICE = 10; + + int REGULAR_SERVICE = 11; 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 36700ff59f..098dc4e83d 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 @@ -240,4 +240,6 @@ public interface AssetDao extends Dao, TenantEntityDao, Exportable PageData findProfileEntityIdInfos(PageLink pageLink); + PageData findProfileEntityIdInfosByTenantId(UUID tenantId, 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 21d5ea7f4e..fed201d403 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 @@ -293,6 +293,14 @@ public class BaseAssetService extends AbstractCachedEntityService findProfileEntityIdInfosByTenantId(TenantId tenantId, PageLink pageLink) { + log.trace("Executing findProfileEntityIdInfosByTenantId, tenantId[{}], pageLink [{}]", tenantId, pageLink); + validateId(tenantId, id -> INCORRECT_TENANT_ID + id); + validatePageLink(pageLink); + return assetDao.findProfileEntityIdInfosByTenantId(tenantId.getId(), 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/cf/BaseCalculatedFieldService.java b/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java index dceef73aba..0c5df18e80 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java @@ -104,6 +104,14 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements return calculatedFieldDao.findAll(pageLink); } + @Override + public PageData findCalculatedFieldsByTenantId(TenantId tenantId, PageLink pageLink) { + log.trace("Executing findAllByTenantId, tenantId [{}], pageLink [{}]", tenantId, pageLink); + validateId(tenantId, id -> INCORRECT_TENANT_ID + id); + validatePageLink(pageLink); + return calculatedFieldDao.findAllByTenantId(tenantId, pageLink); + } + @Override public PageData findAllCalculatedFieldsByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink) { log.trace("Executing findAllByEntityId, entityId [{}], pageLink [{}]", entityId, pageLink); @@ -174,6 +182,14 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements return calculatedFieldLinkDao.findCalculatedFieldLinksByEntityId(tenantId, entityId); } + @Override + public PageData findAllCalculatedFieldLinksByTenantId(TenantId tenantId, PageLink pageLink) { + log.trace("Executing findAllCalculatedFieldLinksByTenantId, tenantId[{}] pageLink [{}]", tenantId, pageLink); + validateId(tenantId, id -> INCORRECT_TENANT_ID + id); + validatePageLink(pageLink); + return calculatedFieldLinkDao.findAllByTenantId(tenantId, pageLink); + } + @Override public PageData findAllCalculatedFieldLinks(PageLink pageLink) { log.trace("Executing findAllCalculatedFieldLinks, pageLink [{}]", pageLink); diff --git a/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldDao.java b/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldDao.java index a966977968..aadae93893 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldDao.java @@ -37,6 +37,8 @@ public interface CalculatedFieldDao extends Dao { PageData findAll(PageLink pageLink); + PageData findAllByTenantId(TenantId tenantId, PageLink pageLink); + PageData findAllByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink); List removeAllByEntityId(TenantId tenantId, EntityId entityId); diff --git a/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldLinkDao.java b/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldLinkDao.java index 8b4a5e7086..dd184289ed 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldLinkDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/cf/CalculatedFieldLinkDao.java @@ -31,8 +31,12 @@ public interface CalculatedFieldLinkDao extends Dao { List findCalculatedFieldLinksByEntityId(TenantId tenantId, EntityId entityId); + List findCalculatedFieldLinksByTenantId(TenantId tenantId); + List findAll(); PageData findAll(PageLink pageLink); + PageData findAllByTenantId(TenantId tenantId, PageLink pageLink); + } 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 efc57119eb..6bc8903d20 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 @@ -23,9 +23,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.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; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.ota.OtaPackageType; import org.thingsboard.server.common.data.page.PageData; @@ -233,6 +231,8 @@ public interface DeviceDao extends Dao, TenantEntityDao, Exporta PageData findProfileEntityIdInfos(PageLink pageLink); + PageData findProfileEntityIdInfosByTenantId(UUID tenantId, 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 2c890082a0..6d993f3e3d 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 @@ -394,6 +394,14 @@ public class DeviceServiceImpl extends CachedVersionedEntityService findProfileEntityIdInfosByTenantId(TenantId tenantId, PageLink pageLink) { + log.trace("Executing findProfileEntityIdInfosByTenantId, tenantId[{}], pageLink [{}]", tenantId, pageLink); + validateId(tenantId, id -> INCORRECT_TENANT_ID + id); + validatePageLink(pageLink); + return deviceDao.findProfileEntityIdInfosByTenantId(tenantId.getId(), 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 c61d894de5..4e99fb57e4 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 @@ -39,14 +39,12 @@ 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; import java.util.List; import java.util.Optional; import java.util.UUID; -import java.util.stream.Collectors; import static org.thingsboard.server.dao.DaoUtil.convertTenantEntityInfosToDto; @@ -262,10 +260,16 @@ public class JpaAssetDao extends JpaAbstractDao implements A @Override public PageData findProfileEntityIdInfos(PageLink pageLink) { - log.debug("Find profile device id infos by pageLink [{}]", pageLink); + log.debug("Find profile asset id infos by pageLink [{}]", pageLink); return nativeAssetRepository.findProfileEntityIdInfos(DaoUtil.toPageable(pageLink)); } + @Override + public PageData findProfileEntityIdInfosByTenantId(UUID tenantId, PageLink pageLink) { + log.debug("Find profile asset id infos by pageLink [{}]", pageLink); + return nativeAssetRepository.findProfileEntityIdInfosByTenantId(tenantId, 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/cf/CalculatedFieldLinkRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldLinkRepository.java index 584a3b5199..6f6a0775f3 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldLinkRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldLinkRepository.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.dao.sql.cf; +import org.springframework.data.domain.Page; +import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; import org.thingsboard.server.dao.model.sql.CalculatedFieldLinkEntity; @@ -27,4 +29,8 @@ public interface CalculatedFieldLinkRepository extends JpaRepository findAllByTenantIdAndEntityId(UUID tenantId, UUID entityId); + List findAllByTenantId(UUID tenantId); + + Page findAllByTenantId(UUID tenantId, Pageable pageable); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldRepository.java index 0f48f3b00d..be122816ba 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/CalculatedFieldRepository.java @@ -32,6 +32,8 @@ public interface CalculatedFieldRepository extends JpaRepository findAllByTenantIdAndEntityId(UUID tenantId, UUID entityId); + Page findAllByTenantId(UUID tenantId, Pageable pageable); + Page findAllByTenantIdAndEntityId(UUID tenantId, UUID entityId, Pageable pageable); List findAllByTenantId(UUID tenantId); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldDao.java index 8922eaca4e..4bb52c29db 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldDao.java @@ -71,6 +71,12 @@ public class JpaCalculatedFieldDao extends JpaAbstractDao findAllByTenantId(TenantId tenantId, PageLink pageLink) { + log.debug("Try to find calculated fields by tenantId[{}] and pageLink [{}]", tenantId, pageLink); + return DaoUtil.toPageData(calculatedFieldRepository.findAllByTenantId(tenantId.getId(), DaoUtil.toPageable(pageLink))); + } + @Override public PageData findAllByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink) { log.debug("Try to find calculated fields by entityId[{}] and pageLink [{}]", entityId, pageLink); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldLinkDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldLinkDao.java index dbb2fd87da..38871f2fb8 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldLinkDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/cf/JpaCalculatedFieldLinkDao.java @@ -54,6 +54,11 @@ public class JpaCalculatedFieldLinkDao extends JpaAbstractDao findCalculatedFieldLinksByTenantId(TenantId tenantId) { + return DaoUtil.convertDataList(calculatedFieldLinkRepository.findAllByTenantId(tenantId.getId())); + } + @Override public List findAll() { return DaoUtil.convertDataList(calculatedFieldLinkRepository.findAll()); @@ -65,6 +70,12 @@ public class JpaCalculatedFieldLinkDao extends JpaAbstractDao findAllByTenantId(TenantId tenantId, PageLink pageLink) { + log.debug("Try to find calculated field links by tenantId [{}], pageLink [{}]", tenantId, pageLink); + return DaoUtil.toPageData(calculatedFieldLinkRepository.findAllByTenantId(tenantId.getId(), DaoUtil.toPageable(pageLink))); + } + @Override protected Class getEntityClass() { return CalculatedFieldLinkEntity.class; 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 index 43d66e2ff0..ec47da3499 100644 --- 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 @@ -20,12 +20,9 @@ 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; @@ -43,8 +40,8 @@ public class DefaultNativeAssetRepository extends AbstractNativeRepository imple @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 -> { + String PROFILE_ASSET_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_ASSET_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"); @@ -52,4 +49,14 @@ public class DefaultNativeAssetRepository extends AbstractNativeRepository imple }); } + @Override + public PageData findProfileEntityIdInfosByTenantId(UUID tenantId, Pageable pageable) { + String PROFILE_ASSET_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, asset_profile_id as profileId, id as id FROM asset WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId); + return find(COUNT_QUERY, PROFILE_ASSET_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 776dedc2d5..78ee2795b0 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 @@ -61,4 +61,15 @@ public class DefaultNativeDeviceRepository extends AbstractNativeRepository impl }); } + @Override + public PageData findProfileEntityIdInfosByTenantId(UUID tenantId, Pageable pageable) { + String PROFILE_DEVICE_ID_INFO_QUERY = String.format("SELECT tenant_id as tenantId, device_profile_id as profileId, id as id FROM device WHERE tenant_id = '%s' ORDER BY created_time ASC LIMIT %%s OFFSET %%s", tenantId); + 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 34835f52f1..9c79637e39 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 @@ -179,11 +179,11 @@ public class JpaDeviceDao extends JpaAbstractDao implement @Override public PageData findDeviceIdsByTenantIdAndDeviceProfileId(UUID tenantId, UUID deviceProfileId, PageLink pageLink) { return DaoUtil.pageToPageData( - deviceRepository.findIdsByTenantIdAndDeviceProfileId( - tenantId, - deviceProfileId, - pageLink.getTextSearch(), - DaoUtil.toPageable(pageLink))) + deviceRepository.findIdsByTenantIdAndDeviceProfileId( + tenantId, + deviceProfileId, + pageLink.getTextSearch(), + DaoUtil.toPageable(pageLink))) .mapData(DeviceId::new); } @@ -281,6 +281,12 @@ public class JpaDeviceDao extends JpaAbstractDao implement return nativeDeviceRepository.findProfileEntityIdInfos(DaoUtil.toPageable(pageLink)); } + @Override + public PageData findProfileEntityIdInfosByTenantId(UUID tenantId, PageLink pageLink) { + log.debug("Find profile device id infos by tenantId[{}], pageLink [{}]", tenantId, pageLink); + return nativeDeviceRepository.findProfileEntityIdInfosByTenantId(tenantId, 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/NativeProfileEntityRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/device/NativeProfileEntityRepository.java index 750f0c8787..3568f01851 100644 --- 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 @@ -19,8 +19,12 @@ import org.springframework.data.domain.Pageable; import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.page.PageData; +import java.util.UUID; + public interface NativeProfileEntityRepository { PageData findProfileEntityIdInfos(Pageable pageable); + PageData findProfileEntityIdInfosByTenantId(UUID tenantId, Pageable pageable); + }