简柏林 / thingskit · Commits

Commit c7b428fbc75fd8315bd7a4b2a60e552d325313af

Authored by Andrii Shvaika
2 parents 3eaae1ef 45bd764e

Merge branch 'develop/2.5' of github.com:thingsboard/thingsboard into develop/2.5

Inline Side-by-side
Showing 28 changed files with 975 additions and 95 deletions
@@ -517,7 +517,7 @@ swagger: @@ -517,7 +517,7 @@ swagger:
517 version: "${SWAGGER_VERSION:2.0}" 517 version: "${SWAGGER_VERSION:2.0}"
518 518
519 queue: 519 queue:
520 - type: "${TB_QUEUE_TYPE:in-memory}" # kafka or in-memory or aws-sqs or pubsub 520 + type: "${TB_QUEUE_TYPE:in-memory}" # kafka or in-memory or aws-sqs or pubsub or service-bus
521 kafka: 521 kafka:
522 bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}" 522 bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}"
523 acks: "${TB_KAFKA_ACKS:all}" 523 acks: "${TB_KAFKA_ACKS:all}"
@@ -537,6 +537,11 @@ queue: @@ -537,6 +537,11 @@ queue:
537 ack_deadline: "${TB_QUEUE_PUBSUB_ACK_DEADLINE:30}" #In seconds. If messages wont commit in this time, messages will poll again 537 ack_deadline: "${TB_QUEUE_PUBSUB_ACK_DEADLINE:30}" #In seconds. If messages wont commit in this time, messages will poll again
538 max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" #in bytes 538 max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" #in bytes
539 max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}" 539 max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
  540 + service_bus:
  541 + namespace_name: "${TB_QUEUE_SERVICE_BUS_NAMESPACE_NAME:YOUR_NAMESPACE_NAME}"
  542 + sas_key_name: "${TB_QUEUE_SERVICE_BUS_SAS_KEY_NAME:YOUR_SAS_KEY_NAME}"
  543 + sas_key: "${TB_QUEUE_SERVICE_BUS_SAS_KEY:YOUR_SAS_KEY}"
  544 + max_messages: "${TB_QUEUE_SERVICE_BUS_MAX_MESSAGES:1000}"
540 partitions: 545 partitions:
541 hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" 546 hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}"
542 virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}" 547 virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}"
@@ -61,6 +61,10 @@ @@ -61,6 +61,10 @@
61 <artifactId>google-cloud-pubsub</artifactId> 61 <artifactId>google-cloud-pubsub</artifactId>
62 </dependency> 62 </dependency>
63 <dependency> 63 <dependency>
  64 + <groupId>com.microsoft.azure</groupId>
  65 + <artifactId>azure-servicebus</artifactId>
  66 + </dependency>
  67 + <dependency>
64 <groupId>org.springframework</groupId> 68 <groupId>org.springframework</groupId>
65 <artifactId>spring-context-support</artifactId> 69 <artifactId>spring-context-support</artifactId>
66 </dependency> 70 </dependency>
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.queue.azure.servicebus;
  17 +
  18 +import com.microsoft.azure.servicebus.management.ManagementClient;
  19 +import com.microsoft.azure.servicebus.primitives.ConnectionStringBuilder;
  20 +import com.microsoft.azure.servicebus.primitives.ServiceBusException;
  21 +import lombok.extern.slf4j.Slf4j;
  22 +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
  23 +import org.springframework.stereotype.Component;
  24 +import org.thingsboard.server.queue.TbQueueAdmin;
  25 +
  26 +import javax.annotation.PreDestroy;
  27 +import java.io.IOException;
  28 +import java.util.Set;
  29 +import java.util.concurrent.ConcurrentHashMap;
  30 +
  31 +@Slf4j
  32 +@Component
  33 +@ConditionalOnExpression("'${queue.type:null}'=='service-bus'")
  34 +public class TbServiceBusAdmin implements TbQueueAdmin {
  35 +
  36 + private final Set<String> queues = ConcurrentHashMap.newKeySet();
  37 +
  38 + private final ManagementClient client;
  39 +
  40 + public TbServiceBusAdmin(TbServiceBusSettings serviceBusSettings) {
  41 + ConnectionStringBuilder builder = new ConnectionStringBuilder(
  42 + serviceBusSettings.getNamespaceName(),
  43 + "queues",
  44 + serviceBusSettings.getSasKeyName(),
  45 + serviceBusSettings.getSasKey());
  46 +
  47 + client = new ManagementClient(builder);
  48 +
  49 + try {
  50 + client.getQueues().forEach(queueDescription -> queues.add(queueDescription.getPath()));
  51 + } catch (ServiceBusException | InterruptedException e) {
  52 + log.error("Failed to get queues.", e);
  53 + throw new RuntimeException("Failed to get queues.", e);
  54 + }
  55 + }
  56 +
  57 + @Override
  58 + public void createTopicIfNotExists(String topic) {
  59 + if (queues.contains(topic)) {
  60 + return;
  61 + }
  62 +
  63 + try {
  64 + client.createQueue(topic);
  65 + queues.add(topic);
  66 + } catch (ServiceBusException | InterruptedException e) {
  67 + log.error("Failed to create queue: [{}]", topic, e);
  68 + }
  69 + }
  70 +
  71 + @PreDestroy
  72 + private void destroy() {
  73 + try {
  74 + client.close();
  75 + } catch (IOException e) {
  76 + log.error("Failed to close ManagementClient.");
  77 + }
  78 + }
  79 +}
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.queue.azure.servicebus;
  17 +
  18 +import com.google.gson.Gson;
  19 +import com.google.protobuf.InvalidProtocolBufferException;
  20 +import com.microsoft.azure.servicebus.TransactionContext;
  21 +import com.microsoft.azure.servicebus.primitives.ConnectionStringBuilder;
  22 +import com.microsoft.azure.servicebus.primitives.CoreMessageReceiver;
  23 +import com.microsoft.azure.servicebus.primitives.MessageWithDeliveryTag;
  24 +import com.microsoft.azure.servicebus.primitives.MessagingEntityType;
  25 +import com.microsoft.azure.servicebus.primitives.MessagingFactory;
  26 +import com.microsoft.azure.servicebus.primitives.SettleModePair;
  27 +import lombok.extern.slf4j.Slf4j;
  28 +import org.apache.qpid.proton.amqp.messaging.Data;
  29 +import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode;
  30 +import org.apache.qpid.proton.amqp.transport.SenderSettleMode;
  31 +import org.springframework.util.CollectionUtils;
  32 +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
  33 +import org.thingsboard.server.queue.TbQueueAdmin;
  34 +import org.thingsboard.server.queue.TbQueueConsumer;
  35 +import org.thingsboard.server.queue.TbQueueMsg;
  36 +import org.thingsboard.server.queue.TbQueueMsgDecoder;
  37 +import org.thingsboard.server.queue.common.DefaultTbQueueMsg;
  38 +
  39 +import java.time.Duration;
  40 +import java.util.Collection;
  41 +import java.util.Collections;
  42 +import java.util.HashSet;
  43 +import java.util.List;
  44 +import java.util.Map;
  45 +import java.util.Set;
  46 +import java.util.concurrent.CompletableFuture;
  47 +import java.util.concurrent.ConcurrentHashMap;
  48 +import java.util.concurrent.ExecutionException;
  49 +import java.util.stream.Collectors;
  50 +import java.util.stream.Stream;
  51 +
  52 +@Slf4j
  53 +public class TbServiceBusConsumerTemplate<T extends TbQueueMsg> implements TbQueueConsumer<T> {
  54 + private final TbQueueAdmin admin;
  55 + private final String topic;
  56 + private final TbQueueMsgDecoder<T> decoder;
  57 + private final TbServiceBusSettings serviceBusSettings;
  58 +
  59 + private final Gson gson = new Gson();
  60 +
  61 + private Set<CoreMessageReceiver> receivers;
  62 + private volatile Set<TopicPartitionInfo> partitions;
  63 + private volatile boolean subscribed;
  64 + private volatile boolean stopped = false;
  65 + private Map<CoreMessageReceiver, Collection<MessageWithDeliveryTag>> pendingMessages = new ConcurrentHashMap<>();
  66 + private volatile int messagesPerQueue;
  67 +
  68 + public TbServiceBusConsumerTemplate(TbQueueAdmin admin, TbServiceBusSettings serviceBusSettings, String topic, TbQueueMsgDecoder<T> decoder) {
  69 + this.admin = admin;
  70 + this.decoder = decoder;
  71 + this.topic = topic;
  72 + this.serviceBusSettings = serviceBusSettings;
  73 + }
  74 +
  75 + @Override
  76 + public String getTopic() {
  77 + return topic;
  78 + }
  79 +
  80 + @Override
  81 + public void subscribe() {
  82 + partitions = Collections.singleton(new TopicPartitionInfo(topic, null, null, true));
  83 + subscribed = false;
  84 + }
  85 +
  86 + @Override
  87 + public void subscribe(Set<TopicPartitionInfo> partitions) {
  88 + this.partitions = partitions;
  89 + subscribed = false;
  90 + }
  91 +
  92 + @Override
  93 + public void unsubscribe() {
  94 + stopped = true;
  95 + receivers.forEach(CoreMessageReceiver::closeAsync);
  96 + }
  97 +
  98 + @Override
  99 + public List<T> poll(long durationInMillis) {
  100 + if (!subscribed && partitions == null) {
  101 + try {
  102 + Thread.sleep(durationInMillis);
  103 + } catch (InterruptedException e) {
  104 + log.debug("Failed to await subscription", e);
  105 + }
  106 + } else {
  107 + if (!subscribed) {
  108 + createReceivers();
  109 + messagesPerQueue = receivers.size() / partitions.size();
  110 + subscribed = true;
  111 + }
  112 +
  113 + List<CompletableFuture<Collection<MessageWithDeliveryTag>>> messageFutures =
  114 + receivers.stream()
  115 + .map(receiver -> receiver
  116 + .receiveAsync(messagesPerQueue, Duration.ofMillis(durationInMillis))
  117 + .whenComplete((messages, err) -> {
  118 + if (!CollectionUtils.isEmpty(messages)) {
  119 + pendingMessages.put(receiver, messages);
  120 + } else if (err != null) {
  121 + log.error("Failed to receive messages.", err);
  122 + }
  123 + }))
  124 + .collect(Collectors.toList());
  125 + try {
  126 + return fromList(messageFutures)
  127 + .get()
  128 + .stream()
  129 + .flatMap(messages -> CollectionUtils.isEmpty(messages) ? Stream.empty() : messages.stream())
  130 + .map(message -> {
  131 + try {
  132 + return decode(message);
  133 + } catch (InvalidProtocolBufferException e) {
  134 + log.error("Failed to parse message.", e);
  135 + throw new RuntimeException("Failed to parse message.", e);
  136 + }
  137 + }).collect(Collectors.toList());
  138 + } catch (InterruptedException | ExecutionException e) {
  139 + if (stopped) {
  140 + log.info("[{}] Service Bus consumer is stopped.", topic);
  141 + } else {
  142 + log.error("Failed to receive messages", e);
  143 + }
  144 + }
  145 + }
  146 + return Collections.emptyList();
  147 + }
  148 +
  149 + private void createReceivers() {
  150 + List<CompletableFuture<CoreMessageReceiver>> receiverFutures = partitions.stream()
  151 + .map(TopicPartitionInfo::getFullTopicName)
  152 + .map(queue -> {
  153 + MessagingFactory factory;
  154 + try {
  155 + factory = MessagingFactory.createFromConnectionStringBuilder(createConnection(queue));
  156 + } catch (InterruptedException | ExecutionException e) {
  157 + log.error("Failed to create factory for the queue [{}]", queue);
  158 + throw new RuntimeException("Failed to create the factory", e);
  159 + }
  160 +
  161 + return CoreMessageReceiver.create(factory, queue, queue, 0,
  162 + new SettleModePair(SenderSettleMode.UNSETTLED, ReceiverSettleMode.SECOND),
  163 + MessagingEntityType.QUEUE);
  164 + }).collect(Collectors.toList());
  165 +
  166 + try {
  167 + receivers = new HashSet<>(fromList(receiverFutures).get());
  168 + } catch (InterruptedException | ExecutionException e) {
  169 + if (stopped) {
  170 + log.info("[{}] Service Bus consumer is stopped.", topic);
  171 + } else {
  172 + log.error("Failed to create receivers", e);
  173 + }
  174 + }
  175 + }
  176 +
  177 + private ConnectionStringBuilder createConnection(String queue) {
  178 + admin.createTopicIfNotExists(queue);
  179 + return new ConnectionStringBuilder(
  180 + serviceBusSettings.getNamespaceName(),
  181 + queue,
  182 + serviceBusSettings.getSasKeyName(),
  183 + serviceBusSettings.getSasKey());
  184 + }
  185 +
  186 + private <V> CompletableFuture<List<V>> fromList(List<CompletableFuture<V>> futures) {
  187 + CompletableFuture<Collection<V>>[] arrayFuture = new CompletableFuture[futures.size()];
  188 + futures.toArray(arrayFuture);
  189 +
  190 + return CompletableFuture
  191 + .allOf(arrayFuture)
  192 + .thenApply(v -> futures
  193 + .stream()
  194 + .map(CompletableFuture::join)
  195 + .collect(Collectors.toList()));
  196 + }
  197 +
  198 + @Override
  199 + public void commit() {
  200 + pendingMessages.forEach((receiver, msgs) ->
  201 + msgs.forEach(msg -> receiver.completeMessageAsync(msg.getDeliveryTag(), TransactionContext.NULL_TXN)));
  202 + pendingMessages.clear();
  203 + }
  204 +
  205 + private T decode(MessageWithDeliveryTag data) throws InvalidProtocolBufferException {
  206 + DefaultTbQueueMsg msg = gson.fromJson(new String(((Data) data.getMessage().getBody()).getValue().getArray()), DefaultTbQueueMsg.class);
  207 + return decoder.decode(msg);
  208 + }
  209 +
  210 +}
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.queue.azure.servicebus;
  17 +
  18 +import com.google.gson.Gson;
  19 +import com.microsoft.azure.servicebus.IMessage;
  20 +import com.microsoft.azure.servicebus.Message;
  21 +import com.microsoft.azure.servicebus.QueueClient;
  22 +import com.microsoft.azure.servicebus.ReceiveMode;
  23 +import com.microsoft.azure.servicebus.primitives.ConnectionStringBuilder;
  24 +import com.microsoft.azure.servicebus.primitives.ServiceBusException;
  25 +import lombok.extern.slf4j.Slf4j;
  26 +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
  27 +import org.thingsboard.server.queue.TbQueueAdmin;
  28 +import org.thingsboard.server.queue.TbQueueCallback;
  29 +import org.thingsboard.server.queue.TbQueueMsg;
  30 +import org.thingsboard.server.queue.TbQueueProducer;
  31 +import org.thingsboard.server.queue.common.DefaultTbQueueMsg;
  32 +
  33 +import java.util.HashMap;
  34 +import java.util.Map;
  35 +import java.util.concurrent.CompletableFuture;
  36 +import java.util.concurrent.ExecutorService;
  37 +import java.util.concurrent.Executors;
  38 +
  39 +@Slf4j
  40 +public class TbServiceBusProducerTemplate<T extends TbQueueMsg> implements TbQueueProducer<T> {
  41 + private final String defaultTopic;
  42 + private final Gson gson = new Gson();
  43 + private final TbQueueAdmin admin;
  44 + private final TbServiceBusSettings serviceBusSettings;
  45 + private final Map<String, QueueClient> clients = new HashMap<>();
  46 + private ExecutorService executorService;
  47 +
  48 + public TbServiceBusProducerTemplate(TbQueueAdmin admin, TbServiceBusSettings serviceBusSettings, String defaultTopic) {
  49 + this.admin = admin;
  50 + this.defaultTopic = defaultTopic;
  51 + this.serviceBusSettings = serviceBusSettings;
  52 + executorService = Executors.newSingleThreadExecutor();
  53 + }
  54 +
  55 + @Override
  56 + public void init() {
  57 +
  58 + }
  59 +
  60 + @Override
  61 + public String getDefaultTopic() {
  62 + return defaultTopic;
  63 + }
  64 +
  65 + @Override
  66 + public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) {
  67 + IMessage message = new Message(gson.toJson(new DefaultTbQueueMsg(msg)));
  68 + CompletableFuture<Void> future = getClient(tpi.getFullTopicName()).sendAsync(message);
  69 + future.whenCompleteAsync((success, err) -> {
  70 + if (err != null) {
  71 + callback.onFailure(err);
  72 + } else {
  73 + callback.onSuccess(null);
  74 + }
  75 + }, executorService);
  76 + }
  77 +
  78 + @Override
  79 + public void stop() {
  80 + clients.forEach((t, client) -> {
  81 + try {
  82 + client.close();
  83 + } catch (ServiceBusException e) {
  84 + log.error("Failed to close QueueClient.", e);
  85 + }
  86 + });
  87 +
  88 + if (executorService != null) {
  89 + executorService.shutdownNow();
  90 + }
  91 + }
  92 +
  93 + private QueueClient getClient(String topic) {
  94 + return clients.computeIfAbsent(topic, k -> {
  95 + admin.createTopicIfNotExists(topic);
  96 + ConnectionStringBuilder builder =
  97 + new ConnectionStringBuilder(
  98 + serviceBusSettings.getNamespaceName(),
  99 + topic,
  100 + serviceBusSettings.getSasKeyName(),
  101 + serviceBusSettings.getSasKey());
  102 + try {
  103 + return new QueueClient(builder, ReceiveMode.PEEKLOCK);
  104 + } catch (InterruptedException | ServiceBusException e) {
  105 + log.error("Failed to create new client for the Queue: [{}]", topic, e);
  106 + throw new RuntimeException("Failed to create new client for the Queue", e);
  107 + }
  108 + });
  109 + }
  110 +}
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.queue.azure.servicebus;
  17 +
  18 +import lombok.Data;
  19 +import lombok.extern.slf4j.Slf4j;
  20 +import org.springframework.beans.factory.annotation.Value;
  21 +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
  22 +import org.springframework.stereotype.Component;
  23 +
  24 +@Slf4j
  25 +@ConditionalOnExpression("'${queue.type:null}'=='service-bus'")
  26 +@Component
  27 +@Data
  28 +public class TbServiceBusSettings {
  29 + @Value("${queue.service_bus.namespace_name}")
  30 + private String namespaceName;
  31 + @Value("${queue.service_bus.sas_key_name}")
  32 + private String sasKeyName;
  33 + @Value("${queue.service_bus.sas_key}")
  34 + private String sasKey;
  35 + @Value("${queue.service_bus.max_messages}")
  36 + private int maxMessages;
  37 +}
@@ -15,7 +15,6 @@ @@ -15,7 +15,6 @@
15 */ 15 */
16 package org.thingsboard.server.queue.common; 16 package org.thingsboard.server.queue.common;
17 17
18 -import com.google.gson.annotations.Expose;  
19 import lombok.Data; 18 import lombok.Data;
20 import org.thingsboard.server.queue.TbQueueMsg; 19 import org.thingsboard.server.queue.TbQueueMsg;
21 import org.thingsboard.server.queue.TbQueueMsgHeaders; 20 import org.thingsboard.server.queue.TbQueueMsgHeaders;
@@ -26,12 +25,20 @@ import java.util.UUID; @@ -26,12 +25,20 @@ import java.util.UUID;
26 public class DefaultTbQueueMsg implements TbQueueMsg { 25 public class DefaultTbQueueMsg implements TbQueueMsg {
27 private final UUID key; 26 private final UUID key;
28 private final byte[] data; 27 private final byte[] data;
  28 + private DefaultTbQueueMsgHeaders headers;
  29 +
29 30
30 public DefaultTbQueueMsg(UUID key, byte[] data) { 31 public DefaultTbQueueMsg(UUID key, byte[] data) {
31 this.key = key; 32 this.key = key;
32 this.data = data; 33 this.data = data;
33 } 34 }
34 35
35 - @Expose(serialize = false, deserialize = false)  
36 - private TbQueueMsgHeaders headers; 36 + public DefaultTbQueueMsg(TbQueueMsg msg) {
  37 + this.key = msg.getKey();
  38 + this.data = msg.getData();
  39 + DefaultTbQueueMsgHeaders headers = new DefaultTbQueueMsgHeaders();
  40 + msg.getHeaders().getData().forEach(headers::put);
  41 + this.headers = headers;
  42 + }
  43 +
37 } 44 }
@@ -36,7 +36,6 @@ import java.nio.charset.StandardCharsets; @@ -36,7 +36,6 @@ import java.nio.charset.StandardCharsets;
36 import java.util.ArrayList; 36 import java.util.ArrayList;
37 import java.util.HashMap; 37 import java.util.HashMap;
38 import java.util.HashSet; 38 import java.util.HashSet;
39 -import java.util.LinkedHashSet;  
40 import java.util.List; 39 import java.util.List;
41 import java.util.Map; 40 import java.util.Map;
42 import java.util.Set; 41 import java.util.Set;
@@ -21,14 +21,14 @@ import org.thingsboard.server.common.msg.queue.ServiceType; @@ -21,14 +21,14 @@ import org.thingsboard.server.common.msg.queue.ServiceType;
21 import org.thingsboard.server.gen.transport.TransportProtos; 21 import org.thingsboard.server.gen.transport.TransportProtos;
22 import org.thingsboard.server.queue.TbQueueAdmin; 22 import org.thingsboard.server.queue.TbQueueAdmin;
23 import org.thingsboard.server.queue.TbQueueConsumer; 23 import org.thingsboard.server.queue.TbQueueConsumer;
24 -import org.thingsboard.server.queue.settings.TbQueueCoreSettings;  
25 import org.thingsboard.server.queue.TbQueueProducer; 24 import org.thingsboard.server.queue.TbQueueProducer;
26 -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;  
27 -import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;  
28 -import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;  
29 import org.thingsboard.server.queue.common.TbProtoQueueMsg; 25 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
30 import org.thingsboard.server.queue.discovery.PartitionService; 26 import org.thingsboard.server.queue.discovery.PartitionService;
31 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; 27 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
  28 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  29 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
  30 +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
  31 +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;
32 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; 32 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
33 import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; 33 import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin;
34 import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate; 34 import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate;
@@ -18,21 +18,22 @@ package org.thingsboard.server.queue.provider; @@ -18,21 +18,22 @@ package org.thingsboard.server.queue.provider;
18 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; 18 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
19 import org.springframework.stereotype.Component; 19 import org.springframework.stereotype.Component;
20 import org.thingsboard.server.common.msg.queue.ServiceType; 20 import org.thingsboard.server.common.msg.queue.ServiceType;
21 -import org.thingsboard.server.gen.transport.TransportProtos;  
22 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; 21 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
  22 +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
23 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; 23 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
  24 +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg;
24 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; 25 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
25 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; 26 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
26 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; 27 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
27 import org.thingsboard.server.queue.TbQueueAdmin; 28 import org.thingsboard.server.queue.TbQueueAdmin;
28 import org.thingsboard.server.queue.TbQueueConsumer; 29 import org.thingsboard.server.queue.TbQueueConsumer;
29 -import org.thingsboard.server.queue.settings.TbQueueCoreSettings;  
30 import org.thingsboard.server.queue.TbQueueProducer; 30 import org.thingsboard.server.queue.TbQueueProducer;
31 -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;  
32 -import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;  
33 import org.thingsboard.server.queue.common.TbProtoQueueMsg; 31 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
34 import org.thingsboard.server.queue.discovery.PartitionService; 32 import org.thingsboard.server.queue.discovery.PartitionService;
35 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; 33 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
  34 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  35 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
  36 +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
36 import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; 37 import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin;
37 import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate; 38 import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate;
38 import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate; 39 import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate;
@@ -48,8 +49,6 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { @@ -48,8 +49,6 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory {
48 private final TbQueueTransportApiSettings transportApiSettings; 49 private final TbQueueTransportApiSettings transportApiSettings;
49 private final PartitionService partitionService; 50 private final PartitionService partitionService;
50 private final TbServiceInfoProvider serviceInfoProvider; 51 private final TbServiceInfoProvider serviceInfoProvider;
51 -  
52 -  
53 private final TbQueueAdmin admin; 52 private final TbQueueAdmin admin;
54 53
55 public AwsSqsTbCoreQueueFactory(TbAwsSqsSettings sqsSettings, 54 public AwsSqsTbCoreQueueFactory(TbAwsSqsSettings sqsSettings,
@@ -78,7 +77,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { @@ -78,7 +77,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory {
78 } 77 }
79 78
80 @Override 79 @Override
81 - public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> createRuleEngineNotificationsMsgProducer() { 80 + public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> createRuleEngineNotificationsMsgProducer() {
82 return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic()); 81 return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic());
83 } 82 }
84 83
@@ -88,7 +87,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { @@ -88,7 +87,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory {
88 } 87 }
89 88
90 @Override 89 @Override
91 - public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> createTbCoreNotificationsMsgProducer() { 90 + public TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> createTbCoreNotificationsMsgProducer() {
92 return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); 91 return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
93 } 92 }
94 93
@@ -99,15 +98,16 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { @@ -99,15 +98,16 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory {
99 } 98 }
100 99
101 @Override 100 @Override
102 - public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> createToCoreNotificationsMsgConsumer() { 101 + public TbQueueConsumer<TbProtoQueueMsg<ToCoreNotificationMsg>> createToCoreNotificationsMsgConsumer() {
103 return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, 102 return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings,
104 partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), 103 partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(),
105 - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); 104 + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
106 } 105 }
107 106
108 @Override 107 @Override
109 public TbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> createTransportApiRequestConsumer() { 108 public TbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> createTransportApiRequestConsumer() {
110 - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); 109 + return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic(),
  110 + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders()));
111 } 111 }
112 112
113 @Override 113 @Override
@@ -24,12 +24,12 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -24,12 +24,12 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
24 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; 24 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
25 import org.thingsboard.server.queue.TbQueueAdmin; 25 import org.thingsboard.server.queue.TbQueueAdmin;
26 import org.thingsboard.server.queue.TbQueueConsumer; 26 import org.thingsboard.server.queue.TbQueueConsumer;
27 -import org.thingsboard.server.queue.settings.TbQueueCoreSettings;  
28 import org.thingsboard.server.queue.TbQueueProducer; 27 import org.thingsboard.server.queue.TbQueueProducer;
29 -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;  
30 import org.thingsboard.server.queue.common.TbProtoQueueMsg; 28 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
31 import org.thingsboard.server.queue.discovery.PartitionService; 29 import org.thingsboard.server.queue.discovery.PartitionService;
32 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; 30 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
  31 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  32 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
33 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; 33 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
34 import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; 34 import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin;
35 import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate; 35 import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate;
@@ -86,7 +86,8 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory @@ -86,7 +86,8 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory
86 86
87 @Override 87 @Override
88 public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) { 88 public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) {
89 - return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); 89 + return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic(),
  90 + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders()));
90 } 91 }
91 92
92 @Override 93 @Override
@@ -18,16 +18,20 @@ package org.thingsboard.server.queue.provider; @@ -18,16 +18,20 @@ package org.thingsboard.server.queue.provider;
18 import lombok.extern.slf4j.Slf4j; 18 import lombok.extern.slf4j.Slf4j;
19 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; 19 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
20 import org.springframework.stereotype.Component; 20 import org.springframework.stereotype.Component;
21 -import org.thingsboard.server.gen.transport.TransportProtos; 21 +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
  22 +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
  23 +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
  24 +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
  25 +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
22 import org.thingsboard.server.queue.TbQueueAdmin; 26 import org.thingsboard.server.queue.TbQueueAdmin;
23 import org.thingsboard.server.queue.TbQueueConsumer; 27 import org.thingsboard.server.queue.TbQueueConsumer;
24 import org.thingsboard.server.queue.TbQueueProducer; 28 import org.thingsboard.server.queue.TbQueueProducer;
25 import org.thingsboard.server.queue.TbQueueRequestTemplate; 29 import org.thingsboard.server.queue.TbQueueRequestTemplate;
26 -import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;  
27 -import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;  
28 import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; 30 import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate;
29 import org.thingsboard.server.queue.common.TbProtoQueueMsg; 31 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
30 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; 32 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
  33 +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
  34 +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;
31 import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; 35 import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin;
32 import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate; 36 import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate;
33 import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate; 37 import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate;
@@ -55,17 +59,17 @@ public class AwsSqsTransportQueueFactory implements TbTransportQueueFactory { @@ -55,17 +59,17 @@ public class AwsSqsTransportQueueFactory implements TbTransportQueueFactory {
55 } 59 }
56 60
57 @Override 61 @Override
58 - public TbQueueRequestTemplate<TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>, TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> createTransportApiRequestTemplate() {  
59 - TbAwsSqsProducerTemplate<TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>> producerTemplate = 62 + public TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> createTransportApiRequestTemplate() {
  63 + TbAwsSqsProducerTemplate<TbProtoQueueMsg<TransportApiRequestMsg>> producerTemplate =
60 new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic()); 64 new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic());
61 65
62 - TbAwsSqsConsumerTemplate<TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> consumerTemplate = 66 + TbAwsSqsConsumerTemplate<TbProtoQueueMsg<TransportApiResponseMsg>> consumerTemplate =
63 new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, 67 new TbAwsSqsConsumerTemplate<>(admin, sqsSettings,
64 transportApiSettings.getResponsesTopic() + "_" + serviceInfoProvider.getServiceId(), 68 transportApiSettings.getResponsesTopic() + "_" + serviceInfoProvider.getServiceId(),
65 - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); 69 + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders()));
66 70
67 DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder 71 DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder
68 - <TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>, TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> templateBuilder = DefaultTbQueueRequestTemplate.builder(); 72 + <TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> templateBuilder = DefaultTbQueueRequestTemplate.builder();
69 templateBuilder.queueAdmin(admin); 73 templateBuilder.queueAdmin(admin);
70 templateBuilder.requestTemplate(producerTemplate); 74 templateBuilder.requestTemplate(producerTemplate);
71 templateBuilder.responseTemplate(consumerTemplate); 75 templateBuilder.responseTemplate(consumerTemplate);
@@ -76,18 +80,18 @@ public class AwsSqsTransportQueueFactory implements TbTransportQueueFactory { @@ -76,18 +80,18 @@ public class AwsSqsTransportQueueFactory implements TbTransportQueueFactory {
76 } 80 }
77 81
78 @Override 82 @Override
79 - public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> createRuleEngineMsgProducer() { 83 + public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> createRuleEngineMsgProducer() {
80 return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic()); 84 return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic());
81 } 85 }
82 86
83 @Override 87 @Override
84 - public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> createTbCoreMsgProducer() { 88 + public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> createTbCoreMsgProducer() {
85 return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic()); 89 return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic());
86 } 90 }
87 91
88 @Override 92 @Override
89 - public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> createTransportNotificationsConsumer() { 93 + public TbQueueConsumer<TbProtoQueueMsg<ToTransportMsg>> createTransportNotificationsConsumer() {
90 return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, transportNotificationSettings.getNotificationsTopic() + "_" + serviceInfoProvider.getServiceId(), 94 return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, transportNotificationSettings.getNotificationsTopic() + "_" + serviceInfoProvider.getServiceId(),
91 - msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); 95 + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders()));
92 } 96 }
93 } 97 }
@@ -26,14 +26,14 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; @@ -26,14 +26,14 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
26 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; 26 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
27 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; 27 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
28 import org.thingsboard.server.queue.TbQueueConsumer; 28 import org.thingsboard.server.queue.TbQueueConsumer;
29 -import org.thingsboard.server.queue.settings.TbQueueCoreSettings;  
30 import org.thingsboard.server.queue.TbQueueProducer; 29 import org.thingsboard.server.queue.TbQueueProducer;
31 -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;  
32 -import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;  
33 -import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;  
34 import org.thingsboard.server.queue.common.TbProtoQueueMsg; 30 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
35 import org.thingsboard.server.queue.memory.InMemoryTbQueueConsumer; 31 import org.thingsboard.server.queue.memory.InMemoryTbQueueConsumer;
36 import org.thingsboard.server.queue.memory.InMemoryTbQueueProducer; 32 import org.thingsboard.server.queue.memory.InMemoryTbQueueProducer;
  33 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  34 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
  35 +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
  36 +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;
37 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; 37 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
38 38
39 @Slf4j 39 @Slf4j
@@ -19,22 +19,22 @@ import com.google.common.util.concurrent.Futures; @@ -19,22 +19,22 @@ import com.google.common.util.concurrent.Futures;
19 import lombok.extern.slf4j.Slf4j; 19 import lombok.extern.slf4j.Slf4j;
20 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; 20 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
21 import org.springframework.stereotype.Component; 21 import org.springframework.stereotype.Component;
22 -import org.thingsboard.server.queue.TbQueueConsumer;  
23 -import org.thingsboard.server.queue.settings.TbQueueCoreSettings;  
24 -import org.thingsboard.server.queue.TbQueueProducer;  
25 -import org.thingsboard.server.queue.TbQueueRequestTemplate;  
26 -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;  
27 -import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;  
28 -import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;  
29 -import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate;  
30 -import org.thingsboard.server.queue.common.TbProtoQueueMsg;  
31 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; 22 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
32 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; 23 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
33 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; 24 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
34 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; 25 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
35 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; 26 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
  27 +import org.thingsboard.server.queue.TbQueueConsumer;
  28 +import org.thingsboard.server.queue.TbQueueProducer;
  29 +import org.thingsboard.server.queue.TbQueueRequestTemplate;
  30 +import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate;
  31 +import org.thingsboard.server.queue.common.TbProtoQueueMsg;
36 import org.thingsboard.server.queue.memory.InMemoryTbQueueConsumer; 32 import org.thingsboard.server.queue.memory.InMemoryTbQueueConsumer;
37 import org.thingsboard.server.queue.memory.InMemoryTbQueueProducer; 33 import org.thingsboard.server.queue.memory.InMemoryTbQueueProducer;
  34 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  35 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
  36 +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
  37 +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;
38 38
39 @Component 39 @Component
40 @ConditionalOnExpression("'${queue.type:null}'=='in-memory' && ('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport')") 40 @ConditionalOnExpression("'${queue.type:null}'=='in-memory' && ('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport')")
@@ -26,17 +26,17 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; @@ -26,17 +26,17 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
26 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; 26 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
27 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; 27 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
28 import org.thingsboard.server.queue.TbQueueConsumer; 28 import org.thingsboard.server.queue.TbQueueConsumer;
29 -import org.thingsboard.server.queue.settings.TbQueueCoreSettings;  
30 import org.thingsboard.server.queue.TbQueueProducer; 29 import org.thingsboard.server.queue.TbQueueProducer;
31 -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;  
32 -import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;  
33 -import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;  
34 import org.thingsboard.server.queue.common.TbProtoQueueMsg; 30 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
35 import org.thingsboard.server.queue.discovery.PartitionService; 31 import org.thingsboard.server.queue.discovery.PartitionService;
36 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; 32 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
37 import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; 33 import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate;
38 import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; 34 import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate;
39 import org.thingsboard.server.queue.kafka.TbKafkaSettings; 35 import org.thingsboard.server.queue.kafka.TbKafkaSettings;
  36 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  37 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
  38 +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
  39 +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;
