refactoring & added tests for alarm status filter node & entity and msg type switch nodes & check field presence

application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java

@@ -852,17 +852,10 @@ class DefaultTbContext implements TbContext {
     }
 
     private static String getFailureMessage(Throwable th) {
-        String failureMessage;
+        if (th == null) {
-        if (th != null) {
+            return null;
-            if (!StringUtils.isEmpty(th.getMessage())) {
-                failureMessage = th.getMessage();
-            } else {
-                failureMessage = th.getClass().getSimpleName();
-            }
-        } else {
-            failureMessage = null;
         }
-        return failureMessage;
+        return StringUtils.isNotEmpty(th.getMessage()) ? th.getMessage() : th.getClass().getSimpleName();
     }
 
     private class SimpleTbQueueCallback implements TbQueueCallback {

application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java

@@ -114,13 +114,14 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.thingsboard.server.common.data.DataConstants.DEFAULT_DEVICE_TYPE;
+
 @Service
 @Profile("install")
 @Slf4j
 public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
 
     public static final String CUSTOMER_CRED = "customer";
-    public static final String DEFAULT_DEVICE_TYPE = "default";
     public static final String ACTIVITY_STATE = "active";
 
     @Autowired

common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java

@@ -65,6 +65,7 @@ public class DataConstants {
     public static final String PROVISION_KEY = "provisionDeviceKey";
     public static final String PROVISION_SECRET = "provisionDeviceSecret";
 
+    public static final String DEFAULT_DEVICE_TYPE = "default";
     public static final String DEVICE_NAME = "deviceName";
     public static final String DEVICE_TYPE = "deviceType";
     public static final String CERT_PUB_KEY = "x509CertPubKey";

common/data/src/main/java/org/thingsboard/server/common/data/StringUtils.java

@@ -42,7 +42,7 @@ public class StringUtils {
     }
 
     public static boolean isNotEmpty(String source) {
-        return source != null && !source.isEmpty();
+        return !isEmpty(source);
     }
 
     public static boolean isNotBlank(String source) {

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java

@@ -71,6 +71,7 @@ import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
 import static org.springframework.util.ConcurrentReferenceHashMap.ReferenceType;
+import static org.thingsboard.server.common.data.DataConstants.DEFAULT_DEVICE_TYPE;
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_CLOSED;
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_OPEN;
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SUBSCRIBE_TO_ATTRIBUTE_UPDATES_ASYNC_MSG;
@@ -85,7 +86,6 @@ import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMess
 @Slf4j
 public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDeviceSessionContext> {
 
-    protected static final String DEFAULT_DEVICE_TYPE = "default";
     private static final String CAN_T_PARSE_VALUE = "Can't parse value: ";
     private static final String DEVICE_PROPERTY = "device";
 

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/external/TbAbstractExternalNode.java

@@ -38,17 +38,9 @@ public abstract class TbAbstractExternalNode implements TbNode {
 
     protected void tellFailure(TbContext ctx, TbMsg tbMsg, Throwable t) {
         if (forceAck) {
-            if (t == null) {
+            ctx.enqueueForTellFailure(tbMsg.copyWithNewCtx(), t);
-                ctx.enqueueForTellNext(tbMsg.copyWithNewCtx(), TbNodeConnectionType.FAILURE);
-            } else {
-                ctx.enqueueForTellFailure(tbMsg.copyWithNewCtx(), t);
-            }
         } else {
-            if (t == null) {
+            ctx.tellFailure(tbMsg, t);
-                ctx.tellNext(tbMsg, TbNodeConnectionType.FAILURE);
-            } else {
-                ctx.tellFailure(tbMsg, t);
-            }
         }
     }
 

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbAssetTypeSwitchNodeTest.java

@@ -124,4 +124,5 @@ class TbAssetTypeSwitchNodeTest {
     private TbMsg getTbMsg(EntityId entityId) {
         return TbMsg.newMsg(POST_ATTRIBUTES_REQUEST.name(), entityId, new TbMsgMetaData(), "{}", callback);
     }
+
 }

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbCheckAlarmStatusNodeTest.java

@@ -163,5 +163,4 @@ class TbCheckAlarmStatusNodeTest {
         return TbMsg.newMsg(POST_ATTRIBUTES_REQUEST.name(), entityId, new TbMsgMetaData(), msgData);
     }
 
-
+}
-}

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbCheckMessageNodeTest.java

@@ -0,0 +1,206 @@
+/**
+ * Copyright © 2016-2023 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.filter;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+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.DeviceId;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbMsgMetaData;
+
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.thingsboard.server.common.data.DataConstants.DEFAULT_DEVICE_TYPE;
+import static org.thingsboard.server.common.data.DataConstants.DEVICE_NAME;
+import static org.thingsboard.server.common.data.DataConstants.DEVICE_TYPE;
+import static org.thingsboard.server.common.data.msg.TbMsgType.POST_ATTRIBUTES_REQUEST;
+
+class TbCheckMessageNodeTest {
+
+    private static final DeviceId DEVICE_ID = new DeviceId(UUID.randomUUID());
+    private static final TbMsgMetaData EMPTY_METADATA = new TbMsgMetaData();
+    private static final String EMPTY_DATA = "{}";
+    private static final TbMsg EMPTY_POST_ATTRIBUTES_MSG = TbMsg.newMsg(POST_ATTRIBUTES_REQUEST.name(), DEVICE_ID, EMPTY_METADATA, EMPTY_DATA);
+
+    private static TbCheckMessageNode node;
+
+    private static TbContext ctx;
+
+    @BeforeEach
+    void setUp() {
+        ctx = mock(TbContext.class);
+        node = new TbCheckMessageNode();
+    }
+
+    @AfterEach
+    void tearDown() {
+        node.destroy();
+    }
+
+    @Test
+    void givenDefaultConfig_whenOnMsg_then_True() throws TbNodeException {
+        // GIVEN
+        var configuration = new TbCheckMessageNodeConfiguration().defaultConfiguration();
+        node.init(ctx, new TbNodeConfiguration(JacksonUtil.valueToTree(configuration)));
+
+        // WHEN
+        node.onMsg(ctx, EMPTY_POST_ATTRIBUTES_MSG);
+
+        // THEN
+        ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        verify(ctx, times(1)).tellNext(newMsgCaptor.capture(), eq(TbNodeConnectionType.TRUE));
+        verify(ctx, never()).tellFailure(any(), any());
+        TbMsg newMsg = newMsgCaptor.getValue();
+        assertThat(newMsg).isNotNull();
+        assertThat(newMsg).isSameAs(EMPTY_POST_ATTRIBUTES_MSG);
+    }
+
+    @Test
+    void givenCustomConfigWithoutCheckAllKeysAndWithEmptyLists_whenOnMsg_then_False() throws TbNodeException {
+        // GIVEN
+        var configuration = new TbCheckMessageNodeConfiguration().defaultConfiguration();
+        configuration.setCheckAllKeys(false);
+        node.init(ctx, new TbNodeConfiguration(JacksonUtil.valueToTree(configuration)));
+
+        // WHEN
+        node.onMsg(ctx, EMPTY_POST_ATTRIBUTES_MSG);
+
+        // THEN
+        ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        verify(ctx, times(1)).tellNext(newMsgCaptor.capture(), eq(TbNodeConnectionType.FALSE));
+        verify(ctx, never()).tellFailure(any(), any());
+        TbMsg newMsg = newMsgCaptor.getValue();
+        assertThat(newMsg).isNotNull();
+        assertThat(newMsg).isSameAs(EMPTY_POST_ATTRIBUTES_MSG);
+    }
+
+    @Test
+    void givenCustomConfigWithCheckAllKeys_whenOnMsg_then_True() throws TbNodeException {
+        // GIVEN
+        var configuration = new TbCheckMessageNodeConfiguration().defaultConfiguration();
+        configuration.setMessageNames(List.of("temperature-0"));
+        configuration.setMetadataNames(List.of("deviceName", "deviceType", "ts"));
+        node.init(ctx, new TbNodeConfiguration(JacksonUtil.valueToTree(configuration)));
+
+        TbMsg tbMsg = getTbMsg();
+
+        // WHEN
+        node.onMsg(ctx, tbMsg);
+
+        // THEN
+        ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        verify(ctx, times(1)).tellNext(newMsgCaptor.capture(), eq(TbNodeConnectionType.TRUE));
+        verify(ctx, never()).tellFailure(any(), any());
+        TbMsg newMsg = newMsgCaptor.getValue();
+        assertThat(newMsg).isNotNull();
+        assertThat(newMsg).isSameAs(tbMsg);
+    }
+
+    @Test
+    void givenCustomConfigWithCheckAllKeys_whenOnMsg_then_False() throws TbNodeException {
+        // GIVEN
+        var configuration = new TbCheckMessageNodeConfiguration().defaultConfiguration();
+        configuration.setMessageNames(List.of("temperature-0", "temperature-1"));
+        configuration.setMetadataNames(List.of("deviceName", "deviceType", "ts"));
+        node.init(ctx, new TbNodeConfiguration(JacksonUtil.valueToTree(configuration)));
+
+        TbMsg tbMsg = getTbMsg();
+
+        // WHEN
+        node.onMsg(ctx, tbMsg);
+
+        // THEN
+        ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        verify(ctx, times(1)).tellNext(newMsgCaptor.capture(), eq(TbNodeConnectionType.FALSE));
+        verify(ctx, never()).tellFailure(any(), any());
+        TbMsg newMsg = newMsgCaptor.getValue();
+        assertThat(newMsg).isNotNull();
+        assertThat(newMsg).isSameAs(tbMsg);
+    }
+
+    @Test
+    void givenCustomConfigWithoutCheckAllKeys_whenOnMsg_then_True() throws TbNodeException {
+        // GIVEN
+        var configuration = new TbCheckMessageNodeConfiguration().defaultConfiguration();
+        configuration.setMessageNames(List.of("temperature-0", "temperature-1"));
+        configuration.setCheckAllKeys(false);
+        node.init(ctx, new TbNodeConfiguration(JacksonUtil.valueToTree(configuration)));
+
+        TbMsg tbMsg = getTbMsg();
+
+        // WHEN
+        node.onMsg(ctx, tbMsg);
+
+        // THEN
+        ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        verify(ctx, times(1)).tellNext(newMsgCaptor.capture(), eq(TbNodeConnectionType.TRUE));
+        verify(ctx, never()).tellFailure(any(), any());
+        TbMsg newMsg = newMsgCaptor.getValue();
+        assertThat(newMsg).isNotNull();
+        assertThat(newMsg).isSameAs(tbMsg);
+    }
+
+    @Test
+    void givenCustomConfigWithoutCheckAllKeysAndEmptyMsg_whenOnMsg_then_False() throws TbNodeException {
+        // GIVEN
+        var configuration = new TbCheckMessageNodeConfiguration().defaultConfiguration();
+        configuration.setMessageNames(List.of("temperature-0", "temperature-1"));
+        configuration.setCheckAllKeys(false);
+        node.init(ctx, new TbNodeConfiguration(JacksonUtil.valueToTree(configuration)));
+
+        TbMsg tbMsg = getTbMsg(true);
+
+        // WHEN
+        node.onMsg(ctx, tbMsg);
+
+        // THEN
+        ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        verify(ctx, times(1)).tellNext(newMsgCaptor.capture(), eq(TbNodeConnectionType.FALSE));
+        verify(ctx, never()).tellFailure(any(), any());
+        TbMsg newMsg = newMsgCaptor.getValue();
+        assertThat(newMsg).isNotNull();
+        assertThat(newMsg).isSameAs(tbMsg);
+    }
+
+    private TbMsg getTbMsg() {
+        return getTbMsg(false);
+    }
+
+    private TbMsg getTbMsg(boolean emptyData) {
+        String data = emptyData ? EMPTY_DATA : "{\"temperature-0\": 25}";
+        var metadata = new TbMsgMetaData();
+        metadata.putValue(DEVICE_NAME, "Test Device");
+        metadata.putValue(DEVICE_TYPE, DEFAULT_DEVICE_TYPE);
+        metadata.putValue("ts", String.valueOf(System.currentTimeMillis()));
+        return TbMsg.newMsg(POST_ATTRIBUTES_REQUEST.name(), DEVICE_ID, metadata, data);
+    }
+
+}

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbMsgTypeSwitchNodeTest.java

@@ -0,0 +1,97 @@
+/**
+ * Copyright © 2016-2023 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.filter;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.msg.TbMsgType;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbMsgMetaData;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+class TbMsgTypeSwitchNodeTest {
+
+    private static final DeviceId DEVICE_ID = new DeviceId(UUID.randomUUID());
+    private static final TbMsgMetaData EMPTY_METADATA = new TbMsgMetaData();
+    private static final String EMPTY_DATA = "{}";
+
+    private static TbMsgTypeSwitchNode node;
+
+    private static TbContext ctx;
+
+    @BeforeEach
+    void setUp() {
+        ctx = mock(TbContext.class);
+        node = new TbMsgTypeSwitchNode();
+    }
+
+    @AfterEach
+    void tearDown() {
+        node.destroy();
+    }
+
+    @Test
+    void givenAllTypes_whenOnMsg_then_allTypesSupported() throws TbNodeException {
+        // GIVEN
+        List<TbMsg> tbMsgList = new ArrayList<>();
+        var tbMsgTypes = TbMsgType.values();
+        for (var msgType : tbMsgTypes) {
+            tbMsgList.add(getTbMsg(msgType));
+        }
+
+        // WHEN
+        for (TbMsg tbMsg : tbMsgList) {
+            node.onMsg(ctx, tbMsg);
+        }
+
+        // THEN
+        ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        ArgumentCaptor<String> nodeConnectionCapture = ArgumentCaptor.forClass(String.class);
+        verify(ctx, times(tbMsgList.size())).tellNext(newMsgCaptor.capture(), nodeConnectionCapture.capture());
+        verify(ctx, never()).tellFailure(any(), any());
+        var resultMsgs = newMsgCaptor.getAllValues();
+        var resultNodeConnections = nodeConnectionCapture.getAllValues();
+        for (int i = 0; i < resultMsgs.size(); i++) {
+            var msg = resultMsgs.get(i);
+            assertThat(msg).isNotNull();
+            assertThat(msg.getType()).isNotNull();
+            assertThat(msg).isSameAs(tbMsgList.get(i));
+            // todo add additional validation that types like ALARM or PROVISION returns OTHER for backward-compatibility.
+            assertThat(resultNodeConnections.get(i))
+                    .isEqualTo(TbMsgType.getRuleNodeConnection(msg.getType()));
+        }
+    }
+
+    private TbMsg getTbMsg(TbMsgType msgType) {
+        return TbMsg.newMsg(msgType.name(), DEVICE_ID, EMPTY_METADATA, EMPTY_DATA);
+    }
+
+}

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbOriginatorTypeSwitchNodeTest.java

@@ -0,0 +1,98 @@
+/**
+ * Copyright © 2016-2023 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.filter;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.EntityIdFactory;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbMsgMetaData;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.thingsboard.server.common.data.msg.TbMsgType.POST_ATTRIBUTES_REQUEST;
+
+class TbOriginatorTypeSwitchNodeTest {
+
+    private static final UUID RANDOM_UUID = UUID.randomUUID();
+    private static final TbMsgMetaData EMPTY_METADATA = new TbMsgMetaData();
+    private static final String EMPTY_DATA = "{}";
+
+    private static TbOriginatorTypeSwitchNode node;
+
+    private static TbContext ctx;
+
+    @BeforeEach
+    void setUp() {
+        ctx = mock(TbContext.class);
+        node = new TbOriginatorTypeSwitchNode();
+    }
+
+    @AfterEach
+    void tearDown() {
+        node.destroy();
+    }
+
+    @Test
+    void givenAllTypes_whenOnMsg_then_allTypesSupported() throws TbNodeException {
+        // GIVEN
+        List<TbMsg> tbMsgList = new ArrayList<>();
+        var entityTypes = EntityType.values();
+        for (var entityType : entityTypes) {
+            var entityId = EntityIdFactory.getByTypeAndUuid(entityType, RANDOM_UUID);
+            tbMsgList.add(getTbMsg(entityId));
+        }
+
+        // WHEN
+        for (TbMsg tbMsg : tbMsgList) {
+            node.onMsg(ctx, tbMsg);
+        }
+
+        // THEN
+        ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
+        ArgumentCaptor<String> nodeConnectionCapture = ArgumentCaptor.forClass(String.class);
+        verify(ctx, times(tbMsgList.size())).tellNext(newMsgCaptor.capture(), nodeConnectionCapture.capture());
+        verify(ctx, never()).tellFailure(any(), any());
+        var resultMsgs = newMsgCaptor.getAllValues();
+        var resultNodeConnections = nodeConnectionCapture.getAllValues();
+        for (int i = 0; i < resultMsgs.size(); i++) {
+            var msg = resultMsgs.get(i);
+            assertThat(msg).isNotNull();
+            assertThat(msg).isSameAs(tbMsgList.get(i));
+            assertThat(resultNodeConnections.get(i))
+                    .isEqualTo(msg.getOriginator().getEntityType().getNormalName());
+        }
+    }
+
+    private TbMsg getTbMsg(EntityId entityId) {
+        return TbMsg.newMsg(POST_ATTRIBUTES_REQUEST.name(), entityId, EMPTY_METADATA, EMPTY_DATA);
+    }
+
+}