From 0d0ca90cf96cd63a2da79ee28e93110434a18734 Mon Sep 17 00:00:00 2001
From: thingsboard648 <irina.m2004@icloud.com>
Date: Tue, 5 Mar 2024 10:38:19 +0200
Subject: [PATCH] added property to ignore delta in output messages if it is
 zero

---
 .../engine/metadata/CalculateDeltaNode.java   | 27 +++++++++-
 .../CalculateDeltaNodeConfiguration.java      |  3 +-
 .../metadata/CalculateDeltaNodeTest.java      | 54 ++++++++++++++++++-
 3 files changed, 80 insertions(+), 4 deletions(-)

diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java
index 609888db84..510d089372 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java
@@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.data.msg.TbMsgType;
 import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
 import org.thingsboard.server.common.data.plugin.ComponentType;
+import org.thingsboard.server.common.data.util.TbPair;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 
@@ -46,7 +47,9 @@ import static org.thingsboard.common.util.DonAsynchron.withCallback;
 
 @Slf4j
 @RuleNode(type = ComponentType.ENRICHMENT,
-        name = "calculate delta", relationTypes = {TbNodeConnectionType.SUCCESS, TbNodeConnectionType.FAILURE, TbNodeConnectionType.OTHER},
+        name = "calculate delta",
+        version = 1,
+        relationTypes = {TbNodeConnectionType.SUCCESS, TbNodeConnectionType.FAILURE, TbNodeConnectionType.OTHER},
         configClazz = CalculateDeltaNodeConfiguration.class,
         nodeDescription = "Calculates delta and amount of time passed between previous timeseries key reading " +
                 "and current value for this key from the incoming message",
@@ -96,6 +99,11 @@ public class CalculateDeltaNode implements TbNode {
 
                     BigDecimal delta = BigDecimal.valueOf(previousData != null ? currentValue - previousData.value : 0.0);
 
+                    if (config.isOnlyComputeTrueDeltas() && delta.doubleValue() == 0) {
+                        ctx.tellSuccess(msg);
+                        return;
+                    }
+
                     if (config.isTellFailureIfDeltaIsNegative() && delta.doubleValue() < 0) {
                         ctx.tellFailure(msg, new IllegalArgumentException("Delta value is negative!"));
                         return;
@@ -128,6 +136,23 @@ public class CalculateDeltaNode implements TbNode {
         }
     }
 
+    @Override
+    public TbPair<Boolean, JsonNode> upgrade(int fromVersion, JsonNode oldConfiguration) throws TbNodeException {
+        boolean hasChanges = false;
+        switch (fromVersion) {
+            case 0:
+                String onlyComputeTrueDeltas = "onlyComputeTrueDeltas";
+                if (!oldConfiguration.has(onlyComputeTrueDeltas)) {
+                    hasChanges = true;
+                    ((ObjectNode) oldConfiguration).put(onlyComputeTrueDeltas, false);
+                }
+                break;
+            default:
+                break;
+        }
+        return new TbPair<>(hasChanges, oldConfiguration);
+    }
+
     private ListenableFuture<ValueWithTs> fetchLatestValueAsync(EntityId entityId) {
         return Futures.transform(timeseriesService.findLatest(ctx.getTenantId(), entityId, Collections.singletonList(config.getInputValueKey())),
                 list -> extractValue(list.get(0))
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeConfiguration.java
index 0c4e6de556..9f30aae22b 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeConfiguration.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeConfiguration.java
@@ -30,6 +30,7 @@ public class CalculateDeltaNodeConfiguration implements NodeConfiguration<Calcul
     private String periodValueKey;
     private Integer round;
     private boolean tellFailureIfDeltaIsNegative;
+    private boolean onlyComputeTrueDeltas;
 
     @Override
     public CalculateDeltaNodeConfiguration defaultConfiguration() {
@@ -40,7 +41,7 @@ public class CalculateDeltaNodeConfiguration implements NodeConfiguration<Calcul
         configuration.setAddPeriodBetweenMsgs(false);
         configuration.setPeriodValueKey("periodInMs");
         configuration.setTellFailureIfDeltaIsNegative(true);
+        configuration.setOnlyComputeTrueDeltas(false);
         return configuration;
     }
-
 }
diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java
index 707de1741f..e65607dbcc 100644
--- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java
+++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java
@@ -21,14 +21,18 @@ import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.provider.Arguments;
 import org.mockito.ArgumentCaptor;
 import org.mockito.ArgumentMatcher;
 import org.mockito.Mock;
+import org.mockito.Spy;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.ListeningExecutor;
+import org.thingsboard.rule.engine.AbstractRuleNodeUpgradeTest;
 import org.thingsboard.rule.engine.TestDbCallbackExecutor;
 import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNode;
 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
 import org.thingsboard.rule.engine.api.TbNodeException;
 import org.thingsboard.server.common.data.id.DeviceId;
@@ -48,6 +52,7 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService;
 
 import java.util.List;
 import java.util.UUID;
+import java.util.stream.Stream;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -66,7 +71,7 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
-public class CalculateDeltaNodeTest {
+public class CalculateDeltaNodeTest extends AbstractRuleNodeUpgradeTest {
 
     private static final DeviceId DUMMY_DEVICE_ORIGINATOR = new DeviceId(UUID.randomUUID());
     private static final TenantId TENANT_ID = new TenantId(UUID.randomUUID());
@@ -75,13 +80,13 @@ public class CalculateDeltaNodeTest {
     private TbContext ctxMock;
     @Mock
     private TimeseriesService timeseriesServiceMock;
+    @Spy
     private CalculateDeltaNode node;
     private CalculateDeltaNodeConfiguration config;
     private TbNodeConfiguration nodeConfiguration;
 
     @BeforeEach
     public void setUp() throws TbNodeException {
-        node = new CalculateDeltaNode();
         config = new CalculateDeltaNodeConfiguration().defaultConfiguration();
         nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
         when(ctxMock.getTimeseriesService()).thenReturn(timeseriesServiceMock);
@@ -424,6 +429,31 @@ public class CalculateDeltaNodeTest {
                 .hasMessage("Calculation failed. JSON values are not supported!");
     }
 
+    @Test
+    public void givenDeltaValueIsZeroAndOnlyComputeTrueDeltasTrue_whenOnMsg_thenShouldReturnMsgWithoutDelta() throws TbNodeException {
+        // GIVEN
+        config.setOnlyComputeTrueDeltas(true);
+        config.setInputValueKey("temperature");
+        nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
+        node.init(ctxMock, nodeConfiguration);
+
+        mockFindLatest(new BasicTsKvEntry(1L, new DoubleDataEntry("temperature", 40.0)));
+
+        var msgData = "{\"temperature\":40,\"airPressure\":123}";
+        var msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DUMMY_DEVICE_ORIGINATOR, TbMsgMetaData.EMPTY, msgData);
+
+        // WHEN
+
+        node.onMsg(ctxMock, msg);
+
+        // THEN
+
+        verify(ctxMock).tellSuccess(eq(msg));
+        verify(ctxMock, never()).tellNext(any(), anyString());
+        verify(ctxMock, never()).tellNext(any(), anySet());
+        verify(ctxMock, never()).tellFailure(any(), any());
+    }
+
     private void mockFindLatest(TsKvEntry tsKvEntry) {
         when(ctxMock.getTenantId()).thenReturn(TENANT_ID);
         when(timeseriesServiceMock.findLatestSync(
@@ -457,4 +487,24 @@ public class CalculateDeltaNodeTest {
 
     }
 
+    private static Stream<Arguments> givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() {
+        return Stream.of(
+                // default config for version 0
+                Arguments.of(0,
+                        "{\"inputValueKey\":\"pulseCounter\",\"outputValueKey\":\"delta\",\"useCache\":true,\"addPeriodBetweenMsgs\":false, \"periodValueKey\":\"periodInMs\", \"round\":null,\"tellFailureIfDeltaIsNegative\":true}",
+                        true,
+                        "{\"inputValueKey\":\"pulseCounter\",\"outputValueKey\":\"delta\",\"useCache\":true,\"addPeriodBetweenMsgs\":false, \"periodValueKey\":\"periodInMs\", \"round\":null,\"tellFailureIfDeltaIsNegative\":true, \"onlyComputeTrueDeltas\":false}"),
+                // default config for version 1 with upgrade from version 0
+                Arguments.of(1,
+                        "{\"inputValueKey\":\"pulseCounter\",\"outputValueKey\":\"delta\",\"useCache\":true,\"addPeriodBetweenMsgs\":false, \"periodValueKey\":\"periodInMs\", \"round\":null,\"tellFailureIfDeltaIsNegative\":true, \"onlyComputeTrueDeltas\":false}",
+                        false,
+                        "{\"inputValueKey\":\"pulseCounter\",\"outputValueKey\":\"delta\",\"useCache\":true,\"addPeriodBetweenMsgs\":false, \"periodValueKey\":\"periodInMs\", \"round\":null,\"tellFailureIfDeltaIsNegative\":true, \"onlyComputeTrueDeltas\":false}")
+        );
+
+    }
+
+    @Override
+    protected TbNode getTestNode() {
+        return node;
+    }
 }