diff --git a/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java b/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java index c4e780af36..0c372fdfd6 100644 --- a/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java @@ -17,6 +17,7 @@ package org.thingsboard.server.service.ota; import com.google.common.util.concurrent.FutureCallback; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Required; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; @@ -49,6 +50,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateSer import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; +import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.cluster.TbClusterService; @@ -57,6 +59,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; import java.util.List; +import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.function.Consumer; @@ -78,7 +81,6 @@ import static org.thingsboard.server.common.data.ota.OtaPackageUtil.getTelemetry @Slf4j @Service -@TbCoreComponent public class DefaultOtaPackageStateService implements OtaPackageStateService { private final TbClusterService tbClusterService; @@ -93,13 +95,18 @@ public class DefaultOtaPackageStateService implements OtaPackageStateService { DeviceService deviceService, DeviceProfileService deviceProfileService, @Lazy RuleEngineTelemetryService telemetryService, - TbCoreQueueFactory coreQueueFactory) { + Optional coreQueueFactory, + Optional reQueueFactory) { this.tbClusterService = tbClusterService; this.otaPackageService = otaPackageService; this.deviceService = deviceService; this.deviceProfileService = deviceProfileService; this.telemetryService = telemetryService; - this.otaPackageStateMsgProducer = coreQueueFactory.createToOtaPackageStateServiceMsgProducer(); + if (coreQueueFactory.isPresent()) { + this.otaPackageStateMsgProducer = coreQueueFactory.get().createToOtaPackageStateServiceMsgProducer(); + } else { + this.otaPackageStateMsgProducer = reQueueFactory.get().createToOtaPackageStateServiceMsgProducer(); + } } @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 eb25b64906..172f27d357 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 @@ -148,6 +148,12 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getUsageStatsTopic()); } + @Override + public TbQueueProducer> createToOtaPackageStateServiceMsgProducer() { + return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getOtaPackageTopic()); + } + + @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 de6ccc1bba..fce429c9a3 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 @@ -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; @@ -67,6 +68,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { private final TbQueueAdmin ruleEngineAdmin; private final TbQueueAdmin jsExecutorAdmin; private final TbQueueAdmin notificationAdmin; + private final TbQueueAdmin fwUpdatesAdmin; private final AtomicLong consumerCount = new AtomicLong(); public KafkaTbRuleEngineQueueFactory(PartitionService partitionService, TbKafkaSettings kafkaSettings, @@ -88,6 +90,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); this.jsExecutorAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getJsExecutorConfigs()); this.notificationAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getNotificationsConfigs()); + this.fwUpdatesAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getFwUpdatesConfigs()); } @Override @@ -131,6 +134,16 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { return requestBuilder.build(); } + @Override + public TbQueueProducer> createToOtaPackageStateServiceMsgProducer() { + TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("tb-rule-engine-ota-producer-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(coreSettings.getOtaPackageTopic()); + requestBuilder.admin(fwUpdatesAdmin); + return requestBuilder.build(); + } + @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); 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 9442049536..e82de20417 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 @@ -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; @@ -152,6 +153,11 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getUsageStatsTopic()); } + @Override + public TbQueueProducer> createToOtaPackageStateServiceMsgProducer() { + return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getOtaPackageTopic()); + } + @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 abfdb4061f..09c0640fb8 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 @@ -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; @@ -150,6 +151,11 @@ public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactor return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getUsageStatsTopic()); } + @Override + public TbQueueProducer> createToOtaPackageStateServiceMsgProducer() { + return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getOtaPackageTopic()); + } + @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 ccdd11e084..52858882cb 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,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; @@ -150,6 +151,11 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getUsageStatsTopic()); } + @Override + public TbQueueProducer> createToOtaPackageStateServiceMsgProducer() { + return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getOtaPackageTopic()); + } + @PreDestroy private void destroy() { if (coreAdmin != null) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java index e4159d3358..5730465cf2 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java @@ -69,6 +69,13 @@ public interface TbRuleEngineQueueFactory extends TbUsageStatsClientQueueFactory */ TbQueueProducer> createTbCoreNotificationsMsgProducer(); + /** + * Used to consume messages about firmware update notifications to TB Core Service + * + * @return + */ + TbQueueProducer> createToOtaPackageStateServiceMsgProducer(); + /** * Used to consume messages by TB Core Service *