简柏林 / thingskit · Commits

Commit b397dfb518e101558790127b5e7ad9a7e7aef34f

Authored by Sergey Matvienko
1 parent f3757ad1

js-script-engine-api: js sync calls replaced with completely async calls (CE API only)

Inline Side-by-side
Showing 7 changed files with 82 additions and 128 deletions
... ... @@ -390,13 +390,13 @@ public class RuleChainController extends BaseController {
390 390 TbMsg inMsg = TbMsg.newMsg(msgType, null, new TbMsgMetaData(metadata), TbMsgDataType.JSON, data);
391 391 switch (scriptType) {
392 392 case "update":
393   - output = msgToOutput(engine.executeUpdate(inMsg));
  393 + output = msgToOutput(engine.executeUpdateAsync(inMsg).get(TIMEOUT, TimeUnit.SECONDS));
394 394 break;
395 395 case "generate":
396 396 output = msgToOutput(engine.executeGenerateAsync(inMsg).get(TIMEOUT, TimeUnit.SECONDS));
397 397 break;
398 398 case "filter":
399   - boolean result = engine.executeFilter(inMsg);
  399 + boolean result = engine.executeFilterAsync(inMsg).get(TIMEOUT, TimeUnit.SECONDS);
400 400 output = Boolean.toString(result);
401 401 break;
402 402 case "switch":
... ... @@ -404,11 +404,11 @@ public class RuleChainController extends BaseController {
404 404 output = objectMapper.writeValueAsString(states);
405 405 break;
406 406 case "json":
407   - JsonNode json = engine.executeJson(inMsg);
  407 + JsonNode json = engine.executeJsonAsync(inMsg).get(TIMEOUT, TimeUnit.SECONDS);
408 408 output = objectMapper.writeValueAsString(json);
409 409 break;
410 410 case "string":
411   - output = engine.executeToString(inMsg);
  411 + output = engine.executeToStringAsync(inMsg).get(TIMEOUT, TimeUnit.SECONDS);
412 412 break;
413 413 default:
414 414 throw new IllegalArgumentException("Unsupported script type: " + scriptType);
... ...
... ... @@ -107,95 +107,76 @@ public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.S
107 107 }
108 108
109 109 @Override
110   - public List<TbMsg> executeUpdate(TbMsg msg) throws ScriptException {
111   - JsonNode result = executeScript(msg);
112   - if (result.isObject()) {
113   - return Collections.singletonList(unbindMsg(result, msg));
114   - } else if (result.isArray()){
115   - List<TbMsg> res = new ArrayList<>(result.size());
116   - result.forEach(jsonObject -> res.add(unbindMsg(jsonObject, msg)));
117   - return res;
118   - } else {
119   - log.warn("Wrong result type: {}", result.getNodeType());
120   - throw new ScriptException("Wrong result type: " + result.getNodeType());
121   - }
122   - }
123   -
124   - @Override
125 110 public ListenableFuture<List<TbMsg>> executeUpdateAsync(TbMsg msg) {
126 111 ListenableFuture<JsonNode> result = executeScriptAsync(msg);
127   - return Futures.transformAsync(result, json -> {
128   - if (json.isObject()) {
129   - return Futures.immediateFuture(Collections.singletonList(unbindMsg(json, msg)));
130   - } else if (json.isArray()){
131   - List<TbMsg> res = new ArrayList<>(json.size());
132   - json.forEach(jsonObject -> res.add(unbindMsg(jsonObject, msg)));
133   - return Futures.immediateFuture(res);
134   - }
135   - else{
136   - log.warn("Wrong result type: {}", json.getNodeType());
137   - return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + json.getNodeType()));
138   - }
139   - }, MoreExecutors.directExecutor());
  112 + return Futures.transformAsync(result,
  113 + json -> executeUpdateTransform(msg, json),
  114 + MoreExecutors.directExecutor());
  115 + }
  116 +
  117 + ListenableFuture<List<TbMsg>> executeUpdateTransform(TbMsg msg, JsonNode json) {
  118 + if (json.isObject()) {
  119 + return Futures.immediateFuture(Collections.singletonList(unbindMsg(json, msg)));
  120 + } else if (json.isArray()) {
  121 + List<TbMsg> res = new ArrayList<>(json.size());
  122 + json.forEach(jsonObject -> res.add(unbindMsg(jsonObject, msg)));
  123 + return Futures.immediateFuture(res);
  124 + }
  125 + log.warn("Wrong result type: {}", json.getNodeType());
  126 + return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + json.getNodeType()));
140 127 }
141 128
142 129 @Override
143 130 public ListenableFuture<TbMsg> executeGenerateAsync(TbMsg prevMsg) {
144   - log.trace("execute generate async, prevMsg {}", prevMsg);
145   - return Futures.transformAsync(executeScriptAsync(prevMsg), result -> {
146   - if (!result.isObject()) {
147   - log.warn("Wrong result type: {}", result.getNodeType());
148   - Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + result.getNodeType()));
149   - }
150   - return Futures.immediateFuture(unbindMsg(result, prevMsg));
151   - }, MoreExecutors.directExecutor());
152   -
  131 + return Futures.transformAsync(executeScriptAsync(prevMsg),
  132 + result -> executeGenerateTransform(prevMsg, result),
  133 + MoreExecutors.directExecutor());
153 134 }
154 135
155   - @Override
156   - public JsonNode executeJson(TbMsg msg) throws ScriptException {
157   - return executeScript(msg);
  136 + ListenableFuture<TbMsg> executeGenerateTransform(TbMsg prevMsg, JsonNode result) {
  137 + if (!result.isObject()) {
  138 + log.warn("Wrong result type: {}", result.getNodeType());
  139 + Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + result.getNodeType()));
  140 + }
  141 + return Futures.immediateFuture(unbindMsg(result, prevMsg));
158 142 }
159 143
160 144 @Override
161   - public ListenableFuture<JsonNode> executeJsonAsync(TbMsg msg) throws ScriptException {
  145 + public ListenableFuture<JsonNode> executeJsonAsync(TbMsg msg) {
162 146 return executeScriptAsync(msg);
163 147 }
164 148
165 149 @Override
166   - public String executeToString(TbMsg msg) throws ScriptException {
167   - JsonNode result = executeScript(msg);
168   - if (!result.isTextual()) {
169   - log.warn("Wrong result type: {}", result.getNodeType());
170   - throw new ScriptException("Wrong result type: " + result.getNodeType());
171   - }
172   - return result.asText();
  150 + public ListenableFuture<String> executeToStringAsync(TbMsg msg) {
  151 + return Futures.transformAsync(executeScriptAsync(msg),
  152 + this::executeToStringTransform,
  153 + MoreExecutors.directExecutor());
173 154 }
174 155
175   - @Override
176   - public boolean executeFilter(TbMsg msg) throws ScriptException {
177   - JsonNode result = executeScript(msg);
178   - if (!result.isBoolean()) {
179   - log.warn("Wrong result type: {}", result.getNodeType());
180   - throw new ScriptException("Wrong result type: " + result.getNodeType());
  156 + ListenableFuture<String> executeToStringTransform(JsonNode result) {
  157 + if (result.isTextual()) {
  158 + return Futures.immediateFuture(result.asText());
181 159 }
182   - return result.asBoolean();
  160 + log.warn("Wrong result type: {}", result.getNodeType());
  161 + return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + result.getNodeType()));
183 162 }
184 163
185 164 @Override
186 165 public ListenableFuture<Boolean> executeFilterAsync(TbMsg msg) {
187   - ListenableFuture<JsonNode> result = executeScriptAsync(msg);
188   - return Futures.transformAsync(result, json -> {
189   - if (!json.isBoolean()) {
190   - log.warn("Wrong result type: {}", json.getNodeType());
191   - return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + json.getNodeType()));
192   - } else {
193   - return Futures.immediateFuture(json.asBoolean());
194   - }
195   - }, MoreExecutors.directExecutor());
  166 + return Futures.transformAsync(executeScriptAsync(msg),
  167 + this::executeFilterTransform,
  168 + MoreExecutors.directExecutor());
  169 + }
  170 +
  171 + ListenableFuture<Boolean> executeFilterTransform(JsonNode json) {
  172 + if (json.isBoolean()) {
  173 + return Futures.immediateFuture(json.asBoolean());
  174 + }
  175 + log.warn("Wrong result type: {}", json.getNodeType());
  176 + return Futures.immediateFailedFuture(new ScriptException("Wrong result type: " + json.getNodeType()));
196 177 }
197 178
198   - ListenableFuture<Set<String>> executeSwitchPostProcessAsyncFunction(JsonNode result) {
  179 + ListenableFuture<Set<String>> executeSwitchTransform(JsonNode result) {
199 180 if (result.isTextual()) {
200 181 return Futures.immediateFuture(Collections.singleton(result.asText()));
201 182 }
... ... @@ -217,34 +198,19 @@ public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.S
217 198
218 199 @Override
219 200 public ListenableFuture<Set<String>> executeSwitchAsync(TbMsg msg) {
220   - log.trace("execute switch async, msg {}", msg);
221 201 return Futures.transformAsync(executeScriptAsync(msg),
222   - this::executeSwitchPostProcessAsyncFunction,
  202 + this::executeSwitchTransform,
223 203 MoreExecutors.directExecutor()); //usually runs in a callbackExecutor
224 204 }
225 205
226   - private JsonNode executeScript(TbMsg msg) throws ScriptException {
227   - try {
228   - String[] inArgs = prepareArgs(msg);
229   - String eval = sandboxService.invokeFunction(tenantId, msg.getCustomerId(), this.scriptId, inArgs[0], inArgs[1], inArgs[2]).get().toString();
230   - return mapper.readTree(eval);
231   - } catch (ExecutionException e) {
232   - if (e.getCause() instanceof ScriptException) {
233   - throw (ScriptException) e.getCause();
234   - } else if (e.getCause() instanceof RuntimeException) {
235   - throw new ScriptException(e.getCause().getMessage());
236   - } else {
237   - throw new ScriptException(e);
238   - }
239   - } catch (Exception e) {
240   - throw new ScriptException(e);
241   - }
242   - }
243   -
244   - private ListenableFuture<JsonNode> executeScriptAsync(TbMsg msg) {
  206 + ListenableFuture<JsonNode> executeScriptAsync(TbMsg msg) {
245 207 log.trace("execute script async, msg {}", msg);
246 208 String[] inArgs = prepareArgs(msg);
247   - return Futures.transformAsync(sandboxService.invokeFunction(tenantId, msg.getCustomerId(), this.scriptId, inArgs[0], inArgs[1], inArgs[2]),
  209 + return executeScriptAsync(msg.getCustomerId(), inArgs[0], inArgs[1], inArgs[2]);
  210 + }
  211 +
  212 + ListenableFuture<JsonNode> executeScriptAsync(CustomerId customerId, Object... args) {
  213 + return Futures.transformAsync(sandboxService.invokeFunction(tenantId, customerId, this.scriptId, args),
248 214 o -> {
249 215 try {
250 216 return Futures.immediateFuture(mapper.readTree(o.toString()));
... ...
... ... @@ -19,14 +19,11 @@ import com.fasterxml.jackson.databind.JsonNode;
19 19 import com.google.common.util.concurrent.ListenableFuture;
20 20 import org.thingsboard.server.common.msg.TbMsg;
21 21
22   -import javax.script.ScriptException;
23 22 import java.util.List;
24 23 import java.util.Set;
25 24
26 25 public interface ScriptEngine {
27 26
28   - List<TbMsg> executeUpdate(TbMsg msg) throws ScriptException;
29   -
30 27 ListenableFuture<List<TbMsg>> executeUpdateAsync(TbMsg msg);
31 28
32 29 ListenableFuture<TbMsg> executeGenerateAsync(TbMsg prevMsg);
... ... @@ -37,11 +34,9 @@ public interface ScriptEngine {
37 34
38 35 ListenableFuture<Set<String>> executeSwitchAsync(TbMsg msg);
39 36
40   - JsonNode executeJson(TbMsg msg) throws ScriptException;
41   -
42   - ListenableFuture<JsonNode> executeJsonAsync(TbMsg msg) throws ScriptException;
  37 + ListenableFuture<JsonNode> executeJsonAsync(TbMsg msg);
43 38
44   - String executeToString(TbMsg msg) throws ScriptException;
  39 + ListenableFuture<String> executeToStringAsync(TbMsg msg);
45 40
46 41 void destroy();
47 42
... ...
... ... @@ -214,6 +214,10 @@ public interface TbContext {
214 214
215 215 EdgeEventService getEdgeEventService();
216 216
  217 + /**
  218 + * Js script executors call are completely asynchronous
  219 + * */
  220 + @Deprecated
217 221 ListeningExecutor getJsExecutor();
218 222
219 223 ListeningExecutor getMailExecutor();
... ...
... ... @@ -15,8 +15,11 @@
15 15 */
16 16 package org.thingsboard.rule.engine.action;
17 17
  18 +import com.google.common.util.concurrent.FutureCallback;
  19 +import com.google.common.util.concurrent.Futures;
  20 +import com.google.common.util.concurrent.MoreExecutors;
18 21 import lombok.extern.slf4j.Slf4j;
19   -import org.thingsboard.common.util.ListeningExecutor;
  22 +import org.checkerframework.checker.nullness.qual.Nullable;
20 23 import org.thingsboard.rule.engine.api.RuleNode;
21 24 import org.thingsboard.rule.engine.api.ScriptEngine;
22 25 import org.thingsboard.rule.engine.api.TbContext;
... ... @@ -55,18 +58,21 @@ public class TbLogNode implements TbNode {
55 58
56 59 @Override
57 60 public void onMsg(TbContext ctx, TbMsg msg) {
58   - ListeningExecutor jsExecutor = ctx.getJsExecutor();
59 61 ctx.logJsEvalRequest();
60   - withCallback(jsExecutor.executeAsync(() -> jsEngine.executeToString(msg)),
61   - toString -> {
62   - ctx.logJsEvalResponse();
63   - log.info(toString);
64   - ctx.tellSuccess(msg);
65   - },
66   - t -> {
67   - ctx.logJsEvalResponse();
68   - ctx.tellFailure(msg, t);
69   - });
  62 + Futures.addCallback(jsEngine.executeToStringAsync(msg), new FutureCallback<String>() {
  63 + @Override
  64 + public void onSuccess(@Nullable String result) {
  65 + ctx.logJsEvalResponse();
  66 + log.info(result);
  67 + ctx.tellSuccess(msg);
  68 + }
  69 +
  70 + @Override
  71 + public void onFailure(Throwable t) {
  72 + ctx.logJsEvalResponse();
  73 + ctx.tellFailure(msg, t);
  74 + }
  75 + }, MoreExecutors.directExecutor()); //usually js responses runs on js callback executor
70 76 }
71 77
72 78 @Override
... ...
... ... @@ -33,8 +33,6 @@ import org.thingsboard.server.common.msg.TbMsg;
33 33
34 34 import java.util.Set;
35 35
36   -import static org.thingsboard.common.util.DonAsynchron.withCallback;
37   -
38 36 @Slf4j
39 37 @RuleNode(
40 38 type = ComponentType.FILTER,
... ...
... ... @@ -64,7 +64,7 @@ public class TbJsSwitchNodeTest {
64 64 private RuleNodeId ruleNodeId = new RuleNodeId(Uuids.timeBased());
65 65
66 66 @Test
67   - public void multipleRoutesAreAllowed() throws TbNodeException, ScriptException {
  67 + public void multipleRoutesAreAllowed() throws TbNodeException {
68 68 initWithScript();
69 69 TbMsgMetaData metaData = new TbMsgMetaData();
70 70 metaData.putValue("temp", "10");
... ... @@ -72,11 +72,9 @@ public class TbJsSwitchNodeTest {
72 72 String rawJson = "{\"name\": \"Vit\", \"passed\": 5}";
73 73
74 74 TbMsg msg = TbMsg.newMsg( "USER", null, metaData, TbMsgDataType.JSON, rawJson, ruleChainId, ruleNodeId);
75   - mockJsExecutor();
76 75 when(scriptEngine.executeSwitchAsync(msg)).thenReturn(Futures.immediateFuture(Sets.newHashSet("one", "three")));
77 76
78 77 node.onMsg(ctx, msg);
79   - verify(ctx).getJsExecutor();
80 78 verify(ctx).tellNext(msg, Sets.newHashSet("one", "three"));
81 79 }
82 80
... ... @@ -92,19 +90,6 @@ public class TbJsSwitchNodeTest {
92 90 node.init(ctx, nodeConfiguration);
93 91 }
94 92
95   - @SuppressWarnings("unchecked")
96   - private void mockJsExecutor() {
97   - when(ctx.getJsExecutor()).thenReturn(executor);
98   - doAnswer((Answer<ListenableFuture<Set<String>>>) invocationOnMock -> {
99   - try {
100   - Callable task = (Callable) (invocationOnMock.getArguments())[0];
101   - return Futures.immediateFuture((Set<String>) task.call());
102   - } catch (Throwable th) {
103   - return Futures.immediateFailedFuture(th);
104   - }
105   - }).when(executor).executeAsync(ArgumentMatchers.any(Callable.class));
106   - }
107   -
108 93 private void verifyError(TbMsg msg, String message, Class expectedClass) {
109 94 ArgumentCaptor<Throwable> captor = ArgumentCaptor.forClass(Throwable.class);
110 95 verify(ctx).tellFailure(same(msg), captor.capture());
... ...