dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

created Aws Sqs Queue (#2534)

Browse Source 
* created Aws Sqs Queue

* improvement AwsSqs providers

* revert package-lock.json

* Aws sqs improvements

* Aws sqs improvements

* Aws sqs improvements

* Aws sqs improvements

* aws improvements

* aws improvements

* aws improvements

* added visibility timeout to aws queue
This commit is contained in:
Yevhen Bondarenko 2020-03-30 19:13:34 +03:00 committed by GitHub
parent c4269023dd
commit 2553cf6b6f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
25 changed files with 1074 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
application/src/main/java/org/thingsboard/server/controller/RpcController.java
View File

@ -88,7 +88,6 @@ public class RpcController extends BaseController {
return handleDeviceRPCRequest(false, new DeviceId(UUID.fromString(deviceIdStr)), requestBody);
}
private DeferredResult<ResponseEntity> handleDeviceRPCRequest(boolean oneWay, DeviceId deviceId, String requestBody) throws ThingsboardException {
try {
JsonNode rpcRequestBody = jsonMapper.readTree(requestBody);

9
application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java
View File

@ -136,6 +136,15 @@ public abstract class AbstractConsumerService<T extends com.google.protobuf.Gene
@PreDestroy
public void destroy() {
stopped = true;
if (mainConsumer != null) {
mainConsumer.unsubscribe();
}
if (nfConsumer != null) {
nfConsumer.unsubscribe();
}
if (mainConsumerExecutor != null) {
mainConsumerExecutor.shutdownNow();
}

8
application/src/main/resources/thingsboard.yml
View File

@ -517,7 +517,7 @@ swagger:
version: "${SWAGGER_VERSION:2.0}"
queue:
type: "${TB_QUEUE_TYPE:in-memory}" # kafka or in-memory
type: "${TB_QUEUE_TYPE:in-memory}" # kafka or in-memory or aws-sqs
kafka:
bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}"
acks: "${TB_KAFKA_ACKS:all}"
@ -525,6 +525,12 @@ queue:
batch.size: "${TB_KAFKA_BATCH_SIZE:16384}"
linger.ms: "${TB_KAFKA_LINGER_MS:1}"
buffer.memory: "${TB_BUFFER_MEMORY:33554432}"
aws_sqs:
access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}"
secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}"
region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}"
threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}"
visibility_timeout: "${TB_QUEUE_AWS_SQS_VISIBILITY_TIMEOUT:30}" #in seconds
partitions:
hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}"
virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}"

5
common/queue/pom.xml
View File

@ -52,6 +52,11 @@
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-sqs</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context-support</artifactId>

24
common/queue/src/main/java/org/thingsboard/server/queue/TbQueueMsgDecoder.java Normal file
View File

@ -0,0 +1,24 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.queue;
import com.google.protobuf.InvalidProtocolBufferException;
import org.thingsboard.server.queue.TbQueueMsg;
public interface TbQueueMsgDecoder<T extends TbQueueMsg> {
T decode(TbQueueMsg msg) throws InvalidProtocolBufferException;
}

1
common/queue/src/main/java/org/thingsboard/server/queue/TbQueueProducer.java
View File

@ -25,4 +25,5 @@ public interface TbQueueProducer<T extends TbQueueMsg> {
void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback);
void stop();
}

11
common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java
View File

