From 0dc31fbdde07dada1e79e8d7ce1c57cab97324b9 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Tue, 28 Apr 2020 16:13:16 +0300 Subject: [PATCH] created awsSqs, pubSub, rabbitmq js-executors --- .../provider/AwsSqsMonolithQueueFactory.java | 36 +++- .../provider/PubSubMonolithQueueFactory.java | 36 +++- .../RabbitMqMonolithQueueFactory.java | 37 +++- .../api/jsInvokeMessageProcessor.js | 59 +++---- .../config/custom-environment-variables.yml | 23 ++- msa/js-executor/config/default.yml | 14 +- msa/js-executor/package.json | 3 + msa/js-executor/queue/awsSqsTemplate.js | 163 ++++++++++++++++++ .../queue/{kafka => }/kafkaTemplate.js | 61 ++++--- msa/js-executor/queue/pubSubTemplate.js | 87 ++++++++++ msa/js-executor/queue/rabbitmqTemplate.js | 149 ++++++++++++++++ msa/js-executor/server.js | 24 ++- 12 files changed, 616 insertions(+), 76 deletions(-) create mode 100644 msa/js-executor/queue/awsSqsTemplate.js rename msa/js-executor/queue/{kafka => }/kafkaTemplate.js (64%) create mode 100644 msa/js-executor/queue/pubSubTemplate.js create mode 100644 msa/js-executor/queue/rabbitmqTemplate.js 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 76ff04c238..c86baf9c48 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 @@ -15,20 +15,26 @@ */ 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.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.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.pubsub.TbPubSubConsumerTemplate; +import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; 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; @@ -40,6 +46,7 @@ import org.thingsboard.server.queue.sqs.TbAwsSqsQueueAttributes; import org.thingsboard.server.queue.sqs.TbAwsSqsSettings; import javax.annotation.PreDestroy; +import java.nio.charset.StandardCharsets; @Component @ConditionalOnExpression("'${queue.type:null}'=='aws-sqs' && '${service.type:null}'=='monolith'") @@ -52,6 +59,7 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbAwsSqsSettings sqsSettings; + private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -65,7 +73,8 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, TbAwsSqsSettings sqsSettings, - TbAwsSqsQueueAttributes sqsQueueAttributes) { + TbAwsSqsQueueAttributes sqsQueueAttributes, + TbQueueRemoteJsInvokeSettings jsInvokeSettings) { this.partitionService = partitionService; this.coreSettings = coreSettings; this.serviceInfoProvider = serviceInfoProvider; @@ -73,6 +82,7 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; this.sqsSettings = sqsSettings; + this.jsInvokeSettings = jsInvokeSettings; this.coreAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getCoreAttributes()); this.ruleEngineAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getRuleEngineAttributes()); @@ -144,8 +154,26 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng } @Override - public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { - return null; + @Bean + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + TbQueueProducer> producer = new TbAwsSqsProducerTemplate<>(jsExecutorAdmin, sqsSettings, jsInvokeSettings.getRequestTopic()); + TbQueueConsumer> consumer = new TbAwsSqsConsumerTemplate<>(jsExecutorAdmin, sqsSettings, + jsInvokeSettings.getResponseTopic() + "_" + serviceInfoProvider.getServiceId(), + msg -> { + RemoteJsResponse.Builder builder = 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(jsExecutorAdmin); + builder.requestTemplate(producer); + builder.responseTemplate(consumer); + builder.maxPendingRequests(jsInvokeSettings.getMaxPendingRequests()); + builder.maxRequestTimeout(jsInvokeSettings.getMaxRequestsTimeout()); + builder.pollInterval(jsInvokeSettings.getResponsePollInterval()); + return builder.build(); } @PreDestroy 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 b1afc61dbd..760fa7708d 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 @@ -15,10 +15,13 @@ */ 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.js.JsInvokeProtos.RemoteJsRequest; +import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -30,6 +33,7 @@ import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; @@ -39,11 +43,14 @@ 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.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}'=='pubsub' && '${service.type:null}'=='monolith'") public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngineQueueFactory { @@ -56,6 +63,7 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng private final TbQueueAdmin admin; private final PartitionService partitionService; private final TbServiceInfoProvider serviceInfoProvider; + private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; public PubSubMonolithQueueFactory(TbPubSubSettings pubSubSettings, TbQueueCoreSettings coreSettings, @@ -63,7 +71,8 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, PartitionService partitionService, - TbServiceInfoProvider serviceInfoProvider) { + TbServiceInfoProvider serviceInfoProvider, + TbQueueRemoteJsInvokeSettings jsInvokeSettings) { this.pubSubSettings = pubSubSettings; this.coreSettings = coreSettings; this.ruleEngineSettings = ruleEngineSettings; @@ -72,6 +81,7 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng this.admin = new TbPubSubAdmin(pubSubSettings); this.partitionService = partitionService; this.serviceInfoProvider = serviceInfoProvider; + this.jsInvokeSettings = jsInvokeSettings; } @Override @@ -138,7 +148,25 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng } @Override - public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { - return null; + @Bean + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + TbQueueProducer> producer = new TbPubSubProducerTemplate<>(admin, pubSubSettings, jsInvokeSettings.getRequestTopic()); + TbQueueConsumer> consumer = new TbPubSubConsumerTemplate<>(admin, pubSubSettings, + jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId(), + msg -> { + RemoteJsResponse.Builder builder = 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(admin); + builder.requestTemplate(producer); + builder.responseTemplate(consumer); + 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/RabbitMqMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java index ff4a69e2e6..2435b30851 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java @@ -15,28 +15,37 @@ */ 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.js.JsInvokeProtos.RemoteJsRequest; +import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; 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.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.pubsub.TbPubSubConsumerTemplate; +import org.thingsboard.server.queue.pubsub.TbPubSubProducerTemplate; import org.thingsboard.server.queue.rabbitmq.TbRabbitMqConsumerTemplate; import org.thingsboard.server.queue.rabbitmq.TbRabbitMqProducerTemplate; import org.thingsboard.server.queue.rabbitmq.TbRabbitMqSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; +import org.thingsboard.server.queue.settings.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}'=='rabbitmq' && '${service.type:null}'=='monolith'") public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngineQueueFactory { @@ -48,6 +57,8 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbRabbitMqSettings rabbitMqSettings; + private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueAdmin admin; public RabbitMqMonolithQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, @@ -56,6 +67,7 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, TbRabbitMqSettings rabbitMqSettings, + TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueAdmin admin) { this.partitionService = partitionService; this.coreSettings = coreSettings; @@ -64,6 +76,7 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; this.rabbitMqSettings = rabbitMqSettings; + this.jsInvokeSettings = jsInvokeSettings; this.admin = admin; } @@ -130,7 +143,25 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE } @Override - public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { - return null; + @Bean + public TbQueueRequestTemplate, TbProtoQueueMsg> createRemoteJsRequestTemplate() { + TbQueueProducer> producer = new TbRabbitMqProducerTemplate<>(admin, rabbitMqSettings, jsInvokeSettings.getRequestTopic()); + TbQueueConsumer> consumer = new TbRabbitMqConsumerTemplate<>(admin, rabbitMqSettings, + jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId(), + msg -> { + RemoteJsResponse.Builder builder = 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(admin); + builder.requestTemplate(producer); + builder.responseTemplate(consumer); + builder.maxPendingRequests(jsInvokeSettings.getMaxPendingRequests()); + builder.maxRequestTimeout(jsInvokeSettings.getMaxRequestsTimeout()); + builder.pollInterval(jsInvokeSettings.getResponsePollInterval()); + return builder.build(); } } diff --git a/msa/js-executor/api/jsInvokeMessageProcessor.js b/msa/js-executor/api/jsInvokeMessageProcessor.js index 4cf20b4227..f6da6886e7 100644 --- a/msa/js-executor/api/jsInvokeMessageProcessor.js +++ b/msa/js-executor/api/jsInvokeMessageProcessor.js @@ -19,7 +19,6 @@ 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'), @@ -31,7 +30,7 @@ const useSandbox = config.get('script.use_sandbox') === 'true'; const maxActiveScripts = Number(config.get('script.max_active_scripts')); function JsInvokeMessageProcessor(producer) { - console.log("Kafka Producer:", producer); + console.log("Producer:", producer); this.producer = producer; this.executor = new JsExecutor(useSandbox); this.scriptMap = {}; @@ -39,26 +38,27 @@ function JsInvokeMessageProcessor(producer) { this.executedScriptsCounter = 0; } -JsInvokeMessageProcessor.prototype.onJsInvokeMessage = function(message) { +JsInvokeMessageProcessor.prototype.onJsInvokeMessage = function(messageStr) { - var requestId; - var responseTopic; + let requestId; + let responseTopic; try { - var request = JSON.parse(message.value.toString('utf8')); - headers = message.headers; - var buf = message.headers['requestId']; + let message = JSON.parse(messageStr); + let request = JSON.parse(Buffer.from(message.data).toString('utf8')); + let headers = message.headers; + let buf = Buffer.from(headers.data['requestId']); requestId = Utils.UUIDFromBuffer(buf); - buf = message.headers['responseTopic']; + buf = Buffer.from(headers.data['responseTopic']); responseTopic = buf.toString('utf8'); logger.debug('[%s] Received request, responseTopic: [%s]', requestId, responseTopic); if (request.compileRequest) { - this.processCompileRequest(requestId, responseTopic, request.compileRequest); + this.processCompileRequest(requestId, responseTopic, headers, request.compileRequest); } else if (request.invokeRequest) { - this.processInvokeRequest(requestId, responseTopic, request.invokeRequest); + this.processInvokeRequest(requestId, responseTopic, headers, request.invokeRequest); } else if (request.releaseRequest) { - this.processReleaseRequest(requestId, responseTopic, request.releaseRequest); + this.processReleaseRequest(requestId, responseTopic, headers, request.releaseRequest); } else { logger.error('[%s] Unknown request recevied!', requestId); } @@ -69,7 +69,7 @@ JsInvokeMessageProcessor.prototype.onJsInvokeMessage = function(message) { } } -JsInvokeMessageProcessor.prototype.processCompileRequest = function(requestId, responseTopic, compileRequest) { +JsInvokeMessageProcessor.prototype.processCompileRequest = function(requestId, responseTopic, headers, compileRequest) { var scriptId = getScriptId(compileRequest); logger.debug('[%s] Processing compile request, scriptId: [%s]', requestId, scriptId); @@ -78,17 +78,17 @@ JsInvokeMessageProcessor.prototype.processCompileRequest = function(requestId, r this.cacheScript(scriptId, script); var compileResponse = createCompileResponse(scriptId, true); logger.debug('[%s] Sending success compile response, scriptId: [%s]', requestId, scriptId); - this.sendResponse(requestId, responseTopic, scriptId, compileResponse); + this.sendResponse(requestId, responseTopic, headers, scriptId, compileResponse); }, (err) => { var compileResponse = createCompileResponse(scriptId, false, COMPILATION_ERROR, err); logger.debug('[%s] Sending failed compile response, scriptId: [%s]', requestId, scriptId); - this.sendResponse(requestId, responseTopic, scriptId, compileResponse); + this.sendResponse(requestId, responseTopic, headers, scriptId, compileResponse); } ); } -JsInvokeMessageProcessor.prototype.processInvokeRequest = function(requestId, responseTopic, invokeRequest) { +JsInvokeMessageProcessor.prototype.processInvokeRequest = function(requestId, responseTopic, headers, invokeRequest) { var scriptId = getScriptId(invokeRequest); logger.debug('[%s] Processing invoke request, scriptId: [%s]', requestId, scriptId); this.executedScriptsCounter++; @@ -104,7 +104,7 @@ JsInvokeMessageProcessor.prototype.processInvokeRequest = function(requestId, re (result) => { var invokeResponse = createInvokeResponse(result, true); logger.debug('[%s] Sending success invoke response, scriptId: [%s]', requestId, scriptId); - this.sendResponse(requestId, responseTopic, scriptId, null, invokeResponse); + this.sendResponse(requestId, responseTopic, headers, scriptId, null, invokeResponse); }, (err) => { var errorCode; @@ -115,19 +115,19 @@ JsInvokeMessageProcessor.prototype.processInvokeRequest = function(requestId, re } var invokeResponse = createInvokeResponse("", false, errorCode, err); logger.debug('[%s] Sending failed invoke response, scriptId: [%s], errorCode: [%s]', requestId, scriptId, errorCode); - this.sendResponse(requestId, responseTopic, scriptId, null, invokeResponse); + this.sendResponse(requestId, responseTopic, headers, scriptId, null, invokeResponse); } ) }, (err) => { var invokeResponse = createInvokeResponse("", false, COMPILATION_ERROR, err); logger.debug('[%s] Sending failed invoke response, scriptId: [%s], errorCode: [%s]', requestId, scriptId, COMPILATION_ERROR); - this.sendResponse(requestId, responseTopic, scriptId, null, invokeResponse); + this.sendResponse(requestId, responseTopic, headers, scriptId, null, invokeResponse); } ); } -JsInvokeMessageProcessor.prototype.processReleaseRequest = function(requestId, responseTopic, releaseRequest) { +JsInvokeMessageProcessor.prototype.processReleaseRequest = function(requestId, responseTopic, headers, releaseRequest) { var scriptId = getScriptId(releaseRequest); logger.debug('[%s] Processing release request, scriptId: [%s]', requestId, scriptId); if (this.scriptMap[scriptId]) { @@ -139,28 +139,17 @@ JsInvokeMessageProcessor.prototype.processReleaseRequest = function(requestId, r } var releaseResponse = createReleaseResponse(scriptId, true); logger.debug('[%s] Sending success release response, scriptId: [%s]', requestId, scriptId); - this.sendResponse(requestId, responseTopic, scriptId, null, null, releaseResponse); + this.sendResponse(requestId, responseTopic, headers, scriptId, null, null, releaseResponse); } -JsInvokeMessageProcessor.prototype.sendResponse = function (requestId, responseTopic, scriptId, compileResponse, invokeResponse, releaseResponse) { +JsInvokeMessageProcessor.prototype.sendResponse = function (requestId, responseTopic, headers, scriptId, compileResponse, invokeResponse, releaseResponse) { var remoteResponse = createRemoteResponse(requestId, compileResponse, invokeResponse, releaseResponse); var rawResponse = Buffer.from(JSON.stringify(remoteResponse), 'utf8'); - this.producer.send(responseTopic, scriptId, rawResponse, headers - // { - // topic: responseTopic, - // messages: [ - // { - // key: scriptId, - // value: rawResponse, - // headers: headers - // } - // ] - // } - ).then( + this.producer.send(responseTopic, scriptId, rawResponse, headers).then( () => {}, (err) => { if (err) { - logger.error('[%s] Failed to send response to kafka: %s', requestId, err.message); + logger.error('[%s] Failed to send response to queue: %s', requestId, err.message); logger.error(err.stack); } } diff --git a/msa/js-executor/config/custom-environment-variables.yml b/msa/js-executor/config/custom-environment-variables.yml index 77e68b0429..3beab91b40 100644 --- a/msa/js-executor/config/custom-environment-variables.yml +++ b/msa/js-executor/config/custom-environment-variables.yml @@ -15,12 +15,33 @@ # service-type: "TB_SERVICE_TYPE" +request_topic: "REMOTE_JS_EVAL_REQUEST_TOPIC" kafka: - request_topic: "REMOTE_JS_EVAL_REQUEST_TOPIC" bootstrap: # Kafka Bootstrap Servers servers: "TB_KAFKA_SERVERS" + +pubsub: + project_id: "TB_QUEUE_PUBSUB_PROJECT_ID" + service_account: "TB_QUEUE_PUBSUB_SERVICE_ACCOUNT" + +aws_sqs: + access_key_id: "TB_QUEUE_AWS_SQS_ACCESS_KEY_ID" + secret_access_key: "TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY" + region: "TB_QUEUE_AWS_SQS_REGION" + +rabbitmq: + exchange_name: "TB_QUEUE_RABBIT_MQ_EXCHANGE_NAME" + host: "TB_QUEUE_RABBIT_MQ_HOST" + port: "TB_QUEUE_RABBIT_MQ_PORT" + virtual_host: "TB_QUEUE_RABBIT_MQ_VIRTUAL_HOST" + username: "TB_QUEUE_RABBIT_MQ_USERNAME" + password: "TB_QUEUE_RABBIT_MQ_PASSWORD" + automatic_recovery_enabled: "TB_QUEUE_RABBIT_MQ_AUTOMATIC_RECOVERY_ENABLED" + connection_timeout: "TB_QUEUE_RABBIT_MQ_CONNECTION_TIMEOUT" + handshake_timeout: "TB_QUEUE_RABBIT_MQ_HANDSHAKE_TIMEOUT" + logger: level: "LOGGER_LEVEL" path: "LOG_FOLDER" diff --git a/msa/js-executor/config/default.yml b/msa/js-executor/config/default.yml index 724901d171..ee38296b31 100644 --- a/msa/js-executor/config/default.yml +++ b/msa/js-executor/config/default.yml @@ -15,13 +15,25 @@ # service-type: "kafka" +request_topic: "js.eval.requests" kafka: - request_topic: "js.eval.requests" bootstrap: # Kafka Bootstrap Servers servers: "localhost:9092" +rabbitmq: + exchange_name: "" + host: "localhost" + port: "5672" + virtual_host: "/" + username: "YOUR_USERNAME" + password: "YOUR_PASSWORD" + automatic_recovery_enabled: "false" + connection_timeout: "60000" + handshake_timeout: "10000" + + logger: level: "info" path: "logs" diff --git a/msa/js-executor/package.json b/msa/js-executor/package.json index c87b62c6eb..6b2ac41a9c 100644 --- a/msa/js-executor/package.json +++ b/msa/js-executor/package.json @@ -15,6 +15,9 @@ "config": "^3.2.2", "js-yaml": "^3.12.0", "kafkajs": "^1.11.0", + "@google-cloud/pubsub": "^1.7.1", + "aws-sdk": "^2.663.0", + "amqplib": "^0.5.5", "long": "^4.0.0", "uuid-parse": "^1.0.0", "winston": "^3.0.0", diff --git a/msa/js-executor/queue/awsSqsTemplate.js b/msa/js-executor/queue/awsSqsTemplate.js new file mode 100644 index 0000000000..6bd2f73510 --- /dev/null +++ b/msa/js-executor/queue/awsSqsTemplate.js @@ -0,0 +1,163 @@ +/* + * 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. + */ + +'use strict'; + +const config = require('config'), + JsInvokeMessageProcessor = require('../api/jsInvokeMessageProcessor'), + logger = require('../config/logger')._logger('awsSqsTemplate'); + +const requestTopic = config.get('request_topic'); + +const accessKeyId = config.get('aws_sqs.access_key_id'); +const secretAccessKey = config.get('aws_sqs.secret_access_key'); +const region = config.get('aws_sqs.region'); +const AWS = require('aws-sdk'); + +let sqsClient; +let queueURL; +let responseTopics = new Map(); +let stopped = false; + +function AwsSqsProducer() { + this.send = async (responseTopic, scriptId, rawResponse, headers) => { + let msgBody = JSON.stringify( + { + key: scriptId, + data: [...rawResponse], + headers: headers + }); + + let responseQueueUrl = responseTopics.get(responseTopic); + + if (!responseQueueUrl) { + responseQueueUrl = await createQueue(responseTopic); + responseTopics.set(responseTopic, responseQueueUrl); + } + + let params = {MessageBody: msgBody, QueueUrl: responseQueueUrl, MessageGroupId: scriptId}; + + return new Promise((resolve, reject) => { + sqsClient.sendMessage(params, function (err, data) { + if (err) { + reject(err); + } else { + resolve(data); + } + }); + }); + } +} + +(async () => { + try { + logger.info('Starting ThingsBoard JavaScript Executor Microservice...'); + AWS.config.update({accessKeyId: accessKeyId, secretAccessKey: secretAccessKey, region: region}); + + sqsClient = new AWS.SQS({apiVersion: '2012-11-05'}); + + queueURL = await createQueue(requestTopic); + const messageProcessor = new JsInvokeMessageProcessor(new AwsSqsProducer()); + + const params = { + MaxNumberOfMessages: 10, + QueueUrl: queueURL, + WaitTimeSeconds: 0.025 + }; + while (!stopped) { + const messages = await new Promise((resolve, reject) => { + sqsClient.receiveMessage(params, function (err, data) { + if (err) { + reject(err); + } else { + resolve(data.Messages); + } + }); + }); + + if (messages && messages.length > 0) { + const entries = []; + + messages.forEach(message => { + entries.push({ + Id: message.MessageId, + ReceiptHandle: message.ReceiptHandle + }); + messageProcessor.onJsInvokeMessage(message.Body); + }); + + const deleteBatch = { + QueueUrl: queueURL, + Entries: entries + }; + sqsClient.deleteMessageBatch(deleteBatch, function (err, data) { + if (err) { + logger.error("Failed to delete messages from queue.", err.message); + } else { + //do nothing + } + }); + } + } + } catch (e) { + logger.error('Failed to start ThingsBoard JavaScript Executor Microservice: %s', e.message); + logger.error(e.stack); + exit(-1); + } +})(); + +function createQueue(topic) { + let queueName = topic.replace(/\./g, '_') + '.fifo'; + let queueParams = { + QueueName: queueName, Attributes: { + FifoQueue: 'true', + ContentBasedDeduplication: 'true' + + } + }; + return new Promise((resolve, reject) => { + sqsClient.createQueue(queueParams, function (err, data) { + if (err) { + reject(err); + } else { + resolve(data.QueueUrl); + } + }); + }); +} + +process.on('exit', () => { + stopped = true; + logger.info('Aws Sqs client stopped.'); + exit(0); +}); + +async function exit(status) { + logger.info('Exiting with status: %d ...', status); + if (sqsClient) { + logger.info('Stopping Aws Sqs client.') + try { + await sqsClient.close(); + logger.info('Aws Sqs client is stopped.') + process.exit(status); + } catch (e) { + logger.info('Aws Sqs client stop error.'); + process.exit(status); + } + } else { + process.exit(status); + } +} diff --git a/msa/js-executor/queue/kafka/kafkaTemplate.js b/msa/js-executor/queue/kafkaTemplate.js similarity index 64% rename from msa/js-executor/queue/kafka/kafkaTemplate.js rename to msa/js-executor/queue/kafkaTemplate.js index bb3f81c48d..38e713a7db 100644 --- a/msa/js-executor/queue/kafka/kafkaTemplate.js +++ b/msa/js-executor/queue/kafkaTemplate.js @@ -13,39 +13,41 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -const { logLevel, Kafka } = require('kafkajs'); +const {logLevel, Kafka} = require('kafkajs'); const config = require('config'), - JsInvokeMessageProcessor = require('../../api/jsInvokeMessageProcessor'), - logger = require('../../config/logger')._logger('main'), - KafkaJsWinstonLogCreator = require('../../config/logger').KafkaJsWinstonLogCreator; + JsInvokeMessageProcessor = require('../api/jsInvokeMessageProcessor'), + logger = require('../config/logger')._logger('kafkaTemplate'), + KafkaJsWinstonLogCreator = require('../config/logger').KafkaJsWinstonLogCreator; -var kafkaClient; -var consumer; -var producer; +let kafkaClient; +let consumer; +let producer; function KafkaProducer() { this.send = async (responseTopic, scriptId, rawResponse, headers) => { + let headersData = headers.data; + headersData = Object.fromEntries(Object.entries(headersData).map(([key, value]) => [key, Buffer.from(value)])); return producer.send( - { - topic: responseTopic, - messages: [ - { - key: scriptId, - value: rawResponse, - headers: headers - } - ] - }); + { + topic: responseTopic, + messages: [ + { + key: scriptId, + value: rawResponse, + headers: headersData + } + ] + }); } } -(async() => { +(async () => { try { logger.info('Starting ThingsBoard JavaScript Executor Microservice...'); const kafkaBootstrapServers = config.get('kafka.bootstrap.servers'); - const kafkaRequestTopic = config.get('kafka.request_topic'); + const kafkaRequestTopic = config.get('request_topic'); logger.info('Kafka Bootstrap Servers: %s', kafkaBootstrapServers); logger.info('Kafka Requests Topic: %s', kafkaRequestTopic); @@ -56,17 +58,28 @@ function KafkaProducer() { logCreator: KafkaJsWinstonLogCreator }); - consumer = kafkaClient.consumer({ groupId: 'js-executor-group' }); + consumer = kafkaClient.consumer({groupId: 'js-executor-group'}); producer = kafkaClient.producer(); const messageProcessor = new JsInvokeMessageProcessor(new KafkaProducer()); await consumer.connect(); await producer.connect(); - await consumer.subscribe({ topic: kafkaRequestTopic}); + await consumer.subscribe({topic: kafkaRequestTopic}); logger.info('Started ThingsBoard JavaScript Executor Microservice.'); await consumer.run({ - eachMessage: async ({ topic, partition, message }) => { - messageProcessor.onJsInvokeMessage(message); + eachMessage: async ({topic, partition, message}) => { + let headers = message.headers; + let key = message.key; + let data = message.value; + let msg = {}; + + headers = Object.fromEntries( + Object.entries(headers).map(([key, value]) => [key, [...value]])); + + msg.key = key.toString('utf8'); + msg.data = [...data]; + msg.headers = {data: headers} + messageProcessor.onJsInvokeMessage(JSON.stringify(msg)); }, }); @@ -85,7 +98,7 @@ async function exit(status) { logger.info('Exiting with status: %d ...', status); if (consumer) { logger.info('Stopping Kafka Consumer...'); - var _consumer = consumer; + let _consumer = consumer; consumer = null; try { await _consumer.disconnect(); diff --git a/msa/js-executor/queue/pubSubTemplate.js b/msa/js-executor/queue/pubSubTemplate.js new file mode 100644 index 0000000000..c8b39f7d6b --- /dev/null +++ b/msa/js-executor/queue/pubSubTemplate.js @@ -0,0 +1,87 @@ +/* + * 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. + */ + +'use strict'; + +const config = require('config'), + JsInvokeMessageProcessor = require('../api/jsInvokeMessageProcessor'), + logger = require('../config/logger')._logger('pubSubTemplate'); +const {PubSub} = require('@google-cloud/pubsub'); + +const projectId = config.get('pubsub.project_id'); +const credentials = JSON.parse(config.get('pubsub.service_account')); +const requestTopic = config.get('request_topic'); + +let pubSubClient; + +function PubSubProducer() { + this.send = async (responseTopic, scriptId, rawResponse, headers) => { + let data = JSON.stringify( + { + key: scriptId, + data: [...rawResponse], + headers: headers + }); + let dataBuffer = Buffer.from(data); + return pubSubClient.topic(responseTopic).publish(dataBuffer); + } +} + +(async () => { + try { + logger.info('Starting ThingsBoard JavaScript Executor Microservice...'); + pubSubClient = new PubSub({projectId: projectId, credentials: credentials}); + + const subscription = pubSubClient.subscription(requestTopic); + + const messageProcessor = new JsInvokeMessageProcessor(new PubSubProducer()); + + const messageHandler = message => { + + messageProcessor.onJsInvokeMessage(message.data.toString('utf8')); + message.ack(); + }; + + subscription.on('message', messageHandler); + + } catch (e) { + logger.error('Failed to start ThingsBoard JavaScript Executor Microservice: %s', e.message); + logger.error(e.stack); + exit(-1); + } +})(); + +process.on('exit', () => { + exit(0); +}); + +async function exit(status) { + logger.info('Exiting with status: %d ...', status); + if (pubSubClient) { + logger.info('Stopping Pub/Sub client.') + try { + await pubSubClient.close(); + logger.info('Pub/Sub client is stopped.') + process.exit(status); + } catch (e) { + logger.info('Pub/Sub client stop error.'); + process.exit(status); + } + } else { + process.exit(status); + } +} + diff --git a/msa/js-executor/queue/rabbitmqTemplate.js b/msa/js-executor/queue/rabbitmqTemplate.js new file mode 100644 index 0000000000..0b48cdd62f --- /dev/null +++ b/msa/js-executor/queue/rabbitmqTemplate.js @@ -0,0 +1,149 @@ +/* + * 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. + */ + +'use strict'; + +const config = require('config'), + JsInvokeMessageProcessor = require('../api/jsInvokeMessageProcessor'), + logger = require('../config/logger')._logger('rabbitmqTemplate'); + +const requestTopic = config.get('request_topic'); +const amqp = require('amqplib/callback_api'); +let connection; +let channel; +let stopped = false; +const responseTopics = []; + +function RabbitMqProducer() { + this.send = async (responseTopic, scriptId, rawResponse, headers) => { + + if (!responseTopics.includes(responseTopic)) { + await createQueue(responseTopic); + responseTopics.push(responseTopic); + } + + let data = JSON.stringify( + { + key: scriptId, + data: [...rawResponse], + headers: headers + }); + let dataBuffer = Buffer.from(data); + channel.sendToQueue(responseTopic, dataBuffer); + return new Promise((resolve, reject) => { + channel.waitForConfirms((err) => { + if (err) { + reject(err); + } else { + resolve(); + } + }); + }); + } +} + +(async () => { + try { + logger.info('Starting ThingsBoard JavaScript Executor Microservice...'); + + amqp.credentials.amqplain('admin', 'password'); + connection = await new Promise((resolve, reject) => { + amqp.connect('amqp://localhost:5672/', function (err, connection) { + if (err) { + reject(err); + } else { + resolve(connection); + } + }); + }); + + channel = await new Promise((resolve, reject) => { + connection.createConfirmChannel(function (err, channel) { + if (err) { + reject(err); + } else { + resolve(channel); + } + }); + }); + + await createQueue(requestTopic); + + const messageProcessor = new JsInvokeMessageProcessor(new RabbitMqProducer()); + + while (!stopped) { + let message = await new Promise((resolve, reject) => { + channel.get(requestTopic, {}, function (err, msg) { + if (err) { + reject(err); + } else { + resolve(msg); + } + }); + }); + + if (message) { + messageProcessor.onJsInvokeMessage(message.content.toString('utf8')); + channel.ack(message); + } + } + } catch (e) { + logger.error('Failed to start ThingsBoard JavaScript Executor Microservice: %s', e.message); + logger.error(e.stack); + exit(-1); + } +})(); + +function createQueue(topic) { + let params = {durable: false}; + return new Promise((resolve, reject) => { + channel.assertQueue(topic, params, function (err, data) { + if (err) { + reject(err); + } else { + resolve(); + } + }); + }); +} + +process.on('exit', () => { + exit(0); +}); + +async function exit(status) { + logger.info('Exiting with status: %d ...', status); + + if (channel) { + logger.info('Stopping RabbitMq chanel.') + await channel.close(); + logger.info('RabbitMq chanel is stopped'); + } + + if (connection) { + logger.info('Stopping RabbitMq connection.') + try { + await connection.close(); + logger.info('RabbitMq client is connection.') + process.exit(status); + } catch (e) { + logger.info('RabbitMq connection stop error.'); + process.exit(status); + } + } else { + process.exit(status); + } +} \ No newline at end of file diff --git a/msa/js-executor/server.js b/msa/js-executor/server.js index 486f9dc187..bd84289200 100644 --- a/msa/js-executor/server.js +++ b/msa/js-executor/server.js @@ -14,16 +14,32 @@ * limitations under the License. */ -const config = require('config'); +const config = require('config'), logger = require('./config/logger')._logger('main'); const serviceType = config.get('service-type'); switch (serviceType) { case 'kafka': - require('./queue/kafka/kafkaTemplate'); - console.log('Used kafka template.'); + logger.info('Starting kafka template.'); + require('./queue/kafkaTemplate'); + logger.info('kafka template is started.'); + break; + case 'pubsub': + logger.info('Starting Pub/Sub template.') + require('./queue/pubSubTemplate'); + logger.info('Pub/Sub template is started.') + break; + case 'aws-sqs': + logger.info('Starting Aws Sqs template.') + require('./queue/awsSqsTemplate'); + logger.info('Aws Sqs template is started.') + break; + case 'rabbitmq': + logger.info('Starting RabbitMq template.') + require('./queue/rabbitmqTemplate'); + logger.info('RabbitMq template is started.') break; default: - console.error('Unknown service type: ', serviceType); + logger.error('Unknown service type: ', serviceType); process.exit(-1); }