diff --git a/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java b/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java index b0554decea..8d1f9d662a 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java @@ -21,14 +21,15 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import lombok.Getter; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; -import org.thingsboard.server.queue.TbQueueProducer; -import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; -import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.gen.js.JsInvokeProtos; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; import javax.annotation.Nullable; import javax.annotation.PostConstruct; @@ -41,28 +42,13 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; @Slf4j -@ConditionalOnProperty(prefix = "js", value = "evaluator", havingValue = "remote", matchIfMissing = true) +@ConditionalOnExpression("'${js.evaluator:null}'=='remote' && ('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-core' || '${service.type:null}'=='tb-rule-engine')") @Service public class RemoteJsInvokeService extends AbstractJsInvokeService { - @Value("${js.remote.request_topic}") - private String requestTopic; - - @Value("${js.remote.response_topic_prefix}") - private String responseTopicPrefix; - - @Value("${js.remote.max_pending_requests}") - private long maxPendingRequests; - @Value("${js.remote.max_requests_timeout}") private long maxRequestsTimeout; - @Value("${js.remote.response_poll_interval}") - private int responsePollDuration; - - @Value("${js.remote.response_auto_commit_interval}") - private int autoCommitInterval; - @Getter @Value("${js.remote.max_errors}") private int maxErrors; @@ -94,43 +80,20 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService { } } - private DefaultTbQueueRequestTemplate, TbProtoQueueMsg> defaultTemplate; + @Autowired + private TbQueueRequestTemplate, TbProtoQueueMsg> requestTemplate; + private Map scriptIdToBodysMap = new ConcurrentHashMap<>(); @PostConstruct public void init() { -// TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); -// requestBuilder.settings(kafkaSettings); -// requestBuilder.clientId("producer-js-invoke-" + nodeIdProvider.getNodeId()); -// requestBuilder.defaultTopic(requestTopic); -// requestBuilder.encoder(new RemoteJsRequestEncoder()); - TbQueueProducer> producer; - -// TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder responseBuilder = TBKafkaConsumerTemplate.builder(); -// responseBuilder.settings(kafkaSettings); -// responseBuilder.topic(responseTopicPrefix + "." + nodeIdProvider.getNodeId()); -// responseBuilder.clientId("js-" + nodeIdProvider.getNodeId()); -// responseBuilder.groupId("rule-engine-node-" + nodeIdProvider.getNodeId()); -// responseBuilder.autoCommit(true); -// responseBuilder.autoCommitIntervalMs(autoCommitInterval); -// responseBuilder.decoder(new RemoteJsResponseDecoder()); -// responseBuilder.requestIdExtractor((response) -> new UUID(response.getRequestIdMSB(), response.getRequestIdLSB())); -// -// TbKafkaRequestTemplate.TbKafkaRequestTemplateBuilder -// builder = TbKafkaRequestTemplate.builder(); -// builder.requestTemplate(requestBuilder.build()); -// builder.responseTemplate(responseBuilder.build()); -// builder.maxPendingRequests(maxPendingRequests); -// builder.maxRequestTimeout(maxRequestsTimeout); -// builder.pollInterval(responsePollDuration); -// defaultTemplate = builder.build(); -// defaultTemplate.init(); + requestTemplate.init(); } @PreDestroy public void destroy() { - if (defaultTemplate != null) { - defaultTemplate.stop(); + if (requestTemplate != null) { + requestTemplate.stop(); } } @@ -147,7 +110,7 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService { .build(); log.trace("Post compile request for scriptId [{}]", scriptId); - ListenableFuture> future = defaultTemplate.send(new TbProtoQueueMsg<>(UUID.randomUUID(), jsRequestWrapper)); + ListenableFuture> future = requestTemplate.send(new TbProtoJsQueueMsg<>(UUID.randomUUID(), jsRequestWrapper)); kafkaPushedMsgs.incrementAndGet(); Futures.addCallback(future, new FutureCallback>() { @@ -199,7 +162,7 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService { .setInvokeRequest(jsRequestBuilder.build()) .build(); - ListenableFuture> future = defaultTemplate.send(new TbProtoQueueMsg<>(UUID.randomUUID(), jsRequestWrapper)); + ListenableFuture> future = requestTemplate.send(new TbProtoJsQueueMsg<>(UUID.randomUUID(), jsRequestWrapper)); kafkaPushedMsgs.incrementAndGet(); Futures.addCallback(future, new FutureCallback>() { @Override @@ -239,7 +202,7 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService { .setReleaseRequest(jsRequest) .build(); - ListenableFuture> future = defaultTemplate.send(new TbProtoQueueMsg<>(UUID.randomUUID(), jsRequestWrapper)); + ListenableFuture> future = requestTemplate.send(new TbProtoJsQueueMsg<>(UUID.randomUUID(), jsRequestWrapper)); JsInvokeProtos.RemoteJsResponse response = future.get().getValue(); JsInvokeProtos.JsReleaseResponse compilationResult = response.getReleaseResponse(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/TbProtoJsQueueMsg.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/TbProtoJsQueueMsg.java new file mode 100644 index 0000000000..07417c4a9e --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/TbProtoJsQueueMsg.java @@ -0,0 +1,43 @@ +/** + * 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.common; + +import com.google.protobuf.InvalidProtocolBufferException; +import com.google.protobuf.util.JsonFormat; +import org.thingsboard.server.queue.TbQueueMsgHeaders; + +import java.nio.charset.StandardCharsets; +import java.util.UUID; + +public class TbProtoJsQueueMsg extends TbProtoQueueMsg { + + public TbProtoJsQueueMsg(UUID key, T value) { + super(key, value); + } + + public TbProtoJsQueueMsg(UUID key, T value, TbQueueMsgHeaders headers) { + super(key, value, headers); + } + + @Override + public byte[] getData() { + try { + return JsonFormat.printer().print(value).getBytes(StandardCharsets.UTF_8); + } catch (InvalidProtocolBufferException e) { + throw new RuntimeException(e); + } + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/TbProtoQueueMsg.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/TbProtoQueueMsg.java index 8823002530..2eb76950dc 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/TbProtoQueueMsg.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/TbProtoQueueMsg.java @@ -25,7 +25,7 @@ import java.util.UUID; public class TbProtoQueueMsg implements TbQueueMsg { private final UUID key; - private final T value; + protected final T value; private final TbQueueMsgHeaders headers; public TbProtoQueueMsg(UUID key, T value) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java index 9f70345301..26da11aa3b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java @@ -18,10 +18,13 @@ 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.js.JsInvokeProtos; 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.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; @@ -125,4 +128,9 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng public TbQueueProducer> createTransportApiResponseProducer() { return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, transportApiSettings.getResponsesTopic()); } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + return null; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java index 770e7fa65c..87aba51843 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java @@ -18,6 +18,7 @@ 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.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -28,6 +29,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponse import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; @@ -114,4 +117,9 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { public TbQueueProducer> createTransportApiResponseProducer() { return new TbAwsSqsProducerTemplate<>(admin, sqsSettings, coreSettings.getTopic()); } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + return null; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java index f87f108453..d83b760c19 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java @@ -18,6 +18,7 @@ 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.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -25,6 +26,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; @@ -96,4 +99,9 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + return null; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java index a23c93e015..baad96182f 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java @@ -18,6 +18,7 @@ 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.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -27,6 +28,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestM import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.memory.InMemoryTbQueueConsumer; import org.thingsboard.server.queue.memory.InMemoryTbQueueProducer; @@ -110,4 +113,9 @@ public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new InMemoryTbQueueConsumer<>(ruleEngineSettings.getTopic() + ".notifications"); } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + return null; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java index fa36f8284b..dae4221e91 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java @@ -15,9 +15,12 @@ */ package org.thingsboard.server.queue.provider; +import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -27,18 +30,25 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestM import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; 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.kafka.TBKafkaAdmin; import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; +import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; +import java.nio.charset.StandardCharsets; + @Component @ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='monolith'") public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngineQueueFactory { @@ -50,13 +60,15 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; + private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; public KafkaMonolithQueueFactory(PartitionService partitionService, TbKafkaSettings kafkaSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbQueueTransportApiSettings transportApiSettings, - TbQueueTransportNotificationSettings transportNotificationSettings) { + TbQueueTransportNotificationSettings transportNotificationSettings, + TbQueueRemoteJsInvokeSettings jsInvokeSettings) { this.partitionService = partitionService; this.kafkaSettings = kafkaSettings; this.serviceInfoProvider = serviceInfoProvider; @@ -64,6 +76,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi this.ruleEngineSettings = ruleEngineSettings; this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; + this.jsInvokeSettings = jsInvokeSettings; } @Override @@ -175,4 +188,37 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi requestBuilder.defaultTopic(transportApiSettings.getResponsesTopic()); return requestBuilder.build(); } + + @Override + @Bean + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-js-invoke-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(jsInvokeSettings.getRequestTopic()); + + TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); + responseBuilder.settings(kafkaSettings); + responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); + responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); + responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId()); +// responseBuilder.autoCommit(true); +// responseBuilder.autoCommitIntervalMs(autoCommitInterval); + responseBuilder.decoder(msg -> { + JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); + JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); + return new TbProtoQueueMsg<>(msg.getKey(), builder.build(), msg.getHeaders()); + } + ); + + DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder + , TbProtoQueueMsg> builder = DefaultTbQueueRequestTemplate.builder(); + builder.queueAdmin(new TBKafkaAdmin(kafkaSettings)); + builder.requestTemplate(requestBuilder.build()); + builder.responseTemplate(responseBuilder.build()); + builder.maxPendingRequests(jsInvokeSettings.getMaxPendingRequests()); + builder.maxRequestTimeout(jsInvokeSettings.getMaxRequestsTimeout()); + builder.pollInterval(jsInvokeSettings.getResponsePollInterval()); + return builder.build(); + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java index 5e46d67eed..e148416bb4 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java @@ -15,9 +15,12 @@ */ package org.thingsboard.server.queue.provider; +import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -27,16 +30,23 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestM import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; 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.kafka.TBKafkaAdmin; import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; +import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; +import java.nio.charset.StandardCharsets; + @Component @ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='tb-core'") public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { @@ -47,18 +57,21 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueCoreSettings coreSettings; private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueTransportApiSettings transportApiSettings; + private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; public KafkaTbCoreQueueFactory(PartitionService partitionService, TbKafkaSettings kafkaSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, - TbQueueTransportApiSettings transportApiSettings) { + TbQueueTransportApiSettings transportApiSettings, + TbQueueRemoteJsInvokeSettings jsInvokeSettings) { this.partitionService = partitionService; this.kafkaSettings = kafkaSettings; this.serviceInfoProvider = serviceInfoProvider; this.coreSettings = coreSettings; this.ruleEngineSettings = ruleEngineSettings; this.transportApiSettings = transportApiSettings; + this.jsInvokeSettings = jsInvokeSettings; } @Override @@ -148,4 +161,37 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { return requestBuilder.build(); } + @Override + @Bean + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-js-invoke-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(jsInvokeSettings.getRequestTopic()); + + TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); + responseBuilder.settings(kafkaSettings); + responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); + responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); + responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId()); +// responseBuilder.autoCommit(true); +// responseBuilder.autoCommitIntervalMs(autoCommitInterval); + responseBuilder.decoder(msg -> { + JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); + JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); + return new TbProtoQueueMsg<>(msg.getKey(), builder.build(), msg.getHeaders()); + } + ); + + DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder + , TbProtoQueueMsg> builder = DefaultTbQueueRequestTemplate.builder(); + builder.queueAdmin(new TBKafkaAdmin(kafkaSettings)); + builder.requestTemplate(requestBuilder.build()); + builder.responseTemplate(responseBuilder.build()); + builder.maxPendingRequests(jsInvokeSettings.getMaxPendingRequests()); + builder.maxRequestTimeout(jsInvokeSettings.getMaxRequestsTimeout()); + builder.pollInterval(jsInvokeSettings.getResponsePollInterval()); + return builder.build(); + } + } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java index eef10f35dc..8262d27354 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java @@ -15,9 +15,12 @@ */ package org.thingsboard.server.queue.provider; +import com.google.protobuf.util.JsonFormat; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -25,16 +28,23 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotifica import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; 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.kafka.TBKafkaAdmin; import org.thingsboard.server.queue.kafka.TBKafkaConsumerTemplate; import org.thingsboard.server.queue.kafka.TBKafkaProducerTemplate; import org.thingsboard.server.queue.kafka.TbKafkaSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; +import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; +import java.nio.charset.StandardCharsets; + @Component @ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='tb-rule-engine'") public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { @@ -44,16 +54,19 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueCoreSettings coreSettings; private final TbQueueRuleEngineSettings ruleEngineSettings; + private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; public KafkaTbRuleEngineQueueFactory(PartitionService partitionService, TbKafkaSettings kafkaSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, - TbQueueRuleEngineSettings ruleEngineSettings) { + TbQueueRuleEngineSettings ruleEngineSettings, + TbQueueRemoteJsInvokeSettings jsInvokeSettings) { this.partitionService = partitionService; this.kafkaSettings = kafkaSettings; this.serviceInfoProvider = serviceInfoProvider; this.coreSettings = coreSettings; this.ruleEngineSettings = ruleEngineSettings; + this.jsInvokeSettings = jsInvokeSettings; } @Override @@ -124,4 +137,37 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); return consumerBuilder.build(); } + + @Override + @Bean + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-js-invoke-" + serviceInfoProvider.getServiceId()); + requestBuilder.defaultTopic(jsInvokeSettings.getRequestTopic()); + + TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); + responseBuilder.settings(kafkaSettings); + responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); + responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); + responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId()); +// responseBuilder.autoCommit(true); +// responseBuilder.autoCommitIntervalMs(autoCommitInterval); + responseBuilder.decoder(msg -> { + JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); + JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); + return new TbProtoQueueMsg<>(msg.getKey(), builder.build(), msg.getHeaders()); + } + ); + + DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder + , TbProtoQueueMsg> builder = DefaultTbQueueRequestTemplate.builder(); + builder.queueAdmin(new TBKafkaAdmin(kafkaSettings)); + builder.requestTemplate(requestBuilder.build()); + builder.responseTemplate(responseBuilder.build()); + builder.maxPendingRequests(jsInvokeSettings.getMaxPendingRequests()); + builder.maxRequestTimeout(jsInvokeSettings.getMaxRequestsTimeout()); + builder.pollInterval(jsInvokeSettings.getResponsePollInterval()); + return builder.build(); + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java index f00bd7caef..b1afc61dbd 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java @@ -18,6 +18,7 @@ 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.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -28,6 +29,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponse import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; @@ -133,4 +136,9 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng public TbQueueProducer> createTransportApiResponseProducer() { return new TbPubSubProducerTemplate<>(admin, pubSubSettings, transportApiSettings.getResponsesTopic()); } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + return null; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java index da2fe3acdf..edfcdc3d8e 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java @@ -18,6 +18,7 @@ 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.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -27,6 +28,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestM 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.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; @@ -110,4 +113,9 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { public TbQueueProducer> createTransportApiResponseProducer() { return new TbPubSubProducerTemplate<>(admin, pubSubSettings, coreSettings.getTopic()); } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + return null; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java index da65d583c1..101f335536 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java @@ -18,6 +18,7 @@ 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.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -25,15 +26,17 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotifica 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.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.TbQueueProducer; -import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; 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; +import org.thingsboard.server.queue.settings.TbQueueCoreSettings; +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; @Component @@ -99,4 +102,9 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + return null; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java index 914e200fc9..8b01c38607 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java @@ -18,6 +18,7 @@ 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.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -28,9 +29,11 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponse 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.azure.servicebus.TbServiceBusConsumerTemplate; import org.thingsboard.server.queue.azure.servicebus.TbServiceBusProducerTemplate; import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; @@ -131,4 +134,9 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul public TbQueueProducer> createTransportApiResponseProducer() { return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, transportApiSettings.getResponsesTopic()); } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + return null; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java similarity index 84% rename from common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueProvider.java rename to common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java index 334af58133..523ac88bd1 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java @@ -18,6 +18,7 @@ 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.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -27,9 +28,11 @@ import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponse 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.azure.servicebus.TbServiceBusConsumerTemplate; import org.thingsboard.server.queue.azure.servicebus.TbServiceBusProducerTemplate; import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; @@ -39,7 +42,7 @@ import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; @Component @ConditionalOnExpression("'${queue.type:null}'=='service-bus' && '${service.type:null}'=='tb-core'") -public class ServiceBusTbCoreQueueProvider implements TbCoreQueueFactory { +public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { private final TbServiceBusSettings serviceBusSettings; private final TbQueueRuleEngineSettings ruleEngineSettings; @@ -49,13 +52,13 @@ public class ServiceBusTbCoreQueueProvider implements TbCoreQueueFactory { private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueAdmin admin; - public ServiceBusTbCoreQueueProvider(TbServiceBusSettings serviceBusSettings, - TbQueueCoreSettings coreSettings, - TbQueueTransportApiSettings transportApiSettings, - TbQueueRuleEngineSettings ruleEngineSettings, - PartitionService partitionService, - TbServiceInfoProvider serviceInfoProvider, - TbQueueAdmin admin) { + public ServiceBusTbCoreQueueFactory(TbServiceBusSettings serviceBusSettings, + TbQueueCoreSettings coreSettings, + TbQueueTransportApiSettings transportApiSettings, + TbQueueRuleEngineSettings ruleEngineSettings, + PartitionService partitionService, + TbServiceInfoProvider serviceInfoProvider, + TbQueueAdmin admin) { this.serviceBusSettings = serviceBusSettings; this.coreSettings = coreSettings; this.transportApiSettings = transportApiSettings; @@ -113,4 +116,9 @@ public class ServiceBusTbCoreQueueProvider implements TbCoreQueueFactory { public TbQueueProducer> createTransportApiResponseProducer() { return new TbServiceBusProducerTemplate<>(admin, serviceBusSettings, coreSettings.getTopic()); } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + return null; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java index 3ceb837606..8a2f9b396b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java @@ -18,6 +18,7 @@ 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.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -25,9 +26,11 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.azure.servicebus.TbServiceBusConsumerTemplate; import org.thingsboard.server.queue.azure.servicebus.TbServiceBusProducerTemplate; import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; @@ -96,4 +99,9 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + return null; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java index 1a3ba1ab98..59a8f449b0 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java @@ -15,15 +15,19 @@ */ package org.thingsboard.server.queue.provider; -import org.thingsboard.server.gen.transport.TransportProtos.*; -import org.thingsboard.server.queue.TbQueueConsumer; -import org.thingsboard.server.queue.TbQueueProducer; -import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.gen.js.JsInvokeProtos; 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.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; /** * Responsible for initialization of various Producers and Consumers used by TB Core Node. @@ -94,5 +98,5 @@ public interface TbCoreQueueFactory { */ TbQueueProducer> createTransportApiResponseProducer(); - + TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java index 300fb986a9..561b3e84ea 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java @@ -15,14 +15,17 @@ */ package org.thingsboard.server.queue.provider; +import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.queue.TbQueueConsumer; -import org.thingsboard.server.queue.TbQueueProducer; -import org.thingsboard.server.queue.common.TbProtoQueueMsg; 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.ToRuleEngineNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; /** @@ -82,4 +85,5 @@ public interface TbRuleEngineQueueFactory { */ TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer(); + TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueRemoteJsInvokeSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueRemoteJsInvokeSettings.java new file mode 100644 index 0000000000..cdad407b19 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueRemoteJsInvokeSettings.java @@ -0,0 +1,42 @@ +/** + * 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.settings; + +import lombok.Data; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +@Data +@Component +public class TbQueueRemoteJsInvokeSettings { + @Value("${js.remote.request_topic}") + private String requestTopic; + + @Value("${js.remote.response_topic_prefix}") + private String responseTopic; + + @Value("${js.remote.max_pending_requests}") + private long maxPendingRequests; + + @Value("${js.remote.response_poll_interval}") + private int responsePollInterval; + + @Value("${js.remote.response_auto_commit_interval}") + private int autoCommitInterval; + + @Value("${js.remote.max_requests_timeout}") + private long maxRequestsTimeout; +} diff --git a/application/src/main/proto/jsinvoke.proto b/common/queue/src/main/proto/jsinvoke.proto similarity index 100% rename from application/src/main/proto/jsinvoke.proto rename to common/queue/src/main/proto/jsinvoke.proto diff --git a/msa/js-executor/api/jsInvokeMessageProcessor.js b/msa/js-executor/api/jsInvokeMessageProcessor.js index c17b1ddde4..f0facf8cc1 100644 --- a/msa/js-executor/api/jsInvokeMessageProcessor.js +++ b/msa/js-executor/api/jsInvokeMessageProcessor.js @@ -19,6 +19,7 @@ const COMPILATION_ERROR = 0; const RUNTIME_ERROR = 1; const TIMEOUT_ERROR = 2; const UNRECOGNIZED = -1; +let headers; const config = require('config'), logger = require('../config/logger')._logger('JsInvokeMessageProcessor'), @@ -43,6 +44,7 @@ JsInvokeMessageProcessor.prototype.onJsInvokeMessage = function(message) { var responseTopic; try { var request = JSON.parse(message.value.toString('utf8')); + headers = message.headers; var buf = message.headers['requestId']; requestId = Utils.UUIDFromBuffer(buf); buf = message.headers['responseTopic']; @@ -148,7 +150,8 @@ JsInvokeMessageProcessor.prototype.sendResponse = function (requestId, responseT messages: [ { key: scriptId, - value: rawResponse + value: rawResponse, + headers: headers } ] }