From f74f3a92933cdfc7497d5b2d8f6f44aee9df8bd0 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Mon, 2 May 2022 08:23:16 +0200 Subject: [PATCH] created TbQueueService and improvements --- .../server/controller/TenantController.java | 13 +- .../entity/queue/DefaultTbQueueService.java | 144 ++++++++++++++++++ .../service/entity/queue/TbQueueService.java | 29 ++++ .../entity/tenant/DefaultTbTenantService.java | 135 ++++++++++++++++ .../entity/tenant/TbTenantService.java | 23 +++ .../DefaultSystemDataLoaderService.java | 29 +++- .../DefaultTbRuleEngineConsumerService.java | 2 +- ...ice.java => TbQueueServiceDeprecated.java} | 2 +- .../server/dao/queue/QueueService.java | 4 +- .../server/dao/tenant/TenantService.java | 8 +- ...a => DefaultTbQueueServiceDeprecated.java} | 3 +- .../server/dao/queue/BaseQueueService.java | 108 +------------ .../server/dao/tenant/TenantServiceImpl.java | 85 +---------- 13 files changed, 383 insertions(+), 202 deletions(-) create mode 100644 application/src/main/java/org/thingsboard/server/service/entity/queue/DefaultTbQueueService.java create mode 100644 application/src/main/java/org/thingsboard/server/service/entity/queue/TbQueueService.java create mode 100644 application/src/main/java/org/thingsboard/server/service/entity/tenant/DefaultTbTenantService.java create mode 100644 application/src/main/java/org/thingsboard/server/service/entity/tenant/TbTenantService.java rename common/cluster-api/src/main/java/org/thingsboard/server/queue/{TbQueueService.java => TbQueueServiceDeprecated.java} (95%) rename common/queue/src/main/java/org/thingsboard/server/queue/{DefaultTbQueueService.java => DefaultTbQueueServiceDeprecated.java} (94%) diff --git a/application/src/main/java/org/thingsboard/server/controller/TenantController.java b/application/src/main/java/org/thingsboard/server/controller/TenantController.java index 496cd60e81..e7fd8e24dc 100644 --- a/application/src/main/java/org/thingsboard/server/controller/TenantController.java +++ b/application/src/main/java/org/thingsboard/server/controller/TenantController.java @@ -18,8 +18,8 @@ package org.thingsboard.server.controller; import com.fasterxml.jackson.databind.node.ObjectNode; import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiParam; +import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.HttpStatus; import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.web.bind.annotation.PathVariable; @@ -39,6 +39,7 @@ import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.queue.util.TbCoreComponent; +import org.thingsboard.server.service.entity.tenant.TbTenantService; import org.thingsboard.server.service.install.InstallScripts; import org.thingsboard.server.service.security.permission.Operation; import org.thingsboard.server.service.security.permission.Resource; @@ -63,14 +64,14 @@ import static org.thingsboard.server.controller.ControllerConstants.UUID_WIKI_LI @TbCoreComponent @RequestMapping("/api") @Slf4j +@RequiredArgsConstructor public class TenantController extends BaseController { private static final String TENANT_INFO_DESCRIPTION = "The Tenant Info object extends regular Tenant object and includes Tenant Profile name. "; - @Autowired - private InstallScripts installScripts; - @Autowired - private TenantService tenantService; + private final InstallScripts installScripts; + private final TenantService tenantService; + private final TbTenantService tbTenantService; @ApiOperation(value = "Get Tenant (getTenantById)", notes = "Fetch the Tenant object based on the provided Tenant Id. " + SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH) @@ -129,7 +130,7 @@ public class TenantController extends BaseController { checkEntity(tenant.getId(), tenant, Resource.TENANT); - tenant = checkNotNull(tenantService.saveTenant(tenant)); + tenant = checkNotNull(tbTenantService.saveTenant(tenant)); if (newTenant) { installScripts.createDefaultRuleChains(tenant.getId()); installScripts.createDefaultEdgeRuleChains(tenant.getId()); diff --git a/application/src/main/java/org/thingsboard/server/service/entity/queue/DefaultTbQueueService.java b/application/src/main/java/org/thingsboard/server/service/entity/queue/DefaultTbQueueService.java new file mode 100644 index 0000000000..006c9025ba --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/entity/queue/DefaultTbQueueService.java @@ -0,0 +1,144 @@ +/** + * 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.server.service.entity.queue; + +import lombok.AllArgsConstructor; +import lombok.SneakyThrows; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.id.QueueId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.queue.Queue; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.dao.queue.QueueService; +import org.thingsboard.server.queue.TbQueueAdmin; +import org.thingsboard.server.queue.TbQueueClusterService; +import org.thingsboard.server.queue.util.TbCoreComponent; + +@Slf4j +@Service +@TbCoreComponent +@AllArgsConstructor +public class DefaultTbQueueService implements TbQueueService { + private final QueueService queueService; + private final TbQueueClusterService queueClusterService; + private final TbQueueAdmin tbQueueAdmin; + + @Override + public Queue saveQueue(Queue queue) { + boolean create = queue.getId() == null; + Queue oldQueue; + + if (create) { + oldQueue = null; + } else { + oldQueue = queueService.findQueueById(queue.getTenantId(), queue.getId()); + } + + //TODO: add checkNotNull + Queue savedQueue = queueService.saveQueue(queue); + + if (create) { + onQueueCreated(savedQueue); + } else { + onQueueUpdated(savedQueue, oldQueue); + } + + return savedQueue; + } + + @Override + public void deleteQueue(TenantId tenantId, QueueId queueId) { + Queue queue = queueService.findQueueById(tenantId, queueId); + queueService.deleteQueue(tenantId, queueId); + onQueueDeleted(tenantId, queue); + } + + @Override + public void deleteQueueByQueueName(TenantId tenantId, String queueName) { + Queue queue = queueService.findQueueByTenantIdAndNameInternal(tenantId, queueName); + queueService.deleteQueue(tenantId, queue.getId()); + onQueueDeleted(tenantId, queue); + } + + private void onQueueCreated(Queue queue) { + if (tbQueueAdmin != null) { + for (int i = 0; i < queue.getPartitions(); i++) { + tbQueueAdmin.createTopicIfNotExists( + new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName()); + } + } + + if (queueClusterService != null) { + queueClusterService.onQueueChange(queue); + } + } + + private void onQueueUpdated(Queue queue, Queue oldQueue) { + int oldPartitions = oldQueue.getPartitions(); + int currentPartitions = queue.getPartitions(); + + if (currentPartitions != oldPartitions && tbQueueAdmin != null) { + if (currentPartitions > oldPartitions) { + log.info("Added [{}] new partitions to [{}] queue", currentPartitions - oldPartitions, queue.getName()); + for (int i = oldPartitions; i < currentPartitions; i++) { + tbQueueAdmin.createTopicIfNotExists( + new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName()); + } + if (queueClusterService != null) { + queueClusterService.onQueueChange(queue); + } + } else { + log.info("Removed [{}] partitions from [{}] queue", oldPartitions - currentPartitions, queue.getName()); + if (queueClusterService != null) { + queueClusterService.onQueueChange(queue); + } + await(); + for (int i = currentPartitions; i < oldPartitions; i++) { + tbQueueAdmin.deleteTopic( + new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName()); + } + } + } else if (!oldQueue.equals(queue) && queueClusterService != null) { + queueClusterService.onQueueChange(queue); + } + } + + private void onQueueDeleted(TenantId tenantId, Queue queue) { + if (queueClusterService != null) { + queueClusterService.onQueueDelete(queue); + await(); + } +// queueStatsService.deleteQueueStatsByQueueId(tenantId, queueId); + if (tbQueueAdmin != null) { + for (int i = 0; i < queue.getPartitions(); i++) { + String fullTopicName = new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName(); + log.debug("Deleting queue [{}]", fullTopicName); + try { + tbQueueAdmin.deleteTopic(fullTopicName); + } catch (Exception e) { + log.error("Failed to delete queue [{}]", fullTopicName); + } + } + } + } + + @SneakyThrows + private void await() { + Thread.sleep(3000); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/entity/queue/TbQueueService.java b/application/src/main/java/org/thingsboard/server/service/entity/queue/TbQueueService.java new file mode 100644 index 0000000000..1b37486ba4 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/entity/queue/TbQueueService.java @@ -0,0 +1,29 @@ +/** + * 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.server.service.entity.queue; + +import org.thingsboard.server.common.data.id.QueueId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.queue.Queue; + +public interface TbQueueService { + + Queue saveQueue(Queue queue); + + void deleteQueue(TenantId tenantId, QueueId queueId); + + void deleteQueueByQueueName(TenantId tenantId, String queueName); +} diff --git a/application/src/main/java/org/thingsboard/server/service/entity/tenant/DefaultTbTenantService.java b/application/src/main/java/org/thingsboard/server/service/entity/tenant/DefaultTbTenantService.java new file mode 100644 index 0000000000..015c88813b --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/entity/tenant/DefaultTbTenantService.java @@ -0,0 +1,135 @@ +/** + * 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.server.service.entity.tenant; + +import lombok.AllArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.Tenant; +import org.thingsboard.server.common.data.TenantProfile; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.queue.Queue; +import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration; +import org.thingsboard.server.dao.queue.QueueService; +import org.thingsboard.server.dao.tenant.TbTenantProfileCache; +import org.thingsboard.server.dao.tenant.TenantProfileService; +import org.thingsboard.server.dao.tenant.TenantService; +import org.thingsboard.server.queue.util.TbCoreComponent; +import org.thingsboard.server.service.entity.queue.TbQueueService; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +@Slf4j +@Service +@TbCoreComponent +@AllArgsConstructor +public class DefaultTbTenantService implements TbTenantService { + + private final TenantService tenantService; + private final TbQueueService tbQueueService; + private final QueueService queueService; + private final TenantProfileService tenantProfileService; + private final TbTenantProfileCache tenantProfileCache; + + @Override + public Tenant saveTenant(Tenant tenant) { + boolean updated = tenant.getId() != null; + Tenant oldTenant = updated ? tenantService.findTenantById(tenant.getId()) : null; + List queues; + if (updated) { + + } + + Tenant savedTenant = tenantService.saveTenant(tenant); + tenantProfileCache.evict(tenant.getId()); + updateQueuesForTenant(oldTenant, savedTenant); + return savedTenant; + } + + public void updateQueuesForTenant(Tenant oldTenant, Tenant newTenant) { + TenantProfile oldTenantProfile = oldTenant != null ? tenantProfileService.findTenantProfileById(TenantId.SYS_TENANT_ID, oldTenant.getTenantProfileId()) : null; + TenantProfile newTenantProfile = tenantProfileService.findTenantProfileById(TenantId.SYS_TENANT_ID, newTenant.getTenantProfileId()); + + TenantId tenantId = newTenant.getId(); + + boolean oldIsolated = oldTenantProfile != null && oldTenantProfile.isIsolatedTbRuleEngine(); + boolean newIsolated = newTenantProfile.isIsolatedTbRuleEngine(); + + if (!oldIsolated && !newIsolated) { + return; + } + + if (newTenantProfile.equals(oldTenantProfile)) { + return; + } + + Map oldQueues; + Map newQueues; + + if (oldIsolated) { + oldQueues = oldTenantProfile.getProfileData().getQueueConfiguration().stream() + .collect(Collectors.toMap(TenantProfileQueueConfiguration::getName, q -> q)); + } else { + oldQueues = Collections.emptyMap(); + } + + if (newIsolated) { + newQueues = newTenantProfile.getProfileData().getQueueConfiguration().stream() + .collect(Collectors.toMap(TenantProfileQueueConfiguration::getName, q -> q)); + } else { + newQueues = Collections.emptyMap(); + } + + List toRemove = new ArrayList<>(); + List toCreate = new ArrayList<>(); + List toUpdate = new ArrayList<>(); + + for (String oldQueue : oldQueues.keySet()) { + if (!newQueues.containsKey(oldQueue)) { + toRemove.add(oldQueue); + } + } + + for (String newQueue : newQueues.keySet()) { + if (oldQueues.containsKey(newQueue)) { + toUpdate.add(newQueue); + } else { + toCreate.add(newQueue); + } + } + + toRemove.forEach(q -> tbQueueService.deleteQueueByQueueName(tenantId, q)); + + toCreate.forEach(key -> tbQueueService.saveQueue(new Queue(tenantId, newQueues.get(key)))); + + toUpdate.forEach(key -> { + Queue queueToUpdate = new Queue(tenantId, newQueues.get(key)); + Queue foundQueue = queueService.findQueueByTenantIdAndName(tenantId, key); + queueToUpdate.setId(foundQueue.getId()); + queueToUpdate.setCreatedTime(foundQueue.getCreatedTime()); + + if (queueToUpdate.equals(foundQueue)) { + //Queue not changed + } else { + tbQueueService.saveQueue(queueToUpdate); + } + }); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/entity/tenant/TbTenantService.java b/application/src/main/java/org/thingsboard/server/service/entity/tenant/TbTenantService.java new file mode 100644 index 0000000000..56b3e5e7ba --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/entity/tenant/TbTenantService.java @@ -0,0 +1,23 @@ +/** + * 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.server.service.entity.tenant; + +import org.thingsboard.server.common.data.Tenant; + +public interface TbTenantService { + + Tenant saveTenant(Tenant tenant); +} diff --git a/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java b/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java index 37982f4a6d..8b2b99b840 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java @@ -79,6 +79,7 @@ import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.data.security.UserCredentials; import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; +import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration; import org.thingsboard.server.common.data.widget.WidgetsBundle; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.customer.CustomerService; @@ -208,13 +209,37 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService { log.warn(e.getMessage()); } + TenantProfileData isolatedRuleEngineTenantProfileData = new TenantProfileData(); + isolatedRuleEngineTenantProfileData.setConfiguration(new DefaultTenantProfileConfiguration()); + + TenantProfileQueueConfiguration mainQueueConfiguration = new TenantProfileQueueConfiguration(); + mainQueueConfiguration.setName("Main"); + mainQueueConfiguration.setTopic("tb_rule_engine.main"); + mainQueueConfiguration.setPollInterval(25); + mainQueueConfiguration.setPartitions(10); + mainQueueConfiguration.setConsumerPerPartition(true); + mainQueueConfiguration.setPackProcessingTimeout(2000); + SubmitStrategy mainQueueSubmitStrategy = new SubmitStrategy(); + mainQueueSubmitStrategy.setType(SubmitStrategyType.BURST); + mainQueueSubmitStrategy.setBatchSize(1000); + mainQueueConfiguration.setSubmitStrategy(mainQueueSubmitStrategy); + ProcessingStrategy mainQueueProcessingStrategy = new ProcessingStrategy(); + mainQueueProcessingStrategy.setType(ProcessingStrategyType.SKIP_ALL_FAILURES); + mainQueueProcessingStrategy.setRetries(3); + mainQueueProcessingStrategy.setFailurePercentage(0); + mainQueueProcessingStrategy.setPauseBetweenRetries(3); + mainQueueProcessingStrategy.setMaxPauseBetweenRetries(3); + mainQueueConfiguration.setProcessingStrategy(mainQueueProcessingStrategy); + + isolatedRuleEngineTenantProfileData.setQueueConfiguration(Collections.singletonList(mainQueueConfiguration)); + TenantProfile isolatedTbRuleEngineProfile = new TenantProfile(); isolatedTbRuleEngineProfile.setDefault(false); isolatedTbRuleEngineProfile.setName("Isolated TB Rule Engine"); isolatedTbRuleEngineProfile.setDescription("Isolated TB Rule Engine tenant profile"); isolatedTbRuleEngineProfile.setIsolatedTbCore(false); isolatedTbRuleEngineProfile.setIsolatedTbRuleEngine(true); - isolatedTbRuleEngineProfile.setProfileData(tenantProfileData); + isolatedTbRuleEngineProfile.setProfileData(isolatedRuleEngineTenantProfileData); try { tenantProfileService.saveTenantProfile(TenantId.SYS_TENANT_ID, isolatedTbRuleEngineProfile); @@ -228,7 +253,7 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService { isolatedTbCoreAndTbRuleEngineProfile.setDescription("Isolated TB Core and TB Rule Engine tenant profile"); isolatedTbCoreAndTbRuleEngineProfile.setIsolatedTbCore(true); isolatedTbCoreAndTbRuleEngineProfile.setIsolatedTbRuleEngine(true); - isolatedTbCoreAndTbRuleEngineProfile.setProfileData(tenantProfileData); + isolatedTbCoreAndTbRuleEngineProfile.setProfileData(isolatedRuleEngineTenantProfileData); try { tenantProfileService.saveTenantProfile(TenantId.SYS_TENANT_ID, isolatedTbCoreAndTbRuleEngineProfile); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index 3f8173b9ac..8e3d07316d 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -407,7 +407,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< } } - private void updateQueue(TransportProtos.QueueUpdateMsg queueUpdateMsg) { + private synchronized void updateQueue(TransportProtos.QueueUpdateMsg queueUpdateMsg) { String queueName = queueUpdateMsg.getQueueName(); TenantId tenantId = new TenantId(new UUID(queueUpdateMsg.getTenantIdMSB(), queueUpdateMsg.getTenantIdLSB())); QueueId queueId = new QueueId(new UUID(queueUpdateMsg.getQueueIdMSB(), queueUpdateMsg.getQueueIdLSB())); diff --git a/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueService.java b/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueServiceDeprecated.java similarity index 95% rename from common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueService.java rename to common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueServiceDeprecated.java index bc7964df7e..2328893abe 100644 --- a/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueService.java +++ b/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueServiceDeprecated.java @@ -19,7 +19,7 @@ import org.thingsboard.server.common.msg.queue.ServiceType; import java.util.Set; -public interface TbQueueService { +public interface TbQueueServiceDeprecated { Set getQueuesByServiceType(ServiceType serviceType); diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/queue/QueueService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/queue/QueueService.java index bdea0ef09c..acc3190da9 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/queue/QueueService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/queue/QueueService.java @@ -29,8 +29,6 @@ public interface QueueService { void deleteQueue(TenantId tenantId, QueueId queueId); - void deleteQueueByQueueName(TenantId tenantId, String queueName); - List findQueuesByTenantId(TenantId tenantId); PageData findQueuesByTenantId(TenantId tenantId, PageLink pageLink); @@ -41,5 +39,7 @@ public interface QueueService { Queue findQueueByTenantIdAndName(TenantId tenantId, String name); + Queue findQueueByTenantIdAndNameInternal(TenantId tenantId, String queueName); + void deleteQueuesByTenantId(TenantId tenantId); } \ No newline at end of file diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/tenant/TenantService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/tenant/TenantService.java index 3471d383f2..b4bf769944 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/tenant/TenantService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/tenant/TenantService.java @@ -29,14 +29,14 @@ public interface TenantService { TenantInfo findTenantInfoById(TenantId tenantId); ListenableFuture findTenantByIdAsync(TenantId callerId, TenantId tenantId); - + Tenant saveTenant(Tenant tenant); - + void deleteTenant(TenantId tenantId); - + PageData findTenants(PageLink pageLink); PageData findTenantInfos(PageLink pageLink); - + void deleteTenants(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/DefaultTbQueueService.java b/common/queue/src/main/java/org/thingsboard/server/queue/DefaultTbQueueServiceDeprecated.java similarity index 94% rename from common/queue/src/main/java/org/thingsboard/server/queue/DefaultTbQueueService.java rename to common/queue/src/main/java/org/thingsboard/server/queue/DefaultTbQueueServiceDeprecated.java index e372fb600c..8f4175dad1 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/DefaultTbQueueService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/DefaultTbQueueServiceDeprecated.java @@ -16,7 +16,6 @@ package org.thingsboard.server.queue; import lombok.RequiredArgsConstructor; -import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; import org.thingsboard.server.common.msg.queue.ServiceQueue; import org.thingsboard.server.common.msg.queue.ServiceType; @@ -31,7 +30,7 @@ import java.util.stream.Collectors; //@Service @RequiredArgsConstructor -public class DefaultTbQueueService implements TbQueueService { +public class DefaultTbQueueServiceDeprecated implements TbQueueServiceDeprecated { private final TbQueueRuleEngineSettings ruleEngineSettings; private Set ruleEngineQueues; diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/BaseQueueService.java b/dao/src/main/java/org/thingsboard/server/dao/queue/BaseQueueService.java index a0ad3c3a4a..c95635a439 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/queue/BaseQueueService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/queue/BaseQueueService.java @@ -16,7 +16,6 @@ package org.thingsboard.server.dao.queue; import lombok.RequiredArgsConstructor; -import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; @@ -30,15 +29,12 @@ import org.thingsboard.server.common.data.queue.ProcessingStrategy; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.data.queue.SubmitStrategy; import org.thingsboard.server.common.data.queue.SubmitStrategyType; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.dao.entity.AbstractEntityService; import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.service.DataValidator; import org.thingsboard.server.dao.service.PaginatedRemover; import org.thingsboard.server.dao.service.Validator; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; -import org.thingsboard.server.queue.TbQueueAdmin; -import org.thingsboard.server.queue.TbQueueClusterService; import java.util.List; @@ -53,12 +49,6 @@ public class BaseQueueService extends AbstractEntityService implements QueueServ @Autowired private TbTenantProfileCache tenantProfileCache; - @Autowired(required = false) - private TbQueueAdmin tbQueueAdmin; - - @Autowired(required = false) - private TbQueueClusterService queueClusterService; - // @Autowired // private QueueStatsService queueStatsService; @@ -66,101 +56,13 @@ public class BaseQueueService extends AbstractEntityService implements QueueServ public Queue saveQueue(Queue queue) { log.trace("Executing createOrUpdateQueue [{}]", queue); queueValidator.validate(queue, Queue::getTenantId); - Queue savedQueue; - if (queue.getId() == null) { - savedQueue = createQueue(queue); - } else { - savedQueue = updateQueue(queue); - } - - return savedQueue; - } - - private Queue createQueue(Queue queue) { - Queue createdQueue = queueDao.save(queue.getTenantId(), queue); - if (tbQueueAdmin != null) { - for (int i = 0; i < queue.getPartitions(); i++) { - tbQueueAdmin.createTopicIfNotExists(new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName()); - } - } - - if (queueClusterService != null) { - queueClusterService.onQueueChange(createdQueue); - } - - return createdQueue; - } - - private Queue updateQueue(Queue queue) { - Queue oldQueue = queueDao.findById(queue.getTenantId(), queue.getUuidId()); - Queue updatedQueue = queueDao.save(queue.getTenantId(), queue); - - int oldPartitions = oldQueue.getPartitions(); - int currentPartitions = queue.getPartitions(); - - if (currentPartitions != oldPartitions && tbQueueAdmin != null) { - if (currentPartitions > oldPartitions) { - log.info("Added [{}] new partitions to [{}] queue", currentPartitions - oldPartitions, queue.getName()); - for (int i = oldPartitions; i < currentPartitions; i++) { - tbQueueAdmin.createTopicIfNotExists(new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName()); - } - if (queueClusterService != null) { - queueClusterService.onQueueChange(updatedQueue); - } - } else { - log.info("Removed [{}] partitions from [{}] queue", oldPartitions - currentPartitions, queue.getName()); - if (queueClusterService != null) { - queueClusterService.onQueueChange(updatedQueue); - } - await(); - for (int i = currentPartitions; i < oldPartitions; i++) { - tbQueueAdmin.deleteTopic(new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName()); - } - } - } else if (!oldQueue.equals(queue) && queueClusterService != null) { - queueClusterService.onQueueChange(updatedQueue); - } - - return updatedQueue; + return queueDao.save(queue.getTenantId(), queue); } @Override public void deleteQueue(TenantId tenantId, QueueId queueId) { log.trace("Executing deleteQueue, queueId: [{}]", queueId); - Queue queue = findQueueById(tenantId, queueId); - doDelete(tenantId, queue); - } - - @Override - public void deleteQueueByQueueName(TenantId tenantId, String queueName) { - log.trace("Executing deleteQueueByQueueName, name: [{}]", queueName); - Queue queue = findQueueByTenantIdAndName(tenantId, queueName); - doDelete(tenantId, queue); - } - - private void doDelete(TenantId tenantId, Queue queue) { - if (queueClusterService != null) { - queueClusterService.onQueueDelete(queue); - await(); - } -// queueStatsService.deleteQueueStatsByQueueId(tenantId, queueId); - boolean result = queueDao.removeById(tenantId, queue.getUuidId()); - if (result && tbQueueAdmin != null) { - for (int i = 0; i < queue.getPartitions(); i++) { - String fullTopicName = new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName(); - log.debug("Deleting queue [{}]", fullTopicName); - try { - tbQueueAdmin.deleteTopic(fullTopicName); - } catch (Exception e) { - log.error("Failed to delete queue [{}]", fullTopicName); - } - } - } - } - - @SneakyThrows - private void await() { - Thread.sleep(3000); + queueDao.removeById(tenantId, queueId.getId()); } @Override @@ -194,6 +96,12 @@ public class BaseQueueService extends AbstractEntityService implements QueueServ return queueDao.findQueueByTenantIdAndName(getSystemOrIsolatedTenantId(tenantId), queueName); } + @Override + public Queue findQueueByTenantIdAndNameInternal(TenantId tenantId, String queueName) { + log.trace("Executing findQueueByTenantIdAndNameInternal, tenantId: [{}] queueName: [{}]", tenantId, queueName); + return queueDao.findQueueByTenantIdAndName(tenantId, queueName); + } + @Override public void deleteQueuesByTenantId(TenantId tenantId) { Validator.validateId(tenantId, "Incorrect tenant id for delete queues request."); diff --git a/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantServiceImpl.java index 4b935ca700..dcacaba478 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantServiceImpl.java @@ -20,6 +20,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.TenantInfo; @@ -27,8 +28,6 @@ import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; -import org.thingsboard.server.common.data.queue.Queue; -import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration; import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.customer.CustomerService; import org.thingsboard.server.dao.dashboard.DashboardService; @@ -49,12 +48,6 @@ import org.thingsboard.server.dao.usagerecord.ApiUsageStateService; import org.thingsboard.server.dao.user.UserService; import org.thingsboard.server.dao.widget.WidgetsBundleService; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.stream.Collectors; - import static org.thingsboard.server.dao.service.Validator.validateId; @Service @@ -145,90 +138,14 @@ public class TenantServiceImpl extends AbstractEntityService implements TenantSe tenant.setTenantProfileId(tenantProfile.getId()); } tenantValidator.validate(tenant, Tenant::getId); - - Tenant oldTenant = tenant.getId() != null ? tenantDao.findById(tenant.getId(), tenant.getUuidId()) : null; - Tenant savedTenant = tenantDao.save(tenant.getId(), tenant); if (tenant.getId() == null) { deviceProfileService.createDefaultDeviceProfile(savedTenant.getId()); apiUsageStateService.createDefaultApiUsageState(savedTenant.getId(), null); } - - updateQueuesForTenant(oldTenant, savedTenant); - return savedTenant; } - private void updateQueuesForTenant(Tenant oldTenant, Tenant newTenant) { - TenantProfile oldTenantProfile = oldTenant != null ? tenantProfileService.findTenantProfileById(TenantId.SYS_TENANT_ID, oldTenant.getTenantProfileId()) : null; - TenantProfile newTenantProfile = tenantProfileService.findTenantProfileById(TenantId.SYS_TENANT_ID, newTenant.getTenantProfileId()); - - TenantId tenantId = newTenant.getId(); - - boolean oldIsolated = oldTenantProfile != null && oldTenantProfile.isIsolatedTbRuleEngine(); - boolean newIsolated = newTenantProfile.isIsolatedTbRuleEngine(); - - if (!oldIsolated && !newIsolated) { - return; - } - - if (newTenantProfile.equals(oldTenantProfile)) { - return; - } - - Map oldQueues; - Map newQueues; - - if (oldIsolated) { - oldQueues = oldTenantProfile.getProfileData().getQueueConfiguration().stream() - .collect(Collectors.toMap(TenantProfileQueueConfiguration::getName, q -> q)); - } else { - oldQueues = Collections.emptyMap(); - } - - if (newIsolated) { - newQueues = newTenantProfile.getProfileData().getQueueConfiguration().stream() - .collect(Collectors.toMap(TenantProfileQueueConfiguration::getName, q -> q)); - } else { - newQueues = Collections.emptyMap(); - } - - List toRemove = new ArrayList<>(); - List toCreate = new ArrayList<>(); - List toUpdate = new ArrayList<>(); - - for (String oldQueue : oldQueues.keySet()) { - if (!newQueues.containsKey(oldQueue)) { - toRemove.add(oldQueue); - } - } - - for (String newQueue : newQueues.keySet()) { - if (oldQueues.containsKey(newQueue)) { - toUpdate.add(newQueue); - } else { - toCreate.add(newQueue); - } - } - - toRemove.forEach(q -> queueService.deleteQueueByQueueName(tenantId, q)); - - toCreate.forEach(key -> queueService.saveQueue(new Queue(tenantId, newQueues.get(key)))); - - toUpdate.forEach(key -> { - Queue queueToUpdate = new Queue(tenantId, newQueues.get(key)); - Queue foundQueue = queueService.findQueueByTenantIdAndName(tenantId, key); - queueToUpdate.setId(foundQueue.getId()); - queueToUpdate.setCreatedTime(foundQueue.getCreatedTime()); - - if (queueToUpdate.equals(foundQueue)) { - //Queue not changed - } else { - queueService.saveQueue(queueToUpdate); - } - }); - } - @Override public void deleteTenant(TenantId tenantId) { log.trace("Executing deleteTenant [{}]", tenantId);