@ -92,6 +92,8 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
List<Response> responses = responseTemplate.poll(pollInterval);
if (responses.size() > 0) {
log.trace("Polling responses completed, consumer records count [{}]", responses.size());
} else {
continue;
}
responses.forEach(response -> {
log.trace("Received response to Kafka Template request: {}", response);
@ -145,6 +147,15 @@ public class DefaultTbQueueRequestTemplate<Request extends TbQueueMsg, Response
@Override
public void stop() {
stopped = true;
if (responseTemplate != null) {
responseTemplate.unsubscribe();
}
if (requestTemplate != null) {
requestTemplate.stop();
}
if (internalExecutor) {
executor.shutdownNow();
}

12
common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueResponseTemplate.java
View File

@ -18,12 +18,12 @@ package org.thingsboard.server.queue.common;
import lombok.Builder;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.errors.InterruptException;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.TbQueueHandler;
import org.thingsboard.server.queue.TbQueueMsg;
import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.TbQueueResponseTemplate;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import java.util.List;
import java.util.UUID;
@ -87,6 +87,10 @@ public class DefaultTbQueueResponseTemplate<Request extends TbQueueMsg, Response
}
List<Request> requests = requestTemplate.poll(pollInterval);
if (requests.isEmpty()) {
continue;
}
requests.forEach(request -> {
long currentTime = System.currentTimeMillis();
long requestTime = bytesToLong(request.getHeaders().get(REQUEST_TIME));
@ -147,6 +151,12 @@ public class DefaultTbQueueResponseTemplate<Request extends TbQueueMsg, Response
public void stop() {
stopped = true;
if (requestTemplate != null) {
requestTemplate.unsubscribe();
}
if (responseTemplate != null) {
responseTemplate.stop();
}
if (timeoutExecutor != null) {
timeoutExecutor.shutdownNow();
}

7
common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaConsumerTemplate.java
View File

@ -22,9 +22,9 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.TbQueueMsg;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import java.io.IOException;
import java.time.Duration;
@ -96,7 +96,7 @@ public class TBKafkaConsumerTemplate<T extends TbQueueMsg> implements TbQueueCon
} else {
if (!subscribed) {
List<String> topicNames = partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList());
topicNames.forEach(this::createTopicIfNotExists);
topicNames.forEach(admin::createTopicIfNotExists);
consumer.subscribe(topicNames);
subscribed = true;
}
@ -130,7 +130,4 @@ public class TBKafkaConsumerTemplate<T extends TbQueueMsg> implements TbQueueCon
return decoder.decode(new KafkaTbQueueMsg(record));
}
private void createTopicIfNotExists(String topic) {
admin.createTopicIfNotExists(topic);
}
}

5
common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java
View File

@ -84,4 +84,9 @@ public class TBKafkaProducerTemplate<T extends TbQueueMsg> implements TbQueuePro
}
});
}
@Override
public void stop() {
}
}

5
common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueProducer.java
View File

@ -50,4 +50,9 @@ public class InMemoryTbQueueProducer<T extends TbQueueMsg> implements TbQueuePro
}
}
}
@Override
public void stop() {
}
}

127
common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueProvider.java Normal file
View File

@ -0,0 +1,127 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.queue.provider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.TbQueueAdmin;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.TbQueueCoreSettings;
import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.TbQueueRuleEngineSettings;
import org.thingsboard.server.queue.TbQueueTransportApiSettings;
import org.thingsboard.server.queue.TbQueueTransportNotificationSettings;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin;
import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate;
import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate;
import org.thingsboard.server.queue.sqs.TbAwsSqsSettings;
@Component
@ConditionalOnExpression("'${queue.type:null}'=='aws-sqs' && '${service.type:null}'=='monolith'")
public class AwsSqsMonolithQueueProvider implements TbCoreQueueProvider, TbRuleEngineQueueProvider {
private final PartitionService partitionService;
private final TbQueueCoreSettings coreSettings;
private final TbServiceInfoProvider serviceInfoProvider;
private final TbQueueRuleEngineSettings ruleEngineSettings;
private final TbQueueTransportApiSettings transportApiSettings;
private final TbQueueTransportNotificationSettings transportNotificationSettings;
private final TbAwsSqsSettings sqsSettings;
private final TbQueueAdmin admin;
public AwsSqsMonolithQueueProvider(PartitionService partitionService, TbQueueCoreSettings coreSettings,
TbQueueRuleEngineSettings ruleEngineSettings,
TbServiceInfoProvider serviceInfoProvider,
TbQueueTransportApiSettings transportApiSettings,
TbQueueTransportNotificationSettings transportNotificationSettings,
TbAwsSqsSettings sqsSettings) {
this.partitionService = partitionService;
this.coreSettings = coreSettings;
this.serviceInfoProvider = serviceInfoProvider;
this.ruleEngineSettings = ruleEngineSettings;
this.transportApiSettings = transportApiSettings;
this.transportNotificationSettings = transportNotificationSettings;
this.sqsSettings = sqsSettings;
admin = new TbAwsSqsAdmin(sqsSettings);
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> getTransportNotificationsMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportNotificationSettings.getNotificationsTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> getRuleEngineMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> getRuleEngineNotificationsMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> getTbCoreMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> getToRuleEngineMsgConsumer() {
return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic(),
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders()));
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> getToRuleEngineNotificationsMsgConsumer() {
return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings,
partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(),
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> getToCoreMsgConsumer() {
return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, coreSettings.getTopic(),
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders()));
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> getToCoreNotificationsMsgConsumer() {
return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings,
partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(),
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>> getTransportApiRequestConsumer() {
return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic(),
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders()));
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> getTransportApiResponseProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getResponsesTopic());
}
}

117
common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueProvider.java Normal file
View File

@ -0,0 +1,117 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.queue.provider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
import org.thingsboard.server.queue.TbQueueAdmin;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.TbQueueCoreSettings;
import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.TbQueueRuleEngineSettings;
import org.thingsboard.server.queue.TbQueueTransportApiSettings;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin;
import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate;
import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate;
import org.thingsboard.server.queue.sqs.TbAwsSqsSettings;
@Component
@ConditionalOnExpression("'${queue.type:null}'=='aws-sqs' && '${service.type:null}'=='tb-core'")
public class AwsSqsTbCoreQueueProvider implements TbCoreQueueProvider {
private final TbAwsSqsSettings sqsSettings;
private final TbQueueRuleEngineSettings ruleEngineSettings;
private final TbQueueCoreSettings coreSettings;
private final TbQueueTransportApiSettings transportApiSettings;
private final PartitionService partitionService;
private final TbServiceInfoProvider serviceInfoProvider;
private final TbQueueAdmin admin;
public AwsSqsTbCoreQueueProvider(TbAwsSqsSettings sqsSettings,
TbQueueCoreSettings coreSettings,
TbQueueTransportApiSettings transportApiSettings,
TbQueueRuleEngineSettings ruleEngineSettings,
PartitionService partitionService,
TbServiceInfoProvider serviceInfoProvider) {
this.sqsSettings = sqsSettings;
this.coreSettings = coreSettings;
this.transportApiSettings = transportApiSettings;
this.ruleEngineSettings = ruleEngineSettings;
this.partitionService = partitionService;
this.serviceInfoProvider = serviceInfoProvider;
this.admin = new TbAwsSqsAdmin(sqsSettings);
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportNotificationsMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> getRuleEngineNotificationsMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> getToCoreMsgConsumer() {
return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, coreSettings.getTopic(),
msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders()));
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> getToCoreNotificationsMsgConsumer() {
return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings,
partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(),
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> getTransportApiRequestConsumer() {
return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders()));
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> getTransportApiResponseProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
}
}

97
common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueProvider.java Normal file
View File

@ -0,0 +1,97 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.queue.provider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
import org.thingsboard.server.queue.TbQueueAdmin;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.TbQueueCoreSettings;
import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.TbQueueRuleEngineSettings;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin;
import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate;
import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate;
import org.thingsboard.server.queue.sqs.TbAwsSqsSettings;
@Component
@ConditionalOnExpression("'${queue.type:null}'=='aws-sqs' && '${service.type:null}'=='tb-rule-engine'")
public class AwsSqsTbRuleEngineQueueProvider implements TbRuleEngineQueueProvider {
private final PartitionService partitionService;
private final TbQueueCoreSettings coreSettings;
private final TbServiceInfoProvider serviceInfoProvider;
private final TbQueueRuleEngineSettings ruleEngineSettings;
private final TbAwsSqsSettings sqsSettings;
private final TbQueueAdmin admin;
public AwsSqsTbRuleEngineQueueProvider(PartitionService partitionService, TbQueueCoreSettings coreSettings,
TbQueueRuleEngineSettings ruleEngineSettings,
TbServiceInfoProvider serviceInfoProvider,
TbAwsSqsSettings sqsSettings) {
this.partitionService = partitionService;
this.coreSettings = coreSettings;
this.serviceInfoProvider = serviceInfoProvider;
this.ruleEngineSettings = ruleEngineSettings;
this.sqsSettings = sqsSettings;
admin = new TbAwsSqsAdmin(sqsSettings);
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportNotificationsMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> getRuleEngineNotificationsMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic());
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> getToRuleEngineMsgConsumer() {
return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, ruleEngineSettings.getTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders()));
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> getToRuleEngineNotificationsMsgConsumer() {
return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings,
partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(),
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
}
}

93
common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTransportQueueProvider.java Normal file
View File

@ -0,0 +1,93 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.queue.provider;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.TbQueueAdmin;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.TbQueueRequestTemplate;
import org.thingsboard.server.queue.TbQueueTransportApiSettings;
import org.thingsboard.server.queue.TbQueueTransportNotificationSettings;
import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin;
import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate;
import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate;
import org.thingsboard.server.queue.sqs.TbAwsSqsSettings;
@Component
@ConditionalOnExpression("'${queue.type:null}'=='aws-sqs' && ('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport')")
@Slf4j
public class AwsSqsTransportQueueProvider implements TbTransportQueueProvider {
private final TbQueueTransportApiSettings transportApiSettings;
private final TbQueueTransportNotificationSettings transportNotificationSettings;
private final TbAwsSqsSettings sqsSettings;
private final TbQueueAdmin admin;
private final TbServiceInfoProvider serviceInfoProvider;
public AwsSqsTransportQueueProvider(TbQueueTransportApiSettings transportApiSettings,
TbQueueTransportNotificationSettings transportNotificationSettings,
TbAwsSqsSettings sqsSettings,
TbServiceInfoProvider serviceInfoProvider) {
this.transportApiSettings = transportApiSettings;
this.transportNotificationSettings = transportNotificationSettings;
this.sqsSettings = sqsSettings;
admin = new TbAwsSqsAdmin(sqsSettings);
this.serviceInfoProvider = serviceInfoProvider;
}
@Override
public TbQueueRequestTemplate<TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>, TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> getTransportApiRequestTemplate() {
TbAwsSqsProducerTemplate<TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>> producerTemplate =
new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic());
TbAwsSqsConsumerTemplate<TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> consumerTemplate =
new TbAwsSqsConsumerTemplate<>(admin, sqsSettings,
transportApiSettings.getResponsesTopic() + "_" + serviceInfoProvider.getServiceId(),
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders()));
DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder
<TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg>, TbProtoQueueMsg<TransportProtos.TransportApiResponseMsg>> templateBuilder = DefaultTbQueueRequestTemplate.builder();
templateBuilder.queueAdmin(admin);
templateBuilder.requestTemplate(producerTemplate);
templateBuilder.responseTemplate(consumerTemplate);
templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests());
templateBuilder.maxRequestTimeout(transportApiSettings.getMaxRequestsTimeout());
templateBuilder.pollInterval(transportApiSettings.getResponsePollInterval());
return templateBuilder.build();
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> getRuleEngineMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic());
}
@Override
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> getTbCoreMsgProducer() {
return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getRequestsTopic());
}
@Override
public TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> getTransportNotificationsConsumer() {
return new TbAwsSqsConsumerTemplate<>(admin, sqsSettings, transportNotificationSettings.getNotificationsTopic() + "_" + serviceInfoProvider.getServiceId(),
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders()));
}
}

28
common/queue/src/main/java/org/thingsboard/server/queue/sqs/AwsSqsTbQueueMsgMetadata.java Normal file
View File

@ -0,0 +1,28 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.queue.sqs;
import com.amazonaws.http.SdkHttpMetadata;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.thingsboard.server.queue.TbQueueMsgMetadata;
@Data
@AllArgsConstructor
public class AwsSqsTbQueueMsgMetadata implements TbQueueMsgMetadata {
private final SdkHttpMetadata metadata;
}

65
common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java Normal file
View File

@ -0,0 +1,65 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.queue.sqs;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
import com.amazonaws.services.sqs.model.CreateQueueRequest;
import com.amazonaws.services.sqs.model.QueueAttributeName;
import org.thingsboard.server.queue.TbQueueAdmin;
import java.util.HashMap;
import java.util.Map;
public class TbAwsSqsAdmin implements TbQueueAdmin {
private final TbAwsSqsSettings sqsSettings;
private final Map<String, String> attributes = new HashMap<>();
private final AWSStaticCredentialsProvider credProvider;
public TbAwsSqsAdmin(TbAwsSqsSettings sqsSettings) {
this.sqsSettings = sqsSettings;
AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey());
this.credProvider = new AWSStaticCredentialsProvider(awsCredentials);
attributes.put("FifoQueue", "true");
attributes.put("ContentBasedDeduplication", "true");
attributes.put(QueueAttributeName.VisibilityTimeout.toString(), sqsSettings.getVisibilityTimeout());
}
@Override
public void createTopicIfNotExists(String topic) {
AmazonSQS sqsClient = AmazonSQSClientBuilder.standard()
.withCredentials(credProvider)
.withRegion(sqsSettings.getRegion())
.build();
final CreateQueueRequest createQueueRequest =
new CreateQueueRequest(topic.replaceAll("\\.", "_") + ".fifo")
.withAttributes(attributes);
try {
sqsClient.createQueue(createQueueRequest);
} finally {
if (sqsClient != null) {
sqsClient.shutdown();
}
}
}
}

239
common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsConsumerTemplate.java Normal file
View File

@ -0,0 +1,239 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.queue.sqs;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
import com.amazonaws.services.sqs.model.DeleteMessageBatchRequestEntry;
import com.amazonaws.services.sqs.model.Message;
import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
import com.google.common.reflect.TypeToken;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.gson.Gson;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.queue.TbQueueAdmin;
import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.TbQueueMsg;
import org.thingsboard.server.queue.TbQueueMsgDecoder;
import org.thingsboard.server.queue.TbQueueMsgHeaders;
import org.thingsboard.server.queue.common.DefaultTbQueueMsgHeaders;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@Slf4j
public class TbAwsSqsConsumerTemplate<T extends TbQueueMsg> implements TbQueueConsumer<T> {
private static final int MAX_NUM_MSGS = 10;
private final Gson gson = new Gson();
private final TbQueueAdmin admin;
private final AmazonSQS sqsClient;
private final String topic;
private final TbQueueMsgDecoder<T> decoder;
private final TbAwsSqsSettings sqsSettings;
private final List<AwsSqsMsgWrapper> pendingMessages = new CopyOnWriteArrayList<>();
private volatile Set<String> queueUrls;
private volatile Set<TopicPartitionInfo> partitions;
private ListeningExecutorService consumerExecutor;
private volatile boolean subscribed;
private volatile boolean stopped = false;
public TbAwsSqsConsumerTemplate(TbQueueAdmin admin, TbAwsSqsSettings sqsSettings, String topic, TbQueueMsgDecoder<T> decoder) {
this.admin = admin;
this.decoder = decoder;
this.topic = topic;
this.sqsSettings = sqsSettings;
AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey());
AWSStaticCredentialsProvider credProvider = new AWSStaticCredentialsProvider(awsCredentials);
this.sqsClient = AmazonSQSClientBuilder.standard()
.withCredentials(credProvider)
.withRegion(sqsSettings.getRegion())
.build();
}
@Override
public String getTopic() {
return topic;
}
@Override
public void subscribe() {
partitions = Collections.singleton(new TopicPartitionInfo(topic, null, null, true));
subscribed = false;
}
@Override
public void subscribe(Set<TopicPartitionInfo> partitions) {
this.partitions = partitions;
subscribed = false;
}
@Override
public void unsubscribe() {
stopped = true;
if (sqsClient != null) {
sqsClient.shutdown();
}
if (consumerExecutor != null) {
consumerExecutor.shutdownNow();
}
}
@Override
public List<T> poll(long durationInMillis) {
if (!subscribed && partitions == null) {
try {
Thread.sleep(durationInMillis);
} catch (InterruptedException e) {
log.debug("Failed to await subscription", e);
}
} else {
if (!subscribed) {
List<String> topicNames = partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList());
queueUrls = topicNames.stream().map(this::getQueueUrl).collect(Collectors.toSet());
consumerExecutor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(queueUrls.size() * sqsSettings.getThreadsPerTopic() + 1));
subscribed = true;
}
if (!pendingMessages.isEmpty()) {
log.warn("Present {} non committed messages.", pendingMessages.size());
return Collections.emptyList();
}
List<ListenableFuture<List<Message>>> futureList = queueUrls
.stream()
.map(url -> poll(url, (int) TimeUnit.MILLISECONDS.toSeconds(durationInMillis)))
.collect(Collectors.toList());
ListenableFuture<List<List<Message>>> futureResult = Futures.allAsList(futureList);
try {
return futureResult.get().stream()
.flatMap(List::stream)
.map(msg -> {
try {
return decode(msg);
} catch (IOException e) {
log.error("Failed to decode message: [{}]", msg);
return null;
}
}).filter(Objects::nonNull)
.collect(Collectors.toList());
} catch (InterruptedException | ExecutionException e) {
if (stopped) {
log.info("[{}] Aws SQS consumer is stopped.", topic);
} else {
log.error("Failed to pool messages.", e);
}
}
}
return Collections.emptyList();
}
private ListenableFuture<List<Message>> poll(String url, int waitTimeSeconds) {
List<ListenableFuture<List<Message>>> result = new ArrayList<>();
for (int i = 0; i < sqsSettings.getThreadsPerTopic(); i++) {
result.add(consumerExecutor.submit(() -> {
ReceiveMessageRequest request = new ReceiveMessageRequest();
request
.withWaitTimeSeconds(waitTimeSeconds)
.withMessageAttributeNames("headers")
.withQueueUrl(url)
.withMaxNumberOfMessages(MAX_NUM_MSGS);
return sqsClient.receiveMessage(request).getMessages();
}));
}
return Futures.transform(Futures.allAsList(result), list -> {
if (!CollectionUtils.isEmpty(list)) {
return list.stream()
.flatMap(messageList -> {
if (!messageList.isEmpty()) {
this.pendingMessages.add(new AwsSqsMsgWrapper(url, messageList));
return messageList.stream();
}
return Stream.empty();
})
.collect(Collectors.toList());
}
return Collections.emptyList();
}, consumerExecutor);
}
@Override
public void commit() {
pendingMessages.forEach(msg ->
consumerExecutor.submit(() -> {
List<DeleteMessageBatchRequestEntry> entries = msg.getMessages()
.stream()
.map(message -> new DeleteMessageBatchRequestEntry(message.getMessageId(), message.getReceiptHandle()))
.collect(Collectors.toList());
sqsClient.deleteMessageBatch(msg.getUrl(), entries);
}));
pendingMessages.clear();
}
public T decode(Message message) throws InvalidProtocolBufferException {
TbAwsSqsMsg msg = gson.fromJson(message.getBody(), TbAwsSqsMsg.class);
TbQueueMsgHeaders headers = new DefaultTbQueueMsgHeaders();
Map<String, byte[]> headerMap = gson.fromJson(message.getMessageAttributes().get("headers").getStringValue(), new TypeToken<Map<String, byte[]>>() {
}.getType());
headerMap.forEach(headers::put);
msg.setHeaders(headers);
return decoder.decode(msg);
}
@Data
private static class AwsSqsMsgWrapper {
private final String url;
private final List<Message> messages;
public AwsSqsMsgWrapper(String url, List<Message> messages) {
this.url = url;
this.messages = messages;
}
}
private String getQueueUrl(String topic) {
admin.createTopicIfNotExists(topic);
return sqsClient.getQueueUrl(topic.replaceAll("\\.", "_") + ".fifo").getQueueUrl();
}
}

38
common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsMsg.java Normal file
View File

@ -0,0 +1,38 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.queue.sqs;
import com.google.gson.annotations.Expose;
import lombok.Data;
import org.thingsboard.server.queue.TbQueueMsg;
import org.thingsboard.server.queue.TbQueueMsgHeaders;
import java.util.UUID;
@Data
public class TbAwsSqsMsg implements TbQueueMsg {
private final UUID key;
private final byte[] data;
public TbAwsSqsMsg(UUID key, byte[] data) {
this.key = key;
this.data = data;
}
@Expose(serialize = false, deserialize = false)
private TbQueueMsgHeaders headers;
}

127
common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java Normal file
View File

@ -0,0 +1,127 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.queue.sqs;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
import com.amazonaws.services.sqs.model.MessageAttributeValue;
import com.amazonaws.services.sqs.model.SendMessageRequest;
import com.amazonaws.services.sqs.model.SendMessageResult;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.queue.TbQueueAdmin;
import org.thingsboard.server.queue.TbQueueCallback;
import org.thingsboard.server.queue.TbQueueMsg;
import org.thingsboard.server.queue.TbQueueProducer;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
@Slf4j
public class TbAwsSqsProducerTemplate<T extends TbQueueMsg> implements TbQueueProducer<T> {
private final String defaultTopic;
private final AmazonSQS sqsClient;
private final Gson gson = new Gson();
private final Map<String, String> queueUrlMap = new ConcurrentHashMap<>();
private final TbQueueAdmin admin;
private ListeningExecutorService producerExecutor;
public TbAwsSqsProducerTemplate(TbQueueAdmin admin, TbAwsSqsSettings sqsSettings, String defaultTopic) {
this.admin = admin;
this.defaultTopic = defaultTopic;
AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey());
AWSStaticCredentialsProvider credProvider = new AWSStaticCredentialsProvider(awsCredentials);
this.sqsClient = AmazonSQSClientBuilder.standard()
.withCredentials(credProvider)
.withRegion(sqsSettings.getRegion())
.build();
producerExecutor = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
}
@Override
public void init() {
}
@Override
public String getDefaultTopic() {
return defaultTopic;
}
@Override
public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) {
SendMessageRequest sendMsgRequest = new SendMessageRequest();
sendMsgRequest.withQueueUrl(getQueueUrl(tpi.getFullTopicName()));
sendMsgRequest.withMessageBody(gson.toJson(new TbAwsSqsMsg(msg.getKey(), msg.getData())));
Map<String, MessageAttributeValue> attributes = new HashMap<>();
attributes.put("headers", new MessageAttributeValue()
.withStringValue(gson.toJson(msg.getHeaders().getData()))
.withDataType("String"));
sendMsgRequest.withMessageAttributes(attributes);
sendMsgRequest.withMessageGroupId(msg.getKey().toString());
ListenableFuture<SendMessageResult> future = producerExecutor.submit(() -> sqsClient.sendMessage(sendMsgRequest));
Futures.addCallback(future, new FutureCallback<SendMessageResult>() {
@Override
public void onSuccess(SendMessageResult result) {
if (callback != null) {
callback.onSuccess(new AwsSqsTbQueueMsgMetadata(result.getSdkHttpMetadata()));
}
}
@Override
public void onFailure(Throwable t) {
if (callback != null) {
callback.onFailure(t);
}
}
});
}
@Override
public void stop() {
if (producerExecutor != null) {
producerExecutor.shutdownNow();
}
if (sqsClient != null) {
sqsClient.shutdown();
}
}
private String getQueueUrl(String topic) {
return queueUrlMap.computeIfAbsent(topic, k -> {
admin.createTopicIfNotExists(topic);
return sqsClient.getQueueUrl(topic.replaceAll("\\.", "_") + ".fifo").getQueueUrl();
});
}
}

44
common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java Normal file
View File

@ -0,0 +1,44 @@
/**
* Copyright © 2016-2020 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.queue.sqs;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
@Slf4j
@ConditionalOnExpression("'${queue.type:null}'=='aws-sqs'")
@Component
@Data
public class TbAwsSqsSettings {
@Value("${queue.aws_sqs.access_key_id}")
private String accessKeyId;
@Value("${queue.aws_sqs.secret_access_key}")
private String secretAccessKey;
@Value("${queue.aws_sqs.region}")
private String region;
@Value("${queue.aws_sqs.threads_per_topic}")
private int threadsPerTopic;
@Value("${queue.aws_sqs.visibility_timeout}")
private String visibilityTimeout;
}

7
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View File

@ -138,6 +138,9 @@ public class DefaultTransportService implements TransportService {
while (!stopped) {
try {
List<TbProtoQueueMsg<ToTransportMsg>> records = transportNotificationsConsumer.poll(notificationsPollDuration);
if (records.size() == 0) {
continue;
}
records.forEach(record -> {
try {
ToTransportMsg toTransportMsg = record.getValue();
@ -170,6 +173,10 @@ public class DefaultTransportService implements TransportService {
perDeviceLimits.clear();
}
stopped = true;
if (transportNotificationsConsumer != null) {
transportNotificationsConsumer.unsubscribe();
}
if (schedulerExecutor != null) {
schedulerExecutor.shutdownNow();
}

6
pom.xml
View File

@ -92,6 +92,7 @@
<fst.version>2.57</fst.version>
<antlr.version>2.7.7</antlr.version>
<snakeyaml.version>1.23</snakeyaml.version>
<amazonaws.sqs.version>1.11.747</amazonaws.sqs.version>
<passay.version>1.5.0</passay.version>
<ua-parser.version>1.4.3</ua-parser.version>
</properties>
@ -886,6 +887,11 @@
<artifactId>jts-core</artifactId>
<version>${jts.version}</version>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-sqs</artifactId>
<version>${amazonaws.sqs.version}</version>
</dependency>
<dependency>
<groupId>org.passay</groupId>
<artifactId>passay</artifactId>

2
rule-engine/rule-engine-components/pom.xml
View File

@ -35,7 +35,7 @@
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<main.dir>${basedir}/../..</main.dir>
<aws.sdk.version>1.11.323</aws.sdk.version>
<aws.sdk.version>1.11.747</aws.sdk.version>
<pubsub.client.version>1.83.0</pubsub.client.version>
<google.common.protos.version>1.16.0</google.common.protos.version>
</properties>

6
transport/mqtt/src/main/resources/tb-mqtt-transport.yml
View File

@ -63,7 +63,7 @@ transport:
max_string_value_length: "${JSON_MAX_STRING_VALUE_LENGTH:0}"
queue:
type: "${TB_QUEUE_TYPE:kafka}" # kafka or ?
type: "${TB_QUEUE_TYPE:kafka}" # kafka or aws-sqs
kafka:
bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}"
acks: "${TB_KAFKA_ACKS:all}"
@ -71,6 +71,10 @@ queue:
batch.size: "${TB_KAFKA_BATCH_SIZE:16384}"
linger.ms: "${TB_KAFKA_LINGER_MS:1}"
buffer.memory: "${TB_BUFFER_MEMORY:33554432}"
aws_sqs:
access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}"
secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}"
region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}"
partitions:
hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}"
virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}"