From ef29aba24c935dc963382a5103e6c04944f6a486 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Tue, 20 Oct 2020 10:11:26 +0300 Subject: [PATCH] created usage stats queue consumer and producer --- .../queue/DefaultTbCoreConsumerService.java | 2 +- .../DefaultTbUsageStatsService.java | 15 ++++++ .../usagestats/TbUsageStatsService.java | 15 ++++++ .../src/main/resources/thingsboard.yml | 1 + .../provider/AwsSqsMonolithQueueFactory.java | 45 ++++++++++------ .../provider/AwsSqsTbCoreQueueFactory.java | 12 +++++ .../AwsSqsTbRuleEngineQueueFactory.java | 6 +++ .../provider/AwsSqsTransportQueueFactory.java | 6 +++ .../InMemoryMonolithQueueFactory.java | 12 +++++ .../InMemoryTbTransportQueueFactory.java | 7 +++ .../provider/KafkaMonolithQueueFactory.java | 23 ++++++++ .../provider/KafkaTbCoreQueueFactory.java | 23 ++++++++ .../KafkaTbRuleEngineQueueFactory.java | 11 ++++ .../KafkaTbTransportQueueFactory.java | 11 ++++ .../provider/PubSubMonolithQueueFactory.java | 12 +++++ .../provider/PubSubTbCoreQueueFactory.java | 12 +++++ .../PubSubTbRuleEngineQueueFactory.java | 6 +++ .../provider/PubSubTransportQueueFactory.java | 6 +++ .../RabbitMqMonolithQueueFactory.java | 52 +++++++++++++------ .../provider/RabbitMqTbCoreQueueFactory.java | 12 +++++ .../RabbitMqTbRuleEngineQueueFactory.java | 6 +++ .../RabbitMqTransportQueueFactory.java | 6 +++ .../ServiceBusMonolithQueueFactory.java | 12 +++++ .../ServiceBusTbCoreQueueFactory.java | 23 ++++++-- .../ServiceBusTbRuleEngineQueueFactory.java | 17 ++++-- .../ServiceBusTransportQueueFactory.java | 36 ++++++++----- .../queue/provider/TbCoreQueueFactory.java | 7 +-- .../TbUsageStatsClientQueueFactory.java | 5 +- .../queue/settings/TbQueueCoreSettings.java | 3 ++ 29 files changed, 337 insertions(+), 67 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 3f03b65416..e806eccf66 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/application/src/main/java/org/thingsboard/server/service/usagestats/DefaultTbUsageStatsService.java b/application/src/main/java/org/thingsboard/server/service/usagestats/DefaultTbUsageStatsService.java index 6611507531..bf20419670 100644 --- a/application/src/main/java/org/thingsboard/server/service/usagestats/DefaultTbUsageStatsService.java +++ b/application/src/main/java/org/thingsboard/server/service/usagestats/DefaultTbUsageStatsService.java @@ -1,3 +1,18 @@ +/** + * 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.service.usagestats; import org.springframework.stereotype.Service; diff --git a/application/src/main/java/org/thingsboard/server/service/usagestats/TbUsageStatsService.java b/application/src/main/java/org/thingsboard/server/service/usagestats/TbUsageStatsService.java index 35a7051896..e735c5e2e2 100644 --- a/application/src/main/java/org/thingsboard/server/service/usagestats/TbUsageStatsService.java +++ b/application/src/main/java/org/thingsboard/server/service/usagestats/TbUsageStatsService.java @@ -1,3 +1,18 @@ +/** + * 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.service.usagestats; import org.thingsboard.server.common.msg.queue.TbCallback; diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index deb28ffd99..f0f4b9dd54 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -673,6 +673,7 @@ queue: poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_CORE_PARTITIONS:10}" pack-processing-timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:2000}" + usage-stats-topic: "${TB_QUEUE_US_TOPIC:tb_usage_stats}" stats: enabled: "${TB_QUEUE_CORE_STATS_ENABLED:true}" print-interval-ms: "${TB_QUEUE_CORE_STATS_PRINT_INTERVAL_MS:60000}" 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 c1d724c207..0299720005 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 @@ -22,7 +22,7 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.gen.transport.TransportProtos.*; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; @@ -91,64 +91,64 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng } @Override - public TbQueueProducer> createTransportNotificationsMsgProducer() { + public TbQueueProducer> createTransportNotificationsMsgProducer() { return new TbAwsSqsProducerTemplate<>(notificationAdmin, sqsSettings, transportNotificationSettings.getNotificationsTopic()); } @Override - public TbQueueProducer> createRuleEngineMsgProducer() { + public TbQueueProducer> createRuleEngineMsgProducer() { return new TbAwsSqsProducerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic()); } @Override - public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { + public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { return new TbAwsSqsProducerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic()); } @Override - public TbQueueProducer> createTbCoreMsgProducer() { + public TbQueueProducer> createTbCoreMsgProducer() { return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override - public TbQueueProducer> createTbCoreNotificationsMsgProducer() { + public TbQueueProducer> createTbCoreNotificationsMsgProducer() { return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { return new TbAwsSqsConsumerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override - public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { + public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { 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())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override - public TbQueueConsumer> createToCoreMsgConsumer() { + public TbQueueConsumer> createToCoreMsgConsumer() { return new TbAwsSqsConsumerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override - public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { + public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { 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())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override - public TbQueueConsumer> createTransportApiRequestConsumer() { + public TbQueueConsumer> createTransportApiRequestConsumer() { return new TbAwsSqsConsumerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getRequestsTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override - public TbQueueProducer> createTransportApiResponseProducer() { + public TbQueueProducer> createTransportApiResponseProducer() { return new TbAwsSqsProducerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getResponsesTopic()); } @@ -175,6 +175,17 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng return builder.build(); } + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getUsageStatsTopic()); + } + + @Override + public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + return new TbAwsSqsConsumerTemplate<>(transportApiAdmin, sqsSettings, coreSettings.getUsageStatsTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { 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 4889a7182d..4625b9d4fc 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 @@ -21,6 +21,7 @@ import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; +import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -163,6 +164,17 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { return builder.build(); } + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getUsageStatsTopic()); + } + + @Override + public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + return new TbAwsSqsConsumerTemplate<>(transportApiAdmin, sqsSettings, coreSettings.getUsageStatsTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { 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 cf43bc5fe4..851b8cc40e 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 @@ -143,6 +143,11 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory return builder.build(); } + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { @@ -158,4 +163,5 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory notificationAdmin.destroy(); } } + } 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 8106ee084b..103711f36b 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 @@ -18,6 +18,7 @@ package org.thingsboard.server.queue.provider; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Component; +import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; @@ -109,6 +110,11 @@ public class AwsSqsTransportQueueFactory implements TbTransportQueueFactory { msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); } + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java index 8ebda6e7b9..f7a47ecf3a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java @@ -125,8 +125,20 @@ public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE return null; } + @Override + public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + return new InMemoryTbQueueConsumer<>(coreSettings.getUsageStatsTopic()); + } + + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new InMemoryTbQueueProducer<>(coreSettings.getUsageStatsTopic()); + } + @Scheduled(fixedRateString = "${queue.in_memory.stats.print-interval-ms:60000}") private void printInMemoryStats() { storage.printStats(); } + + } 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 9e8e326f0d..9ddf72534c 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 @@ -18,6 +18,7 @@ package org.thingsboard.server.queue.provider; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Component; +import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; @@ -96,4 +97,10 @@ public class InMemoryTbTransportQueueFactory implements TbTransportQueueFactory public TbQueueConsumer> createTransportNotificationsConsumer() { return new InMemoryTbQueueConsumer<>(transportNotificationSettings.getNotificationsTopic() + "." + serviceInfoProvider.getServiceId()); } + + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new InMemoryTbQueueProducer<>(coreSettings.getUsageStatsTopic()); + } + } 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 31ef4d3e45..538d91aa0f 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 @@ -26,6 +26,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMs import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueAdmin; @@ -249,6 +250,28 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi return builder.build(); } + @Override + public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); + consumerBuilder.settings(kafkaSettings); + consumerBuilder.topic(coreSettings.getUsageStatsTopic()); + consumerBuilder.clientId("monolith-us-consumer-" + serviceInfoProvider.getServiceId()); + consumerBuilder.groupId("monolith-us-consumer-" + serviceInfoProvider.getServiceId()); + consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(coreAdmin); + return consumerBuilder.build(); + } + + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("monolith-us-producer-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(coreSettings.getUsageStatsTopic()); + requestBuilder.admin(coreAdmin); + return requestBuilder.build(); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { 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 bf2ec6751e..5e1b398e91 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 @@ -26,6 +26,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMs import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueAdmin; @@ -219,6 +220,28 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { return builder.build(); } + @Override + public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); + consumerBuilder.settings(kafkaSettings); + consumerBuilder.topic(coreSettings.getUsageStatsTopic()); + consumerBuilder.clientId("tb-core-us-consumer-" + serviceInfoProvider.getServiceId()); + consumerBuilder.groupId("tb-core-us-consumer-" + serviceInfoProvider.getServiceId()); + consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); + consumerBuilder.admin(coreAdmin); + return consumerBuilder.build(); + } + + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("tb-core-us-producer-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(coreSettings.getUsageStatsTopic()); + requestBuilder.admin(coreAdmin); + return requestBuilder.build(); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { 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 dc44caf3f0..118f0f3316 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 @@ -26,6 +26,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMs import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; @@ -192,6 +193,16 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { return builder.build(); } + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("tb-rule-engine-us-producer-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(coreSettings.getUsageStatsTopic()); + requestBuilder.admin(coreAdmin); + return requestBuilder.build(); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { 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 1cf8bca4b2..7961bb97bb 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 @@ -21,6 +21,7 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueAdmin; @@ -138,6 +139,16 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory { return responseBuilder.build(); } + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("transport-node-us-producer-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(coreSettings.getUsageStatsTopic()); + requestBuilder.admin(coreAdmin); + return requestBuilder.build(); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { 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 29a60af1e2..d7c7542626 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 @@ -27,6 +27,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMs import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueAdmin; @@ -183,6 +184,17 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng return builder.build(); } + @Override + public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + return new TbPubSubConsumerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { 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 7d85b5c3ba..71a1ceaf9b 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 @@ -26,6 +26,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMs import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueAdmin; @@ -157,6 +158,17 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { return builder.build(); } + @Override + public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + return new TbPubSubConsumerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { 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 a6105fea0c..ad679039e4 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 @@ -26,6 +26,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMs import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; @@ -146,6 +147,11 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory return builder.build(); } + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { 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 72665c5163..efde055271 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 @@ -21,6 +21,7 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueAdmin; @@ -114,6 +115,11 @@ public class PubSubTransportQueueFactory implements TbTransportQueueFactory { msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); } + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java index 1f812ea13c..e6eff652be 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java @@ -22,7 +22,14 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; +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; @@ -91,64 +98,64 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE } @Override - public TbQueueProducer> createTransportNotificationsMsgProducer() { + public TbQueueProducer> createTransportNotificationsMsgProducer() { return new TbRabbitMqProducerTemplate<>(notificationAdmin, rabbitMqSettings, transportNotificationSettings.getNotificationsTopic()); } @Override - public TbQueueProducer> createRuleEngineMsgProducer() { + public TbQueueProducer> createRuleEngineMsgProducer() { return new TbRabbitMqProducerTemplate<>(ruleEngineAdmin, rabbitMqSettings, ruleEngineSettings.getTopic()); } @Override - public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { + public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { return new TbRabbitMqProducerTemplate<>(ruleEngineAdmin, rabbitMqSettings, ruleEngineSettings.getTopic()); } @Override - public TbQueueProducer> createTbCoreMsgProducer() { + public TbQueueProducer> createTbCoreMsgProducer() { return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getTopic()); } @Override - public TbQueueProducer> createTbCoreNotificationsMsgProducer() { + public TbQueueProducer> createTbCoreNotificationsMsgProducer() { return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getTopic()); } @Override - public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { return new TbRabbitMqConsumerTemplate<>(ruleEngineAdmin, rabbitMqSettings, ruleEngineSettings.getTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override - public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { + public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbRabbitMqConsumerTemplate<>(notificationAdmin, rabbitMqSettings, partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override - public TbQueueConsumer> createToCoreMsgConsumer() { + public TbQueueConsumer> createToCoreMsgConsumer() { return new TbRabbitMqConsumerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override - public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { + public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbRabbitMqConsumerTemplate<>(notificationAdmin, rabbitMqSettings, partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override - public TbQueueConsumer> createTransportApiRequestConsumer() { + public TbQueueConsumer> createTransportApiRequestConsumer() { return new TbRabbitMqConsumerTemplate<>(transportApiAdmin, rabbitMqSettings, transportApiSettings.getRequestsTopic(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override - public TbQueueProducer> createTransportApiResponseProducer() { + public TbQueueProducer> createTransportApiResponseProducer() { return new TbRabbitMqProducerTemplate<>(transportApiAdmin, rabbitMqSettings, transportApiSettings.getResponsesTopic()); } @@ -175,6 +182,17 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE return builder.build(); } + @Override + public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + return new TbRabbitMqConsumerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getUsageStatsTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java index 65d582c4bc..e3064dd62d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java @@ -26,6 +26,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMs import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueAdmin; @@ -163,6 +164,17 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { return builder.build(); } + @Override + public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + return new TbRabbitMqConsumerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getUsageStatsTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java index 26abf78164..a1097049db 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java @@ -26,6 +26,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMs import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; @@ -144,6 +145,11 @@ public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactor return builder.build(); } + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTransportQueueFactory.java index 03cd6eb557..b28a2cb7e2 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTransportQueueFactory.java @@ -21,6 +21,7 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueAdmin; @@ -111,6 +112,11 @@ public class RabbitMqTransportQueueFactory implements TbTransportQueueFactory { msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); } + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java index 3db6496b7d..a205f9b061 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java @@ -26,6 +26,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMs import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueAdmin; @@ -180,6 +181,17 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul return builder.build(); } + @Override + public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + return new TbServiceBusConsumerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getUsageStatsTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java index a3ba577f06..f00aee331b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java @@ -21,10 +21,12 @@ import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; -import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueAdmin; @@ -101,7 +103,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { } @Override - public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { + public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { return new TbServiceBusProducerTemplate<>(ruleEngineAdmin, serviceBusSettings, ruleEngineSettings.getTopic()); } @@ -111,7 +113,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { } @Override - public TbQueueProducer> createTbCoreNotificationsMsgProducer() { + public TbQueueProducer> createTbCoreNotificationsMsgProducer() { return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getTopic()); } @@ -122,10 +124,10 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { } @Override - public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { + public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbServiceBusConsumerTemplate<>(notificationAdmin, serviceBusSettings, partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override @@ -162,6 +164,17 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { return builder.build(); } + @Override + public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + return new TbServiceBusConsumerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getUsageStatsTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java index e4f6dbfbb4..40a183a254 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java @@ -21,10 +21,12 @@ import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; -import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; @@ -93,7 +95,7 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact } @Override - public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { + public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { return new TbServiceBusProducerTemplate<>(ruleEngineAdmin, serviceBusSettings, ruleEngineSettings.getTopic()); } @@ -103,7 +105,7 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact } @Override - public TbQueueProducer> createTbCoreNotificationsMsgProducer() { + public TbQueueProducer> createTbCoreNotificationsMsgProducer() { return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getTopic()); } @@ -114,10 +116,10 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact } @Override - public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { + public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbServiceBusConsumerTemplate<>(notificationAdmin, serviceBusSettings, partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override @@ -143,6 +145,11 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact return builder.build(); } + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getUsageStatsTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTransportQueueFactory.java index 1e900f37e5..4781d46583 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTransportQueueFactory.java @@ -18,19 +18,24 @@ package org.thingsboard.server.queue.provider; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Component; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; +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; import org.thingsboard.server.queue.azure.servicebus.TbServiceBusAdmin; +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusConsumerTemplate; +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusProducerTemplate; import org.thingsboard.server.queue.azure.servicebus.TbServiceBusQueueConfigs; +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings; 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.azure.servicebus.TbServiceBusConsumerTemplate; -import org.thingsboard.server.queue.azure.servicebus.TbServiceBusProducerTemplate; -import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; @@ -69,17 +74,17 @@ public class ServiceBusTransportQueueFactory implements TbTransportQueueFactory } @Override - public TbQueueRequestTemplate, TbProtoQueueMsg> createTransportApiRequestTemplate() { - TbQueueProducer> producerTemplate = + public TbQueueRequestTemplate, TbProtoQueueMsg> createTransportApiRequestTemplate() { + TbQueueProducer> producerTemplate = new TbServiceBusProducerTemplate<>(transportApiAdmin, serviceBusSettings, transportApiSettings.getRequestsTopic()); - TbQueueConsumer> consumerTemplate = + TbQueueConsumer> consumerTemplate = new TbServiceBusConsumerTemplate<>(transportApiAdmin, serviceBusSettings, transportApiSettings.getResponsesTopic() + "." + serviceInfoProvider.getServiceId(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder - , TbProtoQueueMsg> templateBuilder = DefaultTbQueueRequestTemplate.builder(); + , TbProtoQueueMsg> templateBuilder = DefaultTbQueueRequestTemplate.builder(); templateBuilder.queueAdmin(transportApiAdmin); templateBuilder.requestTemplate(producerTemplate); templateBuilder.responseTemplate(consumerTemplate); @@ -90,20 +95,25 @@ public class ServiceBusTransportQueueFactory implements TbTransportQueueFactory } @Override - public TbQueueProducer> createRuleEngineMsgProducer() { + public TbQueueProducer> createRuleEngineMsgProducer() { return new TbServiceBusProducerTemplate<>(transportApiAdmin, serviceBusSettings, transportApiSettings.getRequestsTopic()); } @Override - public TbQueueProducer> createTbCoreMsgProducer() { + public TbQueueProducer> createTbCoreMsgProducer() { return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getTopic()); } @Override - public TbQueueConsumer> createTransportNotificationsConsumer() { + public TbQueueConsumer> createTransportNotificationsConsumer() { return new TbServiceBusConsumerTemplate<>(notificationAdmin, serviceBusSettings, transportNotificationSettings.getNotificationsTopic() + "." + serviceInfoProvider.getServiceId(), - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueProducer> createToUsageStatsServiceMsgProducer() { + return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getUsageStatsTopic()); } @PreDestroy diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java index d009757997..7c19a1ba2f 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java @@ -16,12 +16,12 @@ package org.thingsboard.server.queue.provider; import org.thingsboard.server.gen.js.JsInvokeProtos; -import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueConsumer; @@ -83,10 +83,7 @@ public interface TbCoreQueueFactory extends TbUsageStatsClientQueueFactory { * * @return */ - default TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { - //TODO: implement - return null; - } + TbQueueConsumer> createToUsageStatsServiceMsgConsumer(); /** * Used to consume high priority messages by TB Core Service diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbUsageStatsClientQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbUsageStatsClientQueueFactory.java index 1811877338..f2ce1a9fe5 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbUsageStatsClientQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbUsageStatsClientQueueFactory.java @@ -21,9 +21,6 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg; public interface TbUsageStatsClientQueueFactory { - default TbQueueProducer> createToUsageStatsServiceMsgProducer() { - //TODO: implement - return null; - } + TbQueueProducer> createToUsageStatsServiceMsgProducer(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueCoreSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueCoreSettings.java index 1c260c754d..d406139dbb 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueCoreSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueCoreSettings.java @@ -26,6 +26,9 @@ public class TbQueueCoreSettings { @Value("${queue.core.topic}") private String topic; + @Value("${queue.core.usage-stats-topic}") + private String usageStatsTopic; + @Value("${queue.core.partitions}") private int partitions; }