diff --git a/application/src/main/java/org/thingsboard/server/controller/QueueController.java b/application/src/main/java/org/thingsboard/server/controller/QueueController.java index fef6fb9027..c48fc8036f 100644 --- a/application/src/main/java/org/thingsboard/server/controller/QueueController.java +++ b/application/src/main/java/org/thingsboard/server/controller/QueueController.java @@ -33,13 +33,14 @@ 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.msg.queue.ServiceType; -import org.thingsboard.server.queue.TbQueueService; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.security.permission.Operation; import org.thingsboard.server.service.security.permission.Resource; +import java.util.Collections; import java.util.Set; import java.util.UUID; +import java.util.stream.Collectors; import static org.thingsboard.server.controller.ControllerConstants.QUEUE_SERVICE_TYPE_ALLOWABLE_VALUES; import static org.thingsboard.server.controller.ControllerConstants.QUEUE_SERVICE_TYPE_DESCRIPTION; @@ -51,8 +52,6 @@ import static org.thingsboard.server.controller.ControllerConstants.TENANT_AUTHO @RequiredArgsConstructor public class QueueController extends BaseController { - private final TbQueueService tbQueueService; - @ApiOperation(value = "Get queue names (getTenantQueuesByServiceType)", notes = "Returns a set of unique queue names based on service type. " + TENANT_AUTHORITY_PARAGRAPH) @PreAuthorize("hasAuthority('TENANT_ADMIN')") @@ -62,8 +61,13 @@ public class QueueController extends BaseController { @RequestParam String serviceType) throws ThingsboardException { checkParameter("serviceType", serviceType); try { - //TODO: replace for using new QueueService - return tbQueueService.getQueuesByServiceType(ServiceType.valueOf(serviceType)); + ServiceType type = ServiceType.valueOf(serviceType); + switch (type) { + case TB_RULE_ENGINE: + return queueService.findQueuesByTenantId(getTenantId()).stream().map(Queue::getName).collect(Collectors.toSet()); + default: + return Collections.emptySet(); + } } catch (Exception e) { throw handleException(e); } diff --git a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java index 597b70275b..49d320e3e9 100644 --- a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java +++ b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java @@ -255,6 +255,7 @@ public class ThingsboardInstallService { systemDataLoaderService.createAdminSettings(); systemDataLoaderService.loadSystemWidgets(); systemDataLoaderService.createOAuth2Templates(); + systemDataLoaderService.createQueues(); // systemDataLoaderService.loadSystemPlugins(); // systemDataLoaderService.loadSystemRules(); 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 c5bdf722af..37982f4a6d 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 @@ -68,6 +68,11 @@ import org.thingsboard.server.common.data.query.DynamicValueSourceType; import org.thingsboard.server.common.data.query.EntityKeyValueType; import org.thingsboard.server.common.data.query.FilterPredicateValue; import org.thingsboard.server.common.data.query.NumericFilterPredicate; +import org.thingsboard.server.common.data.queue.ProcessingStrategy; +import org.thingsboard.server.common.data.queue.ProcessingStrategyType; +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.data.rule.RuleChainType; import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.common.data.security.DeviceCredentials; @@ -81,6 +86,7 @@ import org.thingsboard.server.dao.device.DeviceCredentialsService; import org.thingsboard.server.dao.device.DeviceProfileService; import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.dao.exception.DataValidationException; +import org.thingsboard.server.dao.queue.QueueService; import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.dao.settings.AdminSettingsService; import org.thingsboard.server.dao.tenant.TenantProfileService; @@ -155,6 +161,9 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService { @Getter private boolean persistActivityToTelemetry; + @Autowired + private QueueService queueService; + @Bean protected BCryptPasswordEncoder passwordEncoder() { return new BCryptPasswordEncoder(); @@ -575,4 +584,69 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService { }, tsCallBackExecutor); } + @Override + public void createQueues() { + Queue mainQueue = new Queue(); + mainQueue.setTenantId(TenantId.SYS_TENANT_ID); + mainQueue.setName("Main"); + mainQueue.setTopic("tb_rule_engine.main"); + mainQueue.setPollInterval(25); + mainQueue.setPartitions(10); + mainQueue.setConsumerPerPartition(true); + mainQueue.setPackProcessingTimeout(2000); + SubmitStrategy mainQueueSubmitStrategy = new SubmitStrategy(); + mainQueueSubmitStrategy.setType(SubmitStrategyType.BURST); + mainQueueSubmitStrategy.setBatchSize(1000); + mainQueue.setSubmitStrategy(mainQueueSubmitStrategy); + ProcessingStrategy mainQueueProcessingStrategy = new ProcessingStrategy(); + mainQueueProcessingStrategy.setType(ProcessingStrategyType.SKIP_ALL_FAILURES); + mainQueueProcessingStrategy.setRetries(3); + mainQueueProcessingStrategy.setFailurePercentage(0); + mainQueueProcessingStrategy.setPauseBetweenRetries(3); + mainQueueProcessingStrategy.setMaxPauseBetweenRetries(3); + mainQueue.setProcessingStrategy(mainQueueProcessingStrategy); + queueService.saveQueue(mainQueue); + + Queue highPriorityQueue = new Queue(); + highPriorityQueue.setTenantId(TenantId.SYS_TENANT_ID); + highPriorityQueue.setName("HighPriority"); + highPriorityQueue.setTopic("tb_rule_engine.hp"); + highPriorityQueue.setPollInterval(25); + highPriorityQueue.setPartitions(10); + highPriorityQueue.setConsumerPerPartition(true); + highPriorityQueue.setPackProcessingTimeout(2000); + SubmitStrategy highPriorityQueueSubmitStrategy = new SubmitStrategy(); + highPriorityQueueSubmitStrategy.setType(SubmitStrategyType.BURST); + highPriorityQueueSubmitStrategy.setBatchSize(100); + highPriorityQueue.setSubmitStrategy(highPriorityQueueSubmitStrategy); + ProcessingStrategy highPriorityQueueProcessingStrategy = new ProcessingStrategy(); + highPriorityQueueProcessingStrategy.setType(ProcessingStrategyType.RETRY_FAILED_AND_TIMED_OUT); + highPriorityQueueProcessingStrategy.setRetries(0); + highPriorityQueueProcessingStrategy.setFailurePercentage(0); + highPriorityQueueProcessingStrategy.setPauseBetweenRetries(5); + highPriorityQueueProcessingStrategy.setMaxPauseBetweenRetries(5); + highPriorityQueue.setProcessingStrategy(highPriorityQueueProcessingStrategy); + queueService.saveQueue(highPriorityQueue); + + Queue sequentialByOriginatorQueue = new Queue(); + sequentialByOriginatorQueue.setTenantId(TenantId.SYS_TENANT_ID); + sequentialByOriginatorQueue.setName("SequentialByOriginator"); + sequentialByOriginatorQueue.setTopic("tb_rule_engine.sq"); + sequentialByOriginatorQueue.setPollInterval(25); + sequentialByOriginatorQueue.setPartitions(10); + sequentialByOriginatorQueue.setPackProcessingTimeout(2000); + sequentialByOriginatorQueue.setConsumerPerPartition(true); + SubmitStrategy sequentialByOriginatorQueueSubmitStrategy = new SubmitStrategy(); + sequentialByOriginatorQueueSubmitStrategy.setType(SubmitStrategyType.SEQUENTIAL_BY_ORIGINATOR); + sequentialByOriginatorQueueSubmitStrategy.setBatchSize(100); + sequentialByOriginatorQueue.setSubmitStrategy(sequentialByOriginatorQueueSubmitStrategy); + ProcessingStrategy sequentialByOriginatorQueueProcessingStrategy = new ProcessingStrategy(); + sequentialByOriginatorQueueProcessingStrategy.setType(ProcessingStrategyType.RETRY_FAILED_AND_TIMED_OUT); + sequentialByOriginatorQueueProcessingStrategy.setRetries(3); + sequentialByOriginatorQueueProcessingStrategy.setFailurePercentage(0); + sequentialByOriginatorQueueProcessingStrategy.setPauseBetweenRetries(5); + sequentialByOriginatorQueueProcessingStrategy.setMaxPauseBetweenRetries(5); + sequentialByOriginatorQueue.setProcessingStrategy(sequentialByOriginatorQueueProcessingStrategy); + queueService.saveQueue(sequentialByOriginatorQueue); + } } diff --git a/application/src/main/java/org/thingsboard/server/service/install/SystemDataLoaderService.java b/application/src/main/java/org/thingsboard/server/service/install/SystemDataLoaderService.java index aca458bd15..1ceb1be289 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/SystemDataLoaderService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/SystemDataLoaderService.java @@ -33,4 +33,5 @@ public interface SystemDataLoaderService { void deleteSystemWidgetBundle(String bundleAlias) throws Exception; + void createQueues(); } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java index ebe99d99e0..8f1db53764 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java @@ -96,11 +96,17 @@ public class DefaultTbClusterService implements TbClusterService { @Lazy private PartitionService partitionService; - private final TbQueueProducerProvider producerProvider; + @Autowired + @Lazy + private TbQueueProducerProvider producerProvider; + + @Autowired + @Lazy + private OtaPackageStateService otaPackageStateService; + private final NotificationsTopicService notificationsTopicService; private final DataDecodingEncodingService encodingService; private final TbDeviceProfileCache deviceProfileCache; - private final OtaPackageStateService otaPackageStateService; private final GatewayNotificationsService gatewayNotificationsService; @Override @@ -486,88 +492,81 @@ public class DefaultTbClusterService implements TbClusterService { public void onQueueChange(Queue queue, TbQueueCallback callback) { log.trace("[{}][{}] Processing queue change [{}] event", queue.getTenantId(), queue.getId(), queue.getName()); -// TransportProtos.QueueUpdateMsg queueUpdateMsg = TransportProtos.QueueUpdateMsg.newBuilder() -// .setTenantIdMSB(queue.getTenantId().getId().getMostSignificantBits()) -// .setTenantIdLSB(queue.getTenantId().getId().getLeastSignificantBits()) -// .setQueueName(queue.getName()) -// .setQueueTopic(queue.getTopic()) -// .setPartitions(queue.getPartitions()) -// .build(); -// -// if ("Main".equals(queue.getName())) { -// Set tbTransportServices = partitionService.getAllServices(ServiceType.TB_TRANSPORT); -// ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setQueueUpdateMsg(queueUpdateMsg).build(); -// for (TransportProtos.ServiceInfo transportService : tbTransportServices) { -// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, transportService.getServiceId()); -// producerProvider.getTransportNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), transportMsg), callback); -// toTransportNfs.incrementAndGet(); -// } -// -// Set tbCoreServices = partitionService.getAllServices(ServiceType.TB_CORE); -// ToCoreNotificationMsg coreMsg = ToCoreNotificationMsg.newBuilder().setQueueUpdateMsg(queueUpdateMsg).build(); -// for (TransportProtos.ServiceInfo coreService : tbCoreServices) { -// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, coreService.getServiceId()); -// producerProvider.getTbCoreNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), coreMsg), callback); -// toCoreNfs.incrementAndGet(); -// } -// } else { -// Set tbRuleEngineServices = partitionService.getAllServices(ServiceType.TB_RULE_ENGINE); -// ToRuleEngineNotificationMsg ruleEngineMsg = ToRuleEngineNotificationMsg.newBuilder().setQueueUpdateMsg(queueUpdateMsg).build(); -// for (TransportProtos.ServiceInfo ruleEngineService : tbRuleEngineServices) { -// TenantId tenantId = new TenantId(new UUID(ruleEngineService.getTenantIdMSB(), ruleEngineService.getTenantIdLSB())); -// if (tenantId.equals(queue.getTenantId())) { -// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, ruleEngineService.getServiceId()); -// producerProvider.getRuleEngineNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), ruleEngineMsg), callback); -// toRuleEngineNfs.incrementAndGet(); -// } -// } -// } + TransportProtos.QueueUpdateMsg queueUpdateMsg = TransportProtos.QueueUpdateMsg.newBuilder() + .setTenantIdMSB(queue.getTenantId().getId().getMostSignificantBits()) + .setTenantIdLSB(queue.getTenantId().getId().getLeastSignificantBits()) + .setQueueName(queue.getName()) + .setQueueTopic(queue.getTopic()) + .setPartitions(queue.getPartitions()) + .build(); + + if ("Main".equals(queue.getName())) { + Set tbTransportServices = partitionService.getAllServices(ServiceType.TB_TRANSPORT); + ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setQueueUpdateMsg(queueUpdateMsg).build(); + for (TransportProtos.ServiceInfo transportService : tbTransportServices) { + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, transportService.getServiceId()); + producerProvider.getTransportNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), transportMsg), callback); + toTransportNfs.incrementAndGet(); + } + + Set tbCoreServices = partitionService.getAllServices(ServiceType.TB_CORE); + ToCoreNotificationMsg coreMsg = ToCoreNotificationMsg.newBuilder().setQueueUpdateMsg(queueUpdateMsg).build(); + for (TransportProtos.ServiceInfo coreService : tbCoreServices) { + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, coreService.getServiceId()); + producerProvider.getTbCoreNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), coreMsg), callback); + toCoreNfs.incrementAndGet(); + } + } else { + Set tbRuleEngineServices = partitionService.getAllServices(ServiceType.TB_RULE_ENGINE); + ToRuleEngineNotificationMsg ruleEngineMsg = ToRuleEngineNotificationMsg.newBuilder().setQueueUpdateMsg(queueUpdateMsg).build(); + for (TransportProtos.ServiceInfo ruleEngineService : tbRuleEngineServices) { + TenantId tenantId = new TenantId(new UUID(ruleEngineService.getTenantIdMSB(), ruleEngineService.getTenantIdLSB())); + if (tenantId.equals(queue.getTenantId())) { + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, ruleEngineService.getServiceId()); + producerProvider.getRuleEngineNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), ruleEngineMsg), callback); + toRuleEngineNfs.incrementAndGet(); + } + } + } } @Override public void onQueueDelete(Queue queue, TbQueueCallback callback) { log.trace("[{}][{}] Processing queue delete [{}] event", queue.getTenantId(), queue.getId(), queue.getName()); -// TransportProtos.QueueDeleteMsg queueDeleteMsg = TransportProtos.QueueDeleteMsg.newBuilder() -// .setTenantIdMSB(queue.getTenantId().getId().getMostSignificantBits()) -// .setTenantIdLSB(queue.getTenantId().getId().getLeastSignificantBits()) -// .setQueueName(queue.getName()) -// .build(); -// -// if ("Main".equals(queue.getName())) { -// Set tbTransportServices = partitionService.getAllServices(ServiceType.TB_TRANSPORT); -// ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setQueueDeleteMsg(queueDeleteMsg).build(); -// for (TransportProtos.ServiceInfo transportService : tbTransportServices) { -// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, transportService.getServiceId()); -// producerProvider.getTransportNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), transportMsg), callback); -// toTransportNfs.incrementAndGet(); -// } -// -// Set tbCoreServices = partitionService.getAllServices(ServiceType.TB_CORE); -// ToCoreNotificationMsg coreMsg = ToCoreNotificationMsg.newBuilder().setQueueDeleteMsg(queueDeleteMsg).build(); -// for (TransportProtos.ServiceInfo coreService : tbCoreServices) { -// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, coreService.getServiceId()); -// producerProvider.getTbCoreNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), coreMsg), callback); -// toCoreNfs.incrementAndGet(); -// } -// } else { -// //TODO: 3.2 should be changed with smth like TransportApiRequestTemplate and get responses from all RE's -// Set tbRuleEngineServices = partitionService.getAllServices(ServiceType.TB_RULE_ENGINE); -// ToRuleEngineNotificationMsg ruleEngineMsg = ToRuleEngineNotificationMsg.newBuilder().setQueueDeleteMsg(queueDeleteMsg).build(); -// for (TransportProtos.ServiceInfo ruleEngineService : tbRuleEngineServices) { -// TenantId tenantId = new TenantId(new UUID(ruleEngineService.getTenantIdMSB(), ruleEngineService.getTenantIdLSB())); -// if (tenantId.equals(queue.getTenantId())) { -// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, ruleEngineService.getServiceId()); -// log.info("Send notification about deleting queue [{}] to Rule Engine [{}]", queue.getName(), tpi.getFullTopicName()); -// producerProvider.getRuleEngineNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), ruleEngineMsg), callback); -// toRuleEngineNfs.incrementAndGet(); -// } -// } -// try { -// Thread.sleep(3000); -// } catch (InterruptedException e) { -// e.printStackTrace(); -// } -// } + TransportProtos.QueueDeleteMsg queueDeleteMsg = TransportProtos.QueueDeleteMsg.newBuilder() + .setTenantIdMSB(queue.getTenantId().getId().getMostSignificantBits()) + .setTenantIdLSB(queue.getTenantId().getId().getLeastSignificantBits()) + .setQueueName(queue.getName()) + .build(); + + if ("Main".equals(queue.getName())) { + Set tbTransportServices = partitionService.getAllServices(ServiceType.TB_TRANSPORT); + ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setQueueDeleteMsg(queueDeleteMsg).build(); + for (TransportProtos.ServiceInfo transportService : tbTransportServices) { + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, transportService.getServiceId()); + producerProvider.getTransportNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), transportMsg), callback); + toTransportNfs.incrementAndGet(); + } + + Set tbCoreServices = partitionService.getAllServices(ServiceType.TB_CORE); + ToCoreNotificationMsg coreMsg = ToCoreNotificationMsg.newBuilder().setQueueDeleteMsg(queueDeleteMsg).build(); + for (TransportProtos.ServiceInfo coreService : tbCoreServices) { + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, coreService.getServiceId()); + producerProvider.getTbCoreNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), coreMsg), callback); + toCoreNfs.incrementAndGet(); + } + } else { + Set tbRuleEngineServices = partitionService.getAllServices(ServiceType.TB_RULE_ENGINE); + ToRuleEngineNotificationMsg ruleEngineMsg = ToRuleEngineNotificationMsg.newBuilder().setQueueDeleteMsg(queueDeleteMsg).build(); + for (TransportProtos.ServiceInfo ruleEngineService : tbRuleEngineServices) { + TenantId tenantId = new TenantId(new UUID(ruleEngineService.getTenantIdMSB(), ruleEngineService.getTenantIdLSB())); + if (tenantId.equals(queue.getTenantId())) { + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, ruleEngineService.getServiceId()); + producerProvider.getRuleEngineNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), ruleEngineMsg), callback); + toRuleEngineNfs.incrementAndGet(); + } + } + } } } 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 f1bd296e5f..7113272187 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 @@ -38,6 +38,7 @@ import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse; import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; +import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.DeviceStateServiceMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.EdgeNotificationMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.FromDeviceRPCResponseProto; @@ -57,6 +58,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceM import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.util.TbCoreComponent; @@ -115,6 +117,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService> usageStatsConsumer; private final TbQueueConsumer> firmwareStatesConsumer; @@ -135,7 +138,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService>> consumers = new ConcurrentHashMap<>(); - private final ConcurrentMap consumerConfigurations = new ConcurrentHashMap<>(); + private final ConcurrentMap consumerConfigurations = new ConcurrentHashMap<>(); private final ConcurrentMap consumerStats = new ConcurrentHashMap<>(); private final ConcurrentMap topicsConsumerPerPartition = new ConcurrentHashMap<>(); final ExecutorService submitExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-rule-engine-consumer-submit")); @@ -108,7 +114,6 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< public DefaultTbRuleEngineConsumerService(TbRuleEngineProcessingStrategyFactory processingStrategyFactory, TbRuleEngineSubmitStrategyFactory submitStrategyFactory, - TbQueueRuleEngineSettings ruleEngineSettings, TbRuleEngineQueueFactory tbRuleEngineQueueFactory, RuleEngineStatisticsService statisticsService, ActorSystemContext actorContext, @@ -117,28 +122,37 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< StatsFactory statsFactory, TbDeviceProfileCache deviceProfileCache, TbTenantProfileCache tenantProfileCache, - TbApiUsageStateService apiUsageStateService) { + TbApiUsageStateService apiUsageStateService, + PartitionService partitionService, TbServiceInfoProvider serviceInfoProvider, QueueService queueService) { super(actorContext, encodingService, tenantProfileCache, deviceProfileCache, apiUsageStateService, tbRuleEngineQueueFactory.createToRuleEngineNotificationsMsgConsumer()); this.statisticsService = statisticsService; - this.ruleEngineSettings = ruleEngineSettings; this.tbRuleEngineQueueFactory = tbRuleEngineQueueFactory; this.submitStrategyFactory = submitStrategyFactory; this.processingStrategyFactory = processingStrategyFactory; this.tbDeviceRpcService = tbDeviceRpcService; this.statsFactory = statsFactory; + this.partitionService = partitionService; + this.serviceInfoProvider = serviceInfoProvider; + this.queueService = queueService; + this.tenantId = actorContext.getServiceInfoProvider().getIsolatedTenant().orElse(TenantId.SYS_TENANT_ID); } @PostConstruct public void init() { super.init("tb-rule-engine-consumer", "tb-rule-engine-notifications-consumer"); - for (TbRuleEngineQueueConfiguration configuration : ruleEngineSettings.getQueues()) { - consumerConfigurations.putIfAbsent(configuration.getName(), configuration); - consumerStats.put(configuration.getName(), new TbRuleEngineConsumerStats(configuration.getName(), statsFactory)); - if (!configuration.isConsumerPerPartition()) { - consumers.computeIfAbsent(configuration.getName(), queueName -> tbRuleEngineQueueFactory.createToRuleEngineMsgConsumer(configuration)); - } else { - topicsConsumerPerPartition.computeIfAbsent(configuration.getName(), TbTopicWithConsumerPerPartition::new); - } + List queues = queueService.findQueuesByTenantId(tenantId); + for (Queue configuration : queues) { + initConsumer(configuration); + } + } + + private void initConsumer(Queue configuration) { + consumerConfigurations.putIfAbsent(configuration.getName(), configuration); + consumerStats.putIfAbsent(configuration.getName(), new TbRuleEngineConsumerStats(configuration.getName(), statsFactory)); + if (!configuration.isConsumerPerPartition()) { + consumers.computeIfAbsent(configuration.getName(), queueName -> tbRuleEngineQueueFactory.createToRuleEngineMsgConsumer(configuration)); + } else { + topicsConsumerPerPartition.computeIfAbsent(configuration.getName(), TbTopicWithConsumerPerPartition::new); } } @@ -147,7 +161,6 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< super.destroy(); submitExecutor.shutdownNow(); repartitionExecutor.shutdownNow(); - ruleEngineSettings.getQueues().forEach(config -> consumerConfigurations.put(config.getName(), config)); } @Override @@ -178,7 +191,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< return; } TbTopicWithConsumerPerPartition tbTopicWithConsumerPerPartition = topicsConsumerPerPartition.get(queueName); - Queue> subscribeQueue = tbTopicWithConsumerPerPartition.getSubscribeQueue(); + java.util.Queue> subscribeQueue = tbTopicWithConsumerPerPartition.getSubscribeQueue(); if (subscribeQueue.isEmpty()) { return; } @@ -207,7 +220,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< addedPartitions.forEach((tpi) -> { log.info("[{}] Adding consumer for topic: {}", queueName, tpi); - TbRuleEngineQueueConfiguration configuration = consumerConfigurations.get(queueName); + Queue configuration = consumerConfigurations.get(queueName); TbQueueConsumer> consumer = tbRuleEngineQueueFactory.createToRuleEngineMsgConsumer(configuration); consumers.put(tpi, consumer); launchConsumer(consumer, consumerConfigurations.get(queueName), consumerStats.get(queueName), "" + queueName + "-" + tpi.getPartition().orElse(-999999)); @@ -241,11 +254,11 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< .forEach((tpi) -> removeConsumerForTopicByTpi(tbTopicWithConsumerPerPartition.getTopic(), tbTopicWithConsumerPerPartition.getConsumers(), tpi))); } - void launchConsumer(TbQueueConsumer> consumer, TbRuleEngineQueueConfiguration configuration, TbRuleEngineConsumerStats stats, String threadSuffix) { + void launchConsumer(TbQueueConsumer> consumer, Queue configuration, TbRuleEngineConsumerStats stats, String threadSuffix) { consumersExecutor.execute(() -> consumerLoop(consumer, configuration, stats, threadSuffix)); } - void consumerLoop(TbQueueConsumer> consumer, TbRuleEngineQueueConfiguration configuration, TbRuleEngineConsumerStats stats, String threadSuffix) { + void consumerLoop(TbQueueConsumer> consumer, org.thingsboard.server.common.data.queue.Queue configuration, TbRuleEngineConsumerStats stats, String threadSuffix) { updateCurrentThreadName(threadSuffix); while (!stopped && !consumer.isStopped()) { try { @@ -310,15 +323,15 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< Thread.currentThread().setName(name); } - TbRuleEngineProcessingStrategy getAckStrategy(TbRuleEngineQueueConfiguration configuration) { + TbRuleEngineProcessingStrategy getAckStrategy(Queue configuration) { return processingStrategyFactory.newInstance(configuration.getName(), configuration.getProcessingStrategy()); } - TbRuleEngineSubmitStrategy getSubmitStrategy(TbRuleEngineQueueConfiguration configuration) { + TbRuleEngineSubmitStrategy getSubmitStrategy(Queue configuration) { return submitStrategyFactory.newInstance(configuration.getName(), configuration.getSubmitStrategy()); } - void submitMessage(TbRuleEngineQueueConfiguration configuration, TbRuleEngineConsumerStats stats, TbMsgPackProcessingContext ctx, UUID id, TbProtoQueueMsg msg) { + void submitMessage(Queue configuration, TbRuleEngineConsumerStats stats, TbMsgPackProcessingContext ctx, UUID id, TbProtoQueueMsg msg) { log.trace("[{}] Creating callback for topic {} message: {}", id, configuration.getName(), msg.getValue()); ToRuleEngineMsg toRuleEngineMsg = msg.getValue(); TenantId tenantId = TenantId.fromUUID(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB())); @@ -336,7 +349,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< } } - private void printFirstOrAll(TbRuleEngineQueueConfiguration configuration, TbMsgPackProcessingContext ctx, Map> map, String prefix) { + private void printFirstOrAll(Queue configuration, TbMsgPackProcessingContext ctx, Map> map, String prefix) { boolean printAll = log.isTraceEnabled(); log.info("{} to process [{}] messages", prefix, map.size()); for (Map.Entry> pending : map.entrySet()) { @@ -380,12 +393,67 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< , proto.getResponse(), error); tbDeviceRpcService.processRpcResponseFromDevice(response); callback.onSuccess(); + } else if (nfMsg.hasQueueUpdateMsg()) { + updateQueue(nfMsg.getQueueUpdateMsg()); + callback.onSuccess(); + } else if (nfMsg.hasQueueDeleteMsg()) { + deleteQueue(nfMsg.getQueueDeleteMsg()); + callback.onSuccess(); } else { log.trace("Received notification with missing handler"); callback.onSuccess(); } } + private void updateQueue(TransportProtos.QueueUpdateMsg queueUpdateMsg) { + String queueName = queueUpdateMsg.getQueueName(); + Queue queue = queueService.findQueueByTenantIdAndName(tenantId, queueName); + Queue oldQueue = consumerConfigurations.remove(queueName); + if (oldQueue != null) { + if (oldQueue.isConsumerPerPartition()) { + TbTopicWithConsumerPerPartition consumerPerPartition = topicsConsumerPerPartition.remove(queueName); + ReentrantLock lock = consumerPerPartition.getLock(); + try { + lock.lock(); + consumerPerPartition.getConsumers().values().forEach(TbQueueConsumer::unsubscribe); + } finally { + lock.unlock(); + } + } else { + TbQueueConsumer> consumer = consumers.remove(queueName); + consumer.unsubscribe(); + } + } + + initConsumer(queue); + + if (!queue.isConsumerPerPartition()) { + launchConsumer(consumers.get(queueName), consumerConfigurations.get(queueName), consumerStats.get(queueName), queueName); + } + + partitionService.updateQueue(queueUpdateMsg); + partitionService.recalculatePartitions(serviceInfoProvider.getServiceInfo(), new ArrayList<>(partitionService.getOtherServices(ServiceType.TB_RULE_ENGINE))); + } + + private void deleteQueue(TransportProtos.QueueDeleteMsg queueDeleteMsg) { + Queue queue = consumerConfigurations.remove(queueDeleteMsg.getQueueName()); + if (queue != null) { + if (queue.isConsumerPerPartition()) { + TbTopicWithConsumerPerPartition tbTopicWithConsumerPerPartition = topicsConsumerPerPartition.remove(queueDeleteMsg.getQueueName()); + if (tbTopicWithConsumerPerPartition != null) { + tbTopicWithConsumerPerPartition.getConsumers().values().forEach(TbQueueConsumer::unsubscribe); + tbTopicWithConsumerPerPartition.getConsumers().clear(); + } + } else { + TbQueueConsumer> consumer = consumers.remove(queueDeleteMsg.getQueueName()); + if (consumer != null) { + consumer.unsubscribe(); + } + } + } + partitionService.removeQueue(queueDeleteMsg); + } + private void forwardToRuleEngineActor(String queueName, TenantId tenantId, ToRuleEngineMsg toRuleEngineMsg, TbMsgCallback callback) { TbMsg tbMsg = TbMsg.fromBytes(queueName, toRuleEngineMsg.getTbMsg().toByteArray(), callback); QueueToRuleEngineMsg msg; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbTopicWithConsumerPerPartition.java b/application/src/main/java/org/thingsboard/server/service/queue/TbTopicWithConsumerPerPartition.java index b1696dcc1e..e51b345f98 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbTopicWithConsumerPerPartition.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbTopicWithConsumerPerPartition.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.service.queue; -import lombok.Builder; import lombok.Data; import lombok.Getter; import lombok.RequiredArgsConstructor; @@ -25,7 +24,6 @@ import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import java.util.Collections; -import java.util.Map; import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java index 02774207b0..3056e65c78 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.queue.processing; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; +import org.thingsboard.server.common.data.queue.ProcessingStrategy; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.queue.TbMsgCallback; import org.thingsboard.server.gen.transport.TransportProtos; @@ -33,22 +34,22 @@ import java.util.concurrent.TimeUnit; @Slf4j public class TbRuleEngineProcessingStrategyFactory { - public TbRuleEngineProcessingStrategy newInstance(String name, TbRuleEngineQueueAckStrategyConfiguration configuration) { - switch (configuration.getType()) { - case "SKIP_ALL_FAILURES": + public TbRuleEngineProcessingStrategy newInstance(String name, ProcessingStrategy processingStrategy) { + switch (processingStrategy.getType()) { + case SKIP_ALL_FAILURES: return new SkipStrategy(name, false); - case "SKIP_ALL_FAILURES_AND_TIMED_OUT": + case SKIP_ALL_FAILURES_AND_TIMED_OUT: return new SkipStrategy(name, true); - case "RETRY_ALL": - return new RetryStrategy(name, true, true, true, configuration); - case "RETRY_FAILED": - return new RetryStrategy(name, false, true, false, configuration); - case "RETRY_TIMED_OUT": - return new RetryStrategy(name, false, false, true, configuration); - case "RETRY_FAILED_AND_TIMED_OUT": - return new RetryStrategy(name, false, true, true, configuration); + case RETRY_ALL: + return new RetryStrategy(name, true, true, true, processingStrategy); + case RETRY_FAILED: + return new RetryStrategy(name, false, true, false, processingStrategy); + case RETRY_TIMED_OUT: + return new RetryStrategy(name, false, false, true, processingStrategy); + case RETRY_FAILED_AND_TIMED_OUT: + return new RetryStrategy(name, false, true, true, processingStrategy); default: - throw new RuntimeException("TbRuleEngineProcessingStrategy with type " + configuration.getType() + " is not supported!"); + throw new RuntimeException("TbRuleEngineProcessingStrategy with type " + processingStrategy.getType() + " is not supported!"); } } @@ -66,15 +67,15 @@ public class TbRuleEngineProcessingStrategyFactory { private int initialTotalCount; private int retryCount; - public RetryStrategy(String queueName, boolean retrySuccessful, boolean retryFailed, boolean retryTimeout, TbRuleEngineQueueAckStrategyConfiguration configuration) { + public RetryStrategy(String queueName, boolean retrySuccessful, boolean retryFailed, boolean retryTimeout, ProcessingStrategy processingStrategy) { this.queueName = queueName; this.retrySuccessful = retrySuccessful; this.retryFailed = retryFailed; this.retryTimeout = retryTimeout; - this.maxRetries = configuration.getRetries(); - this.maxAllowedFailurePercentage = configuration.getFailurePercentage(); - this.pauseBetweenRetries = configuration.getPauseBetweenRetries(); - this.maxPauseBetweenRetries = configuration.getMaxPauseBetweenRetries(); + this.maxRetries = processingStrategy.getRetries(); + this.maxAllowedFailurePercentage = processingStrategy.getFailurePercentage(); + this.pauseBetweenRetries = processingStrategy.getPauseBetweenRetries(); + this.maxPauseBetweenRetries = processingStrategy.getMaxPauseBetweenRetries(); } @Override diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineSubmitStrategyFactory.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineSubmitStrategyFactory.java index 3c4d538394..5c4d3b14ab 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineSubmitStrategyFactory.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineSubmitStrategyFactory.java @@ -17,26 +17,27 @@ package org.thingsboard.server.service.queue.processing; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.queue.SubmitStrategy; import org.thingsboard.server.queue.settings.TbRuleEngineQueueSubmitStrategyConfiguration; @Component @Slf4j public class TbRuleEngineSubmitStrategyFactory { - public TbRuleEngineSubmitStrategy newInstance(String name, TbRuleEngineQueueSubmitStrategyConfiguration configuration) { - switch (configuration.getType()) { - case "BURST": + public TbRuleEngineSubmitStrategy newInstance(String name, SubmitStrategy submitStrategy) { + switch (submitStrategy.getType()) { + case BURST: return new BurstTbRuleEngineSubmitStrategy(name); - case "BATCH": - return new BatchTbRuleEngineSubmitStrategy(name, configuration.getBatchSize()); - case "SEQUENTIAL_BY_ORIGINATOR": + case BATCH: + return new BatchTbRuleEngineSubmitStrategy(name, submitStrategy.getBatchSize()); + case SEQUENTIAL_BY_ORIGINATOR: return new SequentialByOriginatorIdTbRuleEngineSubmitStrategy(name); - case "SEQUENTIAL_BY_TENANT": + case SEQUENTIAL_BY_TENANT: return new SequentialByTenantIdTbRuleEngineSubmitStrategy(name); - case "SEQUENTIAL": + case SEQUENTIAL: return new SequentialTbRuleEngineSubmitStrategy(name); default: - throw new RuntimeException("TbRuleEngineProcessingStrategy with type " + configuration.getType() + " is not supported!"); + throw new RuntimeException("TbRuleEngineProcessingStrategy with type " + submitStrategy.getType() + " is not supported!"); } } diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java index 6625bdefc7..157e10295c 100644 --- a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java @@ -69,6 +69,11 @@ 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.page.TimePageLink; +import org.thingsboard.server.common.data.queue.ProcessingStrategy; +import org.thingsboard.server.common.data.queue.ProcessingStrategyType; +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.data.security.Authority; import org.thingsboard.server.config.ThingsboardSecurityConfiguration; import org.thingsboard.server.dao.tenant.TenantProfileService; diff --git a/application/src/test/java/org/thingsboard/server/service/cluster/routing/HashPartitionServiceTest.java b/application/src/test/java/org/thingsboard/server/service/cluster/routing/HashPartitionServiceTest.java index 09fe580661..0c13157623 100644 --- a/application/src/test/java/org/thingsboard/server/service/cluster/routing/HashPartitionServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/cluster/routing/HashPartitionServiceTest.java @@ -26,14 +26,13 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.test.util.ReflectionTestUtils; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.queue.TbQueueService; -import org.thingsboard.server.queue.discovery.HashPartitionService; import org.thingsboard.server.common.msg.queue.ServiceType; -import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.queue.discovery.HashPartitionService; +import org.thingsboard.server.queue.discovery.QueueRoutingInfoService; +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.discovery.TenantRoutingInfoService; -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import java.util.ArrayList; import java.util.Collections; @@ -44,7 +43,6 @@ import java.util.Map; import java.util.stream.Collectors; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; @Slf4j @RunWith(MockitoJUnitRunner.class) @@ -57,26 +55,20 @@ public class HashPartitionServiceTest { private TbServiceInfoProvider discoveryService; private TenantRoutingInfoService routingInfoService; private ApplicationEventPublisher applicationEventPublisher; - private TbQueueRuleEngineSettings ruleEngineSettings; - private TbQueueService queueService; + private QueueRoutingInfoService queueRoutingInfoService; private String hashFunctionName = "sha256"; - @Before public void setup() throws Exception { discoveryService = mock(TbServiceInfoProvider.class); applicationEventPublisher = mock(ApplicationEventPublisher.class); routingInfoService = mock(TenantRoutingInfoService.class); - ruleEngineSettings = mock(TbQueueRuleEngineSettings.class); - queueService = mock(TbQueueService.class); + queueRoutingInfoService = mock(QueueRoutingInfoService.class); clusterRoutingService = new HashPartitionService(discoveryService, routingInfoService, applicationEventPublisher, - ruleEngineSettings, - queueService - ); - when(ruleEngineSettings.getQueues()).thenReturn(Collections.emptyList()); + queueRoutingInfoService); ReflectionTestUtils.setField(clusterRoutingService, "coreTopic", "tb.core"); ReflectionTestUtils.setField(clusterRoutingService, "corePartitions", 10); ReflectionTestUtils.setField(clusterRoutingService, "hashFunctionName", hashFunctionName); diff --git a/common/cluster-api/src/main/proto/queue.proto b/common/cluster-api/src/main/proto/queue.proto index 1084cb2e44..014244cb08 100644 --- a/common/cluster-api/src/main/proto/queue.proto +++ b/common/cluster-api/src/main/proto/queue.proto @@ -20,11 +20,11 @@ package transport; option java_package = "org.thingsboard.server.gen.transport"; option java_outer_classname = "TransportProtos"; -message QueueInfo { - string name = 1; - string topic = 2; - int32 partitions = 3; -} +//message QueueInfo { +// string name = 1; +// string topic = 2; +// int32 partitions = 3; +//} /** * Service Discovery Data Structures; @@ -34,7 +34,7 @@ message ServiceInfo { repeated string serviceTypes = 2; int64 tenantIdMSB = 3; int64 tenantIdLSB = 4; - repeated QueueInfo ruleEngineQueues = 5; +// repeated QueueInfo ruleEngineQueues = 5; repeated string transports = 6; } @@ -741,6 +741,8 @@ message ToCoreNotificationMsg { FromDeviceRPCResponseProto fromDeviceRpcResponse = 2; bytes componentLifecycleMsg = 3; bytes edgeEventUpdateMsg = 4; + QueueUpdateMsg queueUpdateMsg = 5; + QueueDeleteMsg queueDeleteMsg = 6; } /* Messages that are handled by ThingsBoard RuleEngine Service */ @@ -755,6 +757,8 @@ message ToRuleEngineMsg { message ToRuleEngineNotificationMsg { bytes componentLifecycleMsg = 1; FromDeviceRPCResponseProto fromDeviceRpcResponse = 2; + QueueUpdateMsg queueUpdateMsg = 3; + QueueDeleteMsg queueDeleteMsg = 4; } /* Messages that are handled by ThingsBoard Transport Service */ @@ -773,6 +777,8 @@ message ToTransportMsg { ResourceUpdateMsg resourceUpdateMsg = 12; ResourceDeleteMsg resourceDeleteMsg = 13; UplinkNotificationMsg uplinkNotificationMsg = 14; + QueueUpdateMsg queueUpdateMsg = 15; + QueueDeleteMsg queueDeleteMsg = 16; } message UsageStatsKVProto{ diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/DefaultTbQueueService.java b/common/queue/src/main/java/org/thingsboard/server/queue/DefaultTbQueueService.java index 5a0e540ac4..e372fb600c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/DefaultTbQueueService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/DefaultTbQueueService.java @@ -29,7 +29,7 @@ import java.util.LinkedHashSet; import java.util.Set; import java.util.stream.Collectors; -@Service +//@Service @RequiredArgsConstructor public class DefaultTbQueueService implements TbQueueService { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java index 64ca3ccac5..62e9852dd1 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java @@ -25,10 +25,8 @@ import org.springframework.util.StringUtils; import org.thingsboard.server.common.data.TbTransportService; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.queue.ServiceType; -import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ServiceInfo; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import org.thingsboard.server.queue.util.AfterContextReady; import javax.annotation.PostConstruct; @@ -95,16 +93,6 @@ public class DefaultTbServiceInfoProvider implements TbServiceInfoProvider { builder.setTenantIdMSB(tenantId.getMostSignificantBits()); builder.setTenantIdLSB(tenantId.getLeastSignificantBits()); - if (serviceTypes.contains(ServiceType.TB_RULE_ENGINE) && ruleEngineSettings != null) { - for (TbRuleEngineQueueConfiguration queue : ruleEngineSettings.getQueues()) { - TransportProtos.QueueInfo queueInfo = TransportProtos.QueueInfo.newBuilder() - .setName(queue.getName()) - .setTopic(queue.getTopic()) - .setPartitions(queue.getPartitions()).build(); - builder.addRuleEngineQueues(queueInfo); - } - } - serviceInfo = builder.build(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java index 4e27a01760..58ef13ac6d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java @@ -29,11 +29,9 @@ import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ServiceInfo; -import org.thingsboard.server.queue.TbQueueService; import org.thingsboard.server.queue.discovery.event.ClusterTopologyChangeEvent; import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.event.ServiceListChangedEvent; -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import javax.annotation.PostConstruct; import java.util.ArrayList; @@ -63,10 +61,9 @@ public class HashPartitionService implements PartitionService { private final ApplicationEventPublisher applicationEventPublisher; private final TbServiceInfoProvider serviceInfoProvider; private final TenantRoutingInfoService tenantRoutingInfoService; - private final TbQueueRuleEngineSettings tbQueueRuleEngineSettings; - private final TbQueueService queueService; - private final ConcurrentMap partitionTopics = new ConcurrentHashMap<>(); - private final ConcurrentMap partitionSizes = new ConcurrentHashMap<>(); + private final QueueRoutingInfoService queueRoutingInfoService; + private final ConcurrentMap> partitionTopicsMap = new ConcurrentHashMap<>(); + private final ConcurrentMap> partitionSizesMap = new ConcurrentHashMap<>(); private final ConcurrentMap tenantRoutingInfoMap = new ConcurrentHashMap<>(); private ConcurrentMap> myPartitions = new ConcurrentHashMap<>(); @@ -80,26 +77,87 @@ public class HashPartitionService implements PartitionService { public HashPartitionService(TbServiceInfoProvider serviceInfoProvider, TenantRoutingInfoService tenantRoutingInfoService, ApplicationEventPublisher applicationEventPublisher, - TbQueueRuleEngineSettings tbQueueRuleEngineSettings, - TbQueueService queueService) { + QueueRoutingInfoService queueRoutingInfoService) { this.serviceInfoProvider = serviceInfoProvider; this.tenantRoutingInfoService = tenantRoutingInfoService; this.applicationEventPublisher = applicationEventPublisher; - this.tbQueueRuleEngineSettings = tbQueueRuleEngineSettings; - this.queueService = queueService; + this.queueRoutingInfoService = queueRoutingInfoService; } @PostConstruct public void init() { this.hashFunction = forName(hashFunctionName); - partitionSizes.put(new ServiceQueue(ServiceType.TB_CORE), corePartitions); - partitionTopics.put(new ServiceQueue(ServiceType.TB_CORE), coreTopic); - tbQueueRuleEngineSettings.getQueues().forEach(queueConfiguration -> { - partitionTopics.put(new ServiceQueue(ServiceType.TB_RULE_ENGINE, queueConfiguration.getName()), queueConfiguration.getTopic()); - partitionSizes.put(new ServiceQueue(ServiceType.TB_RULE_ENGINE, queueConfiguration.getName()), queueConfiguration.getPartitions()); + partitionsInit(); + } + + private void partitionsInit() { + addPartitionSizeToMap(TenantId.SYS_TENANT_ID, new ServiceQueue(ServiceType.TB_CORE, "Main"), corePartitions); + addPartitionTopicToMap(TenantId.SYS_TENANT_ID, new ServiceQueue(ServiceType.TB_CORE, "Main"), coreTopic); + + List queueRoutingInfoList; + + String serviceType = serviceInfoProvider.getServiceType(); + + if ("tb-rule-engine".equals(serviceType)) { + queueRoutingInfoList = queueRoutingInfoService.getQueuesRoutingInfo(serviceInfoProvider.getIsolatedTenant().orElse(TenantId.SYS_TENANT_ID)); + } else if ("monolith".equals(serviceType)) { + queueRoutingInfoList = queueRoutingInfoService.getAllQueuesRoutingInfo(); + } else if ("tb-core".equals(serviceType)) { + queueRoutingInfoList = queueRoutingInfoService.getMainQueuesRoutingInfo(); + } else { + //If transport started earlier than tb-core + int getQueuesRetries = 10; + while (true) { + if (getQueuesRetries > 0) { + log.info("Try to get queue routing info."); + try { + queueRoutingInfoList = queueRoutingInfoService.getMainQueuesRoutingInfo(); + break; + } catch (Exception e) { + log.info("Failed to get queues routing info!"); + getQueuesRetries--; + } + try { + Thread.sleep(10000); + } catch (InterruptedException e) { + log.info("Failed to await queues routing info!", e); + } + } else { + throw new RuntimeException("Failed to await queues routing info!"); + } + } + } + + queueRoutingInfoList.forEach(queue -> { + addPartitionTopicToMap(queue.getTenantId(), new ServiceQueue(ServiceType.TB_RULE_ENGINE, queue.getQueueName()), queue.getQueueTopic()); + addPartitionSizeToMap(queue.getTenantId(), new ServiceQueue(ServiceType.TB_RULE_ENGINE, queue.getQueueName()), queue.getPartitions()); }); } + @Override + public void updateQueue(TransportProtos.QueueUpdateMsg queueUpdateMsg) { + TenantId tenantId = new TenantId(new UUID(queueUpdateMsg.getTenantIdMSB(), queueUpdateMsg.getTenantIdLSB())); + addPartitionTopicToMap(tenantId, new ServiceQueue(ServiceType.TB_RULE_ENGINE, queueUpdateMsg.getQueueName()), queueUpdateMsg.getQueueTopic()); + addPartitionSizeToMap(tenantId, new ServiceQueue(ServiceType.TB_RULE_ENGINE, queueUpdateMsg.getQueueName()), queueUpdateMsg.getPartitions()); + } + + @Override + public void removeQueue(TransportProtos.QueueDeleteMsg queueDeleteMsg) { + TenantId tenantId = new TenantId(new UUID(queueDeleteMsg.getTenantIdMSB(), queueDeleteMsg.getTenantIdLSB())); + ServiceQueue serviceQueue = new ServiceQueue(ServiceType.TB_RULE_ENGINE, queueDeleteMsg.getQueueName()); + partitionTopicsMap.get(tenantId).remove(serviceQueue); + partitionSizesMap.get(tenantId).remove(serviceQueue); + myPartitions.remove(new ServiceQueueKey(serviceQueue, tenantId)); + } + + private void addPartitionSizeToMap(TenantId tenantId, ServiceQueue serviceQueue, int partitions) { + partitionSizesMap.computeIfAbsent(tenantId, id -> new ConcurrentHashMap<>()).put(serviceQueue, partitions); + } + + private void addPartitionTopicToMap(TenantId tenantId, ServiceQueue serviceQueue, String topic) { + partitionTopicsMap.computeIfAbsent(tenantId, id -> new ConcurrentHashMap<>()).put(serviceQueue, topic); + } + @Override public TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, EntityId entityId) { return resolve(new ServiceQueue(serviceType), tenantId, entityId); @@ -107,23 +165,25 @@ public class HashPartitionService implements PartitionService { @Override public TopicPartitionInfo resolve(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId) { - queueName = queueService.resolve(serviceType, queueName); - return resolve(new ServiceQueue(serviceType, queueName), tenantId, entityId); + ServiceQueue serviceQueue = new ServiceQueue(serviceType, queueName); + ConcurrentMap queues = partitionTopicsMap.get(getSystemIsolatedTenantId(serviceType, tenantId)); + + if (!queues.containsKey(serviceQueue)) { + serviceQueue = new ServiceQueue(serviceType); + } + + return resolve(serviceQueue, tenantId, entityId); } private TopicPartitionInfo resolve(ServiceQueue serviceQueue, TenantId tenantId, EntityId entityId) { int hash = hashFunction.newHasher() .putLong(entityId.getId().getMostSignificantBits()) .putLong(entityId.getId().getLeastSignificantBits()).hash().asInt(); - Integer partitionSize = partitionSizes.get(serviceQueue); - int partition; - if (partitionSize != null) { - partition = Math.abs(hash % partitionSize); - } else { - //TODO: In 2.6/3.1 this should not happen because all Rule Engine Queues will be in the DB and we always know their partition sizes. - partition = 0; - } - boolean isolatedTenant = isIsolated(serviceQueue, tenantId); + + boolean isolatedTenant = isIsolated(serviceQueue.getType(), tenantId); + Integer partitionSize = partitionSizesMap.get(isolatedTenant ? tenantId : TenantId.SYS_TENANT_ID).get(serviceQueue); + int partition = Math.abs(hash % partitionSize); + TopicPartitionInfoKey cacheKey = new TopicPartitionInfoKey(serviceQueue, isolatedTenant ? tenantId : null, partition); return tpiCache.computeIfAbsent(cacheKey, key -> buildTopicPartitionInfo(serviceQueue, tenantId, partition)); } @@ -143,7 +203,7 @@ public class HashPartitionService implements PartitionService { ConcurrentMap> oldPartitions = myPartitions; TenantId myIsolatedOrSystemTenantId = getSystemOrIsolatedTenantId(currentService); myPartitions = new ConcurrentHashMap<>(); - partitionSizes.forEach((serviceQueue, size) -> { + partitionSizesMap.get(myIsolatedOrSystemTenantId).forEach((serviceQueue, size) -> { ServiceQueueKey myServiceQueueKey = new ServiceQueueKey(serviceQueue, myIsolatedOrSystemTenantId); for (int i = 0; i < size; i++) { ServiceInfo serviceInfo = resolveByPartitionIdx(queueServicesMap.get(myServiceQueueKey), i); @@ -154,6 +214,8 @@ public class HashPartitionService implements PartitionService { } }); + tpiCache.clear(); + oldPartitions.forEach((serviceQueueKey, partitions) -> { if (!myPartitions.containsKey(serviceQueueKey)) { log.info("[{}] NO MORE PARTITIONS FOR CURRENT KEY", serviceQueueKey); @@ -170,7 +232,6 @@ public class HashPartitionService implements PartitionService { applicationEventPublisher.publishEvent(new PartitionChangeEvent(this, serviceQueueKey, tpiList)); } }); - tpiCache.clear(); if (currentOtherServices == null) { currentOtherServices = new ArrayList<>(otherServices); @@ -196,21 +257,33 @@ public class HashPartitionService implements PartitionService { @Override public Set getAllServiceIds(ServiceType serviceType) { - Set result = new HashSet<>(); + return getAllServices(serviceType).stream().map(ServiceInfo::getServiceId).collect(Collectors.toSet()); + } + + @Override + public Set getAllServices(ServiceType serviceType) { + Set result = getOtherServices(serviceType); ServiceInfo current = serviceInfoProvider.getServiceInfo(); if (current.getServiceTypesList().contains(serviceType.name())) { - result.add(current.getServiceId()); + result.add(current); } + return result; + } + + @Override + public Set getOtherServices(ServiceType serviceType) { + Set result = new HashSet<>(); if (currentOtherServices != null) { for (ServiceInfo serviceInfo : currentOtherServices) { if (serviceInfo.getServiceTypesList().contains(serviceType.name())) { - result.add(serviceInfo.getServiceId()); + result.add(serviceInfo); } } } return result; } + @Override public int resolvePartitionIndex(UUID entityId, int partitions) { int hash = hashFunction.newHasher() @@ -231,10 +304,10 @@ public class HashPartitionService implements PartitionService { for (String serviceTypeStr : serviceInfo.getServiceTypesList()) { ServiceType serviceType = ServiceType.valueOf(serviceTypeStr.toUpperCase()); if (ServiceType.TB_RULE_ENGINE.equals(serviceType)) { - for (TransportProtos.QueueInfo queue : serviceInfo.getRuleEngineQueuesList()) { - ServiceQueueKey serviceQueueKey = new ServiceQueueKey(new ServiceQueue(serviceType, queue.getName()), getSystemOrIsolatedTenantId(serviceInfo)); - currentMap.computeIfAbsent(serviceQueueKey, key -> new ArrayList<>()).add(serviceInfo); - } +// for (TransportProtos.QueueInfo queue : serviceInfo.getRuleEngineQueuesList()) { +// ServiceQueueKey serviceQueueKey = new ServiceQueueKey(new ServiceQueue(serviceType, queue.getName()), getSystemOrIsolatedTenantId(serviceInfo)); +// currentMap.computeIfAbsent(serviceQueueKey, key -> new ArrayList<>()).add(serviceInfo); +// } } else { ServiceQueueKey serviceQueueKey = new ServiceQueueKey(new ServiceQueue(serviceType), getSystemOrIsolatedTenantId(serviceInfo)); currentMap.computeIfAbsent(serviceQueueKey, key -> new ArrayList<>()).add(serviceInfo); @@ -244,20 +317,18 @@ public class HashPartitionService implements PartitionService { return currentMap; } - private TopicPartitionInfo buildNotificationsTopicPartitionInfo(ServiceType serviceType, String serviceId) { - return new TopicPartitionInfo(serviceType.name().toLowerCase() + ".notifications." + serviceId, null, null, false); - } - private TopicPartitionInfo buildTopicPartitionInfo(ServiceQueueKey serviceQueueKey, int partition) { return buildTopicPartitionInfo(serviceQueueKey.getServiceQueue(), serviceQueueKey.getTenantId(), partition); } private TopicPartitionInfo buildTopicPartitionInfo(ServiceQueue serviceQueue, TenantId tenantId, int partition) { + boolean isolatedTenant = isIsolated(serviceQueue.getType(), tenantId); + TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder(); - tpi.topic(partitionTopics.get(serviceQueue)); + tpi.topic(partitionTopicsMap.get(isolatedTenant ? tenantId : TenantId.SYS_TENANT_ID).get(serviceQueue)); tpi.partition(partition); ServiceQueueKey myPartitionsSearchKey; - if (isIsolated(serviceQueue, tenantId)) { + if (isolatedTenant) { tpi.tenantId(tenantId); myPartitionsSearchKey = new ServiceQueueKey(serviceQueue, tenantId); } else { @@ -272,7 +343,7 @@ public class HashPartitionService implements PartitionService { return tpi.build(); } - private boolean isIsolated(ServiceQueue serviceQueue, TenantId tenantId) { + private boolean isIsolated(ServiceType serviceType, TenantId tenantId) { if (TenantId.SYS_TENANT_ID.equals(tenantId)) { return false; } @@ -289,7 +360,7 @@ public class HashPartitionService implements PartitionService { if (routingInfo == null) { throw new RuntimeException("Tenant not found!"); } - switch (serviceQueue.getType()) { + switch (serviceType) { case TB_CORE: return routingInfo.isIsolatedTbCore(); case TB_RULE_ENGINE: @@ -299,6 +370,10 @@ public class HashPartitionService implements PartitionService { } } + private TenantId getSystemIsolatedTenantId(ServiceType serviceType, TenantId tenantId) { + return isIsolated(serviceType, tenantId) ? tenantId : TenantId.SYS_TENANT_ID; + } + private void logServiceInfo(TransportProtos.ServiceInfo server) { TenantId tenantId = getSystemOrIsolatedTenantId(server); if (tenantId.isNullUid()) { @@ -317,12 +392,12 @@ public class HashPartitionService implements PartitionService { for (String serviceTypeStr : instance.getServiceTypesList()) { ServiceType serviceType = ServiceType.valueOf(serviceTypeStr.toUpperCase()); if (ServiceType.TB_RULE_ENGINE.equals(serviceType)) { - for (TransportProtos.QueueInfo queue : instance.getRuleEngineQueuesList()) { - ServiceQueueKey serviceQueueKey = new ServiceQueueKey(new ServiceQueue(serviceType, queue.getName()), tenantId); - partitionSizes.put(new ServiceQueue(ServiceType.TB_RULE_ENGINE, queue.getName()), queue.getPartitions()); - partitionTopics.put(new ServiceQueue(ServiceType.TB_RULE_ENGINE, queue.getName()), queue.getTopic()); - queueServiceList.computeIfAbsent(serviceQueueKey, key -> new ArrayList<>()).add(instance); - } + partitionTopicsMap.get(tenantId).forEach((serviceQueue, topic) -> { + if (serviceQueue.getType().equals(ServiceType.TB_RULE_ENGINE)) { + ServiceQueueKey serviceQueueKey = new ServiceQueueKey(serviceQueue, tenantId); + queueServiceList.computeIfAbsent(serviceQueueKey, key -> new ArrayList<>()).add(instance); + } + }); } else { ServiceQueueKey serviceQueueKey = new ServiceQueueKey(new ServiceQueue(serviceType), tenantId); queueServiceList.computeIfAbsent(serviceQueueKey, key -> new ArrayList<>()).add(instance); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java index c222b252a5..84065ec283 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java @@ -49,7 +49,15 @@ public interface PartitionService { */ Set getAllServiceIds(ServiceType serviceType); + Set getAllServices(ServiceType serviceType); + + Set getOtherServices(ServiceType serviceType); + int resolvePartitionIndex(UUID entityId, int partitions); int countTransportsByType(String type); + + void updateQueue(TransportProtos.QueueUpdateMsg queueUpdateMsg); + + void removeQueue(TransportProtos.QueueDeleteMsg queueDeleteMsg); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java index 2a691a2531..8e73c23c7e 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java @@ -25,6 +25,8 @@ public interface TbServiceInfoProvider { String getServiceId(); + String getServiceType(); + ServiceInfo getServiceInfo(); boolean isService(ServiceType serviceType); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java index e204025ec5..e62ee850fa 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerStatsService.java @@ -25,7 +25,9 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.common.TopicPartition; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; import org.thingsboard.common.util.ThingsBoardThreadFactory; @@ -55,7 +57,10 @@ public class TbKafkaConsumerStatsService { private final TbKafkaSettings kafkaSettings; private final TbKafkaConsumerStatisticConfig statsConfig; - private final PartitionService partitionService; + + @Lazy + @Autowired + private PartitionService partitionService; private AdminClient adminClient; private Consumer consumer; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java index 873719c832..660173c71d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java @@ -19,6 +19,7 @@ import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; 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.gen.js.JsInvokeProtos.RemoteJsRequest; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; @@ -45,7 +46,6 @@ import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate; import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate; @@ -124,7 +124,7 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbAwsSqsConsumerTemplate<>(ruleEngineAdmin, sqsSettings, configuration.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java index 9c223f08f5..9d9fd29078 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java @@ -19,6 +19,7 @@ import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; 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.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; @@ -38,7 +39,6 @@ import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate; import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate; @@ -112,7 +112,7 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbAwsSqsConsumerTemplate<>(ruleEngineAdmin, sqsSettings, configuration.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } 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 e3c81df1dd..53de7c1397 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 @@ -19,6 +19,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; 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.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; @@ -36,7 +37,6 @@ import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; @Slf4j @Component @@ -91,7 +91,7 @@ public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new InMemoryTbQueueConsumer<>(configuration.getTopic()); } 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 ce171497de..104c531e12 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 @@ -19,6 +19,7 @@ import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; 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.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; @@ -50,7 +51,6 @@ import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -156,7 +156,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { String queueName = configuration.getName(); TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); 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 645cbee906..12a51fc8fd 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 @@ -19,6 +19,7 @@ import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; 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.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; @@ -47,7 +48,6 @@ import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -158,7 +158,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { String queueName = configuration.getName(); TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java index 3388f0d0f3..aceb9d4f0d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java @@ -19,6 +19,7 @@ import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; 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.gen.js.JsInvokeProtos.RemoteJsRequest; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; @@ -50,7 +51,6 @@ import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -126,7 +126,7 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbPubSubConsumerTemplate<>(ruleEngineAdmin, pubSubSettings, configuration.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java index 8a3b88846f..bd9a360536 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java @@ -19,6 +19,7 @@ import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; 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.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; @@ -46,7 +47,6 @@ import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -116,7 +116,7 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbPubSubConsumerTemplate<>(ruleEngineAdmin, pubSubSettings, configuration.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java index a22f6811e3..2a724289bb 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java @@ -19,6 +19,7 @@ import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; 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.gen.js.JsInvokeProtos.RemoteJsRequest; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; @@ -50,7 +51,6 @@ import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -124,7 +124,7 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbRabbitMqConsumerTemplate<>(ruleEngineAdmin, rabbitMqSettings, configuration.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java index 21e152787d..55737d7753 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java @@ -19,6 +19,7 @@ import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; 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.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; @@ -46,7 +47,6 @@ import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -115,7 +115,7 @@ public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactor } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbRabbitMqConsumerTemplate<>(ruleEngineAdmin, rabbitMqSettings, configuration.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java index da219d9084..88bb0a4045 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java @@ -19,6 +19,7 @@ import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; 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.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; @@ -49,7 +50,6 @@ import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -123,7 +123,7 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbServiceBusConsumerTemplate<>(ruleEngineAdmin, serviceBusSettings, configuration.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java index a6ef991f57..622a3cc1fd 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java @@ -19,6 +19,7 @@ import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; 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.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; @@ -46,7 +47,6 @@ import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -115,7 +115,7 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbServiceBusConsumerTemplate<>(ruleEngineAdmin, serviceBusSettings, configuration.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } 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 df64b768ab..07148ee924 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 @@ -15,6 +15,7 @@ */ package org.thingsboard.server.queue.provider; +import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; @@ -26,7 +27,6 @@ import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; /** * Responsible for initialization of various Producers and Consumers used by TB Core Node. @@ -83,7 +83,7 @@ public interface TbRuleEngineQueueFactory extends TbUsageStatsClientQueueFactory * @param configuration */ //TODO 2.5 ybondarenko: make sure you use queueName to distinct consumers where necessary - TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration); + TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration); /** * Used to consume high priority messages by TB Core Service diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbApiUsageClient.java b/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbApiUsageClient.java index 6d8e3c5b0b..60867f314f 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbApiUsageClient.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbApiUsageClient.java @@ -17,7 +17,9 @@ package org.thingsboard.server.queue.usagestats; import lombok.Data; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.ApiUsageRecordKey; import org.thingsboard.server.common.data.EntityType; @@ -57,7 +59,9 @@ public class DefaultTbApiUsageClient implements TbApiUsageClient { private final EnumMap> stats = new EnumMap<>(ApiUsageRecordKey.class); - private final PartitionService partitionService; + @Lazy + @Autowired + private PartitionService partitionService; private final SchedulerComponent scheduler; private final TbQueueProducerProvider producerProvider; private TbQueueProducer> msgProducer; diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 3de2a6a7d5..f06a7f76ea 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -22,8 +22,10 @@ import com.google.gson.Gson; import com.google.gson.JsonObject; import com.google.protobuf.ByteString; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.annotation.Lazy; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardExecutors; @@ -152,18 +154,25 @@ public class DefaultTransportService implements TransportService { @Value("${transport.stats.enabled:false}") private boolean statsEnabled; + @Autowired + @Lazy + private TbApiUsageClient apiUsageClient; + @Autowired + @Lazy + private PartitionService partitionService; + private final Map statsMap = new LinkedHashMap<>(); private final Gson gson = new Gson(); private final TbTransportQueueFactory queueProvider; private final TbQueueProducerProvider producerProvider; - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbServiceInfoProvider serviceInfoProvider; private final StatsFactory statsFactory; private final TransportDeviceProfileCache deviceProfileCache; private final TransportTenantProfileCache tenantProfileCache; - private final TbApiUsageClient apiUsageClient; + private final TransportRateLimitService rateLimitService; private final DataDecodingEncodingService dataDecodingEncodingService; private final SchedulerComponent scheduler; @@ -191,23 +200,20 @@ public class DefaultTransportService implements TransportService { public DefaultTransportService(TbServiceInfoProvider serviceInfoProvider, TbTransportQueueFactory queueProvider, TbQueueProducerProvider producerProvider, - PartitionService partitionService, NotificationsTopicService notificationsTopicService, StatsFactory statsFactory, TransportDeviceProfileCache deviceProfileCache, TransportTenantProfileCache tenantProfileCache, - TbApiUsageClient apiUsageClient, TransportRateLimitService rateLimitService, + TransportRateLimitService rateLimitService, DataDecodingEncodingService dataDecodingEncodingService, SchedulerComponent scheduler, TransportResourceCache transportResourceCache, ApplicationEventPublisher eventPublisher) { this.serviceInfoProvider = serviceInfoProvider; this.queueProvider = queueProvider; this.producerProvider = producerProvider; - this.partitionService = partitionService; this.notificationsTopicService = notificationsTopicService; this.statsFactory = statsFactory; this.deviceProfileCache = deviceProfileCache; this.tenantProfileCache = tenantProfileCache; - this.apiUsageClient = apiUsageClient; this.rateLimitService = rateLimitService; this.dataDecodingEncodingService = dataDecodingEncodingService; this.scheduler = scheduler; @@ -983,6 +989,10 @@ public class DefaultTransportService implements TransportService { log.warn("ResourceDelete - [{}] [{}]", id, mdRez); transportCallbackExecutor.submit(() -> mdRez.getListener().onResourceDelete(msg)); }); + } else if (toSessionMsg.hasQueueUpdateMsg()) { + partitionService.updateQueue(toSessionMsg.getQueueUpdateMsg()); + } else if (toSessionMsg.hasQueueDeleteMsg()) { + partitionService.removeQueue(toSessionMsg.getQueueDeleteMsg()); } else { //TODO: should we notify the device actor about missed session? log.debug("[{}] Missing session.", sessionId); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/TransportQueueRoutingInfoService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/TransportQueueRoutingInfoService.java index bcdc07b5b0..537a86b1d6 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/TransportQueueRoutingInfoService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/TransportQueueRoutingInfoService.java @@ -36,13 +36,9 @@ import java.util.stream.Collectors; @ConditionalOnExpression("'${service.type:null}'=='tb-transport'") public class TransportQueueRoutingInfoService implements QueueRoutingInfoService { - private TransportService transportService; - @Lazy @Autowired - public void setTransportService(TransportService transportService) { - this.transportService = transportService; - } + private TransportService transportService; @Override public List getAllQueuesRoutingInfo() { diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java index 35d4dc2899..7d6412cbea 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java @@ -29,7 +29,6 @@ import com.squareup.wire.schema.internal.parser.ProtoParser; import com.squareup.wire.schema.internal.parser.TypeElement; import lombok.extern.slf4j.Slf4j; import org.eclipse.leshan.core.util.SecurityUtil; -import org.thingsboard.server.common.data.StringUtils; import org.hibernate.exception.ConstraintViolationException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cache.Cache; @@ -45,6 +44,7 @@ import org.thingsboard.server.common.data.DeviceProfileProvisionType; import org.thingsboard.server.common.data.DeviceProfileType; import org.thingsboard.server.common.data.DeviceTransportType; import org.thingsboard.server.common.data.OtaPackage; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.device.credentials.lwm2m.LwM2MSecurityMode; import org.thingsboard.server.common.data.device.profile.CoapDeviceProfileTransportConfiguration; @@ -61,28 +61,28 @@ import org.thingsboard.server.common.data.device.profile.MqttDeviceProfileTransp import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadConfiguration; import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration; import org.thingsboard.server.common.data.device.profile.lwm2m.bootstrap.AbstractLwM2MBootstrapServerCredential; -import org.thingsboard.server.common.data.device.profile.lwm2m.bootstrap.RPKLwM2MBootstrapServerCredential; import org.thingsboard.server.common.data.device.profile.lwm2m.bootstrap.LwM2MBootstrapServerCredential; +import org.thingsboard.server.common.data.device.profile.lwm2m.bootstrap.RPKLwM2MBootstrapServerCredential; import org.thingsboard.server.common.data.device.profile.lwm2m.bootstrap.X509LwM2MBootstrapServerCredential; -import org.thingsboard.server.common.data.ota.OtaPackageType; 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; import org.thingsboard.server.common.data.page.PageLink; +import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.msg.EncryptionUtil; -import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.dao.dashboard.DashboardService; import org.thingsboard.server.dao.entity.AbstractEntityService; import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.exception.DeviceCredentialsValidationException; import org.thingsboard.server.dao.ota.OtaPackageService; +import org.thingsboard.server.dao.queue.QueueService; import org.thingsboard.server.dao.rule.RuleChainService; 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.TenantDao; -import org.thingsboard.server.queue.TbQueueService; import java.util.Arrays; import java.util.Collections; @@ -115,7 +115,7 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D } @Autowired(required = false) - private TbQueueService queueService; + private QueueService queueService; @Autowired private DeviceProfileDao deviceProfileDao; @@ -386,8 +386,13 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D throw new DataValidationException("Another default device profile is present in scope of current tenant!"); } } - if (!StringUtils.isEmpty(deviceProfile.getDefaultQueueName()) && queueService != null){ - if(!queueService.getQueuesByServiceType(ServiceType.TB_RULE_ENGINE).contains(deviceProfile.getDefaultQueueName())){ + if (!StringUtils.isEmpty(deviceProfile.getDefaultQueueName()) && queueService != null) { + //TODO: Yevhen replace queue name to queue id + if (!queueService.findQueuesByTenantId(tenantId) + .stream() + .map(Queue::getName) + .collect(Collectors.toSet()) + .contains(deviceProfile.getDefaultQueueName())) { throw new DataValidationException("Device profile is referencing to non-existent queue!"); } } @@ -748,7 +753,7 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D case PSK: break; case RPK: - RPKLwM2MBootstrapServerCredential rpkServerCredentials = (RPKLwM2MBootstrapServerCredential) bootstrapServerConfig; + RPKLwM2MBootstrapServerCredential rpkServerCredentials = (RPKLwM2MBootstrapServerCredential) bootstrapServerConfig; server = rpkServerCredentials.isBootstrapServerIs() ? "Bootstrap Server" : "LwM2M Server"; if (StringUtils.isEmpty(rpkServerCredentials.getServerPublicKey())) { throw new DeviceCredentialsValidationException(server + " RPK public key must be specified!"); 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 7049d8ae1c..e2fd459189 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,6 +16,7 @@ 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; @@ -64,7 +65,6 @@ public class BaseQueueService extends AbstractEntityService implements QueueServ // private QueueStatsService queueStatsService; @Override - @Transactional public Queue saveQueue(Queue queue) { log.trace("Executing createOrUpdateQueue [{}]", queue); queueValidator.validate(queue, Queue::getTenantId); @@ -75,9 +75,6 @@ public class BaseQueueService extends AbstractEntityService implements QueueServ savedQueue = updateQueue(queue); } -// if (queueClusterService != null) { -// queueClusterService.onQueueChange(savedQueue, null); -// } return savedQueue; } @@ -88,6 +85,11 @@ public class BaseQueueService extends AbstractEntityService implements QueueServ tbQueueAdmin.createTopicIfNotExists(new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName()); } } + + if (queueClusterService != null) { + queueClusterService.onQueueChange(createdQueue, null); + } + return createdQueue; } @@ -98,32 +100,39 @@ public class BaseQueueService extends AbstractEntityService implements QueueServ int oldPartitions = oldQueue.getPartitions(); int currentPartitions = queue.getPartitions(); - //TODO: remove if partitions can't be deleted. if (currentPartitions != oldPartitions && tbQueueAdmin != null) { - queueClusterService.onQueueDelete(queue, 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, null); + } } else { log.info("Removed [{}] partitions from [{}] queue", oldPartitions - currentPartitions, queue.getName()); + if (queueClusterService != null) { + queueClusterService.onQueueChange(updatedQueue, null); + } + 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, null); } return updatedQueue; } @Override - @Transactional public void deleteQueue(TenantId tenantId, QueueId queueId) { log.trace("Executing deleteQueue, queueId: [{}]", queueId); Queue queue = findQueueById(tenantId, queueId); if (queueClusterService != null) { queueClusterService.onQueueDelete(queue, null); + await(); } // queueStatsService.deleteQueueStatsByQueueId(tenantId, queueId); boolean result = queueDao.removeById(tenantId, queueId.getId()); @@ -140,6 +149,11 @@ public class BaseQueueService extends AbstractEntityService implements QueueServ } } + @SneakyThrows + private void await() { + Thread.sleep(3000); + } + @Override public List findQueuesByTenantId(TenantId tenantId) { log.trace("Executing findQueues, tenantId: [{}]", tenantId); diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index ed69fd9b72..82fbaeed74 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -147,6 +147,8 @@ services: - ./tb-transports/mqtt/log:/var/log/tb-mqtt-transport depends_on: - zookeeper + - tb-core1 + - tb-core2 tb-mqtt-transport2: restart: always image: "${DOCKER_REPO}/${MQTT_TRANSPORT_DOCKER_NAME}:${TB_VERSION}" @@ -161,6 +163,8 @@ services: - ./tb-transports/mqtt/log:/var/log/tb-mqtt-transport depends_on: - zookeeper + - tb-core1 + - tb-core2 tb-http-transport1: restart: always image: "${DOCKER_REPO}/${HTTP_TRANSPORT_DOCKER_NAME}:${TB_VERSION}" @@ -175,6 +179,8 @@ services: - ./tb-transports/http/log:/var/log/tb-http-transport depends_on: - zookeeper + - tb-core1 + - tb-core2 tb-http-transport2: restart: always image: "${DOCKER_REPO}/${HTTP_TRANSPORT_DOCKER_NAME}:${TB_VERSION}" @@ -189,6 +195,8 @@ services: - ./tb-transports/http/log:/var/log/tb-http-transport depends_on: - zookeeper + - tb-core1 + - tb-core2 tb-coap-transport: restart: always image: "${DOCKER_REPO}/${COAP_TRANSPORT_DOCKER_NAME}:${TB_VERSION}" @@ -203,6 +211,8 @@ services: - ./tb-transports/coap/log:/var/log/tb-coap-transport depends_on: - zookeeper + - tb-core1 + - tb-core2 tb-lwm2m-transport: restart: always image: "${DOCKER_REPO}/${LWM2M_TRANSPORT_DOCKER_NAME}:${TB_VERSION}" @@ -217,6 +227,8 @@ services: - ./tb-transports/lwm2m/log:/var/log/tb-lwm2m-transport depends_on: - zookeeper + - tb-core1 + - tb-core2 tb-snmp-transport: restart: always image: "${DOCKER_REPO}/${SNMP_TRANSPORT_DOCKER_NAME}:${TB_VERSION}" @@ -229,6 +241,8 @@ services: - ./tb-transports/snmp/log:/var/log/tb-snmp-transport depends_on: - zookeeper + - tb-core1 + - tb-core2 tb-web-ui1: restart: always image: "${DOCKER_REPO}/${WEB_UI_DOCKER_NAME}:${TB_VERSION}" diff --git a/ui-ngx/src/app/shared/components/queue/queue-type-list.component.html b/ui-ngx/src/app/shared/components/queue/queue-type-list.component.html index 6abd58dda9..29de0f2f12 100644 --- a/ui-ngx/src/app/shared/components/queue/queue-type-list.component.html +++ b/ui-ngx/src/app/shared/components/queue/queue-type-list.component.html @@ -16,7 +16,7 @@ --> - {{ 'queue.name' | translate }} + {{ 'queue.queue-name' | translate }}