Browse Source

Merge pull request #10730 from irynamatveieva/rule-chain-input-node

Rule Chain node: added ability to transfer msg to originator's default rule chain
pull/10740/head
Andrew Shvayka 2 years ago
committed by GitHub
parent
commit
77567b8863
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
  1. 22
      application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java
  2. 2
      common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java
  3. 66
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNode.java
  4. 5
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNodeConfiguration.java
  5. 299
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNodeTest.java

22
application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java

@ -39,6 +39,7 @@ import org.thingsboard.server.dao.device.DeviceConnectivityConfiguration;
import org.thingsboard.server.dao.rule.RuleChainService;
import org.thingsboard.server.dao.settings.AdminSettingsService;
import org.thingsboard.server.dao.sql.JpaExecutorService;
import org.thingsboard.server.dao.tenant.TenantProfileService;
import org.thingsboard.server.service.component.ComponentDiscoveryService;
import org.thingsboard.server.service.component.RuleNodeClassInfo;
import org.thingsboard.server.utils.TbNodeUpgradeUtils;
@ -77,6 +78,8 @@ public class DefaultDataUpdateService implements DataUpdateService {
@Autowired
private CustomerService customerService;
@Autowired
private TenantProfileService tenantProfileService;
@Override
public void updateData(String fromVersion) throws Exception {
@ -88,12 +91,31 @@ public class DefaultDataUpdateService implements DataUpdateService {
case "3.6.4":
log.info("Updating data from version 3.6.4 to 3.7.0 ...");
updateCustomersWithTheSameTitle();
updateMaxRuleNodeExecsPerMessage();
break;
default:
throw new RuntimeException("Unable to update data, unsupported fromVersion: " + fromVersion);
}
}
private void updateMaxRuleNodeExecsPerMessage() {
var tenantProfiles = new PageDataIterable<>(
link -> tenantProfileService.findTenantProfiles(TenantId.SYS_TENANT_ID, link), DEFAULT_PAGE_SIZE);
tenantProfiles.forEach(tenantProfile -> {
var configurationOpt = tenantProfile.getProfileConfiguration();
configurationOpt.ifPresent(configuration -> {
if (configuration.getMaxRuleNodeExecsPerMessage() == 0) {
configuration.setMaxRuleNodeExecutionsPerMessage(1000);
try {
tenantProfileService.saveTenantProfile(TenantId.SYS_TENANT_ID, tenantProfile);
} catch (Exception e) {
log.error("Failed to update tenant profile with id: {} due to: ", tenantProfile.getId(), e);
}
}
});
});
}
private void updateCustomersWithTheSameTitle() {
var customers = new ArrayList<Customer>();
new PageDataIterable<>(pageLink ->

2
common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java

@ -59,7 +59,7 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura
private long maxJSExecutions;
private long maxTbelExecutions;
private long maxDPStorageDays;
private int maxRuleNodeExecutionsPerMessage;
private int maxRuleNodeExecutionsPerMessage = 1000;
private long maxEmails;
private Boolean smsEnabled;
private long maxSms;

66
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNode.java

@ -15,6 +15,8 @@
*/
package org.thingsboard.rule.engine.flow;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.TbContext;
@ -22,10 +24,14 @@ import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.data.util.TbPair;
import org.thingsboard.server.common.msg.TbMsg;
import java.util.Optional;
import java.util.UUID;
@Slf4j
@ -33,11 +39,13 @@ import java.util.UUID;
type = ComponentType.FLOW,
name = "rule chain",
configClazz = TbRuleChainInputNodeConfiguration.class,
nodeDescription = "transfers the message to another rule chain",
nodeDetails = "Allows to nest the rule chain similar to single rule node. " +
"The incoming message is forwarded to the input node of the specified target rule chain. " +
"The target rule chain may produce multiple labeled outputs. " +
"You may use the outputs to forward the results of processing to other rule nodes.",
version = 1,
nodeDescription = "Transfers the message to another rule chain",
nodeDetails = "The incoming message is forwarded to the input node of target rule chain. " +
"If 'Forward message to the originator's default rule chain' is enabled, " +
"then target rule chain might be resolved dynamically based on incoming message originator. " +
"In this case rule chain specified in the configuration will be used as fallback rule chain.<br><br>" +
"Output connections: <i>Any connection(s) produced by output node(s) in the target rule chain.</i>",
uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = "tbFlowNodeRuleChainInputConfig",
relationTypes = {},
@ -46,19 +54,57 @@ import java.util.UUID;
)
public class TbRuleChainInputNode implements TbNode {
private TbRuleChainInputNodeConfiguration config;
private RuleChainId ruleChainId;
private boolean forwardMsgToDefaultRuleChain;
@Override
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
this.config = TbNodeUtils.convert(configuration, TbRuleChainInputNodeConfiguration.class);
this.ruleChainId = new RuleChainId(UUID.fromString(config.getRuleChainId()));
TbRuleChainInputNodeConfiguration config = TbNodeUtils.convert(configuration, TbRuleChainInputNodeConfiguration.class);
if (config.getRuleChainId() == null) {
throw new TbNodeException("Rule chain must be set!", true);
}
UUID ruleChainUUID;
try {
ruleChainUUID = UUID.fromString(config.getRuleChainId());
} catch (Exception e) {
throw new TbNodeException("Failed to parse rule chain id: " + config.getRuleChainId(), true);
}
ruleChainId = new RuleChainId(ruleChainUUID);
ctx.checkTenantEntity(ruleChainId);
forwardMsgToDefaultRuleChain = config.isForwardMsgToDefaultRuleChain();
}
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
ctx.input(msg, ruleChainId);
public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException {
RuleChainId targetRuleChainId = forwardMsgToDefaultRuleChain ?
getOriginatorDefaultRuleChainId(ctx, msg).orElse(ruleChainId) : ruleChainId;
ctx.input(msg, targetRuleChainId);
}
@Override
public TbPair<Boolean, JsonNode> upgrade(int fromVersion, JsonNode oldConfiguration) throws TbNodeException {
boolean hasChanges = false;
switch (fromVersion) {
case 0 -> {
if (!oldConfiguration.has("forwardMsgToDefaultRuleChain")) {
hasChanges = true;
((ObjectNode) oldConfiguration).put("forwardMsgToDefaultRuleChain", false);
}
}
default -> {
}
}
return new TbPair<>(hasChanges, oldConfiguration);
}
private Optional<RuleChainId> getOriginatorDefaultRuleChainId(TbContext ctx, TbMsg msg) {
return Optional.ofNullable(
switch (msg.getOriginator().getEntityType()) {
case DEVICE ->
ctx.getDeviceProfileCache().get(ctx.getTenantId(), (DeviceId) msg.getOriginator()).getDefaultRuleChainId();
case ASSET ->
ctx.getAssetProfileCache().get(ctx.getTenantId(), (AssetId) msg.getOriginator()).getDefaultRuleChainId();
default -> null;
});
}
}

5
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNodeConfiguration.java

@ -22,10 +22,13 @@ import org.thingsboard.rule.engine.api.NodeConfiguration;
public class TbRuleChainInputNodeConfiguration implements NodeConfiguration<TbRuleChainInputNodeConfiguration> {
private String ruleChainId;
private boolean forwardMsgToDefaultRuleChain;
@Override
public TbRuleChainInputNodeConfiguration defaultConfiguration() {
return new TbRuleChainInputNodeConfiguration();
TbRuleChainInputNodeConfiguration configuration = new TbRuleChainInputNodeConfiguration();
configuration.setForwardMsgToDefaultRuleChain(false);
return configuration;
}
}

299
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNodeTest.java

@ -0,0 +1,299 @@
/**
* Copyright © 2016-2024 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.flow;
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.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.test.util.ReflectionTestUtils;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.AbstractRuleNodeUpgradeTest;
import org.thingsboard.rule.engine.api.RuleEngineAssetProfileCache;
import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache;
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.DeviceProfile;
import org.thingsboard.server.common.data.asset.AssetProfile;
import org.thingsboard.server.common.data.id.AssetId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
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.UUID;
import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
public class TbRuleChainInputNodeTest extends AbstractRuleNodeUpgradeTest {
private final TenantId TENANT_ID = new TenantId(UUID.fromString("4ba69ea5-6b27-42df-ab66-e7a727a67027"));
private final DeviceId DEVICE_ID = new DeviceId(UUID.fromString("97731954-2147-4176-8f1a-d14f1b73e4e6"));
private final AssetId ASSET_ID = new AssetId(UUID.fromString("841a47bd-4e8e-4ea5-88e6-420da0d70e51"));
private TbRuleChainInputNode node;
private TbRuleChainInputNodeConfiguration config;
private TbNodeConfiguration nodeConfiguration;
@Mock
private TbContext ctxMock;
@Mock
private RuleEngineDeviceProfileCache deviceProfileCacheMock;
@Mock
private RuleEngineAssetProfileCache assetProfileCacheMock;
@BeforeEach
public void setUp() {
node = spy(new TbRuleChainInputNode());
config = new TbRuleChainInputNodeConfiguration().defaultConfiguration();
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
}
@ParameterizedTest
@MethodSource
public void givenValidConfig_whenInit_thenOk(String ruleChainIdStr, boolean forwardMsgToDefaultRuleChain) throws TbNodeException {
//GIVEN
config.setRuleChainId(ruleChainIdStr);
config.setForwardMsgToDefaultRuleChain(forwardMsgToDefaultRuleChain);
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
//WHEN
assertThatCode(() -> node.init(ctxMock, nodeConfiguration))
.doesNotThrowAnyException();
//THEN
verify(ctxMock).checkTenantEntity(new RuleChainId(UUID.fromString(ruleChainIdStr)));
}
private static Stream<Arguments> givenValidConfig_whenInit_thenOk() {
return Stream.of(
Arguments.of("45bba7c4-04bf-419b-ae03-6ceb9724f10e", false),
Arguments.of("52d57e1b-70bb-480e-bcc4-6710e1dcc9d8", true)
);
}
@ParameterizedTest
@ValueSource(strings = {"91acbce0-079fdb", "", " ", "my test string"})
public void givenInvalidRuleChainId_whenInit_thenThrowsException(String ruleChainIdStr) {
//GIVEN
config.setRuleChainId(ruleChainIdStr);
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
//WHEN-THEN
Assertions.assertThatThrownBy(() -> node.init(ctxMock, nodeConfiguration))
.isInstanceOf(TbNodeException.class)
.hasMessage("Failed to parse rule chain id: " + ruleChainIdStr);
}
@Test
public void givenRuleChainIdIsNotSet_whenInit_thenThrowsException() {
assertThatThrownBy(() -> node.init(ctxMock, nodeConfiguration))
.isInstanceOf(TbNodeException.class)
.hasMessage("Rule chain must be set!");
}
@Test
public void givenForwardMsgToDefaultIsTrue_whenOnMsg_thenShouldTransferToDeviceDefaultRuleChain() throws TbNodeException {
//GIVEN
DeviceProfile deviceProfile = new DeviceProfile();
RuleChainId defaultRuleChainId = new RuleChainId(UUID.fromString("196e3cd5-68b8-421e-a0cf-1d44fa377cdf"));
deviceProfile.setDefaultRuleChainId(defaultRuleChainId);
TbMsg msg = getMsg(DEVICE_ID);
String ruleChainIdFromConfigStr = "acbc924f-7f95-4a9b-a854-e4822deb74c7";
config.setRuleChainId(ruleChainIdFromConfigStr);
config.setForwardMsgToDefaultRuleChain(true);
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
when(ctxMock.getTenantId()).thenReturn(TENANT_ID);
when(ctxMock.getDeviceProfileCache()).thenReturn(deviceProfileCacheMock);
when(deviceProfileCacheMock.get(any(TenantId.class), any(DeviceId.class))).thenReturn(deviceProfile);
node.init(ctxMock, nodeConfiguration);
//WHEN
node.onMsg(ctxMock, msg);
//THEN
ArgumentCaptor<RuleChainId> ruleChainArgumentCaptor = ArgumentCaptor.forClass(RuleChainId.class);
verify(ctxMock).input(eq(msg), ruleChainArgumentCaptor.capture());
RuleChainId expectedRuleChainId = ruleChainArgumentCaptor.getValue();
assertThat(expectedRuleChainId).isEqualTo(defaultRuleChainId);
RuleChainId ruleChainId = (RuleChainId) ReflectionTestUtils.getField(node, "ruleChainId");
assertThat(ruleChainId).isEqualTo(new RuleChainId(UUID.fromString(ruleChainIdFromConfigStr)));
}
@Test
public void givenForwardMsgToDefaultIsTrue_whenOnMsg_thenShouldTransferToAssetDefaultRuleChain() throws TbNodeException {
//GIVEN
AssetProfile assetProfile = new AssetProfile();
RuleChainId defaultRuleChainId = new RuleChainId(UUID.fromString("f0a3cd58-980c-4730-a40c-8f59064d2065"));
assetProfile.setDefaultRuleChainId(defaultRuleChainId);
TbMsg msg = getMsg(ASSET_ID);
String ruleChainIdFromConfigStr = "56f1c0b8-1a00-4ce0-b3ab-a1416d7cc429";
config.setRuleChainId(ruleChainIdFromConfigStr);
config.setForwardMsgToDefaultRuleChain(true);
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
when(ctxMock.getTenantId()).thenReturn(TENANT_ID);
when(ctxMock.getAssetProfileCache()).thenReturn(assetProfileCacheMock);
when(assetProfileCacheMock.get(any(TenantId.class), any(AssetId.class))).thenReturn(assetProfile);
node.init(ctxMock, nodeConfiguration);
//WHEN
node.onMsg(ctxMock, msg);
//THEN
ArgumentCaptor<RuleChainId> ruleChainArgumentCaptor = ArgumentCaptor.forClass(RuleChainId.class);
verify(ctxMock).input(eq(msg), ruleChainArgumentCaptor.capture());
RuleChainId expectedRuleChainId = ruleChainArgumentCaptor.getValue();
assertThat(expectedRuleChainId).isEqualTo(defaultRuleChainId);
RuleChainId ruleChainId = (RuleChainId) ReflectionTestUtils.getField(node, "ruleChainId");
assertThat(ruleChainId).isEqualTo(new RuleChainId(UUID.fromString(ruleChainIdFromConfigStr)));
}
@Test
public void givenForwardMsgToDefaultIsTrueWithoutDeviceDefaultRuleChain_whenOnMsg_thenShouldTransferToRuleChainFromConfig() throws TbNodeException {
//GIVEN
DeviceProfile deviceProfile = new DeviceProfile();
TbMsg msg = getMsg(DEVICE_ID);
String ruleChainIdFromConfigStr = "357c2785-e7cc-46a8-9797-957180dabdeb";
RuleChainId ruleChainIdFromConfig = new RuleChainId(UUID.fromString(ruleChainIdFromConfigStr));
config.setRuleChainId(ruleChainIdFromConfigStr);
config.setForwardMsgToDefaultRuleChain(true);
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
when(ctxMock.getTenantId()).thenReturn(TENANT_ID);
when(ctxMock.getDeviceProfileCache()).thenReturn(deviceProfileCacheMock);
when(deviceProfileCacheMock.get(any(TenantId.class), any(DeviceId.class))).thenReturn(deviceProfile);
node.init(ctxMock, nodeConfiguration);
//WHEN
node.onMsg(ctxMock, msg);
//THEN
ArgumentCaptor<RuleChainId> ruleChainArgumentCaptor = ArgumentCaptor.forClass(RuleChainId.class);
verify(ctxMock).input(eq(msg), ruleChainArgumentCaptor.capture());
assertThat(ruleChainArgumentCaptor.getValue()).isEqualTo(ruleChainIdFromConfig);
}
@Test
public void givenForwardMsgToDefaultIsTrueWithoutAssetDefaultRuleChain_whenOnMsg_thenShouldTransferToRuleChainFromConfig() throws TbNodeException {
//GIVEN
AssetProfile assetProfile = new AssetProfile();
TbMsg msg = getMsg(ASSET_ID);
String ruleChainIdFromConfigStr = "12883c3d-c10b-4d5b-b606-a59385a920bc";
RuleChainId ruleChainIdFromConfig = new RuleChainId(UUID.fromString(ruleChainIdFromConfigStr));
config.setRuleChainId(ruleChainIdFromConfigStr);
config.setForwardMsgToDefaultRuleChain(true);
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
when(ctxMock.getTenantId()).thenReturn(TENANT_ID);
when(ctxMock.getAssetProfileCache()).thenReturn(assetProfileCacheMock);
when(assetProfileCacheMock.get(any(TenantId.class), any(AssetId.class))).thenReturn(assetProfile);
node.init(ctxMock, nodeConfiguration);
//WHEN
node.onMsg(ctxMock, msg);
//THEN
ArgumentCaptor<RuleChainId> ruleChainArgumentCaptor = ArgumentCaptor.forClass(RuleChainId.class);
verify(ctxMock).input(eq(msg), ruleChainArgumentCaptor.capture());
assertThat(ruleChainArgumentCaptor.getValue()).isEqualTo(ruleChainIdFromConfig);
}
@Test
public void givenRuleChainInConfig_whenOnMsg_thenShouldTransferToRuleChainFromConfig() throws TbNodeException {
//GIVEN
String ruleChainIdFromConfigStr = "3c02c8b3-645c-4e67-aac5-f984f59471d1";
RuleChainId ruleChainIdFromConfig = new RuleChainId(UUID.fromString(ruleChainIdFromConfigStr));
TbMsg msg = getMsg(DEVICE_ID);
config.setRuleChainId(ruleChainIdFromConfigStr);
config.setForwardMsgToDefaultRuleChain(false);
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config));
node.init(ctxMock, nodeConfiguration);
//WHEN
node.onMsg(ctxMock, msg);
//THEN
ArgumentCaptor<RuleChainId> ruleChainArgumentCaptor = ArgumentCaptor.forClass(RuleChainId.class);
verify(ctxMock).input(eq(msg), ruleChainArgumentCaptor.capture());
assertThat(ruleChainArgumentCaptor.getValue()).isEqualTo(ruleChainIdFromConfig);
}
private static Stream<Arguments> givenFromVersionAndConfig_whenUpgrade_thenVerifyHasChangesAndConfig() {
return Stream.of(
//config for version 0
Arguments.of(0,
"{\"ruleChainId\": null}",
true,
"{\"ruleChainId\": null, \"forwardMsgToDefaultRuleChain\": false}"
),
//config for version 1 with upgrade from version 0
Arguments.of(1,
"{\"ruleChainId\": null, \"forwardMsgToDefaultRuleChain\": false}",
false,
"{\"ruleChainId\": null, \"forwardMsgToDefaultRuleChain\": false}"
)
);
}
@Override
protected TbNode getTestNode() {
return node;
}
private TbMsg getMsg(EntityId entityId) {
return TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, entityId, TbMsgMetaData.EMPTY, TbMsg.EMPTY_STRING);
}
}
Loading…
Cancel
Save