40 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; 40 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
41 41
42 @Component 42 @Component
@@ -26,16 +26,16 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; @@ -26,16 +26,16 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
26 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; 26 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
27 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; 27 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
28 import org.thingsboard.server.queue.TbQueueConsumer; 28 import org.thingsboard.server.queue.TbQueueConsumer;
29 -import org.thingsboard.server.queue.settings.TbQueueCoreSettings;  
30 import org.thingsboard.server.queue.TbQueueProducer; 29 import org.thingsboard.server.queue.TbQueueProducer;
31 -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;  
32 -import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;  
33 import org.thingsboard.server.queue.common.TbProtoQueueMsg; 30 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
34 import org.thingsboard.server.queue.discovery.PartitionService; 31 import org.thingsboard.server.queue.discovery.PartitionService;
35 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; 32 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
36 import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; 33 import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate;
37 import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; 34 import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate;
38 import org.thingsboard.server.queue.kafka.TbKafkaSettings; 35 import org.thingsboard.server.queue.kafka.TbKafkaSettings;
  36 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  37 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
  38 +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
39 39
40 @Component 40 @Component
41 @ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='tb-core'") 41 @ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='tb-core'")
@@ -24,15 +24,15 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -24,15 +24,15 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
24 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; 24 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg;
25 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; 25 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
26 import org.thingsboard.server.queue.TbQueueConsumer; 26 import org.thingsboard.server.queue.TbQueueConsumer;
27 -import org.thingsboard.server.queue.settings.TbQueueCoreSettings;  
28 import org.thingsboard.server.queue.TbQueueProducer; 27 import org.thingsboard.server.queue.TbQueueProducer;
29 -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;  
30 import org.thingsboard.server.queue.common.TbProtoQueueMsg; 28 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
31 import org.thingsboard.server.queue.discovery.PartitionService; 29 import org.thingsboard.server.queue.discovery.PartitionService;
32 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; 30 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
33 import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; 31 import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate;
34 import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; 32 import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate;
35 import org.thingsboard.server.queue.kafka.TbKafkaSettings; 33 import org.thingsboard.server.queue.kafka.TbKafkaSettings;
  34 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  35 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
36 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; 36 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
37 37
38 @Component 38 @Component
@@ -27,11 +27,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestM @@ -27,11 +27,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestM
27 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; 27 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
28 import org.thingsboard.server.queue.TbQueueAdmin; 28 import org.thingsboard.server.queue.TbQueueAdmin;
29 import org.thingsboard.server.queue.TbQueueConsumer; 29 import org.thingsboard.server.queue.TbQueueConsumer;
30 -import org.thingsboard.server.queue.settings.TbQueueCoreSettings;  
31 import org.thingsboard.server.queue.TbQueueProducer; 30 import org.thingsboard.server.queue.TbQueueProducer;
32 -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;  
33 -import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;  
34 -import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;  
35 import org.thingsboard.server.queue.common.TbProtoQueueMsg; 31 import org.thingsboard.server.queue.common.TbProtoQueueMsg;
36 import org.thingsboard.server.queue.discovery.PartitionService; 32 import org.thingsboard.server.queue.discovery.PartitionService;
37 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; 33 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
@@ -39,6 +35,10 @@ import org.thingsboard.server.queue.pubsub.TbPubSubAdmin; @@ -39,6 +35,10 @@ import org.thingsboard.server.queue.pubsub.TbPubSubAdmin;
39 import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate; 35 import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate;
40 import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; 36 import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate;
41 import org.thingsboard.server.queue.pubsub.TbPubSubSettings; 37 import org.thingsboard.server.queue.pubsub.TbPubSubSettings;
  38 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  39 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
  40 +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
  41 +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;
