From fbec7aab25b670b89a97b72fa252fa625247f039 Mon Sep 17 00:00:00 2001 From: dashevchenko Date: Fri, 17 Nov 2023 12:18:03 +0200 Subject: [PATCH] moved pubsub executors to pubsubSettings --- .../PubSubRuleNodeExecutorProvider.java | 9 +++- .../src/main/resources/thingsboard.yml | 5 +- .../provider/PubSubMonolithQueueFactory.java | 50 ++++++++----------- .../provider/PubSubTbCoreQueueFactory.java | 38 ++++++-------- .../PubSubTbRuleEngineQueueFactory.java | 28 +++++------ .../PubSubTbVersionControlQueueFactory.java | 13 ++--- .../provider/PubSubTransportQueueFactory.java | 20 +++----- .../pubsub/TbPubSubConsumerTemplate.java | 4 +- .../pubsub/TbPubSubProducerTemplate.java | 10 +--- .../pubsub/TbPubSubQueueExecutorProvider.java | 2 +- .../server/queue/pubsub/TbPubSubSettings.java | 2 + .../src/main/resources/tb-vc-executor.yml | 2 +- .../rule/engine/gcp/pubsub/TbPubSubNode.java | 6 +-- .../src/main/resources/tb-coap-transport.yml | 2 +- .../src/main/resources/tb-http-transport.yml | 2 +- .../src/main/resources/tb-lwm2m-transport.yml | 2 +- .../src/main/resources/tb-mqtt-transport.yml | 2 +- .../src/main/resources/tb-snmp-transport.yml | 2 +- 18 files changed, 88 insertions(+), 111 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/executors/PubSubRuleNodeExecutorProvider.java b/application/src/main/java/org/thingsboard/server/service/executors/PubSubRuleNodeExecutorProvider.java index f730949f45..520b8f855c 100644 --- a/application/src/main/java/org/thingsboard/server/service/executors/PubSubRuleNodeExecutorProvider.java +++ b/application/src/main/java/org/thingsboard/server/service/executors/PubSubRuleNodeExecutorProvider.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.service.executors; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import org.thingsboard.common.util.ExecutorProvider; import org.thingsboard.common.util.ThingsBoardThreadFactory; @@ -27,6 +28,9 @@ import java.util.concurrent.ScheduledExecutorService; @Component public class PubSubRuleNodeExecutorProvider implements ExecutorProvider { + @Value("${service.rule_engine.pubsub.executor_thread_pool_size}") + private Integer threadPoolSize; + /** * Refers to com.google.cloud.pubsub.v1.Publisher default executor configuration */ @@ -35,7 +39,10 @@ public class PubSubRuleNodeExecutorProvider implements ExecutorProvider { @PostConstruct public void init() { - executor = Executors.newScheduledThreadPool(THREADS_PER_CPU * Runtime.getRuntime().availableProcessors(), ThingsBoardThreadFactory.forName("pubsub-rule-nodes"));; + if (threadPoolSize == null) { + threadPoolSize = THREADS_PER_CPU * Runtime.getRuntime().availableProcessors(); + } + executor = Executors.newScheduledThreadPool(threadPoolSize, ThingsBoardThreadFactory.forName("pubsub-rule-nodes")); } @Override diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 802756c105..304ad96a34 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1411,7 +1411,7 @@ queue: max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" # Number of messages per consumer max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}" - # Number of threads of pubsub executor provider + # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors) executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:}" queue-properties: # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again @@ -1641,6 +1641,9 @@ service: # Comma-separated list of tenant profile ids assigned to this Rule Engine. # This Rule Engine will only be responsible for tenants with these profiles (in case 'isolation' option is enabled in the profile). assigned_tenant_profiles: "${TB_RULE_ENGINE_ASSIGNED_TENANT_PROFILES:}" + pubsub: + # Thread pool size for pubsub rule node executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors) + executor_thread_pool_size: "${TB_RULE_ENGINE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:}" # Metrics parameters metrics: 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 8ecd6564c8..554d452556 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 @@ -44,7 +44,6 @@ import org.thingsboard.server.queue.discovery.TopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.pubsub.TbPubSubAdmin; import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate; -import org.thingsboard.server.queue.pubsub.TbPubSubQueueExecutorProvider; import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; import org.thingsboard.server.queue.pubsub.TbPubSubSettings; import org.thingsboard.server.queue.pubsub.TbPubSubSubscriptionSettings; @@ -78,7 +77,6 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng private final TbQueueAdmin transportApiAdmin; private final TbQueueAdmin notificationAdmin; private final TbQueueAdmin vcAdmin; - private final TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider; public PubSubMonolithQueueFactory(TbPubSubSettings pubSubSettings, TbQueueCoreSettings coreSettings, @@ -89,8 +87,7 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng TbServiceInfoProvider serviceInfoProvider, TbPubSubSubscriptionSettings pubSubSubscriptionSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings, - TbQueueVersionControlSettings vcSettings, - TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider) { + TbQueueVersionControlSettings vcSettings) { this.pubSubSettings = pubSubSettings; this.coreSettings = coreSettings; this.ruleEngineSettings = ruleEngineSettings; @@ -99,7 +96,6 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng this.topicService = topicService; this.serviceInfoProvider = serviceInfoProvider; this.vcSettings = vcSettings; - this.tbPubSubQueueExecutorProvider = tbPubSubQueueExecutorProvider; this.coreAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getCoreSettings()); this.ruleEngineAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getRuleEngineSettings()); @@ -113,91 +109,85 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, transportNotificationSettings.getNotificationsTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, transportNotificationSettings.getNotificationsTopic()); } @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbPubSubProducerTemplate<>(ruleEngineAdmin, pubSubSettings, ruleEngineSettings.getTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(ruleEngineAdmin, pubSubSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, ruleEngineSettings.getTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createTbCoreMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); } @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, coreSettings.getTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, coreSettings.getTopic()); } @Override public TbQueueConsumer> createToVersionControlMsgConsumer() { return new TbPubSubConsumerTemplate<>(vcAdmin, pubSubSettings, vcSettings.getTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToVersionControlServiceMsg.parseFrom(msg.getData()), msg.getHeaders()), - tbPubSubQueueExecutorProvider + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToVersionControlServiceMsg.parseFrom(msg.getData()), msg.getHeaders()) ); } @Override public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbPubSubConsumerTemplate<>(ruleEngineAdmin, pubSubSettings, configuration.getTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders()), - tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbPubSubConsumerTemplate<>(notificationAdmin, pubSubSettings, topicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()), - tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToCoreMsgConsumer() { return new TbPubSubConsumerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders()), - tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbPubSubConsumerTemplate<>(notificationAdmin, pubSubSettings, topicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()), - tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createTransportApiRequestConsumer() { return new TbPubSubConsumerTemplate<>(transportApiAdmin, pubSubSettings, transportApiSettings.getRequestsTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders()), - tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueProducer> createTransportApiResponseProducer() { - return new TbPubSubProducerTemplate<>(transportApiAdmin, pubSubSettings, transportApiSettings.getResponsesTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(transportApiAdmin, pubSubSettings, transportApiSettings.getResponsesTopic()); } @Override @Bean public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { - TbQueueProducer> producer = new TbPubSubProducerTemplate<>(jsExecutorAdmin, pubSubSettings, jsInvokeSettings.getRequestTopic(), tbPubSubQueueExecutorProvider); + TbQueueProducer> producer = new TbPubSubProducerTemplate<>(jsExecutorAdmin, pubSubSettings, jsInvokeSettings.getRequestTopic()); TbQueueConsumer> consumer = new TbPubSubConsumerTemplate<>(jsExecutorAdmin, pubSubSettings, jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId(), msg -> { RemoteJsResponse.Builder builder = RemoteJsResponse.newBuilder(); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); return new TbProtoQueueMsg<>(msg.getKey(), builder.build(), msg.getHeaders()); - }, tbPubSubQueueExecutorProvider); + }); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> builder = DefaultTbQueueRequestTemplate.builder(); @@ -213,28 +203,28 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @Override public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { return new TbPubSubConsumerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders()), tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToOtaPackageStateServiceMsgConsumer() { return new TbPubSubConsumerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getOtaPackageTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToOtaPackageStateServiceMsg.parseFrom(msg.getData()), msg.getHeaders()), tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToOtaPackageStateServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueProducer> createToOtaPackageStateServiceMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getOtaPackageTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getOtaPackageTopic()); } @Override public TbQueueProducer> createToUsageStatsServiceMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic()); } @Override public TbQueueProducer> createVersionControlMsgProducer() { - return new TbPubSubProducerTemplate<>(vcAdmin, pubSubSettings, vcSettings.getTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(vcAdmin, pubSubSettings, vcSettings.getTopic()); } @PreDestroy diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java index 5443339d2a..3b45992e25 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java @@ -42,7 +42,6 @@ import org.thingsboard.server.queue.discovery.TopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.pubsub.TbPubSubAdmin; import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate; -import org.thingsboard.server.queue.pubsub.TbPubSubQueueExecutorProvider; import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; import org.thingsboard.server.queue.pubsub.TbPubSubSettings; import org.thingsboard.server.queue.pubsub.TbPubSubSubscriptionSettings; @@ -73,7 +72,6 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueAdmin transportApiAdmin; private final TbQueueAdmin notificationAdmin; private final TbQueueAdmin ruleEngineAdmin; - private final TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider; public PubSubTbCoreQueueFactory(TbPubSubSettings pubSubSettings, TbQueueCoreSettings coreSettings, @@ -83,8 +81,7 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueTransportNotificationSettings transportNotificationSettings, TbQueueRuleEngineSettings ruleEngineSettings, - TbPubSubSubscriptionSettings pubSubSubscriptionSettings, - TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider) { + TbPubSubSubscriptionSettings pubSubSubscriptionSettings) { this.pubSubSettings = pubSubSettings; this.coreSettings = coreSettings; this.transportApiSettings = transportApiSettings; @@ -93,7 +90,6 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { this.jsInvokeSettings = jsInvokeSettings; this.transportNotificationSettings = transportNotificationSettings; this.ruleEngineSettings = ruleEngineSettings; - this.tbPubSubQueueExecutorProvider = tbPubSubQueueExecutorProvider; this.coreAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getCoreSettings()); this.jsExecutorAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getJsExecutorSettings()); @@ -104,64 +100,64 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, transportNotificationSettings.getNotificationsTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, transportNotificationSettings.getNotificationsTopic()); } @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); } @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, ruleEngineSettings.getTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createTbCoreMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); } @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, coreSettings.getTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, coreSettings.getTopic()); } @Override public TbQueueConsumer> createToCoreMsgConsumer() { return new TbPubSubConsumerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders()), tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbPubSubConsumerTemplate<>(notificationAdmin, pubSubSettings, topicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()), tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createTransportApiRequestConsumer() { return new TbPubSubConsumerTemplate<>(transportApiAdmin, pubSubSettings, transportApiSettings.getRequestsTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders()), tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueProducer> createTransportApiResponseProducer() { - return new TbPubSubProducerTemplate<>(transportApiAdmin, pubSubSettings, transportApiSettings.getResponsesTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(transportApiAdmin, pubSubSettings, transportApiSettings.getResponsesTopic()); } @Override @Bean public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { - TbQueueProducer> producer = new TbPubSubProducerTemplate<>(jsExecutorAdmin, pubSubSettings, jsInvokeSettings.getRequestTopic(), tbPubSubQueueExecutorProvider); + TbQueueProducer> producer = new TbPubSubProducerTemplate<>(jsExecutorAdmin, pubSubSettings, jsInvokeSettings.getRequestTopic()); TbQueueConsumer> consumer = new TbPubSubConsumerTemplate<>(jsExecutorAdmin, pubSubSettings, jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId(), msg -> { JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); return new TbProtoQueueMsg<>(msg.getKey(), builder.build(), msg.getHeaders()); - }, tbPubSubQueueExecutorProvider); + }); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> builder = DefaultTbQueueRequestTemplate.builder(); @@ -177,25 +173,23 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { return new TbPubSubConsumerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders()), - tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToOtaPackageStateServiceMsgConsumer() { return new TbPubSubConsumerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getOtaPackageTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToOtaPackageStateServiceMsg.parseFrom(msg.getData()), msg.getHeaders()), - tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToOtaPackageStateServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueProducer> createToOtaPackageStateServiceMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getOtaPackageTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getOtaPackageTopic()); } @Override public TbQueueProducer> createToUsageStatsServiceMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic(), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic()); } @Override 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 b0fde82e7d..a60f77852f 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 @@ -40,7 +40,6 @@ import org.thingsboard.server.queue.discovery.TopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.pubsub.TbPubSubAdmin; import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate; -import org.thingsboard.server.queue.pubsub.TbPubSubQueueExecutorProvider; import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; import org.thingsboard.server.queue.pubsub.TbPubSubSettings; import org.thingsboard.server.queue.pubsub.TbPubSubSubscriptionSettings; @@ -63,7 +62,6 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; - private final TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -77,8 +75,7 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory TbServiceInfoProvider serviceInfoProvider, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueTransportNotificationSettings transportNotificationSettings, - TbPubSubSubscriptionSettings pubSubSubscriptionSettings, - TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider) { + TbPubSubSubscriptionSettings pubSubSubscriptionSettings) { this.pubSubSettings = pubSubSettings; this.coreSettings = coreSettings; this.ruleEngineSettings = ruleEngineSettings; @@ -86,7 +83,6 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory this.serviceInfoProvider = serviceInfoProvider; this.jsInvokeSettings = jsInvokeSettings; this.transportNotificationSettings = transportNotificationSettings; - this.tbPubSubQueueExecutorProvider = tbPubSubQueueExecutorProvider; this.coreAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getCoreSettings()); this.ruleEngineAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getRuleEngineSettings()); @@ -96,53 +92,53 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, topicService.buildTopicName(transportNotificationSettings.getNotificationsTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, topicService.buildTopicName(transportNotificationSettings.getNotificationsTopic())); } @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbPubSubProducerTemplate<>(ruleEngineAdmin, pubSubSettings, topicService.buildTopicName(ruleEngineSettings.getTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(ruleEngineAdmin, pubSubSettings, topicService.buildTopicName(ruleEngineSettings.getTopic())); } @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, topicService.buildTopicName(ruleEngineSettings.getTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, topicService.buildTopicName(ruleEngineSettings.getTopic())); } @Override public TbQueueProducer> createTbCoreMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getTopic())); } @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getTopic())); } @Override public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbPubSubConsumerTemplate<>(ruleEngineAdmin, pubSubSettings, topicService.buildTopicName(configuration.getTopic()), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders()), tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbPubSubConsumerTemplate<>(notificationAdmin, pubSubSettings, topicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()), tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override @Bean public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { - TbQueueProducer> producer = new TbPubSubProducerTemplate<>(jsExecutorAdmin, pubSubSettings, jsInvokeSettings.getRequestTopic(), tbPubSubQueueExecutorProvider); + TbQueueProducer> producer = new TbPubSubProducerTemplate<>(jsExecutorAdmin, pubSubSettings, jsInvokeSettings.getRequestTopic()); TbQueueConsumer> consumer = new TbPubSubConsumerTemplate<>(jsExecutorAdmin, pubSubSettings, jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId(), msg -> { JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); return new TbProtoQueueMsg<>(msg.getKey(), builder.build(), msg.getHeaders()); - }, tbPubSubQueueExecutorProvider); + }); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> builder = DefaultTbQueueRequestTemplate.builder(); @@ -157,12 +153,12 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory @Override public TbQueueProducer> createToUsageStatsServiceMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getUsageStatsTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getUsageStatsTopic())); } @Override public TbQueueProducer> createToOtaPackageStateServiceMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getOtaPackageTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getOtaPackageTopic())); } @PreDestroy diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbVersionControlQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbVersionControlQueueFactory.java index 29207ed429..8154d04d6d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbVersionControlQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbVersionControlQueueFactory.java @@ -26,7 +26,6 @@ import org.thingsboard.server.queue.discovery.TopicService; import org.thingsboard.server.queue.pubsub.TbPubSubAdmin; import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate; import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; -import org.thingsboard.server.queue.pubsub.TbPubSubQueueExecutorProvider; import org.thingsboard.server.queue.pubsub.TbPubSubSettings; import org.thingsboard.server.queue.pubsub.TbPubSubSubscriptionSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; @@ -42,7 +41,6 @@ public class PubSubTbVersionControlQueueFactory implements TbVersionControlQueue private final TbQueueCoreSettings coreSettings; private final TbQueueVersionControlSettings vcSettings; private final TopicService topicService; - private final TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin notificationAdmin; @@ -52,14 +50,12 @@ public class PubSubTbVersionControlQueueFactory implements TbVersionControlQueue TbQueueCoreSettings coreSettings, TbQueueVersionControlSettings vcSettings, TbPubSubSubscriptionSettings pubSubSubscriptionSettings, - TopicService topicService, - TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider + TopicService topicService ) { this.pubSubSettings = pubSubSettings; this.coreSettings = coreSettings; this.vcSettings = vcSettings; this.topicService = topicService; - this.tbPubSubQueueExecutorProvider = tbPubSubQueueExecutorProvider; this.coreAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getCoreSettings()); this.notificationAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getNotificationsSettings()); @@ -68,19 +64,18 @@ public class PubSubTbVersionControlQueueFactory implements TbVersionControlQueue @Override public TbQueueProducer> createToUsageStatsServiceMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getUsageStatsTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getUsageStatsTopic())); } @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getTopic())); } @Override public TbQueueConsumer> createToVersionControlMsgConsumer() { return new TbPubSubConsumerTemplate<>(vcAdmin, pubSubSettings, topicService.buildTopicName(vcSettings.getTopic()), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToVersionControlServiceMsg.parseFrom(msg.getData()), msg.getHeaders()), - tbPubSubQueueExecutorProvider + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToVersionControlServiceMsg.parseFrom(msg.getData()), msg.getHeaders()) ); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTransportQueueFactory.java index 94a454dd5b..edcad873c4 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTransportQueueFactory.java @@ -37,7 +37,6 @@ import org.thingsboard.server.queue.discovery.TopicService; import org.thingsboard.server.queue.pubsub.TbPubSubAdmin; import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate; import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; -import org.thingsboard.server.queue.pubsub.TbPubSubQueueExecutorProvider; import org.thingsboard.server.queue.pubsub.TbPubSubSettings; import org.thingsboard.server.queue.pubsub.TbPubSubSubscriptionSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; @@ -59,7 +58,6 @@ public class PubSubTransportQueueFactory implements TbTransportQueueFactory { private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TopicService topicService; - private final TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -73,8 +71,7 @@ public class PubSubTransportQueueFactory implements TbTransportQueueFactory { TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, TbPubSubSubscriptionSettings pubSubSubscriptionSettings, - TopicService topicService, - TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider) { + TopicService topicService) { this.pubSubSettings = pubSubSettings; this.serviceInfoProvider = serviceInfoProvider; this.coreSettings = coreSettings; @@ -82,7 +79,6 @@ public class PubSubTransportQueueFactory implements TbTransportQueueFactory { this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; this.topicService = topicService; - this.tbPubSubQueueExecutorProvider = tbPubSubQueueExecutorProvider; this.coreAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getCoreSettings()); this.ruleEngineAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getRuleEngineSettings()); @@ -92,10 +88,10 @@ public class PubSubTransportQueueFactory implements TbTransportQueueFactory { @Override public TbQueueRequestTemplate, TbProtoQueueMsg> createTransportApiRequestTemplate() { - TbQueueProducer> producer = new TbPubSubProducerTemplate<>(transportApiAdmin, pubSubSettings, topicService.buildTopicName(transportApiSettings.getRequestsTopic()), tbPubSubQueueExecutorProvider); + TbQueueProducer> producer = new TbPubSubProducerTemplate<>(transportApiAdmin, pubSubSettings, topicService.buildTopicName(transportApiSettings.getRequestsTopic())); TbQueueConsumer> consumer = new TbPubSubConsumerTemplate<>(transportApiAdmin, pubSubSettings, topicService.buildTopicName(transportApiSettings.getResponsesTopic() + "." + serviceInfoProvider.getServiceId()), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders()), tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> templateBuilder = DefaultTbQueueRequestTemplate.builder(); @@ -110,29 +106,29 @@ public class PubSubTransportQueueFactory implements TbTransportQueueFactory { @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbPubSubProducerTemplate<>(ruleEngineAdmin, pubSubSettings, topicService.buildTopicName(ruleEngineSettings.getTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(ruleEngineAdmin, pubSubSettings, topicService.buildTopicName(ruleEngineSettings.getTopic())); } @Override public TbQueueProducer> createTbCoreMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getTopic())); } @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getTopic())); } @Override public TbQueueConsumer> createTransportNotificationsConsumer() { return new TbPubSubConsumerTemplate<>(notificationAdmin, pubSubSettings, topicService.buildTopicName(transportNotificationSettings.getNotificationsTopic() + "." + serviceInfoProvider.getServiceId()), - msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders()), tbPubSubQueueExecutorProvider); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueProducer> createToUsageStatsServiceMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getUsageStatsTopic()), tbPubSubQueueExecutorProvider); + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, topicService.buildTopicName(coreSettings.getUsageStatsTopic())); } @PreDestroy diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubConsumerTemplate.java index fd01696873..1135053a66 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubConsumerTemplate.java @@ -63,7 +63,7 @@ public class TbPubSubConsumerTemplate extends AbstractPara private final SubscriberStub subscriber; private volatile int messagesPerTopic; - public TbPubSubConsumerTemplate(TbQueueAdmin admin, TbPubSubSettings pubSubSettings, String topic, TbQueueMsgDecoder decoder, TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider) { + public TbPubSubConsumerTemplate(TbQueueAdmin admin, TbPubSubSettings pubSubSettings, String topic, TbQueueMsgDecoder decoder) { super(topic); this.admin = admin; this.pubSubSettings = pubSubSettings; @@ -77,7 +77,7 @@ public class TbPubSubConsumerTemplate extends AbstractPara SubscriberStubSettings.defaultGrpcTransportProviderBuilder() .setMaxInboundMessageSize(pubSubSettings.getMaxMsgSize()) .build()) - .setExecutorProvider(FixedExecutorProvider.create(tbPubSubQueueExecutorProvider.getExecutor())) + .setExecutorProvider(FixedExecutorProvider.create(pubSubSettings.getTbPubSubQueueExecutorProvider().getExecutor())) .build(); this.subscriber = GrpcSubscriberStub.create(subscriberStubSettings); } catch (IOException e) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubProducerTemplate.java index 48a104f20a..bf297d8af4 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubProducerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubProducerTemplate.java @@ -25,8 +25,6 @@ import com.google.protobuf.ByteString; import com.google.pubsub.v1.ProjectTopicName; import com.google.pubsub.v1.PubsubMessage; import lombok.extern.slf4j.Slf4j; -import org.thingsboard.common.util.ExecutorProvider; -import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueCallback; @@ -39,7 +37,6 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @Slf4j @@ -54,14 +51,11 @@ public class TbPubSubProducerTemplate implements TbQueuePr private final Map publisherMap = new ConcurrentHashMap<>(); private final ExecutorService pubExecutor = Executors.newCachedThreadPool(); - private final FixedExecutorProvider fixedExecutorProvider; - public TbPubSubProducerTemplate(TbQueueAdmin admin, TbPubSubSettings pubSubSettings, String defaultTopic, ExecutorProvider executorProvider) { + public TbPubSubProducerTemplate(TbQueueAdmin admin, TbPubSubSettings pubSubSettings, String defaultTopic) { this.defaultTopic = defaultTopic; this.admin = admin; this.pubSubSettings = pubSubSettings; - - fixedExecutorProvider = FixedExecutorProvider.create(executorProvider.getExecutor()); } @Override @@ -129,7 +123,7 @@ public class TbPubSubProducerTemplate implements TbQueuePr ProjectTopicName topicName = ProjectTopicName.of(pubSubSettings.getProjectId(), topic); Publisher publisher = Publisher.newBuilder(topicName) .setCredentialsProvider(pubSubSettings.getCredentialsProvider()) - .setExecutorProvider(fixedExecutorProvider) + .setExecutorProvider(FixedExecutorProvider.create(pubSubSettings.getTbPubSubQueueExecutorProvider().getExecutor())) .build(); publisherMap.put(topic, publisher); return publisher; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubQueueExecutorProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubQueueExecutorProvider.java index 4de75faa56..5e5f19651c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubQueueExecutorProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubQueueExecutorProvider.java @@ -43,7 +43,7 @@ public class TbPubSubQueueExecutorProvider implements ExecutorProvider { if (threadPoolSize == null) { threadPoolSize = THREADS_PER_CPU * Runtime.getRuntime().availableProcessors(); } - executor = Executors.newScheduledThreadPool(threadPoolSize, ThingsBoardThreadFactory.forName("pubsub-queue-executor"));; + executor = Executors.newScheduledThreadPool(threadPoolSize, ThingsBoardThreadFactory.forName("pubsub-queue-executor")); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubSettings.java index fe9d85c306..530453f26b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubSettings.java @@ -46,6 +46,8 @@ public class TbPubSubSettings { @Value("${queue.pubsub.max_messages}") private int maxMessages; + private final TbPubSubQueueExecutorProvider tbPubSubQueueExecutorProvider; + private CredentialsProvider credentialsProvider; @PostConstruct diff --git a/msa/vc-executor/src/main/resources/tb-vc-executor.yml b/msa/vc-executor/src/main/resources/tb-vc-executor.yml index 9fc9b884f0..62957debcf 100644 --- a/msa/vc-executor/src/main/resources/tb-vc-executor.yml +++ b/msa/vc-executor/src/main/resources/tb-vc-executor.yml @@ -172,7 +172,7 @@ queue: max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" #in bytes # Number of messages per consumer max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}" - # Number of threads of pubsub executor provider + # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors) executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:}" queue-properties: # Pub/Sub properties for Core subscribers, messages which will commit after ackDeadlineInSec period can be consumed again diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/gcp/pubsub/TbPubSubNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/gcp/pubsub/TbPubSubNode.java index 39ab7bf525..3067947445 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/gcp/pubsub/TbPubSubNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/gcp/pubsub/TbPubSubNode.java @@ -69,7 +69,7 @@ public class TbPubSubNode extends TbAbstractExternalNode { super.init(ctx); this.config = TbNodeUtils.convert(configuration, TbPubSubNodeConfiguration.class); try { - this.pubSubClient = initPubSubClient(FixedExecutorProvider.create(ctx.getPubSubRuleNodeExecutorProvider().getExecutor())); + this.pubSubClient = initPubSubClient(ctx); } catch (Exception e) { throw new TbNodeException(e); } @@ -129,7 +129,7 @@ public class TbPubSubNode extends TbAbstractExternalNode { return TbMsg.transformMsgMetadata(origMsg, metaData); } - private Publisher initPubSubClient(FixedExecutorProvider fixedExecutorProvider) throws IOException { + private Publisher initPubSubClient(TbContext ctx) throws IOException { ProjectTopicName topicName = ProjectTopicName.of(config.getProjectId(), config.getTopicName()); ServiceAccountCredentials credentials = ServiceAccountCredentials.fromStream( @@ -149,7 +149,7 @@ public class TbPubSubNode extends TbAbstractExternalNode { return Publisher.newBuilder(topicName) .setCredentialsProvider(credProvider) .setRetrySettings(retrySettings) - .setExecutorProvider(fixedExecutorProvider) + .setExecutorProvider(FixedExecutorProvider.create(ctx.getPubSubRuleNodeExecutorProvider().getExecutor())) .build(); } } diff --git a/transport/coap/src/main/resources/tb-coap-transport.yml b/transport/coap/src/main/resources/tb-coap-transport.yml index 27e5c8b273..c376d6a176 100644 --- a/transport/coap/src/main/resources/tb-coap-transport.yml +++ b/transport/coap/src/main/resources/tb-coap-transport.yml @@ -295,7 +295,7 @@ queue: max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" # Number of messages per consumer max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}" - # Number of threads of pubsub executor provider + # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors) executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:}" queue-properties: # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again diff --git a/transport/http/src/main/resources/tb-http-transport.yml b/transport/http/src/main/resources/tb-http-transport.yml index b8070f6575..f3fd100fd7 100644 --- a/transport/http/src/main/resources/tb-http-transport.yml +++ b/transport/http/src/main/resources/tb-http-transport.yml @@ -278,7 +278,7 @@ queue: max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" # Number of messages per a consumer max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}" - # Number of threads of pubsub executor provider + # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors) executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:}" queue-properties: # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consume again diff --git a/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml b/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml index b1d40c720f..968385b5e2 100644 --- a/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml +++ b/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml @@ -374,7 +374,7 @@ queue: max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" # Number of messages per consumer max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}" - # Number of threads of pubsub executor provider + # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors) executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:}" queue-properties: # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index 4d130eecd9..1e97c59253 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -311,7 +311,7 @@ queue: max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" # Number of messages per consumer max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}" - # Number of threads of pubsub executor provider + # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors) executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:}" queue-properties: # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again diff --git a/transport/snmp/src/main/resources/tb-snmp-transport.yml b/transport/snmp/src/main/resources/tb-snmp-transport.yml index 2b89065a55..8d3f842325 100644 --- a/transport/snmp/src/main/resources/tb-snmp-transport.yml +++ b/transport/snmp/src/main/resources/tb-snmp-transport.yml @@ -264,7 +264,7 @@ queue: max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" # Number of messages per consumer max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}" - # Number of threads of pubsub executor provider + # Thread pool size for pubsub queue executor provider. If not set - default pubsub executor provider value will be used (5 * number of available processors) executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_POOL_SIZE:}" queue-properties: # Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again