diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java index a3e0f6becc..267d3fae73 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java @@ -30,6 +30,7 @@ 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.common.msg.TbMsgMetaData; +import org.thingsboard.server.dao.exception.DataValidationException; import java.util.EnumSet; import java.util.List; @@ -38,6 +39,9 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static org.thingsboard.server.dao.service.ConstraintValidator.validateFields; @Slf4j @RuleNode( @@ -57,7 +61,7 @@ import java.util.concurrent.TimeUnit; public class TbMsgDelayNode implements TbNode { private static final Set supportedTimeUnits = EnumSet.of(TimeUnit.SECONDS, TimeUnit.MINUTES, TimeUnit.HOURS); - private static final String supportedTimeUnitsStr = String.join(",", TimeUnit.SECONDS.name(), TimeUnit.MINUTES.name(), TimeUnit.HOURS.name()); + private static final String supportedTimeUnitsStr = supportedTimeUnits.stream().map(TimeUnit::name).collect(Collectors.joining(", ")); private TbMsgDelayNodeConfiguration config; private ConcurrentMap pendingMsgs; @@ -65,7 +69,12 @@ public class TbMsgDelayNode implements TbNode { @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { this.config = TbNodeUtils.convert(configuration, TbMsgDelayNodeConfiguration.class); - validateConfig(); + String errorPrefix = "'" + ctx.getSelf().getName() + "' node configuration is invalid: "; + try { + validateFields(config, errorPrefix); + } catch (DataValidationException e) { + throw new TbNodeException(e, true); + } this.pendingMsgs = new ConcurrentHashMap<>(); } @@ -116,18 +125,6 @@ public class TbMsgDelayNode implements TbNode { } } - private void validateConfig() throws TbNodeException { - if (config.getMaxPendingMsgs() < 1 || config.getMaxPendingMsgs() > 100000) { - throw new TbNodeException("Maximum pending messages should be in a range from 1 to 100000.", true); - } - if (config.getPeriod() == null) { - throw new TbNodeException("Period should be specified.", true); - } - if (config.getTimeUnit() == null) { - throw new TbNodeException("Time unit should be specified.", true); - } - } - @Override public void destroy() { pendingMsgs.clear(); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeConfiguration.java index b1ec01bb10..56df2c8aea 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeConfiguration.java @@ -15,6 +15,9 @@ */ package org.thingsboard.rule.engine.delay; +import jakarta.validation.constraints.Max; +import jakarta.validation.constraints.Min; +import jakarta.validation.constraints.NotNull; import lombok.Data; import org.thingsboard.rule.engine.api.NodeConfiguration; @@ -23,8 +26,12 @@ import java.util.concurrent.TimeUnit; @Data public class TbMsgDelayNodeConfiguration implements NodeConfiguration { + @NotNull private String period; + @NotNull private String timeUnit; + @Min(1) + @Max(100000) private int maxPendingMsgs; @Override diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeTest.java index 70511b03b9..23fb69d8b7 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/delay/TbMsgDelayNodeTest.java @@ -37,6 +37,7 @@ import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.msg.TbNodeConnectionType; +import org.thingsboard.server.common.data.rule.RuleNode; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -46,6 +47,7 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; @@ -65,13 +67,15 @@ public class TbMsgDelayNodeTest extends AbstractRuleNodeUpgradeTest { private final RuleNodeId RULE_NODE_ID = new RuleNodeId(UUID.fromString("1be24225-b669-4b26-ab7e-083aaa82d0a0")); private final Set supportedTimeUnits = EnumSet.of(TimeUnit.SECONDS, TimeUnit.MINUTES, TimeUnit.HOURS); - private final String supportedTimeUnitsStr = String.join(",", TimeUnit.SECONDS.name(), TimeUnit.MINUTES.name(), TimeUnit.HOURS.name()); + private final String supportedTimeUnitsStr = supportedTimeUnits.stream().map(TimeUnit::name).collect(Collectors.joining(", ")); private TbMsgDelayNode node; private TbMsgDelayNodeConfiguration config; @Mock private TbContext ctxMock; + @Mock + private RuleNode ruleNodeMock; @BeforeEach public void setUp() { @@ -88,6 +92,7 @@ public class TbMsgDelayNodeTest extends AbstractRuleNodeUpgradeTest { @Test public void givenDefaultConfig_whenInit_thenOk() { + given(ctxMock.getSelf()).willReturn(ruleNodeMock); assertThatNoException().isThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)))); } @@ -95,34 +100,19 @@ public class TbMsgDelayNodeTest extends AbstractRuleNodeUpgradeTest { @ValueSource(ints = {-1, 0, 5000000}) public void givenInvalidMaxPendingMsgsValue_whenInit_thenThrowsException(int maxPendingMsgs) { config.setMaxPendingMsgs(maxPendingMsgs); - - assertThatThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)))) - .isInstanceOf(TbNodeException.class) - .hasMessage("Maximum pending messages should be in a range from 1 to 100000.") - .extracting(e -> ((TbNodeException) e).isUnrecoverable()) - .isEqualTo(true); + verifyValidationExceptionOnInit(); } @Test public void givenPeriodIsNull_whenInit_thenThrowsException() { config.setPeriod(null); - - assertThatThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)))) - .isInstanceOf(TbNodeException.class) - .hasMessage("Period should be specified.") - .extracting(e -> ((TbNodeException) e).isUnrecoverable()) - .isEqualTo(true); + verifyValidationExceptionOnInit(); } @Test public void givenTimeUnitIsNull_whenInit_thenThrowsException() { config.setTimeUnit(null); - - assertThatThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)))) - .isInstanceOf(TbNodeException.class) - .hasMessage("Time unit should be specified.") - .extracting(e -> ((TbNodeException) e).isUnrecoverable()) - .isEqualTo(true); + verifyValidationExceptionOnInit(); } @ParameterizedTest @@ -131,6 +121,7 @@ public class TbMsgDelayNodeTest extends AbstractRuleNodeUpgradeTest { String periodPattern, String timeUnitPattern, TbMsgMetaData metaData, String data, long expectedDelay) throws TbNodeException { config.setPeriod(periodPattern); config.setTimeUnit(timeUnitPattern); + given(ctxMock.getSelf()).willReturn(ruleNodeMock); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); @@ -170,8 +161,9 @@ public class TbMsgDelayNodeTest extends AbstractRuleNodeUpgradeTest { @EnumSource(TimeUnit.class) public void givenTimeUnit_whenOnMsg_thenVerify(TimeUnit timeUnit) throws TbNodeException { config.setTimeUnit(timeUnit.name()); - node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + given(ctxMock.getSelf()).willReturn(ruleNodeMock); + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); var msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); if (supportedTimeUnits.contains(timeUnit)) { assertThatNoException().isThrownBy(() -> node.onMsg(ctxMock, msg)); @@ -185,8 +177,9 @@ public class TbMsgDelayNodeTest extends AbstractRuleNodeUpgradeTest { @Test public void givenPeriodIsUnparsable_whenOnMsg_thenThrowsException() throws TbNodeException { config.setPeriod("five"); - node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + given(ctxMock.getSelf()).willReturn(ruleNodeMock); + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); var msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); assertThatThrownBy(() -> node.onMsg(ctxMock, msg)) .isInstanceOf(NumberFormatException.class) @@ -196,8 +189,9 @@ public class TbMsgDelayNodeTest extends AbstractRuleNodeUpgradeTest { @Test public void givenInvalidTimeUnit_whenOnMsg_thenThrowsException() throws TbNodeException { config.setTimeUnit("sec"); - node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + given(ctxMock.getSelf()).willReturn(ruleNodeMock); + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); var msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); assertThatThrownBy(() -> node.onMsg(ctxMock, msg)) .isInstanceOf(IllegalArgumentException.class) @@ -207,6 +201,7 @@ public class TbMsgDelayNodeTest extends AbstractRuleNodeUpgradeTest { @Test public void givenMaxLimitOfPendingMsgsReached_whenOnMsg_thenTellFailure() throws TbNodeException { config.setMaxPendingMsgs(1); + given(ctxMock.getSelf()).willReturn(ruleNodeMock); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); var msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); @@ -233,6 +228,18 @@ public class TbMsgDelayNodeTest extends AbstractRuleNodeUpgradeTest { assertThat(actualPendingMsgs).isEmpty(); } + private void verifyValidationExceptionOnInit() { + RuleNode ruleNode = new RuleNode(); + ruleNode.setName("test"); + given(ctxMock.getSelf()).willReturn(ruleNode); + String errorPrefix = "'test' node configuration is invalid: "; + assertThatThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)))) + .isInstanceOf(TbNodeException.class) + .hasMessageContaining(errorPrefix) + .extracting(e -> ((TbNodeException) e).isUnrecoverable()) + .isEqualTo(true); + } + private static Stream givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() { return Stream.of( // config for version 1 with upgrade from version 0 (useMetadataPeriodInSecondsPatterns does not exist and periodInSeconds exists)