diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 7db2224d34..e1296275cd 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -533,7 +533,13 @@ queue: secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}" region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" - visibility_timeout: "${TB_QUEUE_AWS_SQS_VISIBILITY_TIMEOUT:30}" #In seconds. If messages wont commit in this time, messages will poll again + queue-properties: + rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + core: "${TB_QUEUE_AWS_SQS_CORE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + transport-api: "${TB_QUEUE_AWS_SQS_TA_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + notifications: "${TB_QUEUE_AWS_SQS_NOTIFICATIONS_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + js-executor: "${TB_QUEUE_AWS_SQS_JE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + # VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds pubsub: project_id: "${TB_QUEUE_PUBSUB_PROJECT_ID:YOUR_PROJECT_ID}" service_account: "${TB_QUEUE_PUBSUB_SERVICE_ACCOUNT:YOUR_SERVICE_ACCOUNT}" diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/TbQueueAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/TbQueueAdmin.java index deb84f440e..bb541fb626 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/TbQueueAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/TbQueueAdmin.java @@ -19,4 +19,5 @@ public interface TbQueueAdmin { void createTopicIfNotExists(String topic); + void destroy(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/azure/servicebus/TbServiceBusAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/azure/servicebus/TbServiceBusAdmin.java index fb9cd89a8d..336514b7b2 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/azure/servicebus/TbServiceBusAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/azure/servicebus/TbServiceBusAdmin.java @@ -23,7 +23,6 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Component; import org.thingsboard.server.queue.TbQueueAdmin; -import javax.annotation.PreDestroy; import java.io.IOException; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -68,8 +67,7 @@ public class TbServiceBusAdmin implements TbQueueAdmin { } } - @PreDestroy - private void destroy() { + public void destroy() { try { client.close(); } catch (IOException e) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java similarity index 92% rename from common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaAdmin.java rename to common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java index 69ad6286a6..b92b094af1 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java @@ -32,7 +32,7 @@ import java.util.concurrent.ExecutionException; * Created by ashvayka on 24.09.18. */ @Slf4j -public class TBKafkaAdmin implements TbQueueAdmin { +public class TbKafkaAdmin implements TbQueueAdmin { private final AdminClient client; private final Map topicConfigs; @@ -40,7 +40,7 @@ public class TBKafkaAdmin implements TbQueueAdmin { private final short replicationFactor; - public TBKafkaAdmin(TbKafkaSettings settings, Map topicConfigs) { + public TbKafkaAdmin(TbKafkaSettings settings, Map topicConfigs) { client = AdminClient.create(settings.toProps()); this.topicConfigs = topicConfigs; @@ -76,6 +76,13 @@ public class TBKafkaAdmin implements TbQueueAdmin { } + @Override + public void destroy() { + if (client != null) { + client.close(); + } + } + public CreateTopicsResult createTopic(NewTopic topic) { return client.createTopics(Collections.singletonList(topic)); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java similarity index 97% rename from common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaConsumerTemplate.java rename to common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java index 069f4a6455..49d1d74102 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java @@ -42,7 +42,7 @@ import java.util.stream.Collectors; * Created by ashvayka on 24.09.18. */ @Slf4j -public class TBKafkaConsumerTemplate implements TbQueueConsumer { +public class TbKafkaConsumerTemplate implements TbQueueConsumer { private final TbQueueAdmin admin; private final KafkaConsumer consumer; @@ -55,7 +55,7 @@ public class TBKafkaConsumerTemplate implements TbQueueCon private final String topic; @Builder - private TBKafkaConsumerTemplate(TbKafkaSettings settings, TbKafkaDecoder decoder, + private TbKafkaConsumerTemplate(TbKafkaSettings settings, TbKafkaDecoder decoder, String clientId, String groupId, String topic, boolean autoCommit, int autoCommitIntervalMs, int maxPollRecords, diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaHandler.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaHandler.java deleted file mode 100644 index 8c2cbd2771..0000000000 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaHandler.java +++ /dev/null @@ -1,27 +0,0 @@ -/** - * Copyright © 2016-2020 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.queue.kafka; - -import com.google.common.util.concurrent.ListenableFuture; - -/** - * Created by ashvayka on 05.10.18. - */ -public interface TbKafkaHandler { - - ListenableFuture handle(Request request); - -} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaPartitioner.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaPartitioner.java deleted file mode 100644 index b31f87642b..0000000000 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaPartitioner.java +++ /dev/null @@ -1,30 +0,0 @@ -/** - * Copyright © 2016-2020 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.queue.kafka; - -import org.apache.kafka.clients.producer.Partitioner; -import org.apache.kafka.common.PartitionInfo; - -import java.util.List; - -/** - * Created by ashvayka on 25.09.18. - */ -public interface TbKafkaPartitioner extends Partitioner { - - int partition(String topic, String key, T value, byte[] encodedValue, List partitions); - -} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaProducerTemplate.java similarity index 94% rename from common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java rename to common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaProducerTemplate.java index f81381bc38..4f26f51da7 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaProducerTemplate.java @@ -39,7 +39,7 @@ import java.util.stream.Collectors; * Created by ashvayka on 24.09.18. */ @Slf4j -public class TBKafkaProducerTemplate implements TbQueueProducer { +public class TbKafkaProducerTemplate implements TbQueueProducer { private final KafkaProducer producer; @@ -54,7 +54,7 @@ public class TBKafkaProducerTemplate implements TbQueuePro private final Set topics; @Builder - private TBKafkaProducerTemplate(TbKafkaSettings settings, TbKafkaPartitioner partitioner, String defaultTopic, String clientId, TbQueueAdmin admin) { + private TbKafkaProducerTemplate(TbKafkaSettings settings, String defaultTopic, String clientId, TbQueueAdmin admin) { Properties props = settings.toProps(); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java index 8682ca385f..4b68abf929 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java @@ -17,6 +17,7 @@ package org.thingsboard.server.queue.kafka; import lombok.Getter; import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; @@ -24,6 +25,7 @@ import java.util.HashMap; import java.util.Map; @Component +@ConditionalOnExpression("'${queue.type:null}'=='kafka'") public class TbKafkaTopicConfigs { @Value("${queue.kafka.topic-properties.core}") private String coreProperties; 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 26da11aa3b..90fe643505 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 @@ -36,6 +36,7 @@ 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; +import org.thingsboard.server.queue.sqs.TbAwsSqsQueueAttributes; import org.thingsboard.server.queue.sqs.TbAwsSqsSettings; @Component @@ -49,14 +50,20 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbAwsSqsSettings sqsSettings; - private final TbQueueAdmin admin; + + private final TbQueueAdmin coreAdmin; + private final TbQueueAdmin ruleEngineAdmin; + private final TbQueueAdmin jsExecutorAdmin; + private final TbQueueAdmin transportApiAdmin; + private final TbQueueAdmin notificationAdmin; public AwsSqsMonolithQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, - TbAwsSqsSettings sqsSettings) { + TbAwsSqsSettings sqsSettings, + TbAwsSqsQueueAttributes sqsQueueAttributes) { this.partitionService = partitionService; this.coreSettings = coreSettings; this.serviceInfoProvider = serviceInfoProvider; @@ -64,69 +71,74 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; this.sqsSettings = sqsSettings; - admin = new TbAwsSqsAdmin(sqsSettings); + + this.coreAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getCoreAttributes()); + this.ruleEngineAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getRuleEngineAttributes()); + this.jsExecutorAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getJsExecutorAttributes()); + this.transportApiAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getTransportApiAttributes()); + this.notificationAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getNotificationsAttributes()); } @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportNotificationSettings.getNotificationsTopic()); + return new TbAwsSqsProducerTemplate<>(notificationAdmin, sqsSettings, transportNotificationSettings.getNotificationsTopic()); } @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createTbCoreMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic(), + return new TbAwsSqsConsumerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, + return new TbAwsSqsConsumerTemplate<>(notificationAdmin, sqsSettings, partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToCoreMsgConsumer() { - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, coreSettings.getTopic(), + return new TbAwsSqsConsumerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, + return new TbAwsSqsConsumerTemplate<>(notificationAdmin, sqsSettings, partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createTransportApiRequestConsumer() { - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic(), + return new TbAwsSqsConsumerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getRequestsTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueProducer> createTransportApiResponseProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getResponsesTopic()); + return new TbAwsSqsProducerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getResponsesTopic()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java index 87aba51843..e169e6502d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java @@ -40,6 +40,7 @@ import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate; import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate; +import org.thingsboard.server.queue.sqs.TbAwsSqsQueueAttributes; import org.thingsboard.server.queue.sqs.TbAwsSqsSettings; @Component @@ -52,70 +53,81 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueTransportApiSettings transportApiSettings; private final PartitionService partitionService; private final TbServiceInfoProvider serviceInfoProvider; - private final TbQueueAdmin admin; + + private final TbQueueAdmin coreAdmin; + private final TbQueueAdmin ruleEngineAdmin; + private final TbQueueAdmin jsExecutorAdmin; + private final TbQueueAdmin transportApiAdmin; + private final TbQueueAdmin notificationAdmin; public AwsSqsTbCoreQueueFactory(TbAwsSqsSettings sqsSettings, TbQueueCoreSettings coreSettings, TbQueueTransportApiSettings transportApiSettings, TbQueueRuleEngineSettings ruleEngineSettings, PartitionService partitionService, - TbServiceInfoProvider serviceInfoProvider) { + TbServiceInfoProvider serviceInfoProvider, + TbAwsSqsQueueAttributes sqsQueueAttributes) { this.sqsSettings = sqsSettings; this.coreSettings = coreSettings; this.transportApiSettings = transportApiSettings; this.ruleEngineSettings = ruleEngineSettings; this.partitionService = partitionService; this.serviceInfoProvider = serviceInfoProvider; - this.admin = new TbAwsSqsAdmin(sqsSettings); + + this.coreAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getCoreAttributes()); + this.ruleEngineAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getRuleEngineAttributes()); + this.jsExecutorAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getJsExecutorAttributes()); + this.transportApiAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getTransportApiAttributes()); + this.notificationAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getNotificationsAttributes()); } @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createTbCoreMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override public TbQueueConsumer> createToCoreMsgConsumer() { - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, coreSettings.getTopic(), + return new TbAwsSqsConsumerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, + return new TbAwsSqsConsumerTemplate<>(notificationAdmin, sqsSettings, partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createTransportApiRequestConsumer() { - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic(), + return new TbAwsSqsConsumerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getRequestsTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueProducer> createTransportApiResponseProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override 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 d83b760c19..33f1d8c113 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 @@ -37,6 +37,7 @@ 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; +import org.thingsboard.server.queue.sqs.TbAwsSqsQueueAttributes; import org.thingsboard.server.queue.sqs.TbAwsSqsSettings; @Component @@ -48,54 +49,63 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbAwsSqsSettings sqsSettings; - private final TbQueueAdmin admin; + + private final TbQueueAdmin coreAdmin; + private final TbQueueAdmin ruleEngineAdmin; + private final TbQueueAdmin jsExecutorAdmin; + private final TbQueueAdmin notificationAdmin; public AwsSqsTbRuleEngineQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbServiceInfoProvider serviceInfoProvider, - TbAwsSqsSettings sqsSettings) { + TbAwsSqsSettings sqsSettings, + TbAwsSqsQueueAttributes sqsQueueAttributes) { this.partitionService = partitionService; this.coreSettings = coreSettings; this.serviceInfoProvider = serviceInfoProvider; this.ruleEngineSettings = ruleEngineSettings; this.sqsSettings = sqsSettings; - admin = new TbAwsSqsAdmin(sqsSettings); + + this.coreAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getCoreAttributes()); + this.ruleEngineAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getRuleEngineAttributes()); + this.jsExecutorAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getJsExecutorAttributes()); + this.notificationAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getNotificationsAttributes()); } @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createTbCoreMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic(), + return new TbAwsSqsConsumerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, + return new TbAwsSqsConsumerTemplate<>(notificationAdmin, sqsSettings, partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTransportQueueFactory.java index 4196b226ee..644bc6b709 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTransportQueueFactory.java @@ -35,6 +35,7 @@ import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSetting import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate; import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate; +import org.thingsboard.server.queue.sqs.TbAwsSqsQueueAttributes; import org.thingsboard.server.queue.sqs.TbAwsSqsSettings; @Component @@ -44,33 +45,38 @@ public class AwsSqsTransportQueueFactory implements TbTransportQueueFactory { private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbAwsSqsSettings sqsSettings; - private final TbQueueAdmin admin; private final TbServiceInfoProvider serviceInfoProvider; + private final TbQueueAdmin transportApiAdmin; + private final TbQueueAdmin notificationAdmin; + public AwsSqsTransportQueueFactory(TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, TbAwsSqsSettings sqsSettings, - TbServiceInfoProvider serviceInfoProvider) { + TbServiceInfoProvider serviceInfoProvider, + TbAwsSqsQueueAttributes sqsQueueAttributes) { this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; this.sqsSettings = sqsSettings; - admin = new TbAwsSqsAdmin(sqsSettings); this.serviceInfoProvider = serviceInfoProvider; + + this.transportApiAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getTransportApiAttributes()); + this.notificationAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getNotificationsAttributes()); } @Override public TbQueueRequestTemplate, TbProtoQueueMsg> createTransportApiRequestTemplate() { TbAwsSqsProducerTemplate> producerTemplate = - new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic()); + new TbAwsSqsProducerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getRequestsTopic()); TbAwsSqsConsumerTemplate> consumerTemplate = - new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, + new TbAwsSqsConsumerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getResponsesTopic() + "_" + serviceInfoProvider.getServiceId(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> templateBuilder = DefaultTbQueueRequestTemplate.builder(); - templateBuilder.queueAdmin(admin); + templateBuilder.queueAdmin(transportApiAdmin); templateBuilder.requestTemplate(producerTemplate); templateBuilder.responseTemplate(consumerTemplate); templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests()); @@ -81,17 +87,17 @@ public class AwsSqsTransportQueueFactory implements TbTransportQueueFactory { @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic()); + return new TbAwsSqsProducerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getRequestsTopic()); } @Override public TbQueueProducer> createTbCoreMsgProducer() { - return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic()); + return new TbAwsSqsProducerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getRequestsTopic()); } @Override public TbQueueConsumer> createTransportNotificationsConsumer() { - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, transportNotificationSettings.getNotificationsTopic() + "_" + serviceInfoProvider.getServiceId(), + return new TbAwsSqsConsumerTemplate<>(notificationAdmin, sqsSettings, transportNotificationSettings.getNotificationsTopic() + "_" + serviceInfoProvider.getServiceId(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryTbTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryTbTransportQueueFactory.java index 29660f2461..c03f5e52d8 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryTbTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryTbTransportQueueFactory.java @@ -23,6 +23,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; +import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueRequestTemplate; @@ -33,6 +34,7 @@ import org.thingsboard.server.queue.memory.InMemoryTbQueueConsumer; import org.thingsboard.server.queue.memory.InMemoryTbQueueProducer; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; +import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; @Component @ConditionalOnExpression("'${queue.type:null}'=='in-memory' && ('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport')") @@ -60,8 +62,15 @@ public class InMemoryTbTransportQueueFactory implements TbTransportQueueFactory DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> templateBuilder = DefaultTbQueueRequestTemplate.builder(); - templateBuilder.queueAdmin(topic -> { + + templateBuilder.queueAdmin(new TbQueueAdmin() { + @Override + public void createTopicIfNotExists(String topic) {} + + @Override + public void destroy() {} }); + templateBuilder.requestTemplate(producerTemplate); templateBuilder.responseTemplate(consumerTemplate); templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests()); 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 9eb656c2ea..72c6f7b8b9 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 @@ -37,9 +37,9 @@ import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; -import org.thingsboard.server.queue.kafka.TBKafkaAdmin; -import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; -import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; +import org.thingsboard.server.queue.kafka.TbKafkaAdmin; +import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; +import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; @@ -87,16 +87,16 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi this.transportNotificationSettings = transportNotificationSettings; this.jsInvokeSettings = jsInvokeSettings; - this.coreAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); - this.ruleEngineAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); - this.jsExecutorAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getJsExecutorConfigs()); - this.transportApiAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs()); - this.notificationAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); + this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); + this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); + this.jsExecutorAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getJsExecutorConfigs()); + this.transportApiAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs()); + this.notificationAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); } @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-transport-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(transportNotificationSettings.getNotificationsTopic()); @@ -106,7 +106,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi @Override public TbQueueProducer> createRuleEngineMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-rule-engine-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); @@ -116,7 +116,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-rule-engine-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); @@ -126,7 +126,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi @Override public TbQueueProducer> createTbCoreMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-core-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -136,7 +136,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-core-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -147,7 +147,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi @Override public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { String queueName = configuration.getName(); - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> consumerBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(ruleEngineSettings.getTopic()); consumerBuilder.clientId("re-" + queueName + "-consumer-" + serviceInfoProvider.getServiceId()); @@ -159,7 +159,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> consumerBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); @@ -171,7 +171,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi @Override public TbQueueConsumer> createToCoreMsgConsumer() { - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> consumerBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(coreSettings.getTopic()); consumerBuilder.clientId("monolith-core-consumer-" + serviceInfoProvider.getServiceId()); @@ -183,7 +183,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> consumerBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); @@ -195,7 +195,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi @Override public TbQueueConsumer> createTransportApiRequestConsumer() { - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> consumerBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(transportApiSettings.getRequestsTopic()); consumerBuilder.clientId("monolith-transport-api-consumer-" + serviceInfoProvider.getServiceId()); @@ -207,7 +207,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi @Override public TbQueueProducer> createTransportApiResponseProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-transport-api-producer-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(transportApiSettings.getResponsesTopic()); @@ -218,13 +218,13 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi @Override @Bean public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("producer-js-invoke-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(jsInvokeSettings.getRequestTopic()); requestBuilder.admin(jsExecutorAdmin); - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> responseBuilder = TbKafkaConsumerTemplate.builder(); responseBuilder.settings(kafkaSettings); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java index 8800c1c55a..ca10830981 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java @@ -37,9 +37,9 @@ import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; -import org.thingsboard.server.queue.kafka.TBKafkaAdmin; -import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; -import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; +import org.thingsboard.server.queue.kafka.TbKafkaAdmin; +import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; +import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; @@ -82,16 +82,16 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { this.transportApiSettings = transportApiSettings; this.jsInvokeSettings = jsInvokeSettings; - this.coreAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); - this.ruleEngineAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); - this.jsExecutorAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getJsExecutorConfigs()); - this.transportApiAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs()); - this.notificationAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); + this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); + this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); + this.jsExecutorAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getJsExecutorConfigs()); + this.transportApiAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs()); + this.notificationAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); } @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-transport-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -101,7 +101,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createRuleEngineMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-rule-engine-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -111,7 +111,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-rule-engine-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); @@ -121,7 +121,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTbCoreMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-to-core-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -131,7 +131,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-to-core-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -141,7 +141,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueConsumer> createToCoreMsgConsumer() { - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> consumerBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(coreSettings.getTopic()); consumerBuilder.clientId("tb-core-consumer-" + serviceInfoProvider.getServiceId()); @@ -153,7 +153,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> consumerBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("tb-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); @@ -165,7 +165,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueConsumer> createTransportApiRequestConsumer() { - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> consumerBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(transportApiSettings.getRequestsTopic()); consumerBuilder.clientId("tb-core-transport-api-consumer-" + serviceInfoProvider.getServiceId()); @@ -177,7 +177,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTransportApiResponseProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-transport-api-producer-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -188,13 +188,13 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { @Override @Bean public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("producer-js-invoke-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(jsInvokeSettings.getRequestTopic()); requestBuilder.admin(jsExecutorAdmin); - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> responseBuilder = TbKafkaConsumerTemplate.builder(); responseBuilder.settings(kafkaSettings); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); 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 b96d4ebe86..bcac399a64 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 @@ -35,9 +35,9 @@ import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; -import org.thingsboard.server.queue.kafka.TBKafkaAdmin; -import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; -import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; +import org.thingsboard.server.queue.kafka.TbKafkaAdmin; +import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; +import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; @@ -76,15 +76,15 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { this.ruleEngineSettings = ruleEngineSettings; this.jsInvokeSettings = jsInvokeSettings; - this.coreAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); - this.ruleEngineAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); - this.jsExecutorAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getJsExecutorConfigs()); - this.notificationAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); + this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); + this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); + this.jsExecutorAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getJsExecutorConfigs()); + this.notificationAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); } @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-transport-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -94,7 +94,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { @Override public TbQueueProducer> createRuleEngineMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-to-rule-engine-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -104,7 +104,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-to-rule-engine-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); @@ -115,7 +115,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { @Override public TbQueueProducer> createTbCoreMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-to-core-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -125,7 +125,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-to-core-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -136,7 +136,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { @Override public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { String queueName = configuration.getName(); - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> consumerBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(ruleEngineSettings.getTopic()); consumerBuilder.clientId("re-" + queueName + "-consumer-" + serviceInfoProvider.getServiceId()); @@ -148,7 +148,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> consumerBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("tb-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); @@ -161,13 +161,13 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { @Override @Bean public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("producer-js-invoke-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(jsInvokeSettings.getRequestTopic()); requestBuilder.admin(jsExecutorAdmin); - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> responseBuilder = TbKafkaConsumerTemplate.builder(); responseBuilder.settings(kafkaSettings); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java index f5a31cf1c7..2f7aab4779 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java @@ -30,9 +30,9 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; -import org.thingsboard.server.queue.kafka.TBKafkaAdmin; -import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; -import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; +import org.thingsboard.server.queue.kafka.TbKafkaAdmin; +import org.thingsboard.server.queue.kafka.TbKafkaConsumerTemplate; +import org.thingsboard.server.queue.kafka.TbKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; @@ -71,21 +71,21 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; - this.coreAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); - this.ruleEngineAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); - this.transportApiAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs()); - this.notificationAdmin = new TBKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); + this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); + this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); + this.transportApiAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getTransportApiConfigs()); + this.notificationAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); } @Override public TbQueueRequestTemplate, TbProtoQueueMsg> createTransportApiRequestTemplate() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("transport-api-request-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(transportApiSettings.getRequestsTopic()); requestBuilder.admin(transportApiAdmin); - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> responseBuilder = TbKafkaConsumerTemplate.builder(); responseBuilder.settings(kafkaSettings); responseBuilder.topic(transportApiSettings.getResponsesTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("transport-api-response-" + serviceInfoProvider.getServiceId()); @@ -106,7 +106,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { @Override public TbQueueProducer> createRuleEngineMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("transport-node-rule-engine-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); @@ -116,7 +116,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { @Override public TbQueueProducer> createTbCoreMsgProducer() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("transport-node-core-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); @@ -126,7 +126,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { @Override public TbQueueConsumer> createTransportNotificationsConsumer() { - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> responseBuilder = TbKafkaConsumerTemplate.builder(); responseBuilder.settings(kafkaSettings); responseBuilder.topic(transportNotificationSettings.getNotificationsTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.clientId("transport-api-notifications-" + serviceInfoProvider.getServiceId()); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubAdmin.java index f0af639d4d..cd575818f2 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubAdmin.java @@ -121,6 +121,11 @@ public class TbPubSubAdmin implements TbQueueAdmin { } } + @Override + public void destroy() { + + } + private void createSubscriptionIfNotExists(String partition, ProjectTopicName topicName) { ProjectSubscriptionName subscriptionName = ProjectSubscriptionName.of(pubSubSettings.getProjectId(), partition); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqAdmin.java index fbd678045b..9a7e5db342 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqAdmin.java @@ -14,15 +14,14 @@ * limitations under the License. */ package org.thingsboard.server.queue.rabbitmq; + import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; - import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Component; import org.thingsboard.server.queue.TbQueueAdmin; -import javax.annotation.PreDestroy; import java.io.IOException; import java.util.concurrent.TimeoutException; @@ -62,8 +61,8 @@ public class TbRabbitMqAdmin implements TbQueueAdmin { } } - @PreDestroy - private void destroy() { + @Override + public void destroy() { if (channel != null) { try { channel.close(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java index 6080baec26..8e293e6dff 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java @@ -21,45 +21,56 @@ import com.amazonaws.auth.BasicAWSCredentials; import com.amazonaws.services.sqs.AmazonSQS; import com.amazonaws.services.sqs.AmazonSQSClientBuilder; import com.amazonaws.services.sqs.model.CreateQueueRequest; -import com.amazonaws.services.sqs.model.QueueAttributeName; import org.thingsboard.server.queue.TbQueueAdmin; -import java.util.HashMap; import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; public class TbAwsSqsAdmin implements TbQueueAdmin { - private final TbAwsSqsSettings sqsSettings; - private final Map attributes = new HashMap<>(); - private final AWSStaticCredentialsProvider credProvider; + private final Map attributes; + private final AmazonSQS sqsClient; + private final Set queues; - public TbAwsSqsAdmin(TbAwsSqsSettings sqsSettings) { - this.sqsSettings = sqsSettings; + public TbAwsSqsAdmin(TbAwsSqsSettings sqsSettings, Map attributes) { + this.attributes = attributes; AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey()); - this.credProvider = new AWSStaticCredentialsProvider(awsCredentials); + sqsClient = AmazonSQSClientBuilder.standard() + .withCredentials(new AWSStaticCredentialsProvider(awsCredentials)) + .withRegion(sqsSettings.getRegion()) + .build(); - attributes.put("FifoQueue", "true"); - attributes.put("ContentBasedDeduplication", "true"); - attributes.put(QueueAttributeName.VisibilityTimeout.toString(), sqsSettings.getVisibilityTimeout()); + queues = sqsClient + .listQueues() + .getQueueUrls() + .stream() + .map(this::getQueueNameFromUrl) + .collect(Collectors.toCollection(ConcurrentHashMap::newKeySet)); } @Override public void createTopicIfNotExists(String topic) { - AmazonSQS sqsClient = AmazonSQSClientBuilder.standard() - .withCredentials(credProvider) - .withRegion(sqsSettings.getRegion()) - .build(); + String queueName = topic.replaceAll("\\.", "_") + ".fifo"; + if (queues.contains(queueName)) { + return; + } + final CreateQueueRequest createQueueRequest = new CreateQueueRequest(queueName).withAttributes(attributes); + String queueUrl = sqsClient.createQueue(createQueueRequest).getQueueUrl(); + queues.add(getQueueNameFromUrl(queueUrl)); + } + + private String getQueueNameFromUrl(String queueUrl) { + int delimiterIndex = queueUrl.lastIndexOf("/"); + return queueUrl.substring(delimiterIndex + 1); + } - final CreateQueueRequest createQueueRequest = - new CreateQueueRequest(topic.replaceAll("\\.", "_") + ".fifo") - .withAttributes(attributes); - try { - sqsClient.createQueue(createQueueRequest); - } finally { - if (sqsClient != null) { - sqsClient.shutdown(); - } + @Override + public void destroy() { + if (sqsClient != null) { + sqsClient.shutdown(); } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsQueueAttributes.java b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsQueueAttributes.java new file mode 100644 index 0000000000..70a9587bef --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsQueueAttributes.java @@ -0,0 +1,83 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.queue.sqs; + +import com.amazonaws.services.sqs.model.QueueAttributeName; +import lombok.Getter; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; +import java.util.HashMap; +import java.util.Map; + +@Component +@ConditionalOnExpression("'${queue.type:null}'=='aws-sqs'") +public class TbAwsSqsQueueAttributes { + @Value("${queue.aws-sqs.queue-properties.core}") + private String coreProperties; + @Value("${queue.aws-sqs.queue-properties.rule-engine}") + private String ruleEngineProperties; + @Value("${queue.aws-sqs.queue-properties.transport-api}") + private String transportApiProperties; + @Value("${queue.aws-sqs.queue-properties.notifications}") + private String notificationsProperties; + @Value("${queue.aws-sqs.queue-properties.js-executor}") + private String jsExecutorProperties; + + @Getter + private Map coreAttributes; + @Getter + private Map ruleEngineAttributes; + @Getter + private Map transportApiAttributes; + @Getter + private Map notificationsAttributes; + @Getter + private Map jsExecutorAttributes; + + private final Map defaultAttributes = new HashMap<>(); + + @PostConstruct + private void init() { + defaultAttributes.put(QueueAttributeName.FifoQueue.toString(), "true"); + defaultAttributes.put(QueueAttributeName.ContentBasedDeduplication.toString(), "true"); + + coreAttributes = getConfigs(coreProperties); + ruleEngineAttributes = getConfigs(ruleEngineProperties); + transportApiAttributes = getConfigs(transportApiProperties); + notificationsAttributes = getConfigs(notificationsProperties); + jsExecutorAttributes = getConfigs(jsExecutorProperties); + } + + private Map getConfigs(String properties) { + Map configs = new HashMap<>(); + for (String property : properties.split(";")) { + int delimiterPosition = property.indexOf(":"); + String key = property.substring(0, delimiterPosition); + String value = property.substring(delimiterPosition + 1); + validateAttributeName(key); + configs.put(key, value); + } + configs.putAll(defaultAttributes); + return configs; + } + + private void validateAttributeName(String key) { + QueueAttributeName.fromValue(key); + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java index 42a2f1e983..922a2b1062 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java @@ -39,6 +39,4 @@ public class TbAwsSqsSettings { @Value("${queue.aws_sqs.threads_per_topic}") private int threadsPerTopic; - @Value("${queue.aws_sqs.visibility_timeout}") - private String visibilityTimeout; } diff --git a/transport/coap/src/main/resources/tb-coap-transport.yml b/transport/coap/src/main/resources/tb-coap-transport.yml index e2fb7e149a..78dc191e2c 100644 --- a/transport/coap/src/main/resources/tb-coap-transport.yml +++ b/transport/coap/src/main/resources/tb-coap-transport.yml @@ -62,7 +62,13 @@ queue: secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}" region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" - visibility_timeout: "${TB_QUEUE_AWS_SQS_VISIBILITY_TIMEOUT:30}" #In seconds. If messages wont commit in this time, messages will poll again + queue-properties: + rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + core: "${TB_QUEUE_AWS_SQS_CORE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + transport-api: "${TB_QUEUE_AWS_SQS_TA_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + notifications: "${TB_QUEUE_AWS_SQS_NOTIFICATIONS_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + js-executor: "${TB_QUEUE_AWS_SQS_JE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + # VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds pubsub: project_id: "${TB_QUEUE_PUBSUB_PROJECT_ID:YOUR_PROJECT_ID}" service_account: "${TB_QUEUE_PUBSUB_SERVICE_ACCOUNT:YOUR_SERVICE_ACCOUNT}" diff --git a/transport/http/src/main/resources/tb-http-transport.yml b/transport/http/src/main/resources/tb-http-transport.yml index dfd2c14be2..82b548a456 100644 --- a/transport/http/src/main/resources/tb-http-transport.yml +++ b/transport/http/src/main/resources/tb-http-transport.yml @@ -63,7 +63,13 @@ queue: secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}" region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" - visibility_timeout: "${TB_QUEUE_AWS_SQS_VISIBILITY_TIMEOUT:30}" #In seconds. If messages wont commit in this time, messages will poll again + queue-properties: + rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + core: "${TB_QUEUE_AWS_SQS_CORE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + transport-api: "${TB_QUEUE_AWS_SQS_TA_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + notifications: "${TB_QUEUE_AWS_SQS_NOTIFICATIONS_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + js-executor: "${TB_QUEUE_AWS_SQS_JE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + # VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds pubsub: project_id: "${TB_QUEUE_PUBSUB_PROJECT_ID:YOUR_PROJECT_ID}" service_account: "${TB_QUEUE_PUBSUB_SERVICE_ACCOUNT:YOUR_SERVICE_ACCOUNT}" diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index 81f0fb795d..e3faf9cb87 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -93,7 +93,13 @@ queue: secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}" region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" - visibility_timeout: "${TB_QUEUE_AWS_SQS_VISIBILITY_TIMEOUT:30}" #In seconds. If messages wont commit in this time, messages will poll again + queue-properties: + rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + core: "${TB_QUEUE_AWS_SQS_CORE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + transport-api: "${TB_QUEUE_AWS_SQS_TA_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + notifications: "${TB_QUEUE_AWS_SQS_NOTIFICATIONS_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + js-executor: "${TB_QUEUE_AWS_SQS_JE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" + # VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds pubsub: project_id: "${TB_QUEUE_PUBSUB_PROJECT_ID:YOUR_PROJECT_ID}" service_account: "${TB_QUEUE_PUBSUB_SERVICE_ACCOUNT:YOUR_SERVICE_ACCOUNT}"