From b397dfb518e101558790127b5e7ad9a7e7aef34f Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Wed, 16 Jun 2021 20:49:46 +0300 Subject: [PATCH] js-script-engine-api: js sync calls replaced with completely async calls (CE API only) --- .../controller/RuleChainController.java | 8 +- .../script/RuleNodeJsScriptEngine.java | 140 +++++++----------- .../rule/engine/api/ScriptEngine.java | 9 +- .../rule/engine/api/TbContext.java | 4 + .../rule/engine/action/TbLogNode.java | 30 ++-- .../rule/engine/filter/TbJsSwitchNode.java | 2 - .../engine/filter/TbJsSwitchNodeTest.java | 17 +-- 7 files changed, 82 insertions(+), 128 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java index 39692922c1..fbfd0d595f 100644 --- a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java +++ b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java @@ -390,13 +390,13 @@ 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.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": @@ -404,11 +404,11 @@ public class RuleChainController extends BaseController { 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); diff --git a/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java b/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java index 2e6b48693a..17e999a533 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java +++ b/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java @@ -106,96 +106,77 @@ public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.S } } - @Override - public List executeUpdate(TbMsg msg) throws ScriptException { - JsonNode result = executeScript(msg); - if (result.isObject()) { - return Collections.singletonList(unbindMsg(result, msg)); - } else if (result.isArray()){ - List 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()); - } - } - @Override public ListenableFuture> executeUpdateAsync(TbMsg msg) { ListenableFuture result = executeScriptAsync(msg); - return Futures.transformAsync(result, json -> { - if (json.isObject()) { - return Futures.immediateFuture(Collections.singletonList(unbindMsg(json, msg))); - } else if (json.isArray()){ - List 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()); + return Futures.transformAsync(result, + json -> executeUpdateTransform(msg, json), + MoreExecutors.directExecutor()); + } + + ListenableFuture> executeUpdateTransform(TbMsg msg, JsonNode json) { + if (json.isObject()) { + return Futures.immediateFuture(Collections.singletonList(unbindMsg(json, msg))); + } else if (json.isArray()) { + List 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 executeGenerateAsync(TbMsg prevMsg) { - log.trace("execute generate async, prevMsg {}", prevMsg); - return Futures.transformAsync(executeScriptAsync(prevMsg), result -> { - if (!result.isObject()) { - log.warn("Wrong result type: {}", result.getNodeType()); - Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + result.getNodeType())); - } - return Futures.immediateFuture(unbindMsg(result, prevMsg)); - }, MoreExecutors.directExecutor()); - + return Futures.transformAsync(executeScriptAsync(prevMsg), + result -> executeGenerateTransform(prevMsg, result), + MoreExecutors.directExecutor()); } - @Override - public JsonNode executeJson(TbMsg msg) throws ScriptException { - return executeScript(msg); + ListenableFuture executeGenerateTransform(TbMsg prevMsg, JsonNode result) { + if (!result.isObject()) { + log.warn("Wrong result type: {}", result.getNodeType()); + Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + result.getNodeType())); + } + return Futures.immediateFuture(unbindMsg(result, prevMsg)); } @Override - public ListenableFuture executeJsonAsync(TbMsg msg) throws ScriptException { + public ListenableFuture 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 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 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 executeFilterAsync(TbMsg msg) { - ListenableFuture 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()); + } + + ListenableFuture 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> executeSwitchPostProcessAsyncFunction(JsonNode result) { + ListenableFuture> executeSwitchTransform(JsonNode result) { if (result.isTextual()) { return Futures.immediateFuture(Collections.singleton(result.asText())); } @@ -217,34 +198,19 @@ public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.S @Override public ListenableFuture> executeSwitchAsync(TbMsg msg) { - log.trace("execute switch async, msg {}", msg); return Futures.transformAsync(executeScriptAsync(msg), - this::executeSwitchPostProcessAsyncFunction, + this::executeSwitchTransform, MoreExecutors.directExecutor()); //usually runs in a callbackExecutor } - 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); - } - } - - private ListenableFuture executeScriptAsync(TbMsg msg) { + ListenableFuture 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 executeScriptAsync(CustomerId customerId, Object... args) { + return Futures.transformAsync(sandboxService.invokeFunction(tenantId, customerId, this.scriptId, args), o -> { try { return Futures.immediateFuture(mapper.readTree(o.toString())); diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/ScriptEngine.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/ScriptEngine.java index da9bd93f2b..012e135c60 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/ScriptEngine.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/ScriptEngine.java @@ -19,14 +19,11 @@ 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 executeUpdate(TbMsg msg) throws ScriptException; - ListenableFuture> executeUpdateAsync(TbMsg msg); ListenableFuture executeGenerateAsync(TbMsg prevMsg); @@ -37,11 +34,9 @@ public interface ScriptEngine { ListenableFuture> executeSwitchAsync(TbMsg msg); - JsonNode executeJson(TbMsg msg) throws ScriptException; - - ListenableFuture executeJsonAsync(TbMsg msg) throws ScriptException; + ListenableFuture executeJsonAsync(TbMsg msg); - String executeToString(TbMsg msg) throws ScriptException; + ListenableFuture executeToStringAsync(TbMsg msg); void destroy(); diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 43e3a5e329..62ed9b0fa0 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -214,6 +214,10 @@ public interface TbContext { EdgeEventService getEdgeEventService(); + /** + * Js script executors call are completely asynchronous + * */ + @Deprecated ListeningExecutor getJsExecutor(); ListeningExecutor getMailExecutor(); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbLogNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbLogNode.java index 2b2f0d76a4..c981d5062d 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbLogNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbLogNode.java @@ -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() { + @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 diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsSwitchNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsSwitchNode.java index 5af4ccb29d..bc900a0f2d 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsSwitchNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsSwitchNode.java @@ -33,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, diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsSwitchNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsSwitchNodeTest.java index fc2e5aa041..4ac81120b0 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsSwitchNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsSwitchNodeTest.java @@ -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.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>>) invocationOnMock -> { - try { - Callable task = (Callable) (invocationOnMock.getArguments())[0]; - return Futures.immediateFuture((Set) 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 captor = ArgumentCaptor.forClass(Throwable.class); verify(ctx).tellFailure(same(msg), captor.capture());