tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

added property to ignore delta in output messages if it is zero

pull/10310/head
thingsboard648 3 years ago
parent
6b8af63c1e
commit
0d0ca90cf9
3 changed files with 80 additions and 4 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 27
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java
  2. 3
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeConfiguration.java
  3. 54
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java

27
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNode.java
View File

@ -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))

3
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeConfiguration.java
View File

@ -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;
}
}

54
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/CalculateDeltaNodeTest.java
View File

@ -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;
}
}

Write Preview
Loading…
Cancel
Save