Merge branch 'develop/3.5' of https://github.com/thingsboard/thingsboard into allure

application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java

@@ -862,7 +862,10 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
     protected void awaitForDeviceActorToReceiveSubscription(DeviceId deviceId, FeatureType featureType, int subscriptionCount) {
         DeviceActorMessageProcessor processor = getDeviceActorProcessor(deviceId);
         Map<UUID, SessionInfo> subscriptions = (Map<UUID, SessionInfo>) ReflectionTestUtils.getField(processor, getMapName(featureType));
-        Awaitility.await("Device actor received subscription command from the transport").atMost(TIMEOUT, TimeUnit.SECONDS).until(() -> subscriptions.size() == subscriptionCount);
+        Awaitility.await("Device actor received subscription command from the transport").atMost(5, TimeUnit.SECONDS).until(() -> {
+            log.warn("device {}, subscriptions.size() == {}", deviceId, subscriptions.size());
+            return subscriptions.size() == subscriptionCount;
+        });
     }
 
     protected static String getMapName(FeatureType featureType) {

application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java

@@ -671,6 +671,7 @@ abstract public class BaseDeviceEdgeTest extends AbstractEdgeTest {
         client.setCallback(onUpdateCallback);
 
         client.subscribeAndWait("v1/devices/me/attributes", MqttQoS.AT_MOST_ONCE);
+        awaitForDeviceActorToReceiveSubscription(device.getId(), FeatureType.ATTRIBUTES, 1);
 
         edgeImitator.expectResponsesAmount(1);
 

application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java

@@ -354,6 +354,8 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
                 SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk());
         client.publishAndWait(attrPubTopic, CLIENT_ATTRIBUTES_PAYLOAD.getBytes());
         client.subscribeAndWait(attrSubTopic, MqttQoS.AT_MOST_ONCE);
+        //RequestAttributes does not make any subscriptions in device actor
+
         String update = getWsClient().waitForUpdate();
         assertThat(update).as("ws update received").isNotBlank();
         MqttTestCallback callback = new MqttTestCallback(attrSubTopic.replace("+", "1"));
@@ -383,6 +385,8 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
         doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk());
         client.publishAndWait(attrPubTopic, getAttributesProtoPayloadBytes());
         client.subscribeAndWait(attrSubTopic, MqttQoS.AT_MOST_ONCE);
+        //RequestAttributes does not make any subscriptions in device actor
+
         String update = getWsClient().waitForUpdate();
         assertThat(update).as("ws update received").isNotBlank();
         MqttTestCallback callback = new MqttTestCallback(attrSubTopic.replace("+", "1"));
@@ -442,6 +446,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
         assertThat(update).as("ws update received").isNotBlank();
 
         client.subscribeAndWait(GATEWAY_ATTRIBUTES_RESPONSE_TOPIC, MqttQoS.AT_LEAST_ONCE);
+        //RequestAttributes does not make any subscriptions in device actor
 
         MqttTestCallback clientAttributesCallback = new MqttTestCallback(GATEWAY_ATTRIBUTES_RESPONSE_TOPIC);
         client.setCallback(clientAttributesCallback);
@@ -495,6 +500,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
         assertThat(update).as("ws update received").isNotBlank();
 
         client.subscribeAndWait(GATEWAY_ATTRIBUTES_RESPONSE_TOPIC, MqttQoS.AT_LEAST_ONCE);
+        awaitForDeviceActorToReceiveSubscription(device.getId(), FeatureType.ATTRIBUTES, 1);
 
         MqttTestCallback clientAttributesCallback = new MqttTestCallback(GATEWAY_ATTRIBUTES_RESPONSE_TOPIC);
         client.setCallback(clientAttributesCallback);

application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/AbstractMqttServerSideRpcIntegrationTest.java

@@ -223,7 +223,6 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
 
         MqttTestCallback  callback = new MqttTestCallback(GATEWAY_RPC_TOPIC);
         client.setCallback(callback);
-        client.subscribeAndWait(GATEWAY_RPC_TOPIC, MqttQoS.AT_MOST_ONCE);
         subscribeAndCheckSubscription(client, GATEWAY_RPC_TOPIC, savedDevice.getId(), FeatureType.RPC);
 
         String setGpioRequest = "{\"method\": \"toggle_gpio\", \"params\": {\"pin\":1}}";

