Merge pull request #7677 from thingsboard/feature/attribute_nodes_add_options_send_notification

Attribute nodes add options send notification

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesUpdateNodeCallback.java

@@ -24,8 +24,8 @@ import java.util.List;
 
 public class AttributesUpdateNodeCallback extends TelemetryNodeCallback {
 
-    private String scope;
-    private List<AttributeKvEntry> attributes;
+    private final String scope;
+    private final List<AttributeKvEntry> attributes;
 
     public AttributesUpdateNodeCallback(TbContext ctx, TbMsg msg, String scope, List<AttributeKvEntry> attributes) {
         super(ctx, msg);

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java

@@ -17,6 +17,7 @@ package org.thingsboard.rule.engine.telemetry;
 
 import com.google.gson.JsonParser;
 import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.rule.engine.api.RuleNode;
 import org.thingsboard.rule.engine.api.TbContext;
@@ -40,8 +41,9 @@ import java.util.List;
         configClazz = TbMsgAttributesNodeConfiguration.class,
         nodeDescription = "Saves attributes data",
         nodeDetails = "Saves entity attributes based on configurable scope parameter. Expects messages with 'POST_ATTRIBUTES_REQUEST' message type. " +
-                      "If upsert(update/insert) operation is completed successfully, rule node will send the \"Attributes Updated\" " +
-                      "event to the root chain of the message originator and send the incoming message via <b>Success</b> chain, otherwise, <b>Failure</b> chain is used.",
+                      "If upsert(update/insert) operation is completed successfully rule node will send the incoming message via <b>Success</b> chain, otherwise, <b>Failure</b> chain is used. " +
+                      "Additionally if checkbox <b>Send attributes updated notification</b> is set to true, rule node will put the \"Attributes Updated\" " +
+                      "event for <b>SHARED_SCOPE</b> and <b>SERVER_SCOPE</b> attributes updates to the corresponding rule engine queue.",
         uiResources = {"static/rulenode/rulenode-core-config.js"},
         configDirective = "tbActionNodeAttributesConfig",
         icon = "file_upload"
@@ -49,6 +51,8 @@ import java.util.List;
 public class TbMsgAttributesNode implements TbNode {
 
     private TbMsgAttributesNodeConfiguration config;
+    private String scope;
+    private boolean sendAttributesUpdateNotification;
 
     @Override
     public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
@@ -56,6 +60,10 @@ public class TbMsgAttributesNode implements TbNode {
         if (config.getNotifyDevice() == null) {
             config.setNotifyDevice(true);
         }
+        this.scope = config.getScope();
+        this.sendAttributesUpdateNotification = config.isSendAttributesUpdatedNotification()
+                && !DataConstants.CLIENT_SCOPE.equals(scope);
+
     }
 
     @Override
@@ -65,7 +73,7 @@ public class TbMsgAttributesNode implements TbNode {
             return;
         }
         String src = msg.getData();
-        List<AttributeKvEntry> attributes = new ArrayList<>(JsonConverter.convertToAttributes(new JsonParser().parse(src)));
+        List<AttributeKvEntry> attributes = new ArrayList<>(JsonConverter.convertToAttributes(JsonParser.parseString(src)));
         if (attributes.isEmpty()) {
             ctx.tellSuccess(msg);
             return;
@@ -74,10 +82,12 @@ public class TbMsgAttributesNode implements TbNode {
         ctx.getTelemetryService().saveAndNotify(
                 ctx.getTenantId(),
                 msg.getOriginator(),
-                config.getScope(),
+                scope,
                 attributes,
                 config.getNotifyDevice() || StringUtils.isEmpty(notifyDeviceStr) || Boolean.parseBoolean(notifyDeviceStr),
-                new AttributesUpdateNodeCallback(ctx, msg, config.getScope(), attributes)
+                sendAttributesUpdateNotification ?
+                        new AttributesUpdateNodeCallback(ctx, msg, config.getScope(), attributes) :
+                        new TelemetryNodeCallback(ctx, msg)
         );
     }
 

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfiguration.java

@@ -25,12 +25,14 @@ public class TbMsgAttributesNodeConfiguration implements NodeConfiguration<TbMsg
     private String scope;
 
     private Boolean notifyDevice;
+    private boolean sendAttributesUpdatedNotification;
 
     @Override
     public TbMsgAttributesNodeConfiguration defaultConfiguration() {
         TbMsgAttributesNodeConfiguration configuration = new TbMsgAttributesNodeConfiguration();
         configuration.setScope(DataConstants.SERVER_SCOPE);
         configuration.setNotifyDevice(false);
+        configuration.setSendAttributesUpdatedNotification(false);
         return configuration;
     }
 }

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java

@@ -67,7 +67,15 @@ public class TbMsgDeleteAttributes implements TbNode {
         if (keysToDelete.isEmpty()) {
             ctx.tellSuccess(msg);
         } else {
-            ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), scope, keysToDelete, new AttributesDeleteNodeCallback(ctx, msg, scope, keysToDelete));
+            ctx.getTelemetryService().deleteAndNotify(
+                    ctx.getTenantId(),
+                    msg.getOriginator(),
+                    scope,
+                    keysToDelete,
+                    config.isSendAttributesDeletedNotification() ?
+                            new AttributesDeleteNodeCallback(ctx, msg, scope, keysToDelete) :
+                            new TelemetryNodeCallback(ctx, msg)
+            );
         }
     }
 }

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java

@@ -27,12 +27,14 @@ public class TbMsgDeleteAttributesConfiguration implements NodeConfiguration<TbM
 
     private String scope;
     private List<String> keys;
+    private boolean sendAttributesDeletedNotification;
 
     @Override
     public TbMsgDeleteAttributesConfiguration defaultConfiguration() {
         TbMsgDeleteAttributesConfiguration configuration = new TbMsgDeleteAttributesConfiguration();
         configuration.setScope(DataConstants.SERVER_SCOPE);
         configuration.setKeys(Collections.emptyList());
+        configuration.setSendAttributesDeletedNotification(false);
         return configuration;
     }
 }

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java

@@ -99,7 +99,19 @@ public class TbMsgDeleteAttributesTest {
     }
 
     @Test
-    void givenMsg_whenOnMsg_thenVerifyOutput() throws Exception {
+    void givenMsg_whenOnMsg_thenVerifyOutput_NoSendAttributesDeletedNotification() throws Exception {
+        onMsg_thenVerifyOutput(false);
+    }
+
+    @Test
+    void givenMsg_whenOnMsg_thenVerifyOutput_SendAttributesDeletedNotification() throws Exception {
+        config.setSendAttributesDeletedNotification(true);
+        nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config));
+        node.init(ctx, nodeConfiguration);
+        onMsg_thenVerifyOutput(true);
+    }
+
+    void onMsg_thenVerifyOutput(boolean sendAttributesDeletedNotification) throws Exception {
         final Map<String, String> mdMap = Map.of(
                 "TestAttribute_1", "temperature",
                 "city", "NY"
@@ -114,11 +126,12 @@ public class TbMsgDeleteAttributesTest {
         ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class);
         ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
 
-        verify(ctx, times(1)).enqueue(any(), successCaptor.capture(), failureCaptor.capture());
-        successCaptor.getValue().run();
+        if (sendAttributesDeletedNotification) {
+            verify(ctx, times(1)).enqueue(any(), successCaptor.capture(), failureCaptor.capture());
+            successCaptor.getValue().run();
+            verify(ctx, times(1)).attributesDeletedActionMsg(any(), any(), anyString(), anyList());
+        }
         verify(ctx, times(1)).tellSuccess(newMsgCaptor.capture());
-
-        verify(ctx, times(1)).attributesDeletedActionMsg(any(), any(), anyString(), anyList());
         verify(ctx, never()).tellFailure(any(), any());
         verify(telemetryService, times(1)).deleteAndNotify(any(), any(), anyString(), anyList(), any());
     }