42 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; 42 import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
43 43
44 @Component 44 @Component
@@ -54,8 +54,6 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @@ -54,8 +54,6 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng
54 private final PartitionService partitionService; 54 private final PartitionService partitionService;
55 private final TbServiceInfoProvider serviceInfoProvider; 55 private final TbServiceInfoProvider serviceInfoProvider;
56 56
57 - private TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> tbCoreProducer;  
58 -  
59 public PubSubMonolithQueueFactory(TbPubSubSettings pubSubSettings, 57 public PubSubMonolithQueueFactory(TbPubSubSettings pubSubSettings,
60 TbQueueCoreSettings coreSettings, 58 TbQueueCoreSettings coreSettings,
61 TbQueueRuleEngineSettings ruleEngineSettings, 59 TbQueueRuleEngineSettings ruleEngineSettings,
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.queue.provider;
  17 +
  18 +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
  19 +import org.springframework.stereotype.Component;
  20 +import org.thingsboard.server.common.msg.queue.ServiceType;
  21 +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
  22 +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
  23 +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
  24 +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg;
  25 +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
  26 +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
  27 +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
  28 +import org.thingsboard.server.queue.TbQueueAdmin;
  29 +import org.thingsboard.server.queue.TbQueueConsumer;
  30 +import org.thingsboard.server.queue.TbQueueProducer;
  31 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusConsumerTemplate;
  32 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusProducerTemplate;
  33 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings;
  34 +import org.thingsboard.server.queue.common.TbProtoQueueMsg;
  35 +import org.thingsboard.server.queue.discovery.PartitionService;
  36 +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
  37 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  38 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
  39 +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
  40 +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;
  41 +import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
  42 +
  43 +@Component
  44 +@ConditionalOnExpression("'${queue.type:null}'=='service-bus' && '${service.type:null}'=='monolith'")
  45 +public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngineQueueFactory {
  46 +
  47 + private final PartitionService partitionService;
  48 + private final TbQueueCoreSettings coreSettings;
  49 + private final TbServiceInfoProvider serviceInfoProvider;
  50 + private final TbQueueRuleEngineSettings ruleEngineSettings;
  51 + private final TbQueueTransportApiSettings transportApiSettings;
  52 + private final TbQueueTransportNotificationSettings transportNotificationSettings;
  53 + private final TbServiceBusSettings serviceBusSettings;
  54 + private final TbQueueAdmin admin;
  55 +
  56 + public ServiceBusMonolithQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings,
  57 + TbQueueRuleEngineSettings ruleEngineSettings,
  58 + TbServiceInfoProvider serviceInfoProvider,
  59 + TbQueueTransportApiSettings transportApiSettings,
  60 + TbQueueTransportNotificationSettings transportNotificationSettings,
  61 + TbServiceBusSettings serviceBusSettings,
  62 + TbQueueAdmin admin) {
  63 + this.partitionService = partitionService;
  64 + this.coreSettings = coreSettings;
  65 + this.serviceInfoProvider = serviceInfoProvider;
  66 + this.ruleEngineSettings = ruleEngineSettings;
  67 + this.transportApiSettings = transportApiSettings;
  68 + this.transportNotificationSettings = transportNotificationSettings;
  69 + this.serviceBusSettings = serviceBusSettings;
  70 + this.admin = admin;
  71 + }
  72 +
  73 + @Override
  74 + public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> createTransportNotificationsMsgProducer() {
  75 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, transportNotificationSettings.getNotificationsTopic());
  76 + }
  77 +
  78 + @Override
  79 + public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> createRuleEngineMsgProducer() {
  80 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, ruleEngineSettings.getTopic());
  81 + }
  82 +
  83 + @Override
  84 + public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> createRuleEngineNotificationsMsgProducer() {
  85 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, ruleEngineSettings.getTopic());
  86 + }
  87 +
  88 + @Override
  89 + public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> createTbCoreMsgProducer() {
  90 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic());
  91 + }
  92 +
  93 + @Override
  94 + public TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> createTbCoreNotificationsMsgProducer() {
  95 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic());
  96 + }
  97 +
  98 + @Override
  99 + public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) {
  100 + return new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings, ruleEngineSettings.getTopic(),
  101 + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders()));
  102 + }
  103 +
  104 + @Override
  105 + public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> createToRuleEngineNotificationsMsgConsumer() {
  106 + return new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings,
  107 + partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(),
  108 + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
  109 + }
  110 +
  111 + @Override
  112 + public TbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> createToCoreMsgConsumer() {
  113 + return new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic(),
  114 + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders()));
  115 + }
  116 +
  117 + @Override
  118 + public TbQueueConsumer<TbProtoQueueMsg<ToCoreNotificationMsg>> createToCoreNotificationsMsgConsumer() {
  119 + return new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings,
  120 + partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(),
  121 + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
  122 + }
  123 +
  124 + @Override
  125 + public TbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> createTransportApiRequestConsumer() {
  126 + return new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings, transportApiSettings.getRequestsTopic(),
  127 + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders()));
  128 + }
  129 +
  130 + @Override
  131 + public TbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> createTransportApiResponseProducer() {
  132 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, transportApiSettings.getResponsesTopic());
  133 + }
  134 +}
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.queue.provider;
  17 +
  18 +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
  19 +import org.springframework.stereotype.Component;
  20 +import org.thingsboard.server.common.msg.queue.ServiceType;
  21 +import org.thingsboard.server.gen.transport.TransportProtos;
  22 +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
  23 +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
  24 +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
  25 +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
  26 +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
  27 +import org.thingsboard.server.queue.TbQueueAdmin;
  28 +import org.thingsboard.server.queue.TbQueueConsumer;
  29 +import org.thingsboard.server.queue.TbQueueProducer;
  30 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusConsumerTemplate;
  31 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusProducerTemplate;
  32 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings;
  33 +import org.thingsboard.server.queue.common.TbProtoQueueMsg;
  34 +import org.thingsboard.server.queue.discovery.PartitionService;
  35 +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
  36 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  37 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
  38 +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
  39 +
  40 +@Component
  41 +@ConditionalOnExpression("'${queue.type:null}'=='service-bus' && '${service.type:null}'=='tb-core'")
  42 +public class ServiceBusTbCoreQueueProvider implements TbCoreQueueFactory {
  43 +
  44 + private final TbServiceBusSettings serviceBusSettings;
  45 + private final TbQueueRuleEngineSettings ruleEngineSettings;
  46 + private final TbQueueCoreSettings coreSettings;
  47 + private final TbQueueTransportApiSettings transportApiSettings;
  48 + private final PartitionService partitionService;
  49 + private final TbServiceInfoProvider serviceInfoProvider;
  50 + private final TbQueueAdmin admin;
  51 +
  52 + public ServiceBusTbCoreQueueProvider(TbServiceBusSettings serviceBusSettings,
  53 + TbQueueCoreSettings coreSettings,
  54 + TbQueueTransportApiSettings transportApiSettings,
  55 + TbQueueRuleEngineSettings ruleEngineSettings,
  56 + PartitionService partitionService,
  57 + TbServiceInfoProvider serviceInfoProvider,
  58 + TbQueueAdmin admin) {
  59 + this.serviceBusSettings = serviceBusSettings;
  60 + this.coreSettings = coreSettings;
  61 + this.transportApiSettings = transportApiSettings;
  62 + this.ruleEngineSettings = ruleEngineSettings;
  63 + this.partitionService = partitionService;
  64 + this.serviceInfoProvider = serviceInfoProvider;
  65 + this.admin = admin;
  66 + }
  67 +
  68 + @Override
  69 + public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> createTransportNotificationsMsgProducer() {
  70 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic());
  71 + }
  72 +
  73 + @Override
  74 + public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> createRuleEngineMsgProducer() {
  75 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic());
  76 + }
  77 +
  78 + @Override
  79 + public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> createRuleEngineNotificationsMsgProducer() {
  80 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, ruleEngineSettings.getTopic());
  81 + }
  82 +
  83 + @Override
  84 + public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> createTbCoreMsgProducer() {
  85 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic());
  86 + }
  87 +
  88 + @Override
  89 + public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> createTbCoreNotificationsMsgProducer() {
  90 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic());
  91 + }
  92 +
  93 + @Override
  94 + public TbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> createToCoreMsgConsumer() {
  95 + return new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic(),
  96 + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders()));
  97 + }
  98 +
  99 + @Override
  100 + public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> createToCoreNotificationsMsgConsumer() {
  101 + return new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings,
  102 + partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(),
  103 + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
  104 + }
  105 +
  106 + @Override
  107 + public TbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> createTransportApiRequestConsumer() {
  108 + return new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings, transportApiSettings.getRequestsTopic(),
  109 + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders()));
  110 + }
  111 +
  112 + @Override
  113 + public TbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> createTransportApiResponseProducer() {
  114 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic());
  115 + }
  116 +}
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.queue.provider;
  17 +
  18 +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
  19 +import org.springframework.stereotype.Component;
  20 +import org.thingsboard.server.common.msg.queue.ServiceType;
  21 +import org.thingsboard.server.gen.transport.TransportProtos;
  22 +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
  23 +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
  24 +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
  25 +import org.thingsboard.server.queue.TbQueueAdmin;
  26 +import org.thingsboard.server.queue.TbQueueConsumer;
  27 +import org.thingsboard.server.queue.TbQueueProducer;
  28 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusConsumerTemplate;
  29 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusProducerTemplate;
  30 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings;
  31 +import org.thingsboard.server.queue.common.TbProtoQueueMsg;
  32 +import org.thingsboard.server.queue.discovery.PartitionService;
  33 +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
  34 +import org.thingsboard.server.queue.settings.TbQueueCoreSettings;
  35 +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
  36 +import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
  37 +
  38 +@Component
  39 +@ConditionalOnExpression("'${queue.type:null}'=='service-bus' && '${service.type:null}'=='tb-rule-engine'")
  40 +public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
  41 +
  42 + private final PartitionService partitionService;
  43 + private final TbQueueCoreSettings coreSettings;
  44 + private final TbServiceInfoProvider serviceInfoProvider;
  45 + private final TbQueueRuleEngineSettings ruleEngineSettings;
  46 + private final TbServiceBusSettings serviceBusSettings;
  47 + private final TbQueueAdmin admin;
  48 +
  49 + public ServiceBusTbRuleEngineQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings,
  50 + TbQueueRuleEngineSettings ruleEngineSettings,
  51 + TbServiceInfoProvider serviceInfoProvider,
  52 + TbServiceBusSettings serviceBusSettings,
  53 + TbQueueAdmin admin) {
  54 + this.partitionService = partitionService;
  55 + this.coreSettings = coreSettings;
  56 + this.serviceInfoProvider = serviceInfoProvider;
  57 + this.ruleEngineSettings = ruleEngineSettings;
  58 + this.serviceBusSettings = serviceBusSettings;
  59 + this.admin = admin;
  60 + }
  61 +
  62 + @Override
  63 + public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> createTransportNotificationsMsgProducer() {
  64 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic());
  65 + }
  66 +
  67 + @Override
  68 + public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> createRuleEngineMsgProducer() {
  69 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic());
  70 + }
  71 +
  72 + @Override
  73 + public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> createRuleEngineNotificationsMsgProducer() {
  74 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, ruleEngineSettings.getTopic());
  75 + }
  76 +
  77 + @Override
  78 + public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> createTbCoreMsgProducer() {
  79 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic());
  80 + }
  81 +
  82 + @Override
  83 + public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> createTbCoreNotificationsMsgProducer() {
  84 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic());
  85 + }
  86 +
  87 + @Override
  88 + public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> createToRuleEngineMsgConsumer(TbRuleEngineQueueConfiguration configuration) {
  89 + return new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings, ruleEngineSettings.getTopic(),
  90 + msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders()));
  91 + }
  92 +
  93 + @Override
  94 + public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> createToRuleEngineNotificationsMsgConsumer() {
  95 + return new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings,
  96 + partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(),
  97 + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
  98 + }
  99 +}
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.queue.provider;
  17 +
  18 +import lombok.extern.slf4j.Slf4j;
  19 +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
  20 +import org.springframework.stereotype.Component;
  21 +import org.thingsboard.server.gen.transport.TransportProtos;
  22 +import org.thingsboard.server.queue.TbQueueAdmin;
  23 +import org.thingsboard.server.queue.TbQueueConsumer;
  24 +import org.thingsboard.server.queue.TbQueueProducer;
  25 +import org.thingsboard.server.queue.TbQueueRequestTemplate;
  26 +import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate;
  27 +import org.thingsboard.server.queue.common.TbProtoQueueMsg;
  28 +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
  29 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusConsumerTemplate;
  30 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusProducerTemplate;
  31 +import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings;
  32 +import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings;
  33 +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings;
  34 +
  35 +@Component
  36 +@ConditionalOnExpression("'${queue.type:null}'=='service-bus' && ('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport')")
  37 +@Slf4j
  38 +public class ServiceBusTransportQueueFactory implements TbTransportQueueFactory {
  39 + private final TbQueueTransportApiSettings transportApiSettings;
  40 + private final TbQueueTransportNotificationSettings transportNotificationSettings;
  41 + private final TbServiceBusSettings serviceBusSettings;
  42 + private final TbQueueAdmin admin;
  43 + private final TbServiceInfoProvider serviceInfoProvider;
  44 +
  45 + public ServiceBusTransportQueueFactory(TbQueueTransportApiSettings transportApiSettings,
  46 + TbQueueTransportNotificationSettings transportNotificationSettings,
  47 + TbServiceBusSettings serviceBusSettings,
  48 + TbServiceInfoProvider serviceInfoProvider,
  49 + TbQueueAdmin admin) {
  50 + this.transportApiSettings = transportApiSettings;
  51 + this.transportNotificationSettings = transportNotificationSettings;
  52 + this.serviceBusSettings = serviceBusSettings;
  53 + this.admin = admin;
  54 + this.serviceInfoProvider = serviceInfoProvider;
  55 + }
  56 +
  57 + @Override
  58 + public TbQueueRequestTemplate<TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>, TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> createTransportApiRequestTemplate() {
  59 + TbQueueProducer<TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>> producerTemplate =
  60 + new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, transportApiSettings.getRequestsTopic());
  61 +
  62 + TbQueueConsumer<TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> consumerTemplate =
  63 + new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings,
  64 + transportApiSettings.getResponsesTopic() + "." + serviceInfoProvider.getServiceId(),
  65 + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders()));
  66 +
  67 + DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder
  68 + <TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>, TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> templateBuilder = DefaultTbQueueRequestTemplate.builder();
  69 + templateBuilder.queueAdmin(admin);
  70 + templateBuilder.requestTemplate(producerTemplate);
  71 + templateBuilder.responseTemplate(consumerTemplate);
  72 + templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests());
  73 + templateBuilder.maxRequestTimeout(transportApiSettings.getMaxRequestsTimeout());
  74 + templateBuilder.pollInterval(transportApiSettings.getResponsePollInterval());
  75 + return templateBuilder.build();
  76 + }
  77 +
  78 + @Override
  79 + public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> createRuleEngineMsgProducer() {
  80 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, transportApiSettings.getRequestsTopic());
  81 + }
  82 +
  83 + @Override
  84 + public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> createTbCoreMsgProducer() {
  85 + return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, transportApiSettings.getRequestsTopic());
  86 + }
  87 +
  88 + @Override
  89 + public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> createTransportNotificationsConsumer() {
  90 + return new TbServiceBusConsumerTemplate<>(admin, serviceBusSettings,
  91 + transportNotificationSettings.getNotificationsTopic() + "." + serviceInfoProvider.getServiceId(),
  92 + msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders()));
  93 + }
  94 +}
@@ -20,7 +20,6 @@ import com.google.api.core.ApiFutures; @@ -20,7 +20,6 @@ import com.google.api.core.ApiFutures;
20 import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub; 20 import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub;
21 import com.google.cloud.pubsub.v1.stub.SubscriberStub; 21 import com.google.cloud.pubsub.v1.stub.SubscriberStub;
22 import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings; 22 import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings;
23 -import com.google.common.reflect.TypeToken;  
24 import com.google.gson.Gson; 23 import com.google.gson.Gson;
25 import com.google.protobuf.InvalidProtocolBufferException; 24 import com.google.protobuf.InvalidProtocolBufferException;
26 import com.google.pubsub.v1.AcknowledgeRequest; 25 import com.google.pubsub.v1.AcknowledgeRequest;
@@ -36,15 +35,12 @@ import org.thingsboard.server.queue.TbQueueAdmin; @@ -36,15 +35,12 @@ import org.thingsboard.server.queue.TbQueueAdmin;
36 import org.thingsboard.server.queue.TbQueueConsumer; 35 import org.thingsboard.server.queue.TbQueueConsumer;
37 import org.thingsboard.server.queue.TbQueueMsg; 36 import org.thingsboard.server.queue.TbQueueMsg;
38 import org.thingsboard.server.queue.TbQueueMsgDecoder; 37 import org.thingsboard.server.queue.TbQueueMsgDecoder;
39 -import org.thingsboard.server.queue.TbQueueMsgHeaders;  
40 import org.thingsboard.server.queue.common.DefaultTbQueueMsg; 38 import org.thingsboard.server.queue.common.DefaultTbQueueMsg;
41 -import org.thingsboard.server.queue.common.DefaultTbQueueMsgHeaders;  
42 39
43 import java.io.IOException; 40 import java.io.IOException;
44 import java.util.ArrayList; 41 import java.util.ArrayList;
45 import java.util.Collections; 42 import java.util.Collections;
46 import java.util.List; 43 import java.util.List;
47 -import java.util.Map;  
48 import java.util.Objects; 44 import java.util.Objects;
49 import java.util.Set; 45 import java.util.Set;
50 import java.util.concurrent.CopyOnWriteArrayList; 46 import java.util.concurrent.CopyOnWriteArrayList;
@@ -139,7 +135,7 @@ public class TbPubSubConsumerTemplate<T extends TbQueueMsg> implements TbQueueCo @@ -139,7 +135,7 @@ public class TbPubSubConsumerTemplate<T extends TbQueueMsg> implements TbQueueCo
139 subscriptionNames = partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toSet()); 135 subscriptionNames = partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toSet());
140 subscriptionNames.forEach(admin::createTopicIfNotExists); 136 subscriptionNames.forEach(admin::createTopicIfNotExists);
141 consumerExecutor = Executors.newFixedThreadPool(subscriptionNames.size()); 137 consumerExecutor = Executors.newFixedThreadPool(subscriptionNames.size());
142 - messagesPerTopic = pubSubSettings.getMaxMessages()/subscriptionNames.size(); 138 + messagesPerTopic = pubSubSettings.getMaxMessages() / subscriptionNames.size();
143 subscribed = true; 139 subscribed = true;
144 } 140 }
145 List<ReceivedMessage> messages; 141 List<ReceivedMessage> messages;
@@ -217,11 +213,6 @@ public class TbPubSubConsumerTemplate<T extends TbQueueMsg> implements TbQueueCo @@ -217,11 +213,6 @@ public class TbPubSubConsumerTemplate<T extends TbQueueMsg> implements TbQueueCo
217 213
218 public T decode(PubsubMessage message) throws InvalidProtocolBufferException { 214 public T decode(PubsubMessage message) throws InvalidProtocolBufferException {
219 DefaultTbQueueMsg msg = gson.fromJson(message.getData().toStringUtf8(), DefaultTbQueueMsg.class); 215 DefaultTbQueueMsg msg = gson.fromJson(message.getData().toStringUtf8(), DefaultTbQueueMsg.class);
220 - TbQueueMsgHeaders headers = new DefaultTbQueueMsgHeaders();  
221 - Map<String, byte[]> headerMap = gson.fromJson(message.getAttributesMap().get("headers"), new TypeToken<Map<String, byte[]>>() {  
222 - }.getType());  
223 - headerMap.forEach(headers::put);  
224 - msg.setHeaders(headers);  
225 return decoder.decode(msg); 216 return decoder.decode(msg);
226 } 217 }
227 218
@@ -71,7 +71,6 @@ public class TbPubSubProducerTemplate<T extends TbQueueMsg> implements TbQueuePr @@ -71,7 +71,6 @@ public class TbPubSubProducerTemplate<T extends TbQueueMsg> implements TbQueuePr
71 public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) { 71 public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) {
72 PubsubMessage.Builder pubsubMessageBuilder = PubsubMessage.newBuilder(); 72 PubsubMessage.Builder pubsubMessageBuilder = PubsubMessage.newBuilder();
73 pubsubMessageBuilder.setData(getMsg(msg)); 73 pubsubMessageBuilder.setData(getMsg(msg));
74 - pubsubMessageBuilder.putAttributes("headers", gson.toJson(msg.getHeaders().getData()));  
75 74
76 Publisher publisher = getOrCreatePublisher(tpi.getFullTopicName()); 75 Publisher publisher = getOrCreatePublisher(tpi.getFullTopicName());
77 ApiFuture<String> future = publisher.publish(pubsubMessageBuilder.build()); 76 ApiFuture<String> future = publisher.publish(pubsubMessageBuilder.build());
@@ -110,7 +109,7 @@ public class TbPubSubProducerTemplate<T extends TbQueueMsg> implements TbQueuePr @@ -110,7 +109,7 @@ public class TbPubSubProducerTemplate<T extends TbQueueMsg> implements TbQueuePr
110 } 109 }
111 110
112 private ByteString getMsg(T msg) { 111 private ByteString getMsg(T msg) {
113 - String json = gson.toJson(new DefaultTbQueueMsg(msg.getKey(), msg.getData())); 112 + String json = gson.toJson(new DefaultTbQueueMsg(msg));
114 return ByteString.copyFrom(json.getBytes()); 113 return ByteString.copyFrom(json.getBytes());
115 } 114 }
116 115
@@ -23,7 +23,6 @@ import com.amazonaws.services.sqs.AmazonSQSClientBuilder; @@ -23,7 +23,6 @@ import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
23 import com.amazonaws.services.sqs.model.DeleteMessageBatchRequestEntry; 23 import com.amazonaws.services.sqs.model.DeleteMessageBatchRequestEntry;
24 import com.amazonaws.services.sqs.model.Message; 24 import com.amazonaws.services.sqs.model.Message;
25 import com.amazonaws.services.sqs.model.ReceiveMessageRequest; 25 import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
26 -import com.google.common.reflect.TypeToken;  
27 import com.google.common.util.concurrent.Futures; 26 import com.google.common.util.concurrent.Futures;
28 import com.google.common.util.concurrent.ListenableFuture; 27 import com.google.common.util.concurrent.ListenableFuture;
29 import com.google.common.util.concurrent.ListeningExecutorService; 28 import com.google.common.util.concurrent.ListeningExecutorService;
@@ -38,15 +37,12 @@ import org.thingsboard.server.queue.TbQueueAdmin; @@ -38,15 +37,12 @@ import org.thingsboard.server.queue.TbQueueAdmin;
38 import org.thingsboard.server.queue.TbQueueConsumer; 37 import org.thingsboard.server.queue.TbQueueConsumer;
39 import org.thingsboard.server.queue.TbQueueMsg; 38 import org.thingsboard.server.queue.TbQueueMsg;
40 import org.thingsboard.server.queue.TbQueueMsgDecoder; 39 import org.thingsboard.server.queue.TbQueueMsgDecoder;
41 -import org.thingsboard.server.queue.TbQueueMsgHeaders;  
42 import org.thingsboard.server.queue.common.DefaultTbQueueMsg; 40 import org.thingsboard.server.queue.common.DefaultTbQueueMsg;
43 -import org.thingsboard.server.queue.common.DefaultTbQueueMsgHeaders;  
44 41
45 import java.io.IOException; 42 import java.io.IOException;
46 import java.util.ArrayList; 43 import java.util.ArrayList;
47 import java.util.Collections; 44 import java.util.Collections;
48 import java.util.List; 45 import java.util.List;
49 -import java.util.Map;  
50 import java.util.Objects; 46 import java.util.Objects;
51 import java.util.Set; 47 import java.util.Set;
52 import java.util.concurrent.CopyOnWriteArrayList; 48 import java.util.concurrent.CopyOnWriteArrayList;
@@ -161,7 +157,7 @@ public class TbAwsSqsConsumerTemplate<T extends TbQueueMsg> implements TbQueueCo @@ -161,7 +157,7 @@ public class TbAwsSqsConsumerTemplate<T extends TbQueueMsg> implements TbQueueCo
161 if (stopped) { 157 if (stopped) {
162 log.info("[{}] Aws SQS consumer is stopped.", topic); 158 log.info("[{}] Aws SQS consumer is stopped.", topic);
163 } else { 159 } else {
164 - log.error("Failed to pool messages.", e); 160 + log.error("Failed to pool messages.", e);
165 } 161 }
166 } 162 }
167 } 163 }
@@ -214,11 +210,6 @@ public class TbAwsSqsConsumerTemplate<T extends TbQueueMsg> implements TbQueueCo @@ -214,11 +210,6 @@ public class TbAwsSqsConsumerTemplate<T extends TbQueueMsg> implements TbQueueCo
214 210
215 public T decode(Message message) throws InvalidProtocolBufferException { 211 public T decode(Message message) throws InvalidProtocolBufferException {
216 DefaultTbQueueMsg msg = gson.fromJson(message.getBody(), DefaultTbQueueMsg.class); 212 DefaultTbQueueMsg msg = gson.fromJson(message.getBody(), DefaultTbQueueMsg.class);
217 - TbQueueMsgHeaders headers = new DefaultTbQueueMsgHeaders();  
218 - Map<String, byte[]> headerMap = gson.fromJson(message.getMessageAttributes().get("headers").getStringValue(), new TypeToken<Map<String, byte[]>>() {  
219 - }.getType());  
220 - headerMap.forEach(headers::put);  
221 - msg.setHeaders(headers);  
222 return decoder.decode(msg); 213 return decoder.decode(msg);
223 } 214 }
224 215
@@ -20,7 +20,6 @@ import com.amazonaws.auth.AWSStaticCredentialsProvider; @@ -20,7 +20,6 @@ import com.amazonaws.auth.AWSStaticCredentialsProvider;
20 import com.amazonaws.auth.BasicAWSCredentials; 20 import com.amazonaws.auth.BasicAWSCredentials;
21 import com.amazonaws.services.sqs.AmazonSQS; 21 import com.amazonaws.services.sqs.AmazonSQS;
22 import com.amazonaws.services.sqs.AmazonSQSClientBuilder; 22 import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
23 -import com.amazonaws.services.sqs.model.MessageAttributeValue;  
24 import com.amazonaws.services.sqs.model.SendMessageRequest; 23 import com.amazonaws.services.sqs.model.SendMessageRequest;
25 import com.amazonaws.services.sqs.model.SendMessageResult; 24 import com.amazonaws.services.sqs.model.SendMessageResult;
26 import com.google.common.util.concurrent.FutureCallback; 25 import com.google.common.util.concurrent.FutureCallback;
@@ -37,7 +36,6 @@ import org.thingsboard.server.queue.TbQueueMsg; @@ -37,7 +36,6 @@ import org.thingsboard.server.queue.TbQueueMsg;
37 import org.thingsboard.server.queue.TbQueueProducer; 36 import org.thingsboard.server.queue.TbQueueProducer;
38 import org.thingsboard.server.queue.common.DefaultTbQueueMsg; 37 import org.thingsboard.server.queue.common.DefaultTbQueueMsg;
39 38
40 -import java.util.HashMap;  
41 import java.util.Map; 39 import java.util.Map;
42 import java.util.concurrent.ConcurrentHashMap; 40 import java.util.concurrent.ConcurrentHashMap;
43 import java.util.concurrent.Executors; 41 import java.util.concurrent.Executors;
@@ -82,13 +80,6 @@ public class TbAwsSqsProducerTemplate<T extends TbQueueMsg> implements TbQueuePr @@ -82,13 +80,6 @@ public class TbAwsSqsProducerTemplate<T extends TbQueueMsg> implements TbQueuePr
82 sendMsgRequest.withQueueUrl(getQueueUrl(tpi.getFullTopicName())); 80 sendMsgRequest.withQueueUrl(getQueueUrl(tpi.getFullTopicName()));
83 sendMsgRequest.withMessageBody(gson.toJson(new DefaultTbQueueMsg(msg.getKey(), msg.getData()))); 81 sendMsgRequest.withMessageBody(gson.toJson(new DefaultTbQueueMsg(msg.getKey(), msg.getData())));
84 82
85 - Map<String, MessageAttributeValue> attributes = new HashMap<>();  
86 -  
87 - attributes.put("headers", new MessageAttributeValue()  
88 - .withStringValue(gson.toJson(msg.getHeaders().getData()))  
89 - .withDataType("String"));  
90 -  
91 - sendMsgRequest.withMessageAttributes(attributes);  
92 sendMsgRequest.withMessageGroupId(msg.getKey().toString()); 83 sendMsgRequest.withMessageGroupId(msg.getKey().toString());
93 ListenableFuture<SendMessageResult> future = producerExecutor.submit(() -> sqsClient.sendMessage(sendMsgRequest)); 84 ListenableFuture<SendMessageResult> future = producerExecutor.submit(() -> sqsClient.sendMessage(sendMsgRequest));
94 85
@@ -94,6 +94,7 @@ @@ -94,6 +94,7 @@
94 <snakeyaml.version>1.23</snakeyaml.version> 94 <snakeyaml.version>1.23</snakeyaml.version>
95 <amazonaws.sqs.version>1.11.747</amazonaws.sqs.version> 95 <amazonaws.sqs.version>1.11.747</amazonaws.sqs.version>
96 <pubsub.client.version>1.84.0</pubsub.client.version> 96 <pubsub.client.version>1.84.0</pubsub.client.version>
  97 + <azure-servicebus.version>3.2.0</azure-servicebus.version>
97 <passay.version>1.5.0</passay.version> 98 <passay.version>1.5.0</passay.version>
98 <ua-parser.version>1.4.3</ua-parser.version> 99 <ua-parser.version>1.4.3</ua-parser.version>
99 </properties> 100 </properties>
@@ -899,6 +900,11 @@ @@ -899,6 +900,11 @@
899 <version>${pubsub.client.version}</version> 900 <version>${pubsub.client.version}</version>
900 </dependency> 901 </dependency>
901 <dependency> 902 <dependency>
  903 + <groupId>com.microsoft.azure</groupId>
  904 + <artifactId>azure-servicebus</artifactId>
  905 + <version>${azure-servicebus.version}</version>
  906 + </dependency>
  907 + <dependency>
902 <groupId>org.passay</groupId> 908 <groupId>org.passay</groupId>
903 <artifactId>passay</artifactId> 909 <artifactId>passay</artifactId>
904 <version>${passay.version}</version> 910 <version>${passay.version}</version>
@@ -63,7 +63,7 @@ transport: @@ -63,7 +63,7 @@ transport:
63 max_string_value_length: "${JSON_MAX_STRING_VALUE_LENGTH:0}" 63 max_string_value_length: "${JSON_MAX_STRING_VALUE_LENGTH:0}"
64 64
65 queue: 65 queue:
66 - type: "${TB_QUEUE_TYPE:kafka}" # kafka or aws-sqs or pubsub 66 + type: "${TB_QUEUE_TYPE:kafka}" # kafka or aws-sqs or pubsub or service-bus
67 kafka: 67 kafka:
68 bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}" 68 bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}"
69 acks: "${TB_KAFKA_ACKS:all}" 69 acks: "${TB_KAFKA_ACKS:all}"
@@ -83,6 +83,11 @@ queue: @@ -83,6 +83,11 @@ queue:
83 ack_deadline: "${TB_QUEUE_PUBSUB_ACK_DEADLINE:30}" #In seconds. If messages wont commit in this time, messages will poll again 83 ack_deadline: "${TB_QUEUE_PUBSUB_ACK_DEADLINE:30}" #In seconds. If messages wont commit in this time, messages will poll again
84 max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" #in bytes 84 max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" #in bytes
85 max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}" 85 max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
  86 + service_bus:
  87 + namespace_name: "${TB_QUEUE_SERVICE_BUS_NAMESPACE_NAME:YOUR_NAMESPACE_NAME}"
  88 + sas_key_name: "${TB_QUEUE_SERVICE_BUS_SAS_KEY_NAME:YOUR_SAS_KEY_NAME}"
  89 + sas_key: "${TB_QUEUE_SERVICE_BUS_SAS_KEY:YOUR_SAS_KEY}"
  90 + max_messages: "${TB_QUEUE_SERVICE_BUS_MAX_MESSAGES:1000}"
86 partitions: 91 partitions:
87 hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" 92 hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}"
88 virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}" 93 virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}"