tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge branch 'smatvienko-tb-js-script-engine-async'

pull/4772/head
Igor Kulikov 5 years ago
parent
461881b329 48af7d37c3
commit
555075efb6
9 changed files with 158 additions and 167 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 14
      application/src/main/java/org/thingsboard/server/controller/RuleChainController.java
  2. 6
      application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java
  3. 158
      application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java
  4. 15
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/ScriptEngine.java
  5. 4
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java
  6. 30
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbLogNode.java
  7. 50
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java
  8. 29
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsSwitchNode.java
  9. 19
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsSwitchNodeTest.java

14
application/src/main/java/org/thingsboard/server/controller/RuleChainController.java
View File

@ -74,6 +74,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@Slf4j
@ -86,6 +87,7 @@ public class RuleChainController extends BaseController {
public static final String RULE_NODE_ID = "ruleNodeId";
private static final ObjectMapper objectMapper = new ObjectMapper();
public static final int TIMEOUT = 20;
@Autowired
private InstallScripts installScripts;
@ -388,25 +390,25 @@ public class RuleChainController extends BaseController {
TbMsg inMsg = TbMsg.newMsg(msgType, null, new TbMsgMetaData(metadata), TbMsgDataType.JSON, data);
switch (scriptType) {
case "update":
output = msgToOutput(engine.executeUpdate(inMsg));
output = msgToOutput(engine.executeUpdateAsync(inMsg).get(TIMEOUT, TimeUnit.SECONDS));
break;
case "generate":
output = msgToOutput(engine.executeGenerate(inMsg));
output = msgToOutput(engine.executeGenerateAsync(inMsg).get(TIMEOUT, TimeUnit.SECONDS));
break;
case "filter":
boolean result = engine.executeFilter(inMsg);
boolean result = engine.executeFilterAsync(inMsg).get(TIMEOUT, TimeUnit.SECONDS);
output = Boolean.toString(result);
break;
case "switch":
Set<String> states = engine.executeSwitch(inMsg);
Set<String> states = engine.executeSwitchAsync(inMsg).get(TIMEOUT, TimeUnit.SECONDS);
output = objectMapper.writeValueAsString(states);
break;
case "json":
JsonNode json = engine.executeJson(inMsg);
JsonNode json = engine.executeJsonAsync(inMsg).get(TIMEOUT, TimeUnit.SECONDS);
output = objectMapper.writeValueAsString(json);
break;
case "string":
output = engine.executeToString(inMsg);
output = engine.executeToStringAsync(inMsg).get(TIMEOUT, TimeUnit.SECONDS);
break;
default:
throw new IllegalArgumentException("Unsupported script type: " + scriptType);

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

@ -25,6 +25,7 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.gen.js.JsInvokeProtos;
import org.thingsboard.server.queue.TbQueueRequestTemplate;
@ -180,6 +181,9 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService {
.setInvokeRequest(jsRequestBuilder.build())
.build();
StopWatch stopWatch = new StopWatch();
stopWatch.start();
ListenableFuture<TbProtoQueueMsg<JsInvokeProtos.RemoteJsResponse>> future = requestTemplate.send(new TbProtoJsQueueMsg<>(UUID.randomUUID(), jsRequestWrapper));
if (maxRequestsTimeout > 0) {
future = Futures.withTimeout(future, maxRequestsTimeout, TimeUnit.MILLISECONDS, timeoutExecutorService);
@ -201,6 +205,8 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService {
}
}, callbackExecutor);
return Futures.transform(future, response -> {
stopWatch.stop();
log.trace("doInvokeFunction js-response took {}ms for uuid {}", stopWatch.getTotalTimeMillis(), response.getKey());
JsInvokeProtos.JsInvokeResponse invokeResult = response.getValue().getInvokeResponse();
if (invokeResult.getSuccess()) {
return invokeResult.getResult();

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

@ -18,12 +18,12 @@ package org.thingsboard.server.service.script;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.TbMsg;
@ -32,6 +32,7 @@ import org.thingsboard.server.common.msg.TbMsgMetaData;
import javax.script.ScriptException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@ -102,140 +103,115 @@ public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.S
String newMessageType = !StringUtils.isEmpty(messageType) ? messageType : msg.getType();
return TbMsg.transformMsg(msg, newMessageType, msg.getOriginator(), newMetadata, newData);
} catch (Throwable th) {
th.printStackTrace();
throw new RuntimeException("Failed to unbind message data from javascript result", th);
}
}
@Override
public List<TbMsg> executeUpdate(TbMsg msg) throws ScriptException {
JsonNode result = executeScript(msg);
if (result.isObject()) {
return Collections.singletonList(unbindMsg(result, msg));
} else if (result.isArray()){
List<TbMsg> res = new ArrayList<>(result.size());
result.forEach(jsonObject -> res.add(unbindMsg(jsonObject, msg)));
return res;
} else {
log.warn("Wrong result type: {}", result.getNodeType());
throw new ScriptException("Wrong result type: " + result.getNodeType());
public ListenableFuture<List<TbMsg>> executeUpdateAsync(TbMsg msg) {
ListenableFuture<JsonNode> result = executeScriptAsync(msg);
return Futures.transformAsync(result,
json -> executeUpdateTransform(msg, json),
MoreExecutors.directExecutor());
}
ListenableFuture<List<TbMsg>> executeUpdateTransform(TbMsg msg, JsonNode json) {
if (json.isObject()) {
return Futures.immediateFuture(Collections.singletonList(unbindMsg(json, msg)));
} else if (json.isArray()) {
List<TbMsg> res = new ArrayList<>(json.size());
json.forEach(jsonObject -> res.add(unbindMsg(jsonObject, msg)));
return Futures.immediateFuture(res);
}
log.warn("Wrong result type: {}", json.getNodeType());
return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + json.getNodeType()));
}
@Override
public ListenableFuture<List<TbMsg>> executeUpdateAsync(TbMsg msg) {
ListenableFuture<JsonNode> result = executeScriptAsync(msg);
return Futures.transformAsync(result, json -> {
if (json.isObject()) {
return Futures.immediateFuture(Collections.singletonList(unbindMsg(json, msg)));
} else if (json.isArray()){
List<TbMsg> res = new ArrayList<>(json.size());
json.forEach(jsonObject -> res.add(unbindMsg(jsonObject, msg)));
return Futures.immediateFuture(res);
}
else{
log.warn("Wrong result type: {}", json.getNodeType());
return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + json.getNodeType()));
}
}, MoreExecutors.directExecutor());
public ListenableFuture<TbMsg> executeGenerateAsync(TbMsg prevMsg) {
return Futures.transformAsync(executeScriptAsync(prevMsg),
result -> executeGenerateTransform(prevMsg, result),
MoreExecutors.directExecutor());
}
@Override
public TbMsg executeGenerate(TbMsg prevMsg) throws ScriptException {
JsonNode result = executeScript(prevMsg);
ListenableFuture<TbMsg> executeGenerateTransform(TbMsg prevMsg, JsonNode result) {
if (!result.isObject()) {
log.warn("Wrong result type: {}", result.getNodeType());
throw new ScriptException("Wrong result type: " + result.getNodeType());
Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + result.getNodeType()));
}
return unbindMsg(result, prevMsg);
return Futures.immediateFuture(unbindMsg(result, prevMsg));
}
@Override
public JsonNode executeJson(TbMsg msg) throws ScriptException {
return executeScript(msg);
}
@Override
public ListenableFuture<JsonNode> executeJsonAsync(TbMsg msg) throws ScriptException {
public ListenableFuture<JsonNode> executeJsonAsync(TbMsg msg) {
return executeScriptAsync(msg);
}
@Override
public String executeToString(TbMsg msg) throws ScriptException {
JsonNode result = executeScript(msg);
if (!result.isTextual()) {
log.warn("Wrong result type: {}", result.getNodeType());
throw new ScriptException("Wrong result type: " + result.getNodeType());
}
return result.asText();
public ListenableFuture<String> executeToStringAsync(TbMsg msg) {
return Futures.transformAsync(executeScriptAsync(msg),
this::executeToStringTransform,
MoreExecutors.directExecutor());
}
@Override
public boolean executeFilter(TbMsg msg) throws ScriptException {
JsonNode result = executeScript(msg);
if (!result.isBoolean()) {
log.warn("Wrong result type: {}", result.getNodeType());
throw new ScriptException("Wrong result type: " + result.getNodeType());
ListenableFuture<String> executeToStringTransform(JsonNode result) {
if (result.isTextual()) {
return Futures.immediateFuture(result.asText());
}
return result.asBoolean();
log.warn("Wrong result type: {}", result.getNodeType());
return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + result.getNodeType()));
}
@Override
public ListenableFuture<Boolean> executeFilterAsync(TbMsg msg) {
ListenableFuture<JsonNode> result = executeScriptAsync(msg);
return Futures.transformAsync(result, json -> {
if (!json.isBoolean()) {
log.warn("Wrong result type: {}", json.getNodeType());
return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + json.getNodeType()));
} else {
return Futures.immediateFuture(json.asBoolean());
}
}, MoreExecutors.directExecutor());
return Futures.transformAsync(executeScriptAsync(msg),
this::executeFilterTransform,
MoreExecutors.directExecutor());
}
@Override
public Set<String> executeSwitch(TbMsg msg) throws ScriptException {
JsonNode result = executeScript(msg);
ListenableFuture<Boolean> executeFilterTransform(JsonNode json) {
if (json.isBoolean()) {
return Futures.immediateFuture(json.asBoolean());
}
log.warn("Wrong result type: {}", json.getNodeType());
return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + json.getNodeType()));
}
ListenableFuture<Set<String>> executeSwitchTransform(JsonNode result) {
if (result.isTextual()) {
return Collections.singleton(result.asText());
} else if (result.isArray()) {
Set<String> nextStates = Sets.newHashSet();
return Futures.immediateFuture(Collections.singleton(result.asText()));
}
if (result.isArray()) {
Set<String> nextStates = new HashSet<>();
for (JsonNode val : result) {
if (!val.isTextual()) {
log.warn("Wrong result type: {}", val.getNodeType());
throw new ScriptException("Wrong result type: " + val.getNodeType());
return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + val.getNodeType()));
} else {
nextStates.add(val.asText());
}
}
return nextStates;
} else {
log.warn("Wrong result type: {}", result.getNodeType());
throw new ScriptException("Wrong result type: " + result.getNodeType());
return Futures.immediateFuture(nextStates);
}
log.warn("Wrong result type: {}", result.getNodeType());
return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + result.getNodeType()));
}
private JsonNode executeScript(TbMsg msg) throws ScriptException {
try {
String[] inArgs = prepareArgs(msg);
String eval = sandboxService.invokeFunction(tenantId, msg.getCustomerId(), this.scriptId, inArgs[0], inArgs[1], inArgs[2]).get().toString();
return mapper.readTree(eval);
} 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);
}
} catch (Exception e) {
throw new ScriptException(e);
}
@Override
public ListenableFuture<Set<String>> executeSwitchAsync(TbMsg msg) {
return Futures.transformAsync(executeScriptAsync(msg),
this::executeSwitchTransform,
MoreExecutors.directExecutor()); //usually runs in a callbackExecutor
}
private ListenableFuture<JsonNode> executeScriptAsync(TbMsg msg) {
ListenableFuture<JsonNode> executeScriptAsync(TbMsg msg) {
log.trace("execute script async, msg {}", msg);
String[] inArgs = prepareArgs(msg);
return Futures.transformAsync(sandboxService.invokeFunction(tenantId, msg.getCustomerId(), this.scriptId, inArgs[0], inArgs[1], inArgs[2]),
return executeScriptAsync(msg.getCustomerId(), inArgs[0], inArgs[1], inArgs[2]);
}
ListenableFuture<JsonNode> executeScriptAsync(CustomerId customerId, Object... args) {
return Futures.transformAsync(sandboxService.invokeFunction(tenantId, customerId, this.scriptId, args),
o -> {
try {
return Futures.immediateFuture(mapper.readTree(o.toString()));

15
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/ScriptEngine.java
View File

@ -19,29 +19,22 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.util.concurrent.ListenableFuture;
import org.thingsboard.server.common.msg.TbMsg;
import javax.script.ScriptException;
import java.util.List;
import java.util.Set;
public interface ScriptEngine {
List<TbMsg> executeUpdate(TbMsg msg) throws ScriptException;
ListenableFuture<List<TbMsg>> executeUpdateAsync(TbMsg msg);
TbMsg executeGenerate(TbMsg prevMsg) throws ScriptException;
boolean executeFilter(TbMsg msg) throws ScriptException;
ListenableFuture<TbMsg> executeGenerateAsync(TbMsg prevMsg);
ListenableFuture<Boolean> executeFilterAsync(TbMsg msg);
Set<String> executeSwitch(TbMsg msg) throws ScriptException;
JsonNode executeJson(TbMsg msg) throws ScriptException;
ListenableFuture<Set<String>> executeSwitchAsync(TbMsg msg);
ListenableFuture<JsonNode> executeJsonAsync(TbMsg msg) throws ScriptException;
ListenableFuture<JsonNode> executeJsonAsync(TbMsg msg);
String executeToString(TbMsg msg) throws ScriptException;
ListenableFuture<String> executeToStringAsync(TbMsg msg);
void destroy();

4
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java
View File

@ -214,6 +214,10 @@ public interface TbContext {
EdgeEventService getEdgeEventService();
/**
* Js script executors call are completely asynchronous
* */
@Deprecated
ListeningExecutor getJsExecutor();
ListeningExecutor getMailExecutor();

30
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbLogNode.java
View File

@ -15,8 +15,11 @@
*/
package org.thingsboard.rule.engine.action;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.ListeningExecutor;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.ScriptEngine;
import org.thingsboard.rule.engine.api.TbContext;
@ -55,18 +58,21 @@ public class TbLogNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
ListeningExecutor jsExecutor = ctx.getJsExecutor();
ctx.logJsEvalRequest();
withCallback(jsExecutor.executeAsync(() -> jsEngine.executeToString(msg)),
toString -> {
ctx.logJsEvalResponse();
log.info(toString);
ctx.tellSuccess(msg);
},
t -> {
ctx.logJsEvalResponse();
ctx.tellFailure(msg, t);
});
Futures.addCallback(jsEngine.executeToStringAsync(msg), new FutureCallback<String>() {
@Override
public void onSuccess(@Nullable String result) {
ctx.logJsEvalResponse();
log.info(result);
ctx.tellSuccess(msg);
}
@Override
public void onFailure(Throwable t) {
ctx.logJsEvalResponse();
ctx.tellFailure(msg, t);
}
}, MoreExecutors.directExecutor()); //usually js responses runs on js callback executor
}
@Override

50
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java
View File

@ -15,9 +15,12 @@
*/
package org.thingsboard.rule.engine.debug;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;
import org.thingsboard.common.util.TbStopWatch;
import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.ScriptEngine;
import org.thingsboard.rule.engine.api.TbContext;
@ -35,6 +38,7 @@ import org.thingsboard.server.common.msg.queue.ServiceQueue;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.thingsboard.common.util.DonAsynchron.withCallback;
import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
@ -64,10 +68,11 @@ public class TbMsgGeneratorNode implements TbNode {
private EntityId originatorId;
private UUID nextTickId;
private TbMsg prevMsg;
private volatile boolean initialized;
private final AtomicBoolean initialized = new AtomicBoolean(false);
@Override
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
log.trace("init generator with config {}", configuration);
this.config = TbNodeUtils.convert(configuration, TbMsgGeneratorNodeConfiguration.class);
this.delay = TimeUnit.SECONDS.toMillis(config.getPeriodInSeconds());
this.currentMsgCount = 0;
@ -81,35 +86,39 @@ public class TbMsgGeneratorNode implements TbNode {
@Override
public void onPartitionChangeMsg(TbContext ctx, PartitionChangeMsg msg) {
log.trace("onPartitionChangeMsg, PartitionChangeMsg {}, config {}", msg, config);
updateGeneratorState(ctx);
}
private void updateGeneratorState(TbContext ctx) {
log.trace("updateGeneratorState, config {}", config);
if (ctx.isLocalEntity(originatorId)) {
if (!initialized) {
initialized = true;
if (initialized.compareAndSet(false, true)) {
this.jsEngine = ctx.createJsScriptEngine(config.getJsScript(), "prevMsg", "prevMetadata", "prevMsgType");
scheduleTickMsg(ctx);
}
} else if (initialized) {
initialized = false;
} else if (initialized.compareAndSet(true, false)) {
destroy();
}
}
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
if (initialized && msg.getType().equals(TB_MSG_GENERATOR_NODE_MSG) && msg.getId().equals(nextTickId)) {
log.trace("onMsg, config {}, msg {}", config, msg);
if (initialized.get() && msg.getType().equals(TB_MSG_GENERATOR_NODE_MSG) && msg.getId().equals(nextTickId)) {
TbStopWatch sw = TbStopWatch.startNew();
withCallback(generate(ctx, msg),
m -> {
if (initialized && (config.getMsgCount() == TbMsgGeneratorNodeConfiguration.UNLIMITED_MSG_COUNT || currentMsgCount < config.getMsgCount())) {
log.trace("onMsg onSuccess callback, took {}ms, config {}, msg {}", sw.stopAndGetTotalTimeMillis(), config, msg);
if (initialized.get() && (config.getMsgCount() == TbMsgGeneratorNodeConfiguration.UNLIMITED_MSG_COUNT || currentMsgCount < config.getMsgCount())) {
ctx.enqueueForTellNext(m, SUCCESS);
scheduleTickMsg(ctx);
currentMsgCount++;
}
},
t -> {
if (initialized && (config.getMsgCount() == TbMsgGeneratorNodeConfiguration.UNLIMITED_MSG_COUNT || currentMsgCount < config.getMsgCount())) {
log.warn("onMsg onFailure callback, took {}ms, config {}, msg {}, exception {}", sw.stopAndGetTotalTimeMillis(), config, msg, t);
if (initialized.get() && (config.getMsgCount() == TbMsgGeneratorNodeConfiguration.UNLIMITED_MSG_COUNT || currentMsgCount < config.getMsgCount())) {
ctx.tellFailure(msg, t);
scheduleTickMsg(ctx);
currentMsgCount++;
@ -119,6 +128,7 @@ public class TbMsgGeneratorNode implements TbNode {
}
private void scheduleTickMsg(TbContext ctx) {
log.trace("scheduleTickMsg, config {}", config);
long curTs = System.currentTimeMillis();
if (lastScheduledTs == 0L) {
lastScheduledTs = curTs;
@ -131,22 +141,26 @@ public class TbMsgGeneratorNode implements TbNode {
}
private ListenableFuture<TbMsg> generate(TbContext ctx, TbMsg msg) {
return ctx.getJsExecutor().executeAsync(() -> {
if (prevMsg == null) {
prevMsg = ctx.newMsg(ServiceQueue.MAIN, "", originatorId, msg.getCustomerId(), new TbMsgMetaData(), "{}");
}
if (initialized) {
ctx.logJsEvalRequest();
TbMsg generated = jsEngine.executeGenerate(prevMsg);
log.trace("generate, config {}", config);
if (prevMsg == null) {
prevMsg = ctx.newMsg(ServiceQueue.MAIN, "", originatorId, msg.getCustomerId(), new TbMsgMetaData(), "{}");
}
if (initialized.get()) {
ctx.logJsEvalRequest();
return Futures.transformAsync(jsEngine.executeGenerateAsync(prevMsg), generated -> {
log.trace("generate process response, generated {}, config {}", generated, config);
ctx.logJsEvalResponse();
prevMsg = ctx.newMsg(ServiceQueue.MAIN, generated.getType(), originatorId, msg.getCustomerId(), generated.getMetaData(), generated.getData());
}
return prevMsg;
});
return Futures.immediateFuture(prevMsg);
}, MoreExecutors.directExecutor()); //usually it runs on js-executor-remote-callback thread pool
}
return Futures.immediateFuture(prevMsg);
}
@Override
public void destroy() {
log.trace("destroy, config {}", config);
prevMsg = null;
if (jsEngine != null) {
jsEngine.destroy();

29
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsSwitchNode.java
View File

@ -15,7 +15,11 @@
*/
package org.thingsboard.rule.engine.filter;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.thingsboard.common.util.ListeningExecutor;
import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.ScriptEngine;
@ -29,8 +33,6 @@ import org.thingsboard.server.common.msg.TbMsg;
import java.util.Set;
import static org.thingsboard.common.util.DonAsynchron.withCallback;
@Slf4j
@RuleNode(
type = ComponentType.FILTER,
@ -58,17 +60,20 @@ public class TbJsSwitchNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
ListeningExecutor jsExecutor = ctx.getJsExecutor();
ctx.logJsEvalRequest();
withCallback(jsExecutor.executeAsync(() -> jsEngine.executeSwitch(msg)),
result -> {
ctx.logJsEvalResponse();
processSwitch(ctx, msg, result);
},
t -> {
ctx.logJsEvalFailure();
ctx.tellFailure(msg, t);
}, ctx.getDbCallbackExecutor());
Futures.addCallback(jsEngine.executeSwitchAsync(msg), new FutureCallback<Set<String>>() {
@Override
public void onSuccess(@Nullable Set<String> result) {
ctx.logJsEvalResponse();
processSwitch(ctx, msg, result);
}
@Override
public void onFailure(Throwable t) {
ctx.logJsEvalFailure();
ctx.tellFailure(msg, t);
}
}, MoreExecutors.directExecutor()); //usually runs in a callbackExecutor
}
private void processSwitch(TbContext ctx, TbMsg msg, Set<String> nextRelations) {

19
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsSwitchNodeTest.java
View File

@ -64,7 +64,7 @@ public class TbJsSwitchNodeTest {
private RuleNodeId ruleNodeId = new RuleNodeId(Uuids.timeBased());
@Test
public void multipleRoutesAreAllowed() throws TbNodeException, ScriptException {
public void multipleRoutesAreAllowed() throws TbNodeException {
initWithScript();
TbMsgMetaData metaData = new TbMsgMetaData();
metaData.putValue("temp", "10");
@ -72,11 +72,9 @@ public class TbJsSwitchNodeTest {
String rawJson = "{\"name\": \"Vit\", \"passed\": 5}";
TbMsg msg = TbMsg.newMsg( "USER", null, metaData, TbMsgDataType.JSON, rawJson, ruleChainId, ruleNodeId);
mockJsExecutor();
when(scriptEngine.executeSwitch(msg)).thenReturn(Sets.newHashSet("one", "three"));
when(scriptEngine.executeSwitchAsync(msg)).thenReturn(Futures.immediateFuture(Sets.newHashSet("one", "three")));
node.onMsg(ctx, msg);
verify(ctx).getJsExecutor();
verify(ctx).tellNext(msg, Sets.newHashSet("one", "three"));
}
@ -92,19 +90,6 @@ public class TbJsSwitchNodeTest {
node.init(ctx, nodeConfiguration);
}
@SuppressWarnings("unchecked")
private void mockJsExecutor() {
when(ctx.getJsExecutor()).thenReturn(executor);
doAnswer((Answer<ListenableFuture<Set<String>>>) invocationOnMock -> {
try {
Callable task = (Callable) (invocationOnMock.getArguments())[0];
return Futures.immediateFuture((Set<String>) task.call());
} catch (Throwable th) {
return Futures.immediateFailedFuture(th);
}
}).when(executor).executeAsync(ArgumentMatchers.any(Callable.class));
}
private void verifyError(TbMsg msg, String message, Class expectedClass) {
ArgumentCaptor<Throwable> captor = ArgumentCaptor.forClass(Throwable.class);
verify(ctx).tellFailure(same(msg), captor.capture());

Write Preview
Loading…
Cancel
Save