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

Improve JS Executor. Add docker build.

Browse Source
This commit is contained in:
Igor Kulikov 2018-10-02 11:50:55 +03:00
parent 64b50fa8bc
commit 52f04308d4
18 changed files with 475 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java
View File

@ -166,7 +166,7 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService {
.setScriptBody(scriptIdToBodysMap.get(scriptId));
for (int i = 0; i < args.length; i++) {
jsRequestBuilder.setArgs(i, args[i].toString());
jsRequestBuilder.addArgs(args[i].toString());
}
JsInvokeProtos.RemoteJsRequest jsRequestWrapper = JsInvokeProtos.RemoteJsRequest.newBuilder()
@ -180,7 +180,7 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService {
return invokeResult.getResult();
} else {
log.debug("[{}] Failed to compile script due to [{}]: {}", scriptId, invokeResult.getErrorCode().name(), invokeResult.getErrorDetails());
throw new RuntimeException(invokeResult.getErrorCode().name());
throw new RuntimeException(invokeResult.getErrorDetails());
}
});
}

2
application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java
View File

@ -174,6 +174,8 @@ public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.S
} catch (ExecutionException e) {
if (e.getCause() instanceof ScriptException) {
throw (ScriptException)e.getCause();
} else if (e.getCause() instanceof RuntimeException) {
throw new ScriptException(e.getCause().getMessage());
} else {
throw new ScriptException(e);
}

2
application/src/main/java/org/thingsboard/server/service/script/RuleNodeScriptFactory.java
View File

@ -28,7 +28,7 @@ public class RuleNodeScriptFactory {
" var metadata = JSON.parse(metadataStr); " +
" return JSON.stringify(%s(msg, metadata, msgType));" +
" function %s(%s, %s, %s) {";
private static final String JS_WRAPPER_SUFFIX = "}" +
private static final String JS_WRAPPER_SUFFIX = "\n}" +
"\n}";

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

@ -407,7 +407,7 @@ state:
kafka:
enabled: true
bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}"
bootstrap.servers: "${TB_KAFKA_SERVERS:192.168.2.157:9092}"
acks: "${TB_KAFKA_ACKS:all}"
retries: "${TB_KAFKA_RETRIES:1}"
batch.size: "${TB_KAFKA_BATCH_SIZE:16384}"

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

@ -17,6 +17,9 @@ package org.thingsboard.server.kafka;
import lombok.Builder;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
@ -35,6 +38,7 @@ import java.util.function.BiConsumer;
/**
* Created by ashvayka on 24.09.18.
*/
@Slf4j
public class TBKafkaProducerTemplate<T> {
private final KafkaProducer<String, byte[]> producer;
@ -44,7 +48,7 @@ public class TBKafkaProducerTemplate<T> {
private TbKafkaEnricher<T> enricher = ((value, responseTopic, requestId) -> value);
private final TbKafkaPartitioner<T> partitioner;
private final List<PartitionInfo> partitionInfoList;
private List<PartitionInfo> partitionInfoList;
@Getter
private final String defaultTopic;
@ -55,14 +59,24 @@ public class TBKafkaProducerTemplate<T> {
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
this.producer = new KafkaProducer<>(props);
//Maybe this should not be cached, but we don't plan to change size of partitions
this.partitionInfoList = producer.partitionsFor(defaultTopic);
this.encoder = encoder;
this.enricher = enricher;
this.partitioner = partitioner;
this.defaultTopic = defaultTopic;
}
public void init() {
try {
TBKafkaAdmin admin = new TBKafkaAdmin();
CreateTopicsResult result = admin.createTopic(new NewTopic(defaultTopic, 100, (short) 1));
result.all().get();
} catch (Exception e) {
log.trace("Failed to create topic: {}", e.getMessage(), e);
}
//Maybe this should not be cached, but we don't plan to change size of partitions
this.partitionInfoList = producer.partitionsFor(defaultTopic);
}
public T enrich(T value, String responseTopic, UUID requestId) {
return enricher.enrich(value, responseTopic, requestId);
}

8
common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaRequestTemplate.java
View File

@ -23,6 +23,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
@ -33,11 +34,7 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.*;
/**
* Created by ashvayka on 25.09.18.
@ -86,6 +83,7 @@ public class TbKafkaRequestTemplate<Request, Response> {
} catch (Exception e) {
log.trace("Failed to create topic: {}", e.getMessage(), e);
}
this.requestTemplate.init();
tickTs = System.currentTimeMillis();
responseTemplate.subscribe();
executor.submit(() -> {

46
msa/docker/docker-compose.yml Normal file
View File

@ -0,0 +1,46 @@
#
# Copyright © 2016-2018 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.
#
version: '2'
services:
zookeeper:
image: "wurstmeister/zookeeper"
ports:
- "2181"
kafka:
image: "wurstmeister/kafka"
ports:
- "9092:9092"
environment:
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_LISTENERS: INSIDE://:9093,OUTSIDE://:9092
KAFKA_ADVERTISED_LISTENERS: INSIDE://:9093,OUTSIDE://${KAFKA_HOSTNAME}:9092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
KAFKA_CREATE_TOPICS: "${KAFKA_TOPICS}"
KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
depends_on:
- zookeeper
tb-js-executor:
image: "local-maven-build/tb-js-executor:latest"
environment:
TB_KAFKA_SERVERS: kafka:9092
env_file:
- tb-js-executor.env
depends_on:
- kafka

7
msa/docker/tb-js-executor.env Normal file
View File

@ -0,0 +1,7 @@
REMOTE_JS_EVAL_REQUEST_TOPIC=js.eval.requests
TB_KAFKA_SERVERS=localhost:9092
LOGGER_LEVEL=debug
LOG_FOLDER=logs
LOGGER_FILENAME=tb-js-executor-%DATE%.log
DOCKER_MODE=true

48
msa/js-executor/api/jsExecutor.js Normal file
View File

@ -0,0 +1,48 @@
/*
* Copyright © 2016-2018 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 vm = require('vm');
function JsExecutor() {
}
JsExecutor.prototype.compileScript = function(code) {
return new Promise(function(resolve, reject) {
try {
code = "("+code+")(...args)";
var script = new vm.Script(code);
resolve(script);
} catch (err) {
reject(err);
}
});
}
JsExecutor.prototype.executeScript = function(script, args, timeout) {
return new Promise(function(resolve, reject) {
try {
var sandbox = Object.create(null);
sandbox.args = args;
var result = script.runInNewContext(sandbox, {timeout: timeout});
resolve(result);
} catch (err) {
reject(err);
}
});
}
module.exports = JsExecutor;

226
msa/js-executor/api/jsInvokeMessageProcessor.js Normal file
View File

@ -0,0 +1,226 @@
/*
* Copyright © 2016-2018 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 logger = require('../config/logger')('JsInvokeMessageProcessor'),
Utils = require('./utils'),
js = require('./jsinvoke.proto').js,
KeyedMessage = require('kafka-node').KeyedMessage,
JsExecutor = require('./jsExecutor');
function JsInvokeMessageProcessor(producer) {
this.producer = producer;
this.executor = new JsExecutor();
this.scriptMap = {};
}
JsInvokeMessageProcessor.prototype.onJsInvokeMessage = function(message) {
var requestId;
try {
var request = js.RemoteJsRequest.decode(message.value);
requestId = getRequestId(request);
logger.debug('[%s] Received request, responseTopic: [%s]', requestId, request.responseTopic);
if (request.compileRequest) {
this.processCompileRequest(requestId, request.responseTopic, request.compileRequest);
} else if (request.invokeRequest) {
this.processInvokeRequest(requestId, request.responseTopic, request.invokeRequest);
} else if (request.releaseRequest) {
this.processReleaseRequest(requestId, request.responseTopic, request.releaseRequest);
} else {
logger.error('[%s] Unknown request recevied!', requestId);
}
} catch (err) {
logger.error('[%s] Failed to process request: %s', requestId, err.message);
logger.error(err.stack);
}
}
JsInvokeMessageProcessor.prototype.processCompileRequest = function(requestId, responseTopic, compileRequest) {
var scriptId = getScriptId(compileRequest);
logger.debug('[%s] Processing compile request, scriptId: [%s]', requestId, scriptId);
this.executor.compileScript(compileRequest.scriptBody).then(
(script) => {
this.scriptMap[scriptId] = script;
var compileResponse = createCompileResponse(scriptId, true);
logger.debug('[%s] Sending success compile response, scriptId: [%s]', requestId, scriptId);
this.sendResponse(requestId, responseTopic, scriptId, compileResponse);
},
(err) => {
var compileResponse = createCompileResponse(scriptId, false, js.JsInvokeErrorCode.COMPILATION_ERROR, err);
logger.debug('[%s] Sending failed compile response, scriptId: [%s]', requestId, scriptId);
this.sendResponse(requestId, responseTopic, scriptId, compileResponse);
}
);
}
JsInvokeMessageProcessor.prototype.processInvokeRequest = function(requestId, responseTopic, invokeRequest) {
var scriptId = getScriptId(invokeRequest);
logger.debug('[%s] Processing invoke request, scriptId: [%s]', requestId, scriptId);
this.getOrCompileScript(scriptId, invokeRequest.scriptBody).then(
(script) => {
this.executor.executeScript(script, invokeRequest.args, invokeRequest.timeout).then(
(result) => {
var invokeResponse = createInvokeResponse(result, true);
logger.debug('[%s] Sending success invoke response, scriptId: [%s]', requestId, scriptId);
this.sendResponse(requestId, responseTopic, scriptId, null, invokeResponse);
},
(err) => {
var errorCode;
if (err.message.includes('Script execution timed out')) {
errorCode = js.JsInvokeErrorCode.TIMEOUT_ERROR;
} else {
errorCode = js.JsInvokeErrorCode.RUNTIME_ERROR;
}
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);
}
)
},
(err) => {
var invokeResponse = createInvokeResponse("", false, js.JsInvokeErrorCode.COMPILATION_ERROR, err);
logger.debug('[%s] Sending failed invoke response, scriptId: [%s], errorCode: [%s]', requestId, scriptId, js.JsInvokeErrorCode.COMPILATION_ERROR);
this.sendResponse(requestId, responseTopic, scriptId, null, invokeResponse);
}
);
}
JsInvokeMessageProcessor.prototype.processReleaseRequest = function(requestId, responseTopic, releaseRequest) {
var scriptId = getScriptId(releaseRequest);
logger.debug('[%s] Processing release request, scriptId: [%s]', requestId, scriptId);
if (this.scriptMap[scriptId]) {
delete this.scriptMap[scriptId];
}
var releaseResponse = createReleaseResponse(scriptId, true);
logger.debug('[%s] Sending success release response, scriptId: [%s]', requestId, scriptId);
this.sendResponse(requestId, responseTopic, scriptId, null, null, releaseResponse);
}
JsInvokeMessageProcessor.prototype.sendResponse = function (requestId, responseTopic, scriptId, compileResponse, invokeResponse, releaseResponse) {
var remoteResponse = createRemoteResponse(requestId, compileResponse, invokeResponse, releaseResponse);
var rawResponse = js.RemoteJsResponse.encode(remoteResponse).finish();
const message = new KeyedMessage(scriptId, rawResponse);
const payloads = [ { topic: responseTopic, messages: message, key: scriptId } ];
this.producer.send(payloads, function (err, data) {
if (err) {
logger.error('[%s] Failed to send response to kafka: %s', requestId, err.message);
logger.error(err.stack);
}
});
}
JsInvokeMessageProcessor.prototype.getOrCompileScript = function(scriptId, scriptBody) {
var self = this;
return new Promise(function(resolve, reject) {
if (self.scriptMap[scriptId]) {
resolve(self.scriptMap[scriptId]);
} else {
self.executor.compileScript(scriptBody).then(
(script) => {
self.scriptMap[scriptId] = script;
resolve(script);
},
(err) => {
reject(err);
}
);
}
});
}
function createRemoteResponse(requestId, compileResponse, invokeResponse, releaseResponse) {
const requestIdBits = Utils.UUIDToBits(requestId);
return js.RemoteJsResponse.create(
{
requestIdMSB: requestIdBits[0],
requestIdLSB: requestIdBits[1],
compileResponse: compileResponse,
invokeResponse: invokeResponse,
releaseResponse: releaseResponse
}
);
}
function createCompileResponse(scriptId, success, errorCode, err) {
const scriptIdBits = Utils.UUIDToBits(scriptId);
return js.JsCompileResponse.create(
{
errorCode: errorCode,
success: success,
errorDetails: parseJsErrorDetails(err),
scriptIdMSB: scriptIdBits[0],
scriptIdLSB: scriptIdBits[1]
}
);
}
function createInvokeResponse(result, success, errorCode, err) {
return js.JsInvokeResponse.create(
{
errorCode: errorCode,
success: success,
errorDetails: parseJsErrorDetails(err),
result: result
}
);
}
function createReleaseResponse(scriptId, success) {
const scriptIdBits = Utils.UUIDToBits(scriptId);
return js.JsReleaseResponse.create(
{
success: success,
scriptIdMSB: scriptIdBits[0],
scriptIdLSB: scriptIdBits[1]
}
);
}
function parseJsErrorDetails(err) {
if (!err) {
return '';
}
var details = err.name + ': ' + err.message;
if (err.stack) {
var lines = err.stack.split('\n');
if (lines && lines.length) {
var line = lines[0];
var splitted = line.split(':');
if (splitted && splitted.length === 2) {
if (!isNaN(splitted[1])) {
details += ' in at line number ' + splitted[1];
}
}
}
}
return details;
}
function getScriptId(request) {
return Utils.toUUIDString(request.scriptIdMSB, request.scriptIdLSB);
}
function getRequestId(request) {
return Utils.toUUIDString(request.requestIdMSB, request.requestIdLSB);
}
module.exports = JsInvokeMessageProcessor;

78
msa/js-executor/api/jsMessageConsumer.js
View File

@ -1,78 +0,0 @@
/*
* Copyright © 2016-2018 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 logger = require('../config/logger')('JsMessageConsumer');
const Utils = require('./Utils');
const js = require('./jsinvoke.proto').js;
const KeyedMessage = require('kafka-node').KeyedMessage;
exports.onJsInvokeMessage = function(message, producer) {
logger.info('Received message: %s', JSON.stringify(message));
var request = js.RemoteJsRequest.decode(message.value);
logger.info('Received request: %s', JSON.stringify(request));
var requestId = getRequestId(request);
logger.info('Received request, responseTopic: [%s]; requestId: [%s]', request.responseTopic, requestId);
if (request.compileRequest) {
var scriptId = getScriptId(request.compileRequest);
logger.info('Received compile request, scriptId: [%s]', scriptId);
var compileResponse = js.JsCompileResponse.create(
{
errorCode: js.JsInvokeErrorCode.COMPILATION_ERROR,
success: false,
errorDetails: 'Not Implemented!',
scriptIdLSB: request.compileRequest.scriptIdLSB,
scriptIdMSB: request.compileRequest.scriptIdMSB
}
);
const requestIdBits = Utils.UUIDToBits(requestId);
var response = js.RemoteJsResponse.create(
{
requestIdMSB: requestIdBits[0],
requestIdLSB: requestIdBits[1],
compileResponse: compileResponse
}
);
var rawResponse = js.RemoteJsResponse.encode(response).finish();
sendMessage(producer, rawResponse, request.responseTopic, scriptId);
}
}
function sendMessage(producer, rawMessage, responseTopic, scriptId) {
const message = new KeyedMessage(scriptId, rawMessage);
const payloads = [ { topic: responseTopic, messages: rawMessage, key: scriptId } ];
producer.send(payloads, function (err, data) {
console.log(data);
});
}
function getScriptId(request) {
return Utils.toUUIDString(request.scriptIdMSB, request.scriptIdLSB);
}
function getRequestId(request) {
return Utils.toUUIDString(request.requestIdMSB, request.requestIdLSB);
}

4
msa/js-executor/api/Utils.js → msa/js-executor/api/utils.js
View File

@ -16,8 +16,8 @@
'use strict';
const Long = require('long');
const uuidParse = require('uuid-parse');
const Long = require('long'),
uuidParse = require('uuid-parse');
var logger = require('../config/logger')('Utils');

2
msa/js-executor/config/logger.js
View File

@ -22,7 +22,7 @@ const { combine, timestamp, label, printf, splat } = format;
var loggerTransports = [];
if (process.env.NODE_ENV !== 'production') {
if (process.env.NODE_ENV !== 'production' || process.env.DOCKER_MODE === 'true') {
loggerTransports.push(new transports.Console({
handleExceptions: true
}));

26
msa/js-executor/docker/Dockerfile Normal file
View File

@ -0,0 +1,26 @@
#
# Copyright © 2016-2018 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.
#
FROM debian:stretch
COPY start-js-executor.sh ${pkg.name}.deb /tmp/
RUN chmod a+x /tmp/*.sh \
&& mv /tmp/start-js-executor.sh /usr/bin
RUN dpkg -i /tmp/${pkg.name}.deb
CMD ["start-js-executor.sh"]

29
msa/js-executor/docker/start-js-executor.sh Executable file
View File

@ -0,0 +1,29 @@
#!/bin/bash
#
# Copyright © 2016-2018 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.
#
echo "Starting '${project.name}' ..."
CONF_FOLDER="${pkg.installFolder}/conf"
mainfile=${pkg.installFolder}/bin/${pkg.name}
configfile=${pkg.name}.conf
identity=${pkg.name}
source "${CONF_FOLDER}/${configfile}"
su -s /bin/sh -c "$mainfile"

38
msa/js-executor/pom.xml
View File

@ -40,6 +40,7 @@
<pkg.installFolder>/usr/share/${pkg.name}</pkg.installFolder>
<pkg.linux.dist>${project.build.directory}/package/linux</pkg.linux.dist>
<pkg.win.dist>${project.build.directory}/package/windows</pkg.win.dist>
<dockerfile.skip>true</dockerfile.skip>
</properties>
<dependencies>
@ -211,6 +212,22 @@
</filters>
</configuration>
</execution>
<execution>
<id>copy-docker-config</id>
<phase>process-resources</phase>
<goals>
<goal>copy-resources</goal>
</goals>
<configuration>
<outputDirectory>${project.build.directory}</outputDirectory>
<resources>
<resource>
<directory>docker</directory>
<filtering>true</filtering>
</resource>
</resources>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
@ -260,6 +277,27 @@
</execution>
</executions>
</plugin>
<plugin>
<groupId>com.spotify</groupId>
<artifactId>dockerfile-maven-plugin</artifactId>
<version>1.4.4</version>
<executions>
<execution>
<id>build-docker-image</id>
<phase>pre-integration-test</phase>
<goals>
<goal>build</goal>
</goals>
</execution>
</executions>
<configuration>
<skip>${dockerfile.skip}</skip>
<repository>local-maven-build/${pkg.name}</repository>
<verbose>true</verbose>
<googleContainerRegistryEnabled>false</googleContainerRegistryEnabled>
<contextDirectory>${project.build.directory}</contextDirectory>
</configuration>
</plugin>
</plugins>
</build>
<profiles>

48
msa/js-executor/server.js
View File

@ -16,17 +16,10 @@
const config = require('config'),
kafka = require('kafka-node'),
Consumer = kafka.Consumer,
ConsumerGroup = kafka.ConsumerGroup,
Producer = kafka.Producer,
JsMessageConsumer = require('./api/jsMessageConsumer');
const logger = require('./config/logger')('main');
const kafkaBootstrapServers = config.get('kafka.bootstrap.servers');
const kafkaRequestTopic = config.get('kafka.request_topic');
logger.info('Kafka Bootstrap Servers: %s', kafkaBootstrapServers);
logger.info('Kafka Requests Topic: %s', kafkaRequestTopic);
JsInvokeMessageProcessor = require('./api/jsInvokeMessageProcessor'),
logger = require('./config/logger')('main');
var kafkaClient;
@ -34,35 +27,42 @@ var kafkaClient;
try {
logger.info('Starting ThingsBoard JavaScript Executor Microservice...');
const kafkaBootstrapServers = config.get('kafka.bootstrap.servers');
const kafkaRequestTopic = config.get('kafka.request_topic');
logger.info('Kafka Bootstrap Servers: %s', kafkaBootstrapServers);
logger.info('Kafka Requests Topic: %s', kafkaRequestTopic);
kafkaClient = new kafka.KafkaClient({kafkaHost: kafkaBootstrapServers});
var consumer = new Consumer(
kafkaClient,
[
{ topic: kafkaRequestTopic, partition: 0 }
],
var consumer = new ConsumerGroup(
{
kafkaHost: kafkaBootstrapServers,
groupId: 'js-executor-group',
autoCommit: true,
encoding: 'buffer'
}
},
kafkaRequestTopic
);
var producer = new Producer(kafkaClient);
producer.on('error', (err) => {
logger.error('Unexpected kafka producer error');
logger.error(err);
logger.error('Unexpected kafka producer error: %s', err.message);
logger.error(err.stack);
});
var messageProcessor = new JsInvokeMessageProcessor(producer);
producer.on('ready', () => {
consumer.on('message', (message) => {
JsMessageConsumer.onJsInvokeMessage(message, producer);
messageProcessor.onJsInvokeMessage(message);
});
logger.info('Started ThingsBoard JavaScript Executor Microservice.');
});
logger.info('Started ThingsBoard JavaScript Executor Microservice.');
} catch (e) {
logger.error('Failed to start ThingsBoard JavaScript Executor Microservice: %s', e.message);
logger.error(e);
logger.error(e.stack);
exit(-1);
}
})();
@ -75,11 +75,13 @@ function exit(status) {
logger.info('Exiting with status: %d ...', status);
if (kafkaClient) {
logger.info('Stopping Kafka Client...');
kafkaClient.close(() => {
var _kafkaClient = kafkaClient;
kafkaClient = null;
_kafkaClient.close(() => {
logger.info('Kafka Client stopped.');
process.exit(status);
});
} else {
process.exit(status);
}
}
}

2
msa/js-executor/src/main/scripts/control/deb/postinst
View File

@ -2,5 +2,5 @@
chown -R ${pkg.user}: ${pkg.logFolder}
chown -R ${pkg.user}: ${pkg.installFolder}
update-rc.d ${pkg.name} defaults
# update-rc.d ${pkg.name} defaults