Browse Source
* created Aws Sqs Queue * improvement AwsSqs providers * created pubsub queue * revert package-lock.json * Aws sqs improvements * Aws sqs improvements * Aws sqs improvements * Aws sqs improvements * Created pubsub queue * aws improvements * aws improvements * aws improvements * added visibility timeout to aws queue * pub sub improvements * pub sub improvements * aws sqs improvements * pub sub improvements * added comment to transport.yml about ack deadlinepull/2576/head
Yevhen Bondarenko
6 years ago
committed by
GitHub
GitHub
19 changed files with 1078 additions and 32 deletions
@ -0,0 +1,137 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.queue.provider; |
|||
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
import org.thingsboard.server.queue.TbQueueAdmin; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.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.pubsub.TbPubSubAdmin; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubSettings; |
|||
|
|||
@Component |
|||
@ConditionalOnExpression("'${queue.type:null}'=='pubsub' && '${service.type:null}'=='monolith'") |
|||
public class PubSubMonolithQueueProvider implements TbCoreQueueProvider, TbRuleEngineQueueProvider { |
|||
|
|||
private final TbPubSubSettings pubSubSettings; |
|||
private final TbQueueCoreSettings coreSettings; |
|||
private final TbQueueRuleEngineSettings ruleEngineSettings; |
|||
private final TbQueueTransportApiSettings transportApiSettings; |
|||
private final TbQueueTransportNotificationSettings transportNotificationSettings; |
|||
private final TbQueueAdmin admin; |
|||
private final PartitionService partitionService; |
|||
private final TbServiceInfoProvider serviceInfoProvider; |
|||
|
|||
private TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> tbCoreProducer; |
|||
|
|||
public PubSubMonolithQueueProvider(TbPubSubSettings pubSubSettings, |
|||
TbQueueCoreSettings coreSettings, |
|||
TbQueueRuleEngineSettings ruleEngineSettings, |
|||
TbQueueTransportApiSettings transportApiSettings, |
|||
TbQueueTransportNotificationSettings transportNotificationSettings, |
|||
PartitionService partitionService, |
|||
TbServiceInfoProvider serviceInfoProvider) { |
|||
this.pubSubSettings = pubSubSettings; |
|||
this.coreSettings = coreSettings; |
|||
this.ruleEngineSettings = ruleEngineSettings; |
|||
this.transportApiSettings = transportApiSettings; |
|||
this.transportNotificationSettings = transportNotificationSettings; |
|||
this.admin = new TbPubSubAdmin(pubSubSettings); |
|||
this.partitionService = partitionService; |
|||
this.serviceInfoProvider = serviceInfoProvider; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportNotificationsMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, transportNotificationSettings.getNotificationsTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, ruleEngineSettings.getTopic()); |
|||
|
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> getRuleEngineNotificationsMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, ruleEngineSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> getToRuleEngineMsgConsumer() { |
|||
return new TbPubSubConsumerTemplate<>(admin, pubSubSettings, ruleEngineSettings.getTopic(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> getToRuleEngineNotificationsMsgConsumer() { |
|||
return new TbPubSubConsumerTemplate<>(admin, pubSubSettings, |
|||
partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> getToCoreMsgConsumer() { |
|||
return new TbPubSubConsumerTemplate<>(admin, pubSubSettings, coreSettings.getTopic(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToCoreNotificationMsg>> getToCoreNotificationsMsgConsumer() { |
|||
return new TbPubSubConsumerTemplate<>(admin, pubSubSettings, |
|||
partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> getTransportApiRequestConsumer() { |
|||
return new TbPubSubConsumerTemplate<>(admin, pubSubSettings, transportApiSettings.getRequestsTopic(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> getTransportApiResponseProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, transportApiSettings.getResponsesTopic()); |
|||
} |
|||
} |
|||
@ -0,0 +1,113 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.queue.provider; |
|||
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
import org.thingsboard.server.queue.TbQueueAdmin; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.TbQueueCoreSettings; |
|||
import org.thingsboard.server.queue.TbQueueProducer; |
|||
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.pubsub.TbPubSubConsumerTemplate; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubSettings; |
|||
|
|||
@Component |
|||
@ConditionalOnExpression("'${queue.type:null}'=='pubsub' && '${service.type:null}'=='tb-core'") |
|||
public class PubSubTbCoreQueueProvider implements TbCoreQueueProvider { |
|||
|
|||
private final TbPubSubSettings pubSubSettings; |
|||
private final TbQueueCoreSettings coreSettings; |
|||
private final TbQueueTransportApiSettings transportApiSettings; |
|||
private final TbQueueAdmin admin; |
|||
private final PartitionService partitionService; |
|||
private final TbServiceInfoProvider serviceInfoProvider; |
|||
|
|||
public PubSubTbCoreQueueProvider(TbPubSubSettings pubSubSettings, |
|||
TbQueueCoreSettings coreSettings, |
|||
TbQueueTransportApiSettings transportApiSettings, |
|||
TbQueueAdmin admin, |
|||
PartitionService partitionService, |
|||
TbServiceInfoProvider serviceInfoProvider) { |
|||
this.pubSubSettings = pubSubSettings; |
|||
this.coreSettings = coreSettings; |
|||
this.transportApiSettings = transportApiSettings; |
|||
this.admin = admin; |
|||
this.partitionService = partitionService; |
|||
this.serviceInfoProvider = serviceInfoProvider; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportNotificationsMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> getRuleEngineNotificationsMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> getToCoreMsgConsumer() { |
|||
return new TbPubSubConsumerTemplate<>(admin, pubSubSettings, coreSettings.getTopic(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToCoreNotificationMsg>> getToCoreNotificationsMsgConsumer() { |
|||
return new TbPubSubConsumerTemplate<>(admin, pubSubSettings, |
|||
partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<TransportApiRequestMsg>> getTransportApiRequestConsumer() { |
|||
return new TbPubSubConsumerTemplate<>(admin, pubSubSettings, transportApiSettings.getRequestsTopic(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportApiResponseMsg>> getTransportApiResponseProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
} |
|||
@ -0,0 +1,101 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.queue.provider; |
|||
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.queue.TbQueueAdmin; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.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.pubsub.TbPubSubConsumerTemplate; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubSettings; |
|||
|
|||
@Component |
|||
@ConditionalOnExpression("'${queue.type:null}'=='pubsub' && '${service.type:null}'=='tb-rule-engine'") |
|||
public class PubSubTbRuleEngineQueueProvider implements TbRuleEngineQueueProvider { |
|||
|
|||
private final TbPubSubSettings pubSubSettings; |
|||
private final TbQueueCoreSettings coreSettings; |
|||
private final TbQueueRuleEngineSettings ruleEngineSettings; |
|||
private final TbQueueAdmin admin; |
|||
private final PartitionService partitionService; |
|||
private final TbServiceInfoProvider serviceInfoProvider; |
|||
|
|||
public PubSubTbRuleEngineQueueProvider(TbPubSubSettings pubSubSettings, |
|||
TbQueueCoreSettings coreSettings, |
|||
TbQueueRuleEngineSettings ruleEngineSettings, |
|||
TbQueueAdmin admin, |
|||
PartitionService partitionService, |
|||
TbServiceInfoProvider serviceInfoProvider) { |
|||
this.pubSubSettings = pubSubSettings; |
|||
this.coreSettings = coreSettings; |
|||
this.ruleEngineSettings = ruleEngineSettings; |
|||
this.admin = admin; |
|||
this.partitionService = partitionService; |
|||
this.serviceInfoProvider = serviceInfoProvider; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportNotificationsMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> getRuleEngineNotificationsMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, ruleEngineSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
|
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>> getToRuleEngineMsgConsumer() { |
|||
return new TbPubSubConsumerTemplate<>(admin, pubSubSettings, ruleEngineSettings.getTopic(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> getToRuleEngineNotificationsMsgConsumer() { |
|||
return new TbPubSubConsumerTemplate<>(admin, pubSubSettings, |
|||
partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
} |
|||
} |
|||
@ -0,0 +1,104 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.queue.provider; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; |
|||
import org.thingsboard.server.queue.TbQueueAdmin; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.TbQueueCoreSettings; |
|||
import org.thingsboard.server.queue.TbQueueProducer; |
|||
import org.thingsboard.server.queue.TbQueueRequestTemplate; |
|||
import org.thingsboard.server.queue.TbQueueRuleEngineSettings; |
|||
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.pubsub.TbPubSubAdmin; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubSettings; |
|||
|
|||
@Component |
|||
@ConditionalOnExpression("'${queue.type:null}'=='pubsub' && ('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport')") |
|||
@Slf4j |
|||
public class PubSubTransportQueueProvider implements TbTransportQueueProvider { |
|||
|
|||
private final TbPubSubSettings pubSubSettings; |
|||
private final TbServiceInfoProvider serviceInfoProvider; |
|||
private final TbQueueCoreSettings coreSettings; |
|||
private final TbQueueRuleEngineSettings ruleEngineSettings; |
|||
private final TbQueueTransportApiSettings transportApiSettings; |
|||
private final TbQueueTransportNotificationSettings transportNotificationSettings; |
|||
private final TbQueueAdmin admin; |
|||
|
|||
public PubSubTransportQueueProvider(TbPubSubSettings pubSubSettings, |
|||
TbServiceInfoProvider serviceInfoProvider, |
|||
TbQueueCoreSettings coreSettings, |
|||
TbQueueRuleEngineSettings ruleEngineSettings, |
|||
TbQueueTransportApiSettings transportApiSettings, |
|||
TbQueueTransportNotificationSettings transportNotificationSettings) { |
|||
this.pubSubSettings = pubSubSettings; |
|||
this.serviceInfoProvider = serviceInfoProvider; |
|||
this.coreSettings = coreSettings; |
|||
this.ruleEngineSettings = ruleEngineSettings; |
|||
this.transportApiSettings = transportApiSettings; |
|||
this.transportNotificationSettings = transportNotificationSettings; |
|||
this.admin = new TbPubSubAdmin(pubSubSettings); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> getTransportApiRequestTemplate() { |
|||
TbQueueProducer<TbProtoQueueMsg<TransportApiRequestMsg>> producer = new TbPubSubProducerTemplate<>(admin, pubSubSettings, transportApiSettings.getRequestsTopic()); |
|||
TbQueueConsumer<TbProtoQueueMsg<TransportApiResponseMsg>> consumer = new TbPubSubConsumerTemplate<>(admin, pubSubSettings, |
|||
transportApiSettings.getResponsesTopic() + "." + serviceInfoProvider.getServiceId(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
|
|||
DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder |
|||
<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> templateBuilder = DefaultTbQueueRequestTemplate.builder(); |
|||
templateBuilder.queueAdmin(admin); |
|||
templateBuilder.requestTemplate(producer); |
|||
templateBuilder.responseTemplate(consumer); |
|||
templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests()); |
|||
templateBuilder.maxRequestTimeout(transportApiSettings.getMaxRequestsTimeout()); |
|||
templateBuilder.pollInterval(transportApiSettings.getResponsePollInterval()); |
|||
return templateBuilder.build(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, ruleEngineSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer() { |
|||
return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueConsumer<TbProtoQueueMsg<ToTransportMsg>> getTransportNotificationsConsumer() { |
|||
return new TbPubSubConsumerTemplate<>(admin, pubSubSettings, |
|||
transportNotificationSettings.getNotificationsTopic() + "." + serviceInfoProvider.getServiceId(), |
|||
msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); |
|||
} |
|||
} |
|||
@ -0,0 +1,157 @@ |
|||
/** |
|||
* 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.pubsub; |
|||
|
|||
import com.google.cloud.pubsub.v1.SubscriptionAdminClient; |
|||
import com.google.cloud.pubsub.v1.SubscriptionAdminSettings; |
|||
import com.google.cloud.pubsub.v1.TopicAdminClient; |
|||
import com.google.cloud.pubsub.v1.TopicAdminSettings; |
|||
import com.google.pubsub.v1.ListSubscriptionsRequest; |
|||
import com.google.pubsub.v1.ListTopicsRequest; |
|||
import com.google.pubsub.v1.ProjectName; |
|||
import com.google.pubsub.v1.ProjectSubscriptionName; |
|||
import com.google.pubsub.v1.ProjectTopicName; |
|||
import com.google.pubsub.v1.PushConfig; |
|||
import com.google.pubsub.v1.Subscription; |
|||
import com.google.pubsub.v1.Topic; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.queue.TbQueueAdmin; |
|||
|
|||
import java.io.IOException; |
|||
import java.util.Set; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
|
|||
@Slf4j |
|||
public class TbPubSubAdmin implements TbQueueAdmin { |
|||
|
|||
private final TbPubSubSettings pubSubSettings; |
|||
private final SubscriptionAdminSettings subscriptionAdminSettings; |
|||
private final TopicAdminSettings topicAdminSettings; |
|||
private final Set<String> topicSet = ConcurrentHashMap.newKeySet(); |
|||
private final Set<String> subscriptionSet = ConcurrentHashMap.newKeySet(); |
|||
|
|||
public TbPubSubAdmin(TbPubSubSettings pubSubSettings) { |
|||
this.pubSubSettings = pubSubSettings; |
|||
|
|||
try { |
|||
topicAdminSettings = TopicAdminSettings.newBuilder().setCredentialsProvider(pubSubSettings.getCredentialsProvider()).build(); |
|||
} catch (IOException e) { |
|||
log.error("Failed to create TopicAdminSettings"); |
|||
throw new RuntimeException("Failed to create TopicAdminSettings."); |
|||
} |
|||
|
|||
try { |
|||
subscriptionAdminSettings = SubscriptionAdminSettings.newBuilder().setCredentialsProvider(pubSubSettings.getCredentialsProvider()).build(); |
|||
} catch (IOException e) { |
|||
log.error("Failed to create SubscriptionAdminSettings"); |
|||
throw new RuntimeException("Failed to create SubscriptionAdminSettings."); |
|||
} |
|||
|
|||
try (TopicAdminClient topicAdminClient = TopicAdminClient.create(topicAdminSettings)) { |
|||
ListTopicsRequest listTopicsRequest = |
|||
ListTopicsRequest.newBuilder().setProject(ProjectName.format(pubSubSettings.getProjectId())).build(); |
|||
TopicAdminClient.ListTopicsPagedResponse response = topicAdminClient.listTopics(listTopicsRequest); |
|||
for (Topic topic : response.iterateAll()) { |
|||
topicSet.add(topic.getName()); |
|||
} |
|||
} catch (IOException e) { |
|||
log.error("Failed to get topics.", e); |
|||
throw new RuntimeException("Failed to get topics.", e); |
|||
} |
|||
|
|||
try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create(subscriptionAdminSettings)) { |
|||
|
|||
ListSubscriptionsRequest listSubscriptionsRequest = |
|||
ListSubscriptionsRequest.newBuilder() |
|||
.setProject(ProjectName.of(pubSubSettings.getProjectId()).toString()) |
|||
.build(); |
|||
SubscriptionAdminClient.ListSubscriptionsPagedResponse response = |
|||
subscriptionAdminClient.listSubscriptions(listSubscriptionsRequest); |
|||
|
|||
for (Subscription subscription : response.iterateAll()) { |
|||
subscriptionSet.add(subscription.getName()); |
|||
} |
|||
} catch (IOException e) { |
|||
log.error("Failed to get subscriptions.", e); |
|||
throw new RuntimeException("Failed to get subscriptions.", e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void createTopicIfNotExists(String partition) { |
|||
ProjectTopicName topicName = ProjectTopicName.of(pubSubSettings.getProjectId(), partition); |
|||
|
|||
if (topicSet.contains(topicName.toString())) { |
|||
createSubscriptionIfNotExists(partition, topicName); |
|||
return; |
|||
} |
|||
|
|||
try (TopicAdminClient topicAdminClient = TopicAdminClient.create(topicAdminSettings)) { |
|||
ListTopicsRequest listTopicsRequest = |
|||
ListTopicsRequest.newBuilder().setProject(ProjectName.format(pubSubSettings.getProjectId())).build(); |
|||
TopicAdminClient.ListTopicsPagedResponse response = topicAdminClient.listTopics(listTopicsRequest); |
|||
for (Topic topic : response.iterateAll()) { |
|||
if (topic.getName().contains(topicName.toString())) { |
|||
topicSet.add(topic.getName()); |
|||
createSubscriptionIfNotExists(partition, topicName); |
|||
return; |
|||
} |
|||
} |
|||
|
|||
topicAdminClient.createTopic(topicName); |
|||
topicSet.add(topicName.toString()); |
|||
log.info("Created new topic: [{}]", topicName.toString()); |
|||
createSubscriptionIfNotExists(partition, topicName); |
|||
} catch (IOException e) { |
|||
log.error("Failed to create topic: [{}].", topicName.toString(), e); |
|||
throw new RuntimeException("Failed to create topic.", e); |
|||
} |
|||
} |
|||
|
|||
private void createSubscriptionIfNotExists(String partition, ProjectTopicName topicName) { |
|||
ProjectSubscriptionName subscriptionName = |
|||
ProjectSubscriptionName.of(pubSubSettings.getProjectId(), partition); |
|||
|
|||
if (subscriptionSet.contains(subscriptionName.toString())) { |
|||
return; |
|||
} |
|||
|
|||
try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create(subscriptionAdminSettings)) { |
|||
ListSubscriptionsRequest listSubscriptionsRequest = |
|||
ListSubscriptionsRequest.newBuilder() |
|||
.setProject(ProjectName.of(pubSubSettings.getProjectId()).toString()) |
|||
.build(); |
|||
SubscriptionAdminClient.ListSubscriptionsPagedResponse response = |
|||
subscriptionAdminClient.listSubscriptions(listSubscriptionsRequest); |
|||
|
|||
for (Subscription subscription : response.iterateAll()) { |
|||
if (subscription.getName().equals(subscriptionName.toString())) { |
|||
subscriptionSet.add(subscription.getName()); |
|||
return; |
|||
} |
|||
} |
|||
|
|||
subscriptionAdminClient.createSubscription( |
|||
subscriptionName, topicName, PushConfig.getDefaultInstance(), pubSubSettings.getAckDeadline()).getName(); |
|||
subscriptionSet.add(subscriptionName.toString()); |
|||
log.info("Created new subscription: [{}]", subscriptionName.toString()); |
|||
} catch (IOException e) { |
|||
log.error("Failed to create subscription: [{}].", subscriptionName.toString(), e); |
|||
throw new RuntimeException("Failed to create subscription.", e); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,228 @@ |
|||
/** |
|||
* 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.pubsub; |
|||
|
|||
import com.google.api.core.ApiFuture; |
|||
import com.google.api.core.ApiFutures; |
|||
import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub; |
|||
import com.google.cloud.pubsub.v1.stub.SubscriberStub; |
|||
import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings; |
|||
import com.google.common.reflect.TypeToken; |
|||
import com.google.gson.Gson; |
|||
import com.google.protobuf.InvalidProtocolBufferException; |
|||
import com.google.pubsub.v1.AcknowledgeRequest; |
|||
import com.google.pubsub.v1.ProjectSubscriptionName; |
|||
import com.google.pubsub.v1.PubsubMessage; |
|||
import com.google.pubsub.v1.PullRequest; |
|||
import com.google.pubsub.v1.PullResponse; |
|||
import com.google.pubsub.v1.ReceivedMessage; |
|||
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.DefaultTbQueueMsg; |
|||
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.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Slf4j |
|||
public class TbPubSubConsumerTemplate<T extends TbQueueMsg> implements TbQueueConsumer<T> { |
|||
|
|||
private final Gson gson = new Gson(); |
|||
private final TbQueueAdmin admin; |
|||
private final String topic; |
|||
private final TbQueueMsgDecoder<T> decoder; |
|||
private final TbPubSubSettings pubSubSettings; |
|||
|
|||
private volatile boolean subscribed; |
|||
private volatile Set<TopicPartitionInfo> partitions; |
|||
private volatile Set<String> subscriptionNames; |
|||
private final List<AcknowledgeRequest> acknowledgeRequests = new CopyOnWriteArrayList<>(); |
|||
|
|||
private ExecutorService consumerExecutor; |
|||
private final SubscriberStub subscriber; |
|||
private volatile boolean stopped; |
|||
|
|||
private volatile int messagesPerTopic; |
|||
|
|||
public TbPubSubConsumerTemplate(TbQueueAdmin admin, TbPubSubSettings pubSubSettings, String topic, TbQueueMsgDecoder<T> decoder) { |
|||
this.admin = admin; |
|||
this.pubSubSettings = pubSubSettings; |
|||
this.topic = topic; |
|||
this.decoder = decoder; |
|||
|
|||
try { |
|||
SubscriberStubSettings subscriberStubSettings = |
|||
SubscriberStubSettings.newBuilder() |
|||
.setCredentialsProvider(pubSubSettings.getCredentialsProvider()) |
|||
.setTransportChannelProvider( |
|||
SubscriberStubSettings.defaultGrpcTransportProviderBuilder() |
|||
.setMaxInboundMessageSize(pubSubSettings.getMaxMsgSize()) |
|||
.build()) |
|||
.build(); |
|||
|
|||
this.subscriber = GrpcSubscriberStub.create(subscriberStubSettings); |
|||
} catch (IOException e) { |
|||
log.error("Failed to create subscriber.", e); |
|||
throw new RuntimeException("Failed to create subscriber.", e); |
|||
} |
|||
stopped = false; |
|||
} |
|||
|
|||
@Override |
|||
public String getTopic() { |
|||
return topic; |
|||
} |
|||
|
|||
@Override |
|||
public void subscribe() { |
|||
partitions = Collections.singleton(new TopicPartitionInfo(topic, null, null, true)); |
|||
subscribed = false; |
|||
} |
|||
|
|||
@Override |
|||
public void subscribe(Set<TopicPartitionInfo> partitions) { |
|||
this.partitions = partitions; |
|||
subscribed = false; |
|||
} |
|||
|
|||
@Override |
|||
public void unsubscribe() { |
|||
stopped = true; |
|||
if (consumerExecutor != null) { |
|||
consumerExecutor.shutdownNow(); |
|||
} |
|||
|
|||
if (subscriber != null) { |
|||
subscriber.close(); |
|||
} |
|||
} |
|||
|
|||
@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) { |
|||
subscriptionNames = partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toSet()); |
|||
subscriptionNames.forEach(admin::createTopicIfNotExists); |
|||
consumerExecutor = Executors.newFixedThreadPool(subscriptionNames.size()); |
|||
messagesPerTopic = pubSubSettings.getMaxMessages()/subscriptionNames.size(); |
|||
subscribed = true; |
|||
} |
|||
List<ReceivedMessage> messages; |
|||
try { |
|||
messages = receiveMessages(); |
|||
if (!messages.isEmpty()) { |
|||
List<T> result = new ArrayList<>(); |
|||
messages.forEach(msg -> { |
|||
try { |
|||
result.add(decode(msg.getMessage())); |
|||
} catch (InvalidProtocolBufferException e) { |
|||
log.error("Failed decode record: [{}]", msg); |
|||
} |
|||
}); |
|||
return result; |
|||
} |
|||
} catch (ExecutionException | InterruptedException e) { |
|||
if (stopped) { |
|||
log.info("[{}] Pub/Sub consumer is stopped.", topic); |
|||
} else { |
|||
log.error("Failed to receive messages", e); |
|||
} |
|||
} |
|||
} |
|||
return Collections.emptyList(); |
|||
} |
|||
|
|||
@Override |
|||
public void commit() { |
|||
acknowledgeRequests.forEach(subscriber.acknowledgeCallable()::futureCall); |
|||
acknowledgeRequests.clear(); |
|||
} |
|||
|
|||
private List<ReceivedMessage> receiveMessages() throws ExecutionException, InterruptedException { |
|||
List<ApiFuture<List<ReceivedMessage>>> result = subscriptionNames.stream().map(subscriptionId -> { |
|||
String subscriptionName = ProjectSubscriptionName.format(pubSubSettings.getProjectId(), subscriptionId); |
|||
PullRequest pullRequest = |
|||
PullRequest.newBuilder() |
|||
.setMaxMessages(messagesPerTopic) |
|||
.setReturnImmediately(false) // return immediately if messages are not available
|
|||
.setSubscription(subscriptionName) |
|||
.build(); |
|||
|
|||
ApiFuture<PullResponse> pullResponseApiFuture = subscriber.pullCallable().futureCall(pullRequest); |
|||
|
|||
return ApiFutures.transform(pullResponseApiFuture, pullResponse -> { |
|||
if (pullResponse != null && !pullResponse.getReceivedMessagesList().isEmpty()) { |
|||
List<String> ackIds = new ArrayList<>(); |
|||
for (ReceivedMessage message : pullResponse.getReceivedMessagesList()) { |
|||
ackIds.add(message.getAckId()); |
|||
} |
|||
AcknowledgeRequest acknowledgeRequest = |
|||
AcknowledgeRequest.newBuilder() |
|||
.setSubscription(subscriptionName) |
|||
.addAllAckIds(ackIds) |
|||
.build(); |
|||
|
|||
acknowledgeRequests.add(acknowledgeRequest); |
|||
return pullResponse.getReceivedMessagesList(); |
|||
} |
|||
return null; |
|||
}, consumerExecutor); |
|||
|
|||
}).collect(Collectors.toList()); |
|||
|
|||
ApiFuture<List<ReceivedMessage>> transform = ApiFutures.transform(ApiFutures.allAsList(result), listMessages -> { |
|||
if (!CollectionUtils.isEmpty(listMessages)) { |
|||
return listMessages.stream().filter(Objects::nonNull).flatMap(List::stream).collect(Collectors.toList()); |
|||
} |
|||
return Collections.emptyList(); |
|||
}, consumerExecutor); |
|||
|
|||
return transform.get(); |
|||
} |
|||
|
|||
public T decode(PubsubMessage message) throws InvalidProtocolBufferException { |
|||
DefaultTbQueueMsg msg = gson.fromJson(message.getData().toStringUtf8(), DefaultTbQueueMsg.class); |
|||
TbQueueMsgHeaders headers = new DefaultTbQueueMsgHeaders(); |
|||
Map<String, byte[]> headerMap = gson.fromJson(message.getAttributesMap().get("headers"), new TypeToken<Map<String, byte[]>>() { |
|||
}.getType()); |
|||
headerMap.forEach(headers::put); |
|||
msg.setHeaders(headers); |
|||
return decoder.decode(msg); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,135 @@ |
|||
/** |
|||
* 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.pubsub; |
|||
|
|||
import com.google.api.core.ApiFuture; |
|||
import com.google.api.core.ApiFutureCallback; |
|||
import com.google.api.core.ApiFutures; |
|||
import com.google.cloud.pubsub.v1.Publisher; |
|||
import com.google.gson.Gson; |
|||
import com.google.protobuf.ByteString; |
|||
import com.google.pubsub.v1.ProjectTopicName; |
|||
import com.google.pubsub.v1.PubsubMessage; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.queue.TbQueueAdmin; |
|||
import org.thingsboard.server.queue.TbQueueCallback; |
|||
import org.thingsboard.server.queue.TbQueueMsg; |
|||
import org.thingsboard.server.queue.TbQueueProducer; |
|||
import org.thingsboard.server.queue.common.DefaultTbQueueMsg; |
|||
|
|||
import java.io.IOException; |
|||
import java.util.Map; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Slf4j |
|||
public class TbPubSubProducerTemplate<T extends TbQueueMsg> implements TbQueueProducer<T> { |
|||
|
|||
private final Gson gson = new Gson(); |
|||
|
|||
private final String defaultTopic; |
|||
private final TbQueueAdmin admin; |
|||
private final TbPubSubSettings pubSubSettings; |
|||
|
|||
private final Map<String, Publisher> publisherMap = new ConcurrentHashMap<>(); |
|||
|
|||
private ExecutorService pubExecutor = Executors.newCachedThreadPool(); |
|||
|
|||
public TbPubSubProducerTemplate(TbQueueAdmin admin, TbPubSubSettings pubSubSettings, String defaultTopic) { |
|||
this.defaultTopic = defaultTopic; |
|||
this.admin = admin; |
|||
this.pubSubSettings = pubSubSettings; |
|||
} |
|||
|
|||
@Override |
|||
public void init() { |
|||
|
|||
} |
|||
|
|||
@Override |
|||
public String getDefaultTopic() { |
|||
return defaultTopic; |
|||
} |
|||
|
|||
@Override |
|||
public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) { |
|||
PubsubMessage.Builder pubsubMessageBuilder = PubsubMessage.newBuilder(); |
|||
pubsubMessageBuilder.setData(getMsg(msg)); |
|||
pubsubMessageBuilder.putAttributes("headers", gson.toJson(msg.getHeaders().getData())); |
|||
|
|||
Publisher publisher = getOrCreatePublisher(tpi.getFullTopicName()); |
|||
ApiFuture<String> future = publisher.publish(pubsubMessageBuilder.build()); |
|||
|
|||
ApiFutures.addCallback(future, new ApiFutureCallback<String>() { |
|||
public void onSuccess(String messageId) { |
|||
if (callback != null) { |
|||
callback.onSuccess(null); |
|||
} |
|||
} |
|||
|
|||
public void onFailure(Throwable t) { |
|||
if (callback != null) { |
|||
callback.onFailure(t); |
|||
} |
|||
} |
|||
}, pubExecutor); |
|||
} |
|||
|
|||
@Override |
|||
public void stop() { |
|||
publisherMap.forEach((k, v) -> { |
|||
if (v != null) { |
|||
try { |
|||
v.shutdown(); |
|||
v.awaitTermination(1, TimeUnit.SECONDS); |
|||
} catch (Exception e) { |
|||
log.error("Failed to shutdown PubSub client during destroy()", e); |
|||
} |
|||
} |
|||
}); |
|||
|
|||
if (pubExecutor != null) { |
|||
pubExecutor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
private ByteString getMsg(T msg) { |
|||
String json = gson.toJson(new DefaultTbQueueMsg(msg.getKey(), msg.getData())); |
|||
return ByteString.copyFrom(json.getBytes()); |
|||
} |
|||
|
|||
private Publisher getOrCreatePublisher(String topic) { |
|||
if (publisherMap.containsKey(topic)) { |
|||
return publisherMap.get(topic); |
|||
} else { |
|||
try { |
|||
admin.createTopicIfNotExists(topic); |
|||
ProjectTopicName topicName = ProjectTopicName.of(pubSubSettings.getProjectId(), topic); |
|||
Publisher publisher = Publisher.newBuilder(topicName).setCredentialsProvider(pubSubSettings.getCredentialsProvider()).build(); |
|||
publisherMap.put(topic, publisher); |
|||
return publisher; |
|||
} catch (IOException e) { |
|||
log.error("Failed to create topic [{}].", topic, e); |
|||
throw new RuntimeException("Failed to create topic.", e); |
|||
} |
|||
} |
|||
|
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,61 @@ |
|||
/** |
|||
* 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.pubsub; |
|||
|
|||
import com.google.api.gax.core.CredentialsProvider; |
|||
import com.google.api.gax.core.FixedCredentialsProvider; |
|||
import com.google.auth.oauth2.ServiceAccountCredentials; |
|||
import lombok.Data; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import java.io.ByteArrayInputStream; |
|||
import java.io.IOException; |
|||
|
|||
@Slf4j |
|||
@ConditionalOnExpression("'${queue.type:null}'=='pubsub'") |
|||
@Component |
|||
@Data |
|||
public class TbPubSubSettings { |
|||
|
|||
@Value("${queue.pubsub.project_id}") |
|||
private String projectId; |
|||
|
|||
@Value("${queue.pubsub.service_account}") |
|||
private String serviceAccount; |
|||
|
|||
@Value("${queue.pubsub.ack_deadline}") |
|||
private int ackDeadline; |
|||
|
|||
@Value("${queue.pubsub.max_msg_size}") |
|||
private int maxMsgSize; |
|||
|
|||
@Value("${queue.pubsub.max_messages}") |
|||
private int maxMessages; |
|||
|
|||
private CredentialsProvider credentialsProvider; |
|||
|
|||
@PostConstruct |
|||
private void init() throws IOException { |
|||
ServiceAccountCredentials credentials = ServiceAccountCredentials.fromStream( |
|||
new ByteArrayInputStream(serviceAccount.getBytes())); |
|||
credentialsProvider = FixedCredentialsProvider.create(credentials); |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue