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 426fc705c9..3ccc9bbaef 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -76,6 +76,8 @@ import org.thingsboard.server.dao.tenant.TenantProfileService; import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.user.UserService; +import org.thingsboard.server.dao.widget.WidgetTypeService; +import org.thingsboard.server.dao.widget.WidgetsBundleService; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.util.DataDecodingEncodingService; @@ -379,6 +381,16 @@ public class ActorSystemContext { @Getter private QueueService queueService; + @Lazy + @Autowired(required = false) + @Getter + private WidgetsBundleService widgetsBundleService; + + @Lazy + @Autowired(required = false) + @Getter + private WidgetTypeService widgetTypeService; + @Value("${actors.session.max_concurrent_sessions_per_device:1}") @Getter private long maxConcurrentSessionsPerDevice; diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 3afcf551ef..3e4e459192 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -26,6 +26,7 @@ import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ListeningExecutor; import org.thingsboard.rule.engine.api.MailService; import org.thingsboard.rule.engine.api.RuleEngineAlarmService; +import org.thingsboard.rule.engine.api.RuleEngineApiUsageStateService; import org.thingsboard.rule.engine.api.RuleEngineAssetProfileCache; import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; import org.thingsboard.rule.engine.api.RuleEngineRpcService; @@ -35,6 +36,7 @@ import org.thingsboard.rule.engine.api.SmsService; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbRelationTypes; import org.thingsboard.rule.engine.api.sms.SmsSenderFactory; +import org.thingsboard.rule.engine.util.EntitiesTenantIdAsyncLoader; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.TbActorRef; import org.thingsboard.server.cluster.TbClusterService; @@ -89,6 +91,8 @@ import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.user.UserService; +import org.thingsboard.server.dao.widget.WidgetTypeService; +import org.thingsboard.server.dao.widget.WidgetsBundleService; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueMsgMetadata; @@ -751,12 +755,41 @@ class DefaultTbContext implements TbContext { return mainCtx.getTenantProfileCache().get(getTenantId()); } + @Override + public WidgetsBundleService getWidgetBundleService() { + return mainCtx.getWidgetsBundleService(); + } + + @Override + public WidgetTypeService getWidgetTypeService() { + return mainCtx.getWidgetTypeService(); + } + + @Override + public RuleEngineApiUsageStateService getRuleEngineApiUsageStateService() { + return mainCtx.getApiUsageStateService(); + } + private TbMsgMetaData getActionMetaData(RuleNodeId ruleNodeId) { TbMsgMetaData metaData = new TbMsgMetaData(); metaData.putValue("ruleNodeId", ruleNodeId.toString()); return metaData; } + public void isTenantEntity(EntityId entityId) { + try { + TenantId entityTenantId = EntitiesTenantIdAsyncLoader.findEntityIdAsync(this, entityId).get(); + if (entityTenantId == null) { + throw new RuntimeException("Entity with id '" + entityId + "'not found!"); + } + if (!entityTenantId.equals(this.getTenantId())) { + throw new RuntimeException("Entity with id: '" + entityId + "' specified in the configuration doesn't belong to the current tenant."); + } + } catch (Exception e) { + throw new RuntimeException(e); + } + } + private class SimpleTbQueueCallback implements TbQueueCallback { private final Runnable onSuccess; private final Consumer onFailure; diff --git a/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java b/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java index f6a590eee2..b8ca8b37c0 100644 --- a/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java @@ -395,6 +395,11 @@ public class DefaultTbApiUsageStateService extends AbstractPartitionBasedService } } + @Override + public ApiUsageState findApiUsageStateById(TenantId tenantId, ApiUsageStateId id) { + return apiUsageStateService.findApiUsageStateById(tenantId, id); + } + private interface StateChecker { boolean check(long threshold, long warnThreshold, long value); } diff --git a/application/src/main/java/org/thingsboard/server/service/apiusage/TbApiUsageStateService.java b/application/src/main/java/org/thingsboard/server/service/apiusage/TbApiUsageStateService.java index 172820a667..668dc288a0 100644 --- a/application/src/main/java/org/thingsboard/server/service/apiusage/TbApiUsageStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/apiusage/TbApiUsageStateService.java @@ -16,6 +16,7 @@ package org.thingsboard.server.service.apiusage; import org.springframework.context.ApplicationListener; +import org.thingsboard.rule.engine.api.RuleEngineApiUsageStateService; import org.thingsboard.server.common.data.ApiUsageState; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.TenantId; @@ -26,10 +27,12 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceM import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; -public interface TbApiUsageStateService extends TbApiUsageStateClient, ApplicationListener { +public interface TbApiUsageStateService extends TbApiUsageStateClient, RuleEngineApiUsageStateService, ApplicationListener { void process(TbProtoQueueMsg msg, TbCallback callback); + ApiUsageState getApiUsageState(TenantId tenantId); + void onTenantProfileUpdate(TenantProfileId tenantProfileId); void onTenantUpdate(TenantId tenantId); diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java index 493ab7ef43..96cc19ef70 100644 --- a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java @@ -15,10 +15,14 @@ */ package org.thingsboard.server.service.rpc; +import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.common.data.id.RpcId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.rpc.Rpc; import org.thingsboard.server.common.data.rpc.RpcError; import org.thingsboard.rule.engine.api.RuleEngineDeviceRpcRequest; import org.thingsboard.rule.engine.api.RuleEngineDeviceRpcResponse; @@ -27,6 +31,7 @@ import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse; import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; +import org.thingsboard.server.dao.rpc.RpcService; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; @@ -52,6 +57,7 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi private final PartitionService partitionService; private final TbClusterService clusterService; private final TbServiceInfoProvider serviceInfoProvider; + private final RpcService rpcService; private final ConcurrentMap> toDeviceRpcRequests = new ConcurrentHashMap<>(); @@ -61,10 +67,11 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi public DefaultTbRuleEngineRpcService(PartitionService partitionService, TbClusterService clusterService, - TbServiceInfoProvider serviceInfoProvider) { + TbServiceInfoProvider serviceInfoProvider, RpcService rpcService) { this.partitionService = partitionService; this.clusterService = clusterService; this.serviceInfoProvider = serviceInfoProvider; + this.rpcService = rpcService; } @Autowired(required = false) @@ -119,6 +126,11 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi }); } + @Override + public ListenableFuture findRpcByIdAsync(TenantId tenantId, RpcId id) { + return rpcService.findRpcByIdAsync(tenantId, id); + } + @Override public void processRpcResponseFromDevice(FromDeviceRpcResponse response) { log.trace("[{}] Received response to server-side RPC request from Core RPC Service", response.getId()); diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineApiUsageStateService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineApiUsageStateService.java new file mode 100644 index 0000000000..03734ffcad --- /dev/null +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineApiUsageStateService.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2022 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.rule.engine.api; + +import org.thingsboard.server.common.data.ApiUsageState; +import org.thingsboard.server.common.data.id.ApiUsageStateId; +import org.thingsboard.server.common.data.id.TenantId; + +public interface RuleEngineApiUsageStateService { + + ApiUsageState findApiUsageStateById(TenantId tenantId, ApiUsageStateId id); + +} diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineRpcService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineRpcService.java index abb5ace0f8..aec2cd4e00 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineRpcService.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineRpcService.java @@ -15,7 +15,11 @@ */ package org.thingsboard.rule.engine.api; +import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.RpcId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.rpc.Rpc; import java.util.UUID; import java.util.function.Consumer; @@ -29,4 +33,5 @@ public interface RuleEngineRpcService { void sendRpcRequestToDevice(RuleEngineDeviceRpcRequest request, Consumer consumer); + ListenableFuture findRpcByIdAsync(TenantId tenantId, RpcId id); } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 2f87f0699b..4a47586f64 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -62,6 +62,8 @@ import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.user.UserService; +import org.thingsboard.server.dao.widget.WidgetTypeService; +import org.thingsboard.server.dao.widget.WidgetsBundleService; import java.util.List; import java.util.Set; @@ -314,4 +316,9 @@ public interface TbContext { TenantProfile getTenantProfile(); + WidgetsBundleService getWidgetBundleService(); + + WidgetTypeService getWidgetTypeService(); + + RuleEngineApiUsageStateService getRuleEngineApiUsageStateService(); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java index 03df255e5d..f41ba4d8cf 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java @@ -78,6 +78,7 @@ public class TbMsgGeneratorNode implements TbNode { this.currentMsgCount = 0; if (!StringUtils.isEmpty(config.getOriginatorId())) { originatorId = EntityIdFactory.getByTypeAndUuid(config.getOriginatorType(), config.getOriginatorId()); + ctx.isTenantEntity(originatorId); } else { originatorId = ctx.getSelfId(); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNode.java index 4832a0af35..c6707cbd1e 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNode.java @@ -55,10 +55,15 @@ import static org.thingsboard.common.util.DonAsynchron.withCallback; public class TbCheckRelationNode implements TbNode { private TbCheckRelationNodeConfiguration config; + private EntityId singleEntityId; @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { this.config = TbNodeUtils.convert(configuration, TbCheckRelationNodeConfiguration.class); + if (config.isCheckForSingleEntity()) { + this.singleEntityId = EntityIdFactory.getByTypeAndId(config.getEntityType(), config.getEntityId()); + ctx.isTenantEntity(singleEntityId); + } } @Override @@ -76,10 +81,10 @@ public class TbCheckRelationNode implements TbNode { EntityId from; EntityId to; if (EntitySearchDirection.FROM.name().equals(config.getDirection())) { - from = EntityIdFactory.getByTypeAndId(config.getEntityType(), config.getEntityId()); + from = singleEntityId; to = msg.getOriginator(); } else { - to = EntityIdFactory.getByTypeAndId(config.getEntityType(), config.getEntityId()); + to = singleEntityId; from = msg.getOriginator(); } return ctx.getRelationService().checkRelationAsync(ctx.getTenantId(), from, to, config.getRelationType(), RelationTypeGroup.COMMON); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesTenantIdAsyncLoader.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesTenantIdAsyncLoader.java index 4d57e89fda..6eb4355dd3 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesTenantIdAsyncLoader.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/util/EntitiesTenantIdAsyncLoader.java @@ -17,50 +17,123 @@ package org.thingsboard.rule.engine.util; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.MoreExecutors; +import org.thingsboard.common.util.ListeningExecutor; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.HasTenantId; import org.thingsboard.server.common.data.id.AlarmId; +import org.thingsboard.server.common.data.id.ApiUsageStateId; 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.DashboardId; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.DeviceProfileId; +import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityViewId; +import org.thingsboard.server.common.data.id.OtaPackageId; +import org.thingsboard.server.common.data.id.QueueId; +import org.thingsboard.server.common.data.id.RpcId; import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; +import org.thingsboard.server.common.data.id.TbResourceId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; +import org.thingsboard.server.common.data.id.WidgetTypeId; +import org.thingsboard.server.common.data.id.WidgetsBundleId; +import org.thingsboard.server.common.data.rule.RuleNode; + +import java.util.UUID; public class EntitiesTenantIdAsyncLoader { - /** - * @deprecated consider to remove since tenantId is already defined in the TbContext. - */ - @Deprecated - public static ListenableFuture findEntityIdAsync(TbContext ctx, EntityId original) { - switch (original.getEntityType()) { + public static ListenableFuture findEntityIdAsync(TbContext ctx, EntityId original) { + ListenableFuture hasTenantId; + UUID id = original.getId(); + EntityType entityType = original.getEntityType(); + TenantId tenantId = ctx.getTenantId(); + switch (entityType) { case TENANT: - return Futures.immediateFuture((TenantId) original); + return Futures.immediateFuture(new TenantId(id)); case CUSTOMER: - return getTenantAsync(ctx.getCustomerService().findCustomerByIdAsync(ctx.getTenantId(), (CustomerId) original)); + hasTenantId = ctx.getCustomerService().findCustomerByIdAsync(tenantId, new CustomerId(id)); + break; case USER: - return getTenantAsync(ctx.getUserService().findUserByIdAsync(ctx.getTenantId(), (UserId) original)); + hasTenantId = ctx.getUserService().findUserByIdAsync(tenantId, new UserId(id)); + break; case ASSET: - return getTenantAsync(ctx.getAssetService().findAssetByIdAsync(ctx.getTenantId(), (AssetId) original)); + hasTenantId = ctx.getAssetService().findAssetByIdAsync(tenantId, new AssetId(id)); + break; case DEVICE: - return getTenantAsync(ctx.getDeviceService().findDeviceByIdAsync(ctx.getTenantId(), (DeviceId) original)); + hasTenantId = ctx.getDeviceService().findDeviceByIdAsync(tenantId, new DeviceId(id)); + break; case ALARM: - return getTenantAsync(ctx.getAlarmService().findAlarmByIdAsync(ctx.getTenantId(), (AlarmId) original)); + hasTenantId = ctx.getAlarmService().findAlarmByIdAsync(tenantId, new AlarmId(id)); + break; case RULE_CHAIN: - return getTenantAsync(ctx.getRuleChainService().findRuleChainByIdAsync(ctx.getTenantId(), (RuleChainId) original)); + hasTenantId = ctx.getRuleChainService().findRuleChainByIdAsync(tenantId, new RuleChainId(id)); + break; + case ENTITY_VIEW: + hasTenantId = ctx.getEntityViewService().findEntityViewByIdAsync(tenantId, new EntityViewId(id)); + break; + case DASHBOARD: + hasTenantId = ctx.getDashboardService().findDashboardByIdAsync(tenantId, new DashboardId(id)); + break; + case EDGE: + hasTenantId = ctx.getEdgeService().findEdgeByIdAsync(tenantId, new EdgeId(id)); + break; + case OTA_PACKAGE: + hasTenantId = ctx.getOtaPackageService().findOtaPackageInfoByIdAsync(tenantId, new OtaPackageId(id)); + break; + case ASSET_PROFILE: + hasTenantId = Futures.immediateFuture(ctx.getAssetProfileCache().get(tenantId, new AssetProfileId(id))); + break; + case DEVICE_PROFILE: + hasTenantId = Futures.immediateFuture(ctx.getDeviceProfileCache().get(tenantId, new DeviceProfileId(id))); + break; + case WIDGET_TYPE: + hasTenantId = Futures.immediateFuture(ctx.getWidgetTypeService().findWidgetTypeById(tenantId, new WidgetTypeId(id))); + break; + case WIDGETS_BUNDLE: + hasTenantId = Futures.immediateFuture(ctx.getWidgetBundleService().findWidgetsBundleById(tenantId, new WidgetsBundleId(id))); + break; + case RPC: + hasTenantId = ctx.getRpcService().findRpcByIdAsync(tenantId, new RpcId(id)); + break; + case QUEUE: + hasTenantId = Futures.immediateFuture(ctx.getQueueService().findQueueById(tenantId, new QueueId(id))); + break; + case API_USAGE_STATE: + hasTenantId = Futures.immediateFuture(ctx.getRuleEngineApiUsageStateService().findApiUsageStateById(tenantId, new ApiUsageStateId(id))); + break; + case TB_RESOURCE: + hasTenantId = ctx.getResourceService().findResourceInfoByIdAsync(tenantId, new TbResourceId(id)); + break; + case RULE_NODE: + RuleNode ruleNode = ctx.getRuleChainService().findRuleNodeById(tenantId, new RuleNodeId(id)); + if (ruleNode != null) { + hasTenantId = ctx.getRuleChainService().findRuleChainByIdAsync(tenantId, ruleNode.getRuleChainId()); + } else { + hasTenantId = Futures.immediateFuture(null); + } + break; + case TENANT_PROFILE: + if (ctx.getTenantProfile().getId().equals(id)) { + return Futures.immediateFuture(tenantId); + } else { + hasTenantId = Futures.immediateFuture(null); + } + break; default: - return Futures.immediateFailedFuture(new TbNodeException("Unexpected original EntityType " + original.getEntityType())); + hasTenantId = Futures.immediateFailedFuture(new TbNodeException("Unexpected original EntityType " + original.getEntityType())); } + return getTenantIdAsync(hasTenantId, ctx.getDbCallbackExecutor()); } - private static ListenableFuture getTenantAsync(ListenableFuture future) { - return Futures.transformAsync(future, in -> { - return in != null ? Futures.immediateFuture(in.getTenantId()) - : Futures.immediateFuture(null); - }, MoreExecutors.directExecutor()); + private static ListenableFuture getTenantIdAsync(ListenableFuture future, ListeningExecutor executor) { + return Futures.transformAsync(future, in -> in != null ? Futures.immediateFuture(in.getTenantId()) + : Futures.immediateFuture(null), executor); } } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/util/EntitiesTenantIdAsyncLoaderTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/util/EntitiesTenantIdAsyncLoaderTest.java new file mode 100644 index 0000000000..885fa7ec40 --- /dev/null +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/util/EntitiesTenantIdAsyncLoaderTest.java @@ -0,0 +1,347 @@ +/** + * Copyright © 2016-2022 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.rule.engine.util; + +import com.google.common.util.concurrent.Futures; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; +import org.thingsboard.common.util.AbstractListeningExecutor; +import org.thingsboard.rule.engine.api.RuleEngineAlarmService; +import org.thingsboard.rule.engine.api.RuleEngineApiUsageStateService; +import org.thingsboard.rule.engine.api.RuleEngineAssetProfileCache; +import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; +import org.thingsboard.rule.engine.api.RuleEngineRpcService; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.server.common.data.ApiUsageState; +import org.thingsboard.server.common.data.Customer; +import org.thingsboard.server.common.data.Dashboard; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.EntityView; +import org.thingsboard.server.common.data.OtaPackage; +import org.thingsboard.server.common.data.TbResource; +import org.thingsboard.server.common.data.TenantProfile; +import org.thingsboard.server.common.data.User; +import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.asset.Asset; +import org.thingsboard.server.common.data.asset.AssetProfile; +import org.thingsboard.server.common.data.edge.Edge; +import org.thingsboard.server.common.data.id.AssetProfileId; +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.queue.Queue; +import org.thingsboard.server.common.data.rpc.Rpc; +import org.thingsboard.server.common.data.rule.RuleChain; +import org.thingsboard.server.common.data.rule.RuleNode; +import org.thingsboard.server.common.data.widget.WidgetType; +import org.thingsboard.server.common.data.widget.WidgetsBundle; +import org.thingsboard.server.dao.asset.AssetService; +import org.thingsboard.server.dao.customer.CustomerService; +import org.thingsboard.server.dao.dashboard.DashboardService; +import org.thingsboard.server.dao.device.DeviceService; +import org.thingsboard.server.dao.edge.EdgeService; +import org.thingsboard.server.dao.entityview.EntityViewService; +import org.thingsboard.server.dao.ota.OtaPackageService; +import org.thingsboard.server.dao.queue.QueueService; +import org.thingsboard.server.dao.resource.ResourceService; +import org.thingsboard.server.dao.rule.RuleChainService; +import org.thingsboard.server.dao.user.UserService; +import org.thingsboard.server.dao.widget.WidgetTypeService; +import org.thingsboard.server.dao.widget.WidgetsBundleService; + +import java.util.UUID; +import java.util.concurrent.ExecutionException; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.when; + +@RunWith(MockitoJUnitRunner.class) +public class EntitiesTenantIdAsyncLoaderTest { + + @Mock + private TbContext ctx; + @Mock + private CustomerService customerService; + @Mock + private UserService userService; + @Mock + private AssetService assetService; + @Mock + private DeviceService deviceService; + @Mock + private RuleEngineAlarmService alarmService; + @Mock + private RuleChainService ruleChainService; + @Mock + private EntityViewService entityViewService; + @Mock + private DashboardService dashboardService; + @Mock + private EdgeService edgeService; + @Mock + private OtaPackageService otaPackageService; + @Mock + private RuleEngineAssetProfileCache assetProfileCache; + @Mock + private RuleEngineDeviceProfileCache deviceProfileCache; + @Mock + private WidgetTypeService widgetTypeService; + @Mock + private WidgetsBundleService widgetsBundleService; + @Mock + private QueueService queueService; + @Mock + private ResourceService resourceService; + @Mock + private RuleEngineRpcService rpcService; + @Mock + private RuleEngineApiUsageStateService ruleEngineApiUsageStateService; + + private TenantId tenantId; + private AbstractListeningExecutor dbExecutor; + + @Before + public void before() { + dbExecutor = new AbstractListeningExecutor() { + @Override + protected int getThreadPollSize() { + return 3; + } + }; + dbExecutor.init(); + this.tenantId = new TenantId(UUID.randomUUID()); + + when(ctx.getTenantId()).thenReturn(tenantId); + when(ctx.getDbCallbackExecutor()).thenReturn(dbExecutor); + + for (EntityType entityType : EntityType.values()) { + initMocks(entityType, tenantId); + } + } + + @After + public void after() { + dbExecutor.destroy(); + } + + private void initMocks(EntityType entityType, TenantId tenantId) { + switch (entityType) { + case TENANT: + break; + case CUSTOMER: + Customer customer = new Customer(); + customer.setTenantId(tenantId); + + when(ctx.getCustomerService()).thenReturn(customerService); + doReturn(Futures.immediateFuture(customer)).when(customerService).findCustomerByIdAsync(eq(tenantId), any()); + + break; + case USER: + User user = new User(); + user.setTenantId(tenantId); + + when(ctx.getUserService()).thenReturn(userService); + doReturn(Futures.immediateFuture(user)).when(userService).findUserByIdAsync(eq(tenantId), any()); + + break; + case ASSET: + Asset asset = new Asset(); + asset.setTenantId(tenantId); + + when(ctx.getAssetService()).thenReturn(assetService); + doReturn(Futures.immediateFuture(asset)).when(assetService).findAssetByIdAsync(eq(tenantId), any()); + + break; + case DEVICE: + Device device = new Device(); + device.setTenantId(tenantId); + + when(ctx.getDeviceService()).thenReturn(deviceService); + doReturn(Futures.immediateFuture(device)).when(deviceService).findDeviceByIdAsync(eq(tenantId), any()); + + break; + case ALARM: + Alarm alarm = new Alarm(); + alarm.setTenantId(tenantId); + + when(ctx.getAlarmService()).thenReturn(alarmService); + doReturn(Futures.immediateFuture(alarm)).when(alarmService).findAlarmByIdAsync(eq(tenantId), any()); + + break; + case RULE_CHAIN: + RuleChain ruleChain = new RuleChain(); + ruleChain.setTenantId(tenantId); + + when(ctx.getRuleChainService()).thenReturn(ruleChainService); + doReturn(Futures.immediateFuture(ruleChain)).when(ruleChainService).findRuleChainByIdAsync(eq(tenantId), any()); + + break; + case ENTITY_VIEW: + EntityView entityView = new EntityView(); + entityView.setTenantId(tenantId); + + when(ctx.getEntityViewService()).thenReturn(entityViewService); + doReturn(Futures.immediateFuture(entityView)).when(entityViewService).findEntityViewByIdAsync(eq(tenantId), any()); + + break; + case DASHBOARD: + Dashboard dashboard = new Dashboard(); + dashboard.setTenantId(tenantId); + + when(ctx.getDashboardService()).thenReturn(dashboardService); + doReturn(Futures.immediateFuture(dashboard)).when(dashboardService).findDashboardByIdAsync(eq(tenantId), any()); + + break; + case EDGE: + Edge edge = new Edge(); + edge.setTenantId(tenantId); + + when(ctx.getEdgeService()).thenReturn(edgeService); + doReturn(Futures.immediateFuture(edge)).when(edgeService).findEdgeByIdAsync(eq(tenantId), any()); + + break; + case OTA_PACKAGE: + OtaPackage otaPackage = new OtaPackage(); + otaPackage.setTenantId(tenantId); + + when(ctx.getOtaPackageService()).thenReturn(otaPackageService); + doReturn(Futures.immediateFuture(otaPackage)).when(otaPackageService).findOtaPackageInfoByIdAsync(eq(tenantId), any()); + + break; + case ASSET_PROFILE: + AssetProfile assetProfile = new AssetProfile(); + assetProfile.setTenantId(tenantId); + + when(ctx.getAssetProfileCache()).thenReturn(assetProfileCache); + doReturn(assetProfile).when(assetProfileCache).get(eq(tenantId), any(AssetProfileId.class)); + + break; + case DEVICE_PROFILE: + DeviceProfile deviceProfile = new DeviceProfile(); + deviceProfile.setTenantId(tenantId); + + when(ctx.getDeviceProfileCache()).thenReturn(deviceProfileCache); + doReturn(deviceProfile).when(deviceProfileCache).get(eq(tenantId), any(DeviceProfileId.class)); + + break; + case WIDGET_TYPE: + WidgetType widgetType = new WidgetType(); + widgetType.setTenantId(tenantId); + + when(ctx.getWidgetTypeService()).thenReturn(widgetTypeService); + doReturn(widgetType).when(widgetTypeService).findWidgetTypeById(eq(tenantId), any()); + + break; + case WIDGETS_BUNDLE: + WidgetsBundle widgetsBundle = new WidgetsBundle(); + widgetsBundle.setTenantId(tenantId); + + when(ctx.getWidgetBundleService()).thenReturn(widgetsBundleService); + doReturn(widgetsBundle).when(widgetsBundleService).findWidgetsBundleById(eq(tenantId), any()); + + break; + case RPC: + Rpc rps = new Rpc(); + rps.setTenantId(tenantId); + + when(ctx.getRpcService()).thenReturn(rpcService); + doReturn(Futures.immediateFuture(rps)).when(rpcService).findRpcByIdAsync(eq(tenantId), any()); + + break; + case QUEUE: + Queue queue = new Queue(); + queue.setTenantId(tenantId); + + when(ctx.getQueueService()).thenReturn(queueService); + doReturn(queue).when(queueService).findQueueById(eq(tenantId), any()); + + break; + case API_USAGE_STATE: + ApiUsageState apiUsageState = new ApiUsageState(); + apiUsageState.setTenantId(tenantId); + + when(ctx.getRuleEngineApiUsageStateService()).thenReturn(ruleEngineApiUsageStateService); + doReturn(apiUsageState).when(ruleEngineApiUsageStateService).findApiUsageStateById(eq(tenantId), any()); + + break; + case TB_RESOURCE: + TbResource tbResource = new TbResource(); + tbResource.setTenantId(tenantId); + + when(ctx.getResourceService()).thenReturn(resourceService); + doReturn(Futures.immediateFuture(tbResource)).when(resourceService).findResourceInfoByIdAsync(eq(tenantId), any()); + + break; + case RULE_NODE: + RuleNode ruleNode = new RuleNode(); + + when(ctx.getRuleChainService()).thenReturn(ruleChainService); + doReturn(ruleNode).when(ruleChainService).findRuleNodeById(eq(tenantId), any()); + + break; + case TENANT_PROFILE: + TenantProfile tenantProfile = new TenantProfile(); + + when(ctx.getTenantProfile()).thenReturn(tenantProfile); + when(ctx.getTenantProfile(any(TenantId.class))).thenReturn(tenantProfile); + + break; + default: + throw new RuntimeException("Unexpected original EntityType " + entityType); + } + + } + + private EntityId getEntityId(EntityType entityType) { + return EntityIdFactory.getByTypeAndUuid(entityType, UUID.randomUUID()); + } + + private void checkTenant(TenantId checkTenantId, boolean equals) throws ExecutionException, InterruptedException { + for (EntityType entityType : EntityType.values()) { + EntityId entityId = EntityType.TENANT.equals(entityType) ? tenantId : getEntityId(entityType); + TenantId targetTenantId = EntitiesTenantIdAsyncLoader.findEntityIdAsync(ctx, entityId).get(); + + Assert.assertNotNull(targetTenantId); + String msg = "Check entity type <" + entityType.name() + ">:"; + if (equals) { + Assert.assertEquals(msg, targetTenantId, checkTenantId); + } else { + Assert.assertNotEquals(msg, targetTenantId, checkTenantId); + } + } + } + + @Test + public void test_findEntityIdAsync_current_tenant() throws ExecutionException, InterruptedException { + checkTenant(tenantId, true); + } + + @Test + public void test_findEntityIdAsync_other_tenant() throws ExecutionException, InterruptedException { + checkTenant(new TenantId(UUID.randomUUID()), false); + } + +}