From da7da1ef76ce84e49f8169554cb9d984d099e64f Mon Sep 17 00:00:00 2001 From: Andrew Shvayka Date: Wed, 11 Dec 2019 14:27:13 +0200 Subject: [PATCH] Async JS Execution --- .../script/RuleNodeJsScriptEngine.java | 5 +++ .../rule/engine/api/ScriptEngine.java | 2 + .../engine/action/TbAbstractAlarmNode.java | 15 ++++++-- .../rule/engine/action/TbAlarmNodeTest.java | 38 +++---------------- 4 files changed, 24 insertions(+), 36 deletions(-) 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 12150b806a..8b08333d3b 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 @@ -139,6 +139,11 @@ public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.S return executeScript(msg); } + @Override + public ListenableFuture executeJsonAsync(TbMsg msg) throws ScriptException { + return executeScriptAsync(msg); + } + @Override public String executeToString(TbMsg msg) throws ScriptException { JsonNode result = executeScript(msg); 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 f72adf04ab..1ffaabd6d1 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 @@ -38,6 +38,8 @@ public interface ScriptEngine { JsonNode executeJson(TbMsg msg) throws ScriptException; + ListenableFuture executeJsonAsync(TbMsg msg) throws ScriptException; + String executeToString(TbMsg msg) throws ScriptException; void destroy(); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java index 5fc482348d..b8fbeaf74e 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java @@ -17,6 +17,7 @@ package org.thingsboard.rule.engine.action; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.thingsboard.rule.engine.api.ScriptEngine; @@ -27,6 +28,9 @@ import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; + +import javax.script.ScriptException; + import static org.thingsboard.common.util.DonAsynchron.withCallback; @@ -67,21 +71,24 @@ public abstract class TbAbstractAlarmNode ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); + t -> ctx.tellFailure(msg, t) + , ctx.getDbCallbackExecutor()); } protected abstract ListenableFuture processAlarm(TbContext ctx, TbMsg msg); protected ListenableFuture buildAlarmDetails(TbContext ctx, TbMsg msg, JsonNode previousDetails) { - return ctx.getJsExecutor().executeAsync(() -> { + try { TbMsg dummyMsg = msg; if (previousDetails != null) { TbMsgMetaData metaData = msg.getMetaData().copy(); metaData.putValue(PREV_ALARM_DETAILS, mapper.writeValueAsString(previousDetails)); dummyMsg = ctx.transformMsg(msg, msg.getType(), msg.getOriginator(), metaData, msg.getData()); } - return buildDetailsJsEngine.executeJson(dummyMsg); - }); + return buildDetailsJsEngine.executeJsonAsync(dummyMsg); + } catch (Exception e) { + return Futures.immediateFailedFuture(e); + } } private TbMsg toAlarmMsg(TbContext ctx, AlarmResult alarmResult, TbMsg originalMsg) { diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java index fb07951368..1884ff2330 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java @@ -26,6 +26,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.ArgumentCaptor; import org.mockito.Mock; +import org.mockito.Mockito; import org.mockito.runners.MockitoJUnitRunner; import org.mockito.stubbing.Answer; import org.thingsboard.common.util.ListeningExecutor; @@ -60,8 +61,6 @@ public class TbAlarmNodeTest { @Mock private TbContext ctx; @Mock - private ListeningExecutor executor; - @Mock private AlarmService alarmService; @Mock @@ -102,7 +101,7 @@ public class TbAlarmNodeTest { metaData.putValue("key", "value"); TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson, ruleChainId, ruleNodeId, 0L); - when(detailsJs.executeJson(msg)).thenReturn(null); + when(detailsJs.executeJsonAsync(msg)).thenReturn(Futures.immediateFuture(null)); when(alarmService.findLatestByOriginatorAndType(tenantId, originator, "SomeType")).thenReturn(Futures.immediateFuture(null)); doAnswer((Answer) invocationOnMock -> (Alarm) (invocationOnMock.getArguments())[0]).when(alarmService).createOrUpdateAlarm(any(Alarm.class)); @@ -136,8 +135,6 @@ public class TbAlarmNodeTest { .build(); assertEquals(expectedAlarm, actualAlarm); - - verify(executor, times(1)).executeAsync(any(Callable.class)); } @Test @@ -146,7 +143,7 @@ public class TbAlarmNodeTest { metaData.putValue("key", "value"); TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson, ruleChainId, ruleNodeId, 0L); - when(detailsJs.executeJson(msg)).thenThrow(new NotImplementedException("message")); + when(detailsJs.executeJsonAsync(msg)).thenReturn(Futures.immediateFailedFuture(new NotImplementedException("message"))); when(alarmService.findLatestByOriginatorAndType(tenantId, originator, "SomeType")).thenReturn(Futures.immediateFuture(null)); node.onMsg(ctx, msg); @@ -154,7 +151,6 @@ public class TbAlarmNodeTest { verifyError(msg, "message", NotImplementedException.class); verify(ctx).createJsScriptEngine("DETAILS"); - verify(ctx, times(1)).getJsExecutor(); verify(ctx).getAlarmService(); verify(ctx, times(3)).getDbCallbackExecutor(); verify(ctx).logJsEvalRequest(); @@ -172,7 +168,7 @@ public class TbAlarmNodeTest { Alarm clearedAlarm = Alarm.builder().status(CLEARED_ACK).build(); - when(detailsJs.executeJson(msg)).thenReturn(null); + when(detailsJs.executeJsonAsync(msg)).thenReturn(Futures.immediateFuture(null)); when(alarmService.findLatestByOriginatorAndType(tenantId, originator, "SomeType")).thenReturn(Futures.immediateFuture(clearedAlarm)); doAnswer((Answer) invocationOnMock -> (Alarm) (invocationOnMock.getArguments())[0]).when(alarmService).createOrUpdateAlarm(any(Alarm.class)); @@ -207,8 +203,6 @@ public class TbAlarmNodeTest { .build(); assertEquals(expectedAlarm, actualAlarm); - - verify(executor, times(1)).executeAsync(any(Callable.class)); } @Test @@ -220,7 +214,7 @@ public class TbAlarmNodeTest { long oldEndDate = System.currentTimeMillis(); Alarm activeAlarm = Alarm.builder().type("SomeType").tenantId(tenantId).originator(originator).status(ACTIVE_UNACK).severity(WARNING).endTs(oldEndDate).build(); - when(detailsJs.executeJson(msg)).thenReturn(null); + when(detailsJs.executeJsonAsync(msg)).thenReturn(Futures.immediateFuture(null)); when(alarmService.findLatestByOriginatorAndType(tenantId, originator, "SomeType")).thenReturn(Futures.immediateFuture(activeAlarm)); doAnswer((Answer) invocationOnMock -> (Alarm) (invocationOnMock.getArguments())[0]).when(alarmService).createOrUpdateAlarm(activeAlarm); @@ -256,8 +250,6 @@ public class TbAlarmNodeTest { .build(); assertEquals(expectedAlarm, actualAlarm); - - verify(executor, times(1)).executeAsync(any(Callable.class)); } @Test @@ -269,7 +261,7 @@ public class TbAlarmNodeTest { long oldEndDate = System.currentTimeMillis(); Alarm activeAlarm = Alarm.builder().type("SomeType").tenantId(tenantId).originator(originator).status(ACTIVE_UNACK).severity(WARNING).endTs(oldEndDate).build(); -// when(detailsJs.executeJson(msg)).thenReturn(null); + when(detailsJs.executeJsonAsync(msg)).thenReturn(Futures.immediateFuture(null)); when(alarmService.findLatestByOriginatorAndType(tenantId, originator, "SomeType")).thenReturn(Futures.immediateFuture(activeAlarm)); when(alarmService.clearAlarm(eq(activeAlarm.getTenantId()), eq(activeAlarm.getId()), org.mockito.Mockito.any(JsonNode.class), anyLong())).thenReturn(Futures.immediateFuture(true)); when(alarmService.findAlarmByIdAsync(eq(activeAlarm.getTenantId()), eq(activeAlarm.getId()))).thenReturn(Futures.immediateFuture(activeAlarm)); @@ -320,12 +312,9 @@ public class TbAlarmNodeTest { when(ctx.createJsScriptEngine("DETAILS")).thenReturn(detailsJs); when(ctx.getTenantId()).thenReturn(tenantId); - when(ctx.getJsExecutor()).thenReturn(executor); when(ctx.getAlarmService()).thenReturn(alarmService); when(ctx.getDbCallbackExecutor()).thenReturn(dbExecutor); - mockJsExecutor(); - node = new TbCreateAlarmNode(); node.init(ctx, nodeConfiguration); } catch (TbNodeException ex) { @@ -344,12 +333,9 @@ public class TbAlarmNodeTest { when(ctx.createJsScriptEngine("DETAILS")).thenReturn(detailsJs); when(ctx.getTenantId()).thenReturn(tenantId); - when(ctx.getJsExecutor()).thenReturn(executor); when(ctx.getAlarmService()).thenReturn(alarmService); when(ctx.getDbCallbackExecutor()).thenReturn(dbExecutor); - mockJsExecutor(); - node = new TbClearAlarmNode(); node.init(ctx, nodeConfiguration); } catch (TbNodeException ex) { @@ -357,18 +343,6 @@ public class TbAlarmNodeTest { } } - private void mockJsExecutor() { - when(ctx.getJsExecutor()).thenReturn(executor); - doAnswer((Answer>) invocationOnMock -> { - try { - Callable task = (Callable) (invocationOnMock.getArguments())[0]; - return Futures.immediateFuture((Boolean) task.call()); - } catch (Throwable th) { - return Futures.immediateFailedFuture(th); - } - }).when(executor).executeAsync(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());