application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv5/attributes/AbstractAttributesMqttV5Test.java

@@ -21,6 +21,7 @@ import org.junit.Before;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.device.profile.MqttTopics;
 import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.msg.session.FeatureType;
 import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties;
 import org.thingsboard.server.transport.mqtt.mqttv5.AbstractMqttV5Test;
 import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestCallback;
@@ -104,6 +105,7 @@ public abstract class AbstractAttributesMqttV5Test extends AbstractMqttV5Test {
         MqttV5TestCallback onUpdateCallback = new MqttV5TestCallback();
         client.setCallback(onUpdateCallback);
         client.subscribeAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttQoS.AT_MOST_ONCE);
+        awaitForDeviceActorToReceiveSubscription(savedDevice.getId(), FeatureType.ATTRIBUTES, 1);
 
         doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk());
         onUpdateCallback.getSubscribeLatch().await(3, TimeUnit.SECONDS);

application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv5/client/connection/AbstractMqttV5ClientConnectionTest.java

@@ -25,6 +25,7 @@ import org.eclipse.paho.mqttv5.common.packet.MqttReturnCode;
 import org.eclipse.paho.mqttv5.common.packet.MqttWireMessage;
 import org.junit.Assert;
 import org.thingsboard.server.common.data.device.profile.MqttTopics;
+import org.thingsboard.server.common.msg.session.FeatureType;
 import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest;
 import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestCallback;
 import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient;
@@ -101,6 +102,7 @@ public abstract class AbstractMqttV5ClientConnectionTest extends AbstractMqttInt
         MqttV5TestCallback onUpdateCallback = new MqttV5TestCallback();
         client.setCallback(onUpdateCallback);
         client.subscribeAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttQoS.AT_MOST_ONCE);
+        awaitForDeviceActorToReceiveSubscription(savedDevice.getId(), FeatureType.ATTRIBUTES, 1);
 
         String payload = "{\"sharedStr\":\"" + StringUtils.repeat("*", valueLen) + "\"}";
 

application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv5/client/subscribe/AbstractMqttV5ClientSubscriptionTest.java

@@ -22,6 +22,7 @@ import org.eclipse.paho.mqttv5.common.packet.MqttSubAck;
 import org.eclipse.paho.mqttv5.common.packet.MqttWireMessage;
 import org.junit.Assert;
 import org.thingsboard.server.common.data.device.profile.MqttTopics;
+import org.thingsboard.server.common.msg.session.FeatureType;
 import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest;
 import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient;
 
@@ -34,6 +35,7 @@ public abstract class AbstractMqttV5ClientSubscriptionTest extends AbstractMqttI
         client.connectAndWait(accessToken);
 
         IMqttToken subscriptionResult = client.subscribeAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttQoS.AT_MOST_ONCE);
+        awaitForDeviceActorToReceiveSubscription(savedDevice.getId(), FeatureType.ATTRIBUTES, 1);
 
         MqttWireMessage response = subscriptionResult.getResponse();
 
@@ -52,6 +54,7 @@ public abstract class AbstractMqttV5ClientSubscriptionTest extends AbstractMqttI
         client.connectAndWait(accessToken);
 
         IMqttToken iMqttToken = client.subscribeAndWait("wrong/topic/+", MqttQoS.AT_MOST_ONCE);
+        awaitForDeviceActorToReceiveSubscription(savedDevice.getId(), FeatureType.ATTRIBUTES, 0);
         Assert.assertEquals(MESSAGE_TYPE_SUBACK,iMqttToken.getResponse().getType());
         MqttSubAck subAck = (MqttSubAck) iMqttToken.getResponse();
         Assert.assertEquals(1, subAck.getReturnCodes().length);

application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv5/client/unsubscribe/AbstractMqttV5ClientUnsubscribeTest.java

@@ -22,6 +22,7 @@ import org.eclipse.paho.mqttv5.common.packet.MqttUnsubAck;
 import org.eclipse.paho.mqttv5.common.packet.MqttWireMessage;
 import org.junit.Assert;
 import org.thingsboard.server.common.data.device.profile.MqttTopics;
+import org.thingsboard.server.common.msg.session.FeatureType;
 import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest;
 import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient;
 
@@ -34,6 +35,7 @@ public abstract class AbstractMqttV5ClientUnsubscribeTest extends AbstractMqttIn
         client.connectAndWait(accessToken);
 
         client.subscribeAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttQoS.AT_MOST_ONCE);
+        awaitForDeviceActorToReceiveSubscription(savedDevice.getId(), FeatureType.ATTRIBUTES, 1);
         IMqttToken unsubscribeResult = client.unsubscribeAndWait(MqttTopics.DEVICE_ATTRIBUTES_TOPIC);
         MqttWireMessage response = unsubscribeResult.getResponse();
         Assert.assertEquals(MESSAGE_TYPE_UNSUBACK, response.getType());

application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv5/rpc/AbstractMqttV5RpcTest.java

@@ -22,6 +22,7 @@ import org.eclipse.paho.mqttv5.common.MqttException;
 import org.eclipse.paho.mqttv5.common.MqttMessage;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.server.common.data.StringUtils;
+import org.thingsboard.server.common.msg.session.FeatureType;
 import org.thingsboard.server.transport.mqtt.mqttv5.AbstractMqttV5Test;
 import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestCallback;
 import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient;
@@ -45,6 +46,7 @@ public abstract class AbstractMqttV5RpcTest extends AbstractMqttV5Test {
         MqttV5TestCallback callback = new MqttV5TestCallback(DEVICE_RPC_REQUESTS_SUB_TOPIC.replace("+", "0"));
         client.setCallback(callback);
         client.subscribeAndWait(DEVICE_RPC_REQUESTS_SUB_TOPIC, MqttQoS.AT_MOST_ONCE);
+        awaitForDeviceActorToReceiveSubscription(savedDevice.getId(), FeatureType.RPC, 1);
 
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"23\",\"value\": 1}}";
         String result = doPostAsync("/api/rpc/oneway/" + savedDevice.getId(), setGpioRequest, String.class, status().isOk());
@@ -59,6 +61,7 @@ public abstract class AbstractMqttV5RpcTest extends AbstractMqttV5Test {
         MqttV5TestClient client = new MqttV5TestClient();
         client.connectAndWait(accessToken);
         client.subscribeAndWait(DEVICE_RPC_REQUESTS_SUB_TOPIC, MqttQoS.AT_LEAST_ONCE);
+        awaitForDeviceActorToReceiveSubscription(savedDevice.getId(), FeatureType.RPC, 1);
         MqttV5TestRpcCallback callback = new MqttV5TestRpcCallback(client, DEVICE_RPC_REQUESTS_SUB_TOPIC.replace("+", "0"));
         client.setCallback(callback);
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"26\",\"value\": 1}}";

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

@@ -32,6 +32,8 @@ import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.data.script.ScriptLanguage;
 import org.thingsboard.server.common.msg.TbMsg;
 
+import java.util.Objects;
+
 @Slf4j
 @RuleNode(
         type = ComponentType.ACTION,
@@ -54,8 +56,12 @@ public class TbLogNode implements TbNode {
     @Override
     public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
         this.config = TbNodeUtils.convert(configuration, TbLogNodeConfiguration.class);
-        this.standard = new TbLogNodeConfiguration().defaultConfiguration().getJsScript().equals(config.getJsScript());
+        this.standard = isStandard(config);
-        this.scriptEngine = this.standard ? null : ctx.createScriptEngine(config.getScriptLang(),
+        this.scriptEngine = this.standard ? null : createScriptEngine(ctx, config);
+    }
+
+    ScriptEngine createScriptEngine(TbContext ctx, TbLogNodeConfiguration config) {
+        return ctx.createScriptEngine(config.getScriptLang(),
                 ScriptLanguage.TBEL.equals(config.getScriptLang()) ? config.getTbelScript() : config.getJsScript());
     }
 
@@ -83,6 +89,18 @@ public class TbLogNode implements TbNode {
         }, MoreExecutors.directExecutor()); //usually js responses runs on js callback executor
     }
 
+    boolean isStandard(TbLogNodeConfiguration conf) {
+        Objects.requireNonNull(conf, "node config is null");
+        final TbLogNodeConfiguration defaultConfig = new TbLogNodeConfiguration().defaultConfiguration();
+        switch (conf.getScriptLang()) {
+            case JS: return defaultConfig.getJsScript().equals(conf.getJsScript());
+            case TBEL: return defaultConfig.getTbelScript().equals(conf.getTbelScript());
+            default:
+                log.warn("No rule to define isStandard script for script language [{}], assuming that is non-standard", conf.getScriptLang());
+                return false;
+        }
+    }
+
     void logStandard(TbContext ctx, TbMsg msg) {
         log.info(toLogMessage(msg));
         ctx.tellSuccess(msg);

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbLogNodeTest.java

@@ -17,14 +17,29 @@ package org.thingsboard.rule.engine.action;
 
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.rule.engine.api.TbNodeException;
 import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.script.ScriptLanguage;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
 
 import java.util.Collections;
 import java.util.Map;
+import java.util.UUID;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 
 @Slf4j
 public class TbLogNodeTest {
@@ -61,6 +76,7 @@ public class TbLogNodeTest {
                 "Incoming metadata:\n" +
                 "{}");
     }
+
     @Test
     void givenNullDataMsg_whenToLog_thenReturnString() {
         TbLogNode node = new TbLogNode();
@@ -77,4 +93,50 @@ public class TbLogNodeTest {
                 "{}");
     }
 
+    @ParameterizedTest
+    @EnumSource(ScriptLanguage.class)
+    void givenDefaultConfig_whenIsStandardForEachScriptLanguage_thenTrue(ScriptLanguage scriptLanguage) throws TbNodeException {
+
+        TbLogNodeConfiguration config = new TbLogNodeConfiguration().defaultConfiguration();
+        config.setScriptLang(scriptLanguage);
+        TbLogNode node = spy(new TbLogNode());
+        TbNodeConfiguration tbNodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
+        TbContext ctx = mock(TbContext.class);
+        node.init(ctx, tbNodeConfiguration);
+
+        assertThat(node.isStandard(config)).as("Script is standard for language " + scriptLanguage).isTrue();
+        verify(node, never()).createScriptEngine(any(), any());
+        verify(ctx, never()).createScriptEngine(any(), anyString());
+
+    }
+
+    @Test
+    void givenScriptEngineEnum_whenNewAdded_thenFailed() {
+        assertThat(ScriptLanguage.values().length).as("only two ScriptLanguage supported").isEqualTo(2);
+    }
+
+    @Test
+    void givenScriptEngineLangJs_whenCreateScriptEngine_thenSupplyJsScript(){
+        TbLogNodeConfiguration configJs = new TbLogNodeConfiguration().defaultConfiguration();
+        configJs.setScriptLang(ScriptLanguage.JS);
+        configJs.setJsScript(configJs.getJsScript() + " // This is JS script " + UUID.randomUUID());
+        TbLogNode node = new TbLogNode();
+        TbContext ctx = mock(TbContext.class);
+        node.createScriptEngine(ctx, configJs);
+        verify(ctx).createScriptEngine(ScriptLanguage.JS, configJs.getJsScript());
+        verifyNoMoreInteractions(ctx);
+    }
+
+    @Test
+    void givenScriptEngineLangTbel_whenCreateScriptEngine_thenSupplyTbelScript(){
+        TbLogNodeConfiguration configTbel = new TbLogNodeConfiguration().defaultConfiguration();
+        configTbel.setScriptLang(ScriptLanguage.TBEL);
+        configTbel.setTbelScript(configTbel.getTbelScript() + " // This is TBEL script " + UUID.randomUUID());
+        TbLogNode node = new TbLogNode();
+        TbContext ctx = mock(TbContext.class);
+        node.createScriptEngine(ctx, configTbel);
+        verify(ctx).createScriptEngine(ScriptLanguage.TBEL, configTbel.getTbelScript());
+        verifyNoMoreInteractions(ctx);
+    }
+
 }