diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index 38cc7b8dbd..c8f126b26c 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -119,6 +119,8 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< if (submitExecutor != null) { submitExecutor.shutdownNow(); } + + ruleEngineSettings.getQueues().forEach(config -> consumerConfigurations.put(config.getName(), config)); } @Override diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 4c4955fea6..affd7e7908 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -550,6 +550,16 @@ queue: sas_key_name: "${TB_QUEUE_SERVICE_BUS_SAS_KEY_NAME:YOUR_SAS_KEY_NAME}" sas_key: "${TB_QUEUE_SERVICE_BUS_SAS_KEY:YOUR_SAS_KEY}" max_messages: "${TB_QUEUE_SERVICE_BUS_MAX_MESSAGES:1000}" + rabbitmq: + exchange_name: "${TB_QUEUE_RABBIT_MQ_EXCHANGE_NAME:}" + host: "${TB_QUEUE_RABBIT_MQ_HOST:localhost}" + port: "${TB_QUEUE_RABBIT_MQ_PORT:5672}" + virtual_host: "${TB_QUEUE_RABBIT_MQ_VIRTUAL_HOST:/}" + username: "${TB_QUEUE_RABBIT_MQ_USERNAME:YOUR_USERNAME}" + password: "${TB_QUEUE_RABBIT_MQ_PASSWORD:YOUR_PASSWORD}" + automatic_recovery_enabled: "${TB_QUEUE_RABBIT_MQ_AUTOMATIC_RECOVERY_ENABLED:false}" + connection_timeout: "${TB_QUEUE_RABBIT_MQ_CONNECTION_TIMEOUT:60000}" + handshake_timeout: "${TB_QUEUE_RABBIT_MQ_HANDSHAKE_TIMEOUT:10000}" partitions: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}" diff --git a/common/queue/pom.xml b/common/queue/pom.xml index f55f69ece1..668da07e29 100644 --- a/common/queue/pom.xml +++ b/common/queue/pom.xml @@ -64,6 +64,10 @@ com.microsoft.azure azure-servicebus + + com.rabbitmq + amqp-client + org.springframework spring-context-support diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueMsg.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueMsg.java index c7e439ef7d..7584e8c2d0 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueMsg.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueMsg.java @@ -17,7 +17,6 @@ package org.thingsboard.server.queue.common; import lombok.Data; import org.thingsboard.server.queue.TbQueueMsg; -import org.thingsboard.server.queue.TbQueueMsgHeaders; import java.util.UUID; @@ -25,13 +24,7 @@ import java.util.UUID; public class DefaultTbQueueMsg implements TbQueueMsg { private final UUID key; private final byte[] data; - private DefaultTbQueueMsgHeaders headers; - - - public DefaultTbQueueMsg(UUID key, byte[] data) { - this.key = key; - this.data = data; - } + private final DefaultTbQueueMsgHeaders headers; public DefaultTbQueueMsg(TbQueueMsg msg) { this.key = msg.getKey(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java index 6ae9b9a33c..e8ebb934d3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java @@ -20,7 +20,7 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.SettableFuture; import lombok.Builder; import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.common.errors.InterruptException; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueConsumer; @@ -28,7 +28,6 @@ import org.thingsboard.server.queue.TbQueueMsg; import org.thingsboard.server.queue.TbQueueMsgMetadata; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueRequestTemplate; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import java.util.List; import java.util.UUID; @@ -128,10 +127,6 @@ public class DefaultTbQueueRequestTemplate> createTransportNotificationsMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, transportNotificationSettings.getNotificationsTopic()); + } + + @Override + public TbQueueProducer> createRuleEngineMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, ruleEngineSettings.getTopic()); + } + + @Override + public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, ruleEngineSettings.getTopic()); + } + + @Override + public TbQueueProducer> createTbCoreMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic()); + } + + @Override + public TbQueueProducer> createTbCoreNotificationsMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic()); + } + + @Override + public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + return new TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, ruleEngineSettings.getTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { + return new TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, + 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 TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { + return new TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, + 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 TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, transportApiSettings.getRequestsTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueProducer> createTransportApiResponseProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, transportApiSettings.getResponsesTopic()); + } +} 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 new file mode 100644 index 0000000000..3bf6d4667c --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java @@ -0,0 +1,117 @@ +/** + * 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.provider; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.msg.queue.ServiceType; +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.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.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.queue.rabbitmq.TbRabbitMqConsumerTemplate; +import org.thingsboard.server.queue.rabbitmq.TbRabbitMqProducerTemplate; +import org.thingsboard.server.queue.rabbitmq.TbRabbitMqSettings; +import org.thingsboard.server.queue.settings.TbQueueCoreSettings; +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; + +@Component +@ConditionalOnExpression("'${queue.type:null}'=='rabbitmq' && '${service.type:null}'=='tb-core'") +public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { + + private final TbRabbitMqSettings rabbitMqSettings; + private final TbQueueRuleEngineSettings ruleEngineSettings; + private final TbQueueCoreSettings coreSettings; + private final TbQueueTransportApiSettings transportApiSettings; + private final PartitionService partitionService; + private final TbServiceInfoProvider serviceInfoProvider; + private final TbQueueAdmin admin; + + public RabbitMqTbCoreQueueFactory(TbRabbitMqSettings rabbitMqSettings, + TbQueueCoreSettings coreSettings, + TbQueueTransportApiSettings transportApiSettings, + TbQueueRuleEngineSettings ruleEngineSettings, + PartitionService partitionService, + TbServiceInfoProvider serviceInfoProvider, + TbQueueAdmin admin) { + this.rabbitMqSettings = rabbitMqSettings; + this.coreSettings = coreSettings; + this.transportApiSettings = transportApiSettings; + this.ruleEngineSettings = ruleEngineSettings; + this.partitionService = partitionService; + this.serviceInfoProvider = serviceInfoProvider; + this.admin = admin; + } + + @Override + public TbQueueProducer> createTransportNotificationsMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic()); + } + + @Override + public TbQueueProducer> createRuleEngineMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic()); + } + + @Override + public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, ruleEngineSettings.getTopic()); + } + + @Override + public TbQueueProducer> createTbCoreMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic()); + } + + @Override + public TbQueueProducer> createTbCoreNotificationsMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic()); + } + + @Override + public TbQueueConsumer> createToCoreMsgConsumer() { + return new TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { + return new TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, + 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 TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, transportApiSettings.getRequestsTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueProducer> createTransportApiResponseProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic()); + } +} 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 new file mode 100644 index 0000000000..52d28cccd7 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java @@ -0,0 +1,100 @@ +/** + * 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.provider; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.msg.queue.ServiceType; +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.queue.TbQueueAdmin; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueProducer; +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.rabbitmq.TbRabbitMqConsumerTemplate; +import org.thingsboard.server.queue.rabbitmq.TbRabbitMqProducerTemplate; +import org.thingsboard.server.queue.rabbitmq.TbRabbitMqSettings; +import org.thingsboard.server.queue.settings.TbQueueCoreSettings; +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; +import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; + +@Component +@ConditionalOnExpression("'${queue.type:null}'=='rabbitmq' && '${service.type:null}'=='tb-rule-engine'") +public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { + + private final PartitionService partitionService; + private final TbQueueCoreSettings coreSettings; + private final TbServiceInfoProvider serviceInfoProvider; + private final TbQueueRuleEngineSettings ruleEngineSettings; + private final TbRabbitMqSettings rabbitMqSettings; + private final TbQueueAdmin admin; + + public RabbitMqTbRuleEngineQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, + TbQueueRuleEngineSettings ruleEngineSettings, + TbServiceInfoProvider serviceInfoProvider, + TbRabbitMqSettings rabbitMqSettings, + TbQueueAdmin admin) { + this.partitionService = partitionService; + this.coreSettings = coreSettings; + this.serviceInfoProvider = serviceInfoProvider; + this.ruleEngineSettings = ruleEngineSettings; + this.rabbitMqSettings = rabbitMqSettings; + this.admin = admin; + } + + @Override + public TbQueueProducer> createTransportNotificationsMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic()); + } + + @Override + public TbQueueProducer> createRuleEngineMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic()); + } + + @Override + public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, ruleEngineSettings.getTopic()); + } + + @Override + public TbQueueProducer> createTbCoreMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic()); + } + + @Override + public TbQueueProducer> createTbCoreNotificationsMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, coreSettings.getTopic()); + } + + @Override + public TbQueueConsumer> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { + return new TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, ruleEngineSettings.getTopic(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); + } + + @Override + public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { + return new TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, + partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); + } +} 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 new file mode 100644 index 0000000000..dc6154255c --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTransportQueueFactory.java @@ -0,0 +1,97 @@ +/** + * 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.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.ToCoreMsg; +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; +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.rabbitmq.TbRabbitMqConsumerTemplate; +import org.thingsboard.server.queue.rabbitmq.TbRabbitMqProducerTemplate; +import org.thingsboard.server.queue.rabbitmq.TbRabbitMqSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; + +@Component +@ConditionalOnExpression("'${queue.type:null}'=='rabbitmq' && ('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport')") +@Slf4j +public class RabbitMqTransportQueueFactory implements TbTransportQueueFactory { + private final TbQueueTransportApiSettings transportApiSettings; + private final TbQueueTransportNotificationSettings transportNotificationSettings; + private final TbRabbitMqSettings rabbitMqSettings; + private final TbQueueAdmin admin; + private final TbServiceInfoProvider serviceInfoProvider; + + public RabbitMqTransportQueueFactory(TbQueueTransportApiSettings transportApiSettings, + TbQueueTransportNotificationSettings transportNotificationSettings, + TbRabbitMqSettings rabbitMqSettings, + TbServiceInfoProvider serviceInfoProvider, + TbQueueAdmin admin) { + this.transportApiSettings = transportApiSettings; + this.transportNotificationSettings = transportNotificationSettings; + this.rabbitMqSettings = rabbitMqSettings; + this.admin = admin; + this.serviceInfoProvider = serviceInfoProvider; + } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> createTransportApiRequestTemplate() { + TbRabbitMqProducerTemplate> producerTemplate = + new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, transportApiSettings.getRequestsTopic()); + + TbRabbitMqConsumerTemplate> consumerTemplate = + new TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, + 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.requestTemplate(producerTemplate); + templateBuilder.responseTemplate(consumerTemplate); + templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests()); + templateBuilder.maxRequestTimeout(transportApiSettings.getMaxRequestsTimeout()); + templateBuilder.pollInterval(transportApiSettings.getResponsePollInterval()); + return templateBuilder.build(); + } + + @Override + public TbQueueProducer> createRuleEngineMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, transportApiSettings.getRequestsTopic()); + } + + @Override + public TbQueueProducer> createTbCoreMsgProducer() { + return new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, transportApiSettings.getRequestsTopic()); + } + + @Override + public TbQueueConsumer> createTransportNotificationsConsumer() { + return new TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, 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/rabbitmq/TbRabbitMqAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqAdmin.java new file mode 100644 index 0000000000..fbd678045b --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqAdmin.java @@ -0,0 +1,82 @@ +/** + * 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.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; + +@Slf4j +@Component +@ConditionalOnExpression("'${queue.type:null}'=='rabbitmq'") +public class TbRabbitMqAdmin implements TbQueueAdmin { + + private final TbRabbitMqSettings rabbitMqSettings; + private final Channel channel; + private final Connection connection; + + public TbRabbitMqAdmin(TbRabbitMqSettings rabbitMqSettings) { + this.rabbitMqSettings = rabbitMqSettings; + + try { + connection = rabbitMqSettings.getConnectionFactory().newConnection(); + } catch (IOException | TimeoutException e) { + log.error("Failed to create connection.", e); + throw new RuntimeException("Failed to create connection.", e); + } + + try { + channel = connection.createChannel(); + } catch (IOException e) { + log.error("Failed to create chanel.", e); + throw new RuntimeException("Failed to create chanel.", e); + } + } + + @Override + public void createTopicIfNotExists(String topic) { + try { + channel.queueDeclare(topic, false, false, false, null); + } catch (IOException e) { + log.error("Failed to bind queue: [{}]", topic, e); + } + } + + @PreDestroy + private void destroy() { + if (channel != null) { + try { + channel.close(); + } catch (IOException | TimeoutException e) { + log.error("Failed to close Chanel.", e); + } + } + if (connection != null) { + try { + connection.close(); + } catch (IOException e) { + log.error("Failed to close Connection.", e); + } + } + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqConsumerTemplate.java new file mode 100644 index 0000000000..25d7719163 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqConsumerTemplate.java @@ -0,0 +1,173 @@ +/** + * 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.rabbitmq; + +import com.google.gson.Gson; +import com.google.protobuf.InvalidProtocolBufferException; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.GetResponse; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.queue.TbQueueAdmin; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueMsg; +import org.thingsboard.server.queue.TbQueueMsgDecoder; +import org.thingsboard.server.queue.common.DefaultTbQueueMsg; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeoutException; +import java.util.stream.Collectors; + +@Slf4j +public class TbRabbitMqConsumerTemplate implements TbQueueConsumer { + + private final Gson gson = new Gson(); + private final TbQueueAdmin admin; + private final String topic; + private final TbQueueMsgDecoder decoder; + private final TbRabbitMqSettings rabbitMqSettings; + private final Channel channel; + private final Connection connection; + + private volatile Set partitions; + private volatile boolean subscribed; + private volatile Set queues; + private volatile boolean stopped; + + public TbRabbitMqConsumerTemplate(TbQueueAdmin admin, TbRabbitMqSettings rabbitMqSettings, String topic, TbQueueMsgDecoder decoder) { + this.admin = admin; + this.decoder = decoder; + this.topic = topic; + this.rabbitMqSettings = rabbitMqSettings; + try { + connection = rabbitMqSettings.getConnectionFactory().newConnection(); + } catch (IOException | TimeoutException e) { + log.error("Failed to create connection.", e); + throw new RuntimeException("Failed to create connection.", e); + } + + try { + channel = connection.createChannel(); + } catch (IOException e) { + log.error("Failed to create chanel.", e); + throw new RuntimeException("Failed to create chanel.", e); + } + stopped = false; + } + + @Override + public String getTopic() { + return topic; + } + + @Override + public void subscribe() { + partitions = Collections.singleton(new TopicPartitionInfo(topic, null, null, true)); + subscribed = false; + } + + @Override + public void subscribe(Set partitions) { + this.partitions = partitions; + subscribed = false; + } + + @Override + public void unsubscribe() { + stopped = true; + if (channel != null) { + try { + channel.close(); + } catch (IOException | TimeoutException e) { + log.error("Failed to close the channel."); + } + } + if (connection != null) { + try { + connection.close(); + } catch (IOException e) { + log.error("Failed to close the connection."); + } + } + } + + @Override + public List poll(long durationInMillis) { + if (!subscribed && partitions == null) { + try { + Thread.sleep(durationInMillis); + } catch (InterruptedException e) { + log.debug("Failed to await subscription", e); + } + } else { + if (!subscribed) { + queues = partitions.stream() + .map(TopicPartitionInfo::getFullTopicName) + .collect(Collectors.toSet()); + + queues.forEach(admin::createTopicIfNotExists); + subscribed = true; + } + + List result = queues.stream() + .map(queue -> { + try { + return channel.basicGet(queue, false); + } catch (IOException e) { + log.error("Failed to get messages from queue: [{}]", queue); + throw new RuntimeException("Failed to get messages from queue.", e); + } + }).filter(Objects::nonNull).map(message -> { + try { + return decode(message); + } catch (InvalidProtocolBufferException e) { + log.error("Failed to decode message: [{}].", message); + throw new RuntimeException("Failed to decode message.", e); + } + }).collect(Collectors.toList()); + if (result.size() > 0) { + return result; + } + } + try { + Thread.sleep(durationInMillis); + } catch (InterruptedException e) { + if (!stopped) { + log.error("Failed to wait.", e); + } + } + return Collections.emptyList(); + } + + @Override + public void commit() { + try { + channel.basicAck(0, true); + } catch (IOException e) { + log.error("Failed to ack messages.", e); + } + } + + public T decode(GetResponse message) throws InvalidProtocolBufferException { + DefaultTbQueueMsg msg = gson.fromJson(new String(message.getBody()), DefaultTbQueueMsg.class); + return decoder.decode(msg); + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqProducerTemplate.java new file mode 100644 index 0000000000..91b46213a5 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqProducerTemplate.java @@ -0,0 +1,113 @@ +/** + * 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.rabbitmq; + +import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.MoreExecutors; +import com.google.gson.Gson; +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.queue.TbQueueAdmin; +import org.thingsboard.server.queue.TbQueueCallback; +import org.thingsboard.server.queue.TbQueueMsg; +import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.common.DefaultTbQueueMsg; + +import java.io.IOException; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeoutException; + +@Slf4j +public class TbRabbitMqProducerTemplate implements TbQueueProducer { + private final String defaultTopic; + private final Gson gson = new Gson(); + private final TbQueueAdmin admin; + private final TbRabbitMqSettings rabbitMqSettings; + private ListeningExecutorService producerExecutor; + private final Channel channel; + private final Connection connection; + + public TbRabbitMqProducerTemplate(TbQueueAdmin admin, TbRabbitMqSettings rabbitMqSettings, String defaultTopic) { + this.admin = admin; + this.defaultTopic = defaultTopic; + this.rabbitMqSettings = rabbitMqSettings; + producerExecutor = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool()); + try { + connection = rabbitMqSettings.getConnectionFactory().newConnection(); + } catch (IOException | TimeoutException e) { + log.error("Failed to create connection.", e); + throw new RuntimeException("Failed to create connection.", e); + } + + try { + channel = connection.createChannel(); + } catch (IOException e) { + log.error("Failed to create chanel.", e); + throw new RuntimeException("Failed to create chanel.", e); + } + } + + @Override + public void init() { + + } + + @Override + public String getDefaultTopic() { + return defaultTopic; + } + + @Override + public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) { + AMQP.BasicProperties properties = new AMQP.BasicProperties(); + try { + channel.basicPublish(rabbitMqSettings.getExchangeName(), tpi.getFullTopicName(), properties, gson.toJson(new DefaultTbQueueMsg(msg)).getBytes()); + if (callback != null) { + callback.onSuccess(null); + } + } catch (IOException e) { + log.error("Failed publish message: [{}].", msg, e); + if (callback != null) { + callback.onFailure(e); + } + } + } + + @Override + public void stop() { + if (producerExecutor != null) { + producerExecutor.shutdownNow(); + } + if (channel != null) { + try { + channel.close(); + } catch (IOException | TimeoutException e) { + log.error("Failed to close the channel."); + } + } + if (connection != null) { + try { + connection.close(); + } catch (IOException e) { + log.error("Failed to close the connection."); + } + } + } + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqSettings.java new file mode 100644 index 0000000000..e0156e6dc8 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/rabbitmq/TbRabbitMqSettings.java @@ -0,0 +1,65 @@ +/** + * 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.rabbitmq; + +import com.rabbitmq.client.ConnectionFactory; +import lombok.Data; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; + +@Slf4j +@ConditionalOnExpression("'${queue.type:null}'=='rabbitmq'") +@Component +@Data +public class TbRabbitMqSettings { + @Value("${queue.rabbitmq.exchange_name:}") + private String exchangeName; + @Value("${queue.rabbitmq.host:}") + private String host; + @Value("${queue.rabbitmq.port:}") + private int port; + @Value("${queue.rabbitmq.virtual_host:}") + private String virtualHost; + @Value("${queue.rabbitmq.username:}") + private String username; + @Value("${queue.rabbitmq.password:}") + private String password; + @Value("${queue.rabbitmq.automatic_recovery_enabled:}") + private boolean automaticRecoveryEnabled; + @Value("${queue.rabbitmq.connection_timeout:}") + private int connectionTimeout; + @Value("${queue.rabbitmq.handshake_timeout:}") + private int handshakeTimeout; + + private ConnectionFactory connectionFactory; + + @PostConstruct + private void init() { + connectionFactory = new ConnectionFactory(); + connectionFactory.setHost(host); + connectionFactory.setPort(port); + connectionFactory.setVirtualHost(virtualHost); + connectionFactory.setUsername(username); + connectionFactory.setPassword(password); + connectionFactory.setAutomaticRecoveryEnabled(automaticRecoveryEnabled); + connectionFactory.setConnectionTimeout(connectionTimeout); + connectionFactory.setHandshakeTimeout(handshakeTimeout); + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java index 6eed92d296..2d85539184 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java @@ -78,7 +78,7 @@ public class TbAwsSqsProducerTemplate implements TbQueuePr public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) { SendMessageRequest sendMsgRequest = new SendMessageRequest(); sendMsgRequest.withQueueUrl(getQueueUrl(tpi.getFullTopicName())); - sendMsgRequest.withMessageBody(gson.toJson(new DefaultTbQueueMsg(msg.getKey(), msg.getData()))); + sendMsgRequest.withMessageBody(gson.toJson(new DefaultTbQueueMsg(msg))); sendMsgRequest.withMessageGroupId(msg.getKey().toString()); ListenableFuture future = producerExecutor.submit(() -> sqsClient.sendMessage(sendMsgRequest)); diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index 9b04867cfe..395f1d09e0 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -88,6 +88,16 @@ queue: sas_key_name: "${TB_QUEUE_SERVICE_BUS_SAS_KEY_NAME:YOUR_SAS_KEY_NAME}" sas_key: "${TB_QUEUE_SERVICE_BUS_SAS_KEY:YOUR_SAS_KEY}" max_messages: "${TB_QUEUE_SERVICE_BUS_MAX_MESSAGES:1000}" + rabbitmq: + exchange_name: "${TB_QUEUE_RABBIT_MQ_EXCHANGE_NAME:}" + host: "${TB_QUEUE_RABBIT_MQ_HOST:localhost}" + port: "${TB_QUEUE_RABBIT_MQ_PORT:5672}" + virtual_host: "${TB_QUEUE_RABBIT_MQ_VIRTUAL_HOST:/}" + username: "${TB_QUEUE_RABBIT_MQ_USERNAME:YOUR_USERNAME}" + password: "${TB_QUEUE_RABBIT_MQ_PASSWORD:YOUR_PASSWORD}" + automatic_recovery_enabled: "${TB_QUEUE_RABBIT_MQ_AUTOMATIC_RECOVERY_ENABLED:false}" + connection_timeout: "${TB_QUEUE_RABBIT_MQ_CONNECTION_TIMEOUT:60000}" + handshake_timeout: "${TB_QUEUE_RABBIT_MQ_HANDSHAKE_TIMEOUT:10000}" partitions: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}"