From d745fd21fb72f6bd5561b10b97c9dbea3f490fd9 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Mon, 25 Jul 2022 12:31:58 +0300 Subject: [PATCH 01/24] new RuleNode "delete attributes" --- .../telemetry/TbMsgDeleteAttributes.java | 85 ++++++++++ .../TbMsgDeleteAttributesConfiguration.java | 38 +++++ .../telemetry/TbMsgDeleteAttributesTest.java | 148 ++++++++++++++++++ 3 files changed, 271 insertions(+) create mode 100644 rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java create mode 100644 rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java create mode 100644 rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java new file mode 100644 index 0000000000..edd6f8df17 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java @@ -0,0 +1,85 @@ +/** + * Copyright © 2016-2022 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.telemetry; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.util.CollectionUtils; +import org.thingsboard.rule.engine.api.RuleNode; +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.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.StringUtils; +import org.thingsboard.server.common.data.plugin.ComponentType; +import org.thingsboard.server.common.msg.TbMsg; + +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.stream.Collectors; + +@Slf4j +@RuleNode( + type = ComponentType.ACTION, + name = "delete attributes", + configClazz = TbMsgDeleteAttributesConfiguration.class, + nodeDescription = "Delete attributes for Message Originator.", + nodeDetails = "Allowed scope parameter values: SERVER/CLIENT/SHARED. If no attributes are selected - " + + "message send via Failure chain. If selected attributes successfully deleted - message send via " + + "Success chain, otherwise Failure chain will be used.", + uiResources = {"static/rulenode/rulenode-core-config.js"}, + configDirective = "tbActionNodeDeleteAttributesConfig", + icon = "remove_circle" +) +public class TbMsgDeleteAttributes implements TbNode { + + TbMsgDeleteAttributesConfiguration config; + + @Override + public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { + this.config = TbNodeUtils.convert(configuration, TbMsgDeleteAttributesConfiguration.class); + if (CollectionUtils.isEmpty(config.getKeysPatterns())) { + throw new IllegalArgumentException("Attribute keys list is empty!"); + } + } + + @Override + public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException { + List keysPatterns = config.getKeysPatterns(); + String scope = TbNodeUtils.processPattern(config.getScopePattern(), msg); + if (DataConstants.SERVER_SCOPE.equals(scope) || + DataConstants.CLIENT_SCOPE.equals(scope) || + DataConstants.SHARED_SCOPE.equals(scope)) { + List keysToDelete = keysPatterns.stream() + .map(keyPattern -> TbNodeUtils.processPattern(keyPattern, msg)) + .distinct() + .filter(StringUtils::isNotBlank) + .collect(Collectors.toList()); + if (keysToDelete.isEmpty()) { + throw new RuntimeException("Selected keys patterns have invalid values!"); + } + ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), scope, keysToDelete, new TelemetryNodeCallback(ctx, msg)); + } else { + ctx.tellFailure(msg, new IllegalArgumentException("Unsupported attributes scope '" + scope + "'! Only 'SERVER_SCOPE', 'CLIENT_SCOPE' or 'SHARED_SCOPE' are allowed!")); + } + } + + @Override + public void destroy() { + + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java new file mode 100644 index 0000000000..607791561c --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java @@ -0,0 +1,38 @@ +/** + * Copyright © 2016-2022 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.telemetry; + +import lombok.Data; +import org.thingsboard.rule.engine.api.NodeConfiguration; +import org.thingsboard.server.common.data.DataConstants; + +import java.util.Collections; +import java.util.List; + +@Data +public class TbMsgDeleteAttributesConfiguration implements NodeConfiguration { + + private String scopePattern; + private List keysPatterns; + + @Override + public TbMsgDeleteAttributesConfiguration defaultConfiguration() { + TbMsgDeleteAttributesConfiguration configuration = new TbMsgDeleteAttributesConfiguration(); + configuration.setScopePattern(DataConstants.SERVER_SCOPE); + configuration.setKeysPatterns(Collections.emptyList()); + return configuration; + } +} diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java new file mode 100644 index 0000000000..a04a66fd72 --- /dev/null +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java @@ -0,0 +1,148 @@ +/** + * Copyright © 2016-2022 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.telemetry; + +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.api.TbNodeConfiguration; +import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.queue.TbMsgCallback; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.BDDMockito.willReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +@Slf4j +public class TbMsgDeleteAttributesTest { + final ObjectMapper mapper = new ObjectMapper(); + + DeviceId deviceId; + TbMsgDeleteAttributes node; + TbMsgDeleteAttributesConfiguration config; + TbNodeConfiguration nodeConfiguration; + TbContext ctx; + TbMsgCallback callback; + + RuleEngineTelemetryService telemetryService; + + @BeforeEach + void setUp() throws TbNodeException { + deviceId = new DeviceId(UUID.randomUUID()); + callback = mock(TbMsgCallback.class); + ctx = mock(TbContext.class); + config = new TbMsgDeleteAttributesConfiguration().defaultConfiguration(); + config.setKeysPatterns(List.of("${TestAttribute_1}", "$[TestAttribute_2]", "$[TestAttribute_3]")); + nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); + node = spy(new TbMsgDeleteAttributes()); + node.init(ctx, nodeConfiguration); + telemetryService = mock(RuleEngineTelemetryService.class); + + willReturn(telemetryService).given(ctx).getTelemetryService(); + willAnswer(invocation -> { + TelemetryNodeCallback callBack = invocation.getArgument(4); + callBack.onSuccess(null); + return null; + }).given(telemetryService).deleteAndNotify( + any(), any(), anyString(), anyList(), any()); + } + + @AfterEach + void tearDown() { + node.destroy(); + } + + @Test + void givenDefaultConfig_whenInit_thenOK() { + assertThat(node.config).isEqualTo(config); + } + + @Test + void givenDefaultConfig_whenInit_thenFail() { + config.setKeysPatterns(Collections.emptyList()); + nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); + assertThatThrownBy(() -> node.init(ctx, nodeConfiguration)).isInstanceOf(IllegalArgumentException.class); + } + + @Test + void givenDefaultConfig_whenVerify_thenOK() { + TbMsgDeleteAttributesConfiguration defaultConfig = new TbMsgDeleteAttributesConfiguration().defaultConfiguration(); + assertThat(defaultConfig.getScopePattern()).isEqualTo(DataConstants.SERVER_SCOPE); + assertThat(defaultConfig.getKeysPatterns()).isEqualTo(Collections.emptyList()); + } + + @Test + void givenMsg_whenOnMsg_thenVerifyOutput() throws Exception { + final Map mdMap = Map.of( + "TestAttribute_1", "temperature", + "city", "NY" + ); + final TbMsgMetaData metaData = new TbMsgMetaData(mdMap); + final String data = "{\"TestAttribute_2\": \"humidity\", \"TestAttribute_3\": \"voltage\"}"; + + TbMsg msg = TbMsg.newMsg("POST_ATTRIBUTES_REQUEST", deviceId, metaData, data, callback); + node.onMsg(ctx, msg); + + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + verify(ctx, times(1)).tellSuccess(newMsgCaptor.capture()); + verify(ctx, never()).tellFailure(any(), any()); + verify(telemetryService, times(1)).deleteAndNotify(any(), any(), anyString(), anyList(), any()); + } + + @Test + void givenEmptyMsg_whenOnMsg_thenTellFailure() throws Exception { + final TbMsgMetaData metaData = new TbMsgMetaData(); + final String data = "{}"; + + config.setScopePattern("NOT_SERVER_SCOPE"); + nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); + node.init(ctx, nodeConfiguration); + + TbMsg msg = TbMsg.newMsg("POST_ATTRIBUTES_REQUEST", deviceId, metaData, data, callback); + node.onMsg(ctx, msg); + + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + ArgumentCaptor exceptionCaptor = ArgumentCaptor.forClass(Exception.class); + verify(ctx, never()).tellSuccess(any()); + verify(ctx, times(1)).tellFailure(newMsgCaptor.capture(), exceptionCaptor.capture()); + + assertThat(newMsgCaptor.getValue()).isSameAs(msg); + assertThat(exceptionCaptor.getValue()).isInstanceOf(IllegalArgumentException.class); + } +} \ No newline at end of file From 2cc0a95fde0a33b845f5df925300f75a111096cf Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Mon, 25 Jul 2022 16:52:33 +0300 Subject: [PATCH 02/24] license:format --- .../rule/engine/telemetry/TbMsgDeleteAttributesTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java index a04a66fd72..a5c6dc9c83 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java @@ -126,11 +126,11 @@ public class TbMsgDeleteAttributesTest { } @Test - void givenEmptyMsg_whenOnMsg_thenTellFailure() throws Exception { + void giveInvalidScope_whenOnMsg_thenTellFailure() throws Exception { final TbMsgMetaData metaData = new TbMsgMetaData(); final String data = "{}"; - config.setScopePattern("NOT_SERVER_SCOPE"); + config.setScopePattern("INVALID_SCOPE"); nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); node.init(ctx, nodeConfiguration); From e29b91c3e03a0062eefeebbe60a6117ec937c71a Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Fri, 5 Aug 2022 09:53:04 +0300 Subject: [PATCH 03/24] fix fields config --- .../rule/engine/telemetry/TbMsgDeleteAttributes.java | 6 +++--- .../telemetry/TbMsgDeleteAttributesConfiguration.java | 8 ++++---- .../engine/telemetry/TbMsgDeleteAttributesTest.java | 10 +++++----- 3 files changed, 12 insertions(+), 12 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java index edd6f8df17..8e2a91a9c3 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java @@ -52,15 +52,15 @@ public class TbMsgDeleteAttributes implements TbNode { @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { this.config = TbNodeUtils.convert(configuration, TbMsgDeleteAttributesConfiguration.class); - if (CollectionUtils.isEmpty(config.getKeysPatterns())) { + if (CollectionUtils.isEmpty(config.getKeys())) { throw new IllegalArgumentException("Attribute keys list is empty!"); } } @Override public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException { - List keysPatterns = config.getKeysPatterns(); - String scope = TbNodeUtils.processPattern(config.getScopePattern(), msg); + List keysPatterns = config.getKeys(); + String scope = TbNodeUtils.processPattern(config.getScope(), msg); if (DataConstants.SERVER_SCOPE.equals(scope) || DataConstants.CLIENT_SCOPE.equals(scope) || DataConstants.SHARED_SCOPE.equals(scope)) { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java index 607791561c..03bef2a4e6 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java @@ -25,14 +25,14 @@ import java.util.List; @Data public class TbMsgDeleteAttributesConfiguration implements NodeConfiguration { - private String scopePattern; - private List keysPatterns; + private String scope; + private List keys; @Override public TbMsgDeleteAttributesConfiguration defaultConfiguration() { TbMsgDeleteAttributesConfiguration configuration = new TbMsgDeleteAttributesConfiguration(); - configuration.setScopePattern(DataConstants.SERVER_SCOPE); - configuration.setKeysPatterns(Collections.emptyList()); + configuration.setScope(DataConstants.SERVER_SCOPE); + configuration.setKeys(Collections.emptyList()); return configuration; } } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java index a5c6dc9c83..28af7637aa 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java @@ -68,7 +68,7 @@ public class TbMsgDeleteAttributesTest { callback = mock(TbMsgCallback.class); ctx = mock(TbContext.class); config = new TbMsgDeleteAttributesConfiguration().defaultConfiguration(); - config.setKeysPatterns(List.of("${TestAttribute_1}", "$[TestAttribute_2]", "$[TestAttribute_3]")); + config.setKeys(List.of("${TestAttribute_1}", "$[TestAttribute_2]", "$[TestAttribute_3]")); nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); node = spy(new TbMsgDeleteAttributes()); node.init(ctx, nodeConfiguration); @@ -95,7 +95,7 @@ public class TbMsgDeleteAttributesTest { @Test void givenDefaultConfig_whenInit_thenFail() { - config.setKeysPatterns(Collections.emptyList()); + config.setKeys(Collections.emptyList()); nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); assertThatThrownBy(() -> node.init(ctx, nodeConfiguration)).isInstanceOf(IllegalArgumentException.class); } @@ -103,8 +103,8 @@ public class TbMsgDeleteAttributesTest { @Test void givenDefaultConfig_whenVerify_thenOK() { TbMsgDeleteAttributesConfiguration defaultConfig = new TbMsgDeleteAttributesConfiguration().defaultConfiguration(); - assertThat(defaultConfig.getScopePattern()).isEqualTo(DataConstants.SERVER_SCOPE); - assertThat(defaultConfig.getKeysPatterns()).isEqualTo(Collections.emptyList()); + assertThat(defaultConfig.getScope()).isEqualTo(DataConstants.SERVER_SCOPE); + assertThat(defaultConfig.getKeys()).isEqualTo(Collections.emptyList()); } @Test @@ -130,7 +130,7 @@ public class TbMsgDeleteAttributesTest { final TbMsgMetaData metaData = new TbMsgMetaData(); final String data = "{}"; - config.setScopePattern("INVALID_SCOPE"); + config.setScope("INVALID_SCOPE"); nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); node.init(ctx, nodeConfiguration); From 8974ce3e12c89f6f15c636bd2732f4375ed94e1d Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Fri, 5 Aug 2022 14:02:59 +0300 Subject: [PATCH 04/24] add boolean setScopeAsPattern; --- .../engine/telemetry/TbMsgDeleteAttributesConfiguration.java | 1 + 1 file changed, 1 insertion(+) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java index 03bef2a4e6..8457e9795f 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java @@ -27,6 +27,7 @@ public class TbMsgDeleteAttributesConfiguration implements NodeConfiguration keys; + private boolean setScopeAsPattern; @Override public TbMsgDeleteAttributesConfiguration defaultConfiguration() { From f4cf1d41d493338d68be31de621a17fe96d8927f Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Fri, 5 Aug 2022 14:09:26 +0300 Subject: [PATCH 05/24] add boolean setScopeAsPattern; --- .../engine/telemetry/TbMsgDeleteAttributesConfiguration.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java index 8457e9795f..2c2d4db8ad 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java @@ -27,13 +27,14 @@ public class TbMsgDeleteAttributesConfiguration implements NodeConfiguration keys; - private boolean setScopeAsPattern; + private Boolean setScopeAsPattern; @Override public TbMsgDeleteAttributesConfiguration defaultConfiguration() { TbMsgDeleteAttributesConfiguration configuration = new TbMsgDeleteAttributesConfiguration(); configuration.setScope(DataConstants.SERVER_SCOPE); configuration.setKeys(Collections.emptyList()); + configuration.setSetScopeAsPattern(false); return configuration; } } From 352581a19943e2c5db088f8ab4e768f2c6961a3c Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Fri, 5 Aug 2022 14:20:21 +0300 Subject: [PATCH 06/24] setScopeAsPattern -> useScopeAsPattern; --- .../engine/telemetry/TbMsgDeleteAttributesConfiguration.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java index 2c2d4db8ad..785a3d271d 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java @@ -27,14 +27,14 @@ public class TbMsgDeleteAttributesConfiguration implements NodeConfiguration keys; - private Boolean setScopeAsPattern; + private Boolean useScopeAsPattern; @Override public TbMsgDeleteAttributesConfiguration defaultConfiguration() { TbMsgDeleteAttributesConfiguration configuration = new TbMsgDeleteAttributesConfiguration(); configuration.setScope(DataConstants.SERVER_SCOPE); configuration.setKeys(Collections.emptyList()); - configuration.setSetScopeAsPattern(false); + configuration.setUseScopeAsPattern(false); return configuration; } } From ef28b19f80ad80db8a37f18db4d0326123ded170 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Mon, 8 Aug 2022 17:04:10 +0300 Subject: [PATCH 07/24] Node modification --- .../telemetry/TbMsgDeleteAttributes.java | 32 ++++++------------- .../TbMsgDeleteAttributesConfiguration.java | 2 -- .../telemetry/TbMsgDeleteAttributesTest.java | 14 ++------ 3 files changed, 12 insertions(+), 36 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java index 8e2a91a9c3..3fc3eb5bd3 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java @@ -16,14 +16,12 @@ package org.thingsboard.rule.engine.telemetry; import lombok.extern.slf4j.Slf4j; -import org.springframework.util.CollectionUtils; import org.thingsboard.rule.engine.api.RuleNode; 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.rule.engine.api.util.TbNodeUtils; -import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; @@ -38,9 +36,7 @@ import java.util.stream.Collectors; name = "delete attributes", configClazz = TbMsgDeleteAttributesConfiguration.class, nodeDescription = "Delete attributes for Message Originator.", - nodeDetails = "Allowed scope parameter values: SERVER/CLIENT/SHARED. If no attributes are selected - " + - "message send via Failure chain. If selected attributes successfully deleted - message send via " + - "Success chain, otherwise Failure chain will be used.", + nodeDetails = "Allowed scope parameter values: SERVER/CLIENT/SHARED. Will try to remove attributes by keys from the list", uiResources = {"static/rulenode/rulenode-core-config.js"}, configDirective = "tbActionNodeDeleteAttributesConfig", icon = "remove_circle" @@ -52,29 +48,19 @@ public class TbMsgDeleteAttributes implements TbNode { @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { this.config = TbNodeUtils.convert(configuration, TbMsgDeleteAttributesConfiguration.class); - if (CollectionUtils.isEmpty(config.getKeys())) { - throw new IllegalArgumentException("Attribute keys list is empty!"); - } } @Override public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException { - List keysPatterns = config.getKeys(); - String scope = TbNodeUtils.processPattern(config.getScope(), msg); - if (DataConstants.SERVER_SCOPE.equals(scope) || - DataConstants.CLIENT_SCOPE.equals(scope) || - DataConstants.SHARED_SCOPE.equals(scope)) { - List keysToDelete = keysPatterns.stream() - .map(keyPattern -> TbNodeUtils.processPattern(keyPattern, msg)) - .distinct() - .filter(StringUtils::isNotBlank) - .collect(Collectors.toList()); - if (keysToDelete.isEmpty()) { - throw new RuntimeException("Selected keys patterns have invalid values!"); - } - ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), scope, keysToDelete, new TelemetryNodeCallback(ctx, msg)); + List keysToDelete = config.getKeys().stream() + .map(keyPattern -> TbNodeUtils.processPattern(keyPattern, msg)) + .distinct() + .filter(StringUtils::isNotBlank) + .collect(Collectors.toList()); + if (keysToDelete.isEmpty()) { + ctx.tellFailure(msg, new RuntimeException("Selected keys patterns have invalid values!")); } else { - ctx.tellFailure(msg, new IllegalArgumentException("Unsupported attributes scope '" + scope + "'! Only 'SERVER_SCOPE', 'CLIENT_SCOPE' or 'SHARED_SCOPE' are allowed!")); + ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), config.getScope(), keysToDelete, new TelemetryNodeCallback(ctx, msg)); } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java index 785a3d271d..03bef2a4e6 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesConfiguration.java @@ -27,14 +27,12 @@ public class TbMsgDeleteAttributesConfiguration implements NodeConfiguration keys; - private Boolean useScopeAsPattern; @Override public TbMsgDeleteAttributesConfiguration defaultConfiguration() { TbMsgDeleteAttributesConfiguration configuration = new TbMsgDeleteAttributesConfiguration(); configuration.setScope(DataConstants.SERVER_SCOPE); configuration.setKeys(Collections.emptyList()); - configuration.setUseScopeAsPattern(false); return configuration; } } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java index 28af7637aa..fbca272e67 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java @@ -37,7 +37,6 @@ import java.util.Map; import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyString; @@ -93,13 +92,6 @@ public class TbMsgDeleteAttributesTest { assertThat(node.config).isEqualTo(config); } - @Test - void givenDefaultConfig_whenInit_thenFail() { - config.setKeys(Collections.emptyList()); - nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); - assertThatThrownBy(() -> node.init(ctx, nodeConfiguration)).isInstanceOf(IllegalArgumentException.class); - } - @Test void givenDefaultConfig_whenVerify_thenOK() { TbMsgDeleteAttributesConfiguration defaultConfig = new TbMsgDeleteAttributesConfiguration().defaultConfiguration(); @@ -128,9 +120,9 @@ public class TbMsgDeleteAttributesTest { @Test void giveInvalidScope_whenOnMsg_thenTellFailure() throws Exception { final TbMsgMetaData metaData = new TbMsgMetaData(); - final String data = "{}"; + final String data = "{\"TestAttribute_1\": \"\", \"TestAttribute_2\": \"\"}"; - config.setScope("INVALID_SCOPE"); + config.setKeys(List.of("$[TestAttribute_1]", "$[TestAttribute_2]")); nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); node.init(ctx, nodeConfiguration); @@ -143,6 +135,6 @@ public class TbMsgDeleteAttributesTest { verify(ctx, times(1)).tellFailure(newMsgCaptor.capture(), exceptionCaptor.capture()); assertThat(newMsgCaptor.getValue()).isSameAs(msg); - assertThat(exceptionCaptor.getValue()).isInstanceOf(IllegalArgumentException.class); + assertThat(exceptionCaptor.getValue()).isInstanceOf(RuntimeException.class); } } \ No newline at end of file From 0e9dc6a19e8dc78606612853c986b3d4ecb0bbf8 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Tue, 9 Aug 2022 10:44:45 +0300 Subject: [PATCH 08/24] Node nodeDetails --- .../telemetry/TbMsgDeleteAttributes.java | 5 ++-- .../telemetry/TbMsgDeleteAttributesTest.java | 23 +------------------ 2 files changed, 4 insertions(+), 24 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java index 3fc3eb5bd3..1a2fa21988 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java @@ -36,7 +36,8 @@ import java.util.stream.Collectors; name = "delete attributes", configClazz = TbMsgDeleteAttributesConfiguration.class, nodeDescription = "Delete attributes for Message Originator.", - nodeDetails = "Allowed scope parameter values: SERVER/CLIENT/SHARED. Will try to remove attributes by keys from the list", + nodeDetails = "Attempt to remove attributes by selected keys. If msg originator doesn't have an attribute with " + + " a key selected in the configuration, it will be ignored.", uiResources = {"static/rulenode/rulenode-core-config.js"}, configDirective = "tbActionNodeDeleteAttributesConfig", icon = "remove_circle" @@ -58,7 +59,7 @@ public class TbMsgDeleteAttributes implements TbNode { .filter(StringUtils::isNotBlank) .collect(Collectors.toList()); if (keysToDelete.isEmpty()) { - ctx.tellFailure(msg, new RuntimeException("Selected keys patterns have invalid values!")); + ctx.tellSuccess(msg); } else { ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), config.getScope(), keysToDelete, new TelemetryNodeCallback(ctx, msg)); } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java index fbca272e67..b763ba6215 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java @@ -67,7 +67,7 @@ public class TbMsgDeleteAttributesTest { callback = mock(TbMsgCallback.class); ctx = mock(TbContext.class); config = new TbMsgDeleteAttributesConfiguration().defaultConfiguration(); - config.setKeys(List.of("${TestAttribute_1}", "$[TestAttribute_2]", "$[TestAttribute_3]")); + config.setKeys(List.of("${TestAttribute_1}", "$[TestAttribute_2]", "$[TestAttribute_3]", "TestAttribute_4")); nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); node = spy(new TbMsgDeleteAttributes()); node.init(ctx, nodeConfiguration); @@ -116,25 +116,4 @@ public class TbMsgDeleteAttributesTest { verify(ctx, never()).tellFailure(any(), any()); verify(telemetryService, times(1)).deleteAndNotify(any(), any(), anyString(), anyList(), any()); } - - @Test - void giveInvalidScope_whenOnMsg_thenTellFailure() throws Exception { - final TbMsgMetaData metaData = new TbMsgMetaData(); - final String data = "{\"TestAttribute_1\": \"\", \"TestAttribute_2\": \"\"}"; - - config.setKeys(List.of("$[TestAttribute_1]", "$[TestAttribute_2]")); - nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); - node.init(ctx, nodeConfiguration); - - TbMsg msg = TbMsg.newMsg("POST_ATTRIBUTES_REQUEST", deviceId, metaData, data, callback); - node.onMsg(ctx, msg); - - ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); - ArgumentCaptor exceptionCaptor = ArgumentCaptor.forClass(Exception.class); - verify(ctx, never()).tellSuccess(any()); - verify(ctx, times(1)).tellFailure(newMsgCaptor.capture(), exceptionCaptor.capture()); - - assertThat(newMsgCaptor.getValue()).isSameAs(msg); - assertThat(exceptionCaptor.getValue()).isInstanceOf(RuntimeException.class); - } } \ No newline at end of file From ca581f68e1caf1dcbaee3b9671ee50eab30ab84f Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Thu, 25 Aug 2022 12:16:56 +0300 Subject: [PATCH 09/24] refactor code --- .../rule/engine/telemetry/TbMsgDeleteAttributes.java | 4 +++- .../rule/engine/telemetry/TbMsgDeleteAttributesTest.java | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java index 1a2fa21988..be82be5ecc 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java @@ -45,15 +45,17 @@ import java.util.stream.Collectors; public class TbMsgDeleteAttributes implements TbNode { TbMsgDeleteAttributesConfiguration config; + List keys; @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { this.config = TbNodeUtils.convert(configuration, TbMsgDeleteAttributesConfiguration.class); + this.keys = config.getKeys(); } @Override public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException { - List keysToDelete = config.getKeys().stream() + List keysToDelete = keys.stream() .map(keyPattern -> TbNodeUtils.processPattern(keyPattern, msg)) .distinct() .filter(StringUtils::isNotBlank) diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java index b763ba6215..cc5dcf4c85 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java @@ -116,4 +116,4 @@ public class TbMsgDeleteAttributesTest { verify(ctx, never()).tellFailure(any(), any()); verify(telemetryService, times(1)).deleteAndNotify(any(), any(), anyString(), anyList(), any()); } -} \ No newline at end of file +} From 12148c8a60fb77e147730cc31f817847e1624ec7 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Fri, 23 Sep 2022 16:39:53 +0300 Subject: [PATCH 10/24] modify callback --- .../actors/ruleChain/DefaultTbContext.java | 64 ++++++++++++++++++- .../rule/engine/api/TbContext.java | 7 +- .../AttributeDeleteNodeCallback.java | 45 +++++++++++++ .../AttributeUpdateNodeCallback.java | 44 +++++++++++++ .../telemetry/TbMsgDeleteAttributes.java | 9 +-- 5 files changed, 161 insertions(+), 8 deletions(-) create mode 100644 rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeDeleteNodeCallback.java create mode 100644 rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeUpdateNodeCallback.java diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 92fdfc6e48..9404c50160 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -17,8 +17,11 @@ package org.thingsboard.server.actors.ruleChain; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; import io.netty.channel.EventLoopGroup; import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.ListeningExecutor; import org.thingsboard.rule.engine.api.MailService; import org.thingsboard.rule.engine.api.RuleEngineAlarmService; @@ -49,6 +52,9 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.DataType; +import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.rule.RuleNode; @@ -84,6 +90,7 @@ import org.thingsboard.server.queue.TbQueueMsgMetadata; import org.thingsboard.server.service.script.RuleNodeJsScriptEngine; import java.util.Collections; +import java.util.List; import java.util.Set; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -360,6 +367,57 @@ class DefaultTbContext implements TbContext { return entityActionMsg(alarm, alarm.getId(), ruleNodeId, action, queueName, ruleChainId); } + public TbMsg attributeUpdateActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes) { + ObjectNode entityNode = JacksonUtil.newObjectNode(); + if (attributes != null) { + attributes.forEach(attributeKvEntry -> addKvEntry(entityNode, attributeKvEntry)); + } + return attributeActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_UPDATED, JacksonUtil.toString(entityNode)); + } + + public TbMsg attributeDeleteActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List keys) { + ObjectNode entityNode = JacksonUtil.newObjectNode(); + ArrayNode attrsArrayNode = entityNode.putArray("attributes"); + if (keys != null) { + keys.forEach(attrsArrayNode::add); + } + return attributeActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_DELETED, JacksonUtil.toString(entityNode)); + } + + private TbMsg attributeActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, String action, String msgData) { + RuleChainId ruleChainId = null; + String queueName = null; + if (EntityType.DEVICE.equals(originator.getEntityType())) { + DeviceId deviceId = new DeviceId(originator.getId()); + DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); + if (deviceProfile == null) { + log.warn("[{}] Device profile is null!", deviceId); + } else { + ruleChainId = deviceProfile.getDefaultRuleChainId(); + queueName = deviceProfile.getDefaultQueueName(); + } + } + TbMsgMetaData tbMsgMetaData = getActionMetaData(ruleNodeId); + tbMsgMetaData.putValue("scope", scope); + return entityActionMsg(originator, tbMsgMetaData, msgData, action, queueName, ruleChainId); + } + + private void addKvEntry(ObjectNode entityNode, KvEntry kvEntry) { + if (kvEntry.getDataType() == DataType.BOOLEAN) { + kvEntry.getBooleanValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); + } else if (kvEntry.getDataType() == DataType.DOUBLE) { + kvEntry.getDoubleValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); + } else if (kvEntry.getDataType() == DataType.LONG) { + kvEntry.getLongValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); + } else if (kvEntry.getDataType() == DataType.JSON) { + if (kvEntry.getJsonValue().isPresent()) { + entityNode.set(kvEntry.getKey(), JacksonUtil.valueToTree(kvEntry.getJsonValue().get())); + } + } else { + entityNode.put(kvEntry.getKey(), kvEntry.getValueAsString()); + } + } + @Override public void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId) { mainCtx.getClusterService().onEdgeEventUpdate(tenantId, edgeId); @@ -371,12 +429,16 @@ class DefaultTbContext implements TbContext { public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, String queueName, RuleChainId ruleChainId) { try { - return TbMsg.newMsg(queueName, action, id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), ruleChainId, null); + return entityActionMsg(id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), action, queueName, ruleChainId); } catch (JsonProcessingException | IllegalArgumentException e) { throw new RuntimeException("Failed to process " + id.getEntityType().name().toLowerCase() + " " + action + " msg: " + e); } } + public TbMsg entityActionMsg(I id, TbMsgMetaData msgMetaData, String msgData, String action, String queueName, RuleChainId ruleChainId) { + return TbMsg.newMsg(queueName, action, id, msgMetaData, msgData, ruleChainId, null); + } + @Override public RuleNodeId getSelfId() { return nodeCtx.getSelf().getId(); diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 1dac01f788..cbc4191275 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -29,10 +29,10 @@ import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.QueueId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.rule.RuleNode; @@ -59,6 +59,7 @@ import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.user.UserService; +import java.util.List; import java.util.Set; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -180,6 +181,10 @@ public interface TbContext { // TODO: Does this changes the message? TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action); + TbMsg attributeUpdateActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes); + + TbMsg attributeDeleteActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List keys); + void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId); /* diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeDeleteNodeCallback.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeDeleteNodeCallback.java new file mode 100644 index 0000000000..dc0c9c36f4 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeDeleteNodeCallback.java @@ -0,0 +1,45 @@ +/** + * Copyright © 2016-2022 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.telemetry; + +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.server.common.msg.TbMsg; + +import javax.annotation.Nullable; +import java.util.List; + +@Slf4j +public class AttributeDeleteNodeCallback extends TelemetryNodeCallback { + + private String scope; + private List keys; + + public AttributeDeleteNodeCallback(TbContext ctx, TbMsg msg, String scope, List keys) { + super(ctx, msg); + this.scope = scope; + this.keys = keys; + } + + @Override + public void onSuccess(@Nullable Void result) { + TbContext ctx = this.getCtx(); + TbMsg tbMsg = this.getMsg(); + ctx.enqueue(ctx.attributeDeleteActionMsg(tbMsg.getOriginator(), ctx.getSelfId(), scope, keys), + () -> ctx.tellSuccess(tbMsg), + throwable -> ctx.tellFailure(tbMsg, throwable)); + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeUpdateNodeCallback.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeUpdateNodeCallback.java new file mode 100644 index 0000000000..fbbe44738c --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeUpdateNodeCallback.java @@ -0,0 +1,44 @@ +/** + * Copyright © 2016-2022 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.telemetry; + +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.msg.TbMsg; + +import javax.annotation.Nullable; +import java.util.List; + +public class AttributeUpdateNodeCallback extends TelemetryNodeCallback { + + private String scope; + private List attributes; + + public AttributeUpdateNodeCallback(TbContext ctx, TbMsg msg, String scope, List attributes) { + super(ctx, msg); + this.scope = scope; + this.attributes = attributes; + } + + @Override + public void onSuccess(@Nullable Void result) { + TbContext ctx = this.getCtx(); + TbMsg tbMsg = this.getMsg(); + ctx.enqueue(ctx.attributeUpdateActionMsg(tbMsg.getOriginator(), ctx.getSelfId(), scope, attributes), + () -> ctx.tellSuccess(tbMsg), + throwable -> ctx.tellFailure(tbMsg, throwable)); + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java index be82be5ecc..32c6ce79fa 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java @@ -45,11 +45,13 @@ import java.util.stream.Collectors; public class TbMsgDeleteAttributes implements TbNode { TbMsgDeleteAttributesConfiguration config; + String scope; List keys; @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { this.config = TbNodeUtils.convert(configuration, TbMsgDeleteAttributesConfiguration.class); + this.scope = config.getScope(); this.keys = config.getKeys(); } @@ -63,12 +65,7 @@ public class TbMsgDeleteAttributes implements TbNode { if (keysToDelete.isEmpty()) { ctx.tellSuccess(msg); } else { - ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), config.getScope(), keysToDelete, new TelemetryNodeCallback(ctx, msg)); + ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), scope, keysToDelete, new AttributeDeleteNodeCallback(ctx, msg, scope, keys)); } } - - @Override - public void destroy() { - - } } From 87599ec5f9295e71addcea36e04778aa75f8c501 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Fri, 23 Sep 2022 16:49:17 +0300 Subject: [PATCH 11/24] check test --- .../rule/engine/telemetry/TbMsgDeleteAttributes.java | 2 +- .../rule/engine/telemetry/TbMsgDeleteAttributesTest.java | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java index 32c6ce79fa..0d9e9a1a75 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java @@ -65,7 +65,7 @@ public class TbMsgDeleteAttributes implements TbNode { if (keysToDelete.isEmpty()) { ctx.tellSuccess(msg); } else { - ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), scope, keysToDelete, new AttributeDeleteNodeCallback(ctx, msg, scope, keys)); + ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), scope, keysToDelete, new AttributeDeleteNodeCallback(ctx, msg, scope, keysToDelete)); } } } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java index cc5dcf4c85..22e745291f 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java @@ -111,8 +111,7 @@ public class TbMsgDeleteAttributesTest { TbMsg msg = TbMsg.newMsg("POST_ATTRIBUTES_REQUEST", deviceId, metaData, data, callback); node.onMsg(ctx, msg); - ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctx, times(1)).tellSuccess(newMsgCaptor.capture()); + verify(ctx, times(1)).attributeDeleteActionMsg(any(), any(), anyString(), anyList()); verify(ctx, never()).tellFailure(any(), any()); verify(telemetryService, times(1)).deleteAndNotify(any(), any(), anyString(), anyList(), any()); } From 5a6cf5c92fc9edea46c23dba58bf9b7029479ce4 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Mon, 26 Sep 2022 12:54:17 +0300 Subject: [PATCH 12/24] refactor code & test --- .../server/actors/ruleChain/DefaultTbContext.java | 10 +++++----- .../org/thingsboard/rule/engine/api/TbContext.java | 4 ++-- ...back.java => AttributesDeleteNodeCallback.java} | 6 +++--- ...back.java => AttributesUpdateNodeCallback.java} | 6 +++--- .../engine/telemetry/TbMsgDeleteAttributes.java | 2 +- .../telemetry/TbMsgDeleteAttributesTest.java | 14 +++++++++++++- 6 files changed, 27 insertions(+), 15 deletions(-) rename rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/{AttributeDeleteNodeCallback.java => AttributesDeleteNodeCallback.java} (81%) rename rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/{AttributeUpdateNodeCallback.java => AttributesUpdateNodeCallback.java} (81%) diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 9404c50160..9130b51fd6 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -367,24 +367,24 @@ class DefaultTbContext implements TbContext { return entityActionMsg(alarm, alarm.getId(), ruleNodeId, action, queueName, ruleChainId); } - public TbMsg attributeUpdateActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes) { + public TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes) { ObjectNode entityNode = JacksonUtil.newObjectNode(); if (attributes != null) { attributes.forEach(attributeKvEntry -> addKvEntry(entityNode, attributeKvEntry)); } - return attributeActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_UPDATED, JacksonUtil.toString(entityNode)); + return attributesActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_UPDATED, JacksonUtil.toString(entityNode)); } - public TbMsg attributeDeleteActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List keys) { + public TbMsg attributesDeletedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List keys) { ObjectNode entityNode = JacksonUtil.newObjectNode(); ArrayNode attrsArrayNode = entityNode.putArray("attributes"); if (keys != null) { keys.forEach(attrsArrayNode::add); } - return attributeActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_DELETED, JacksonUtil.toString(entityNode)); + return attributesActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_DELETED, JacksonUtil.toString(entityNode)); } - private TbMsg attributeActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, String action, String msgData) { + private TbMsg attributesActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, String action, String msgData) { RuleChainId ruleChainId = null; String queueName = null; if (EntityType.DEVICE.equals(originator.getEntityType())) { diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index cbc4191275..ec6febdbf4 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -181,9 +181,9 @@ public interface TbContext { // TODO: Does this changes the message? TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action); - TbMsg attributeUpdateActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes); + TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes); - TbMsg attributeDeleteActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List keys); + TbMsg attributesDeletedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List keys); void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeDeleteNodeCallback.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesDeleteNodeCallback.java similarity index 81% rename from rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeDeleteNodeCallback.java rename to rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesDeleteNodeCallback.java index dc0c9c36f4..57bfc5d129 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeDeleteNodeCallback.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesDeleteNodeCallback.java @@ -23,12 +23,12 @@ import javax.annotation.Nullable; import java.util.List; @Slf4j -public class AttributeDeleteNodeCallback extends TelemetryNodeCallback { +public class AttributesDeleteNodeCallback extends TelemetryNodeCallback { private String scope; private List keys; - public AttributeDeleteNodeCallback(TbContext ctx, TbMsg msg, String scope, List keys) { + public AttributesDeleteNodeCallback(TbContext ctx, TbMsg msg, String scope, List keys) { super(ctx, msg); this.scope = scope; this.keys = keys; @@ -38,7 +38,7 @@ public class AttributeDeleteNodeCallback extends TelemetryNodeCallback { public void onSuccess(@Nullable Void result) { TbContext ctx = this.getCtx(); TbMsg tbMsg = this.getMsg(); - ctx.enqueue(ctx.attributeDeleteActionMsg(tbMsg.getOriginator(), ctx.getSelfId(), scope, keys), + ctx.enqueue(ctx.attributesDeletedActionMsg(tbMsg.getOriginator(), ctx.getSelfId(), scope, keys), () -> ctx.tellSuccess(tbMsg), throwable -> ctx.tellFailure(tbMsg, throwable)); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeUpdateNodeCallback.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesUpdateNodeCallback.java similarity index 81% rename from rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeUpdateNodeCallback.java rename to rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesUpdateNodeCallback.java index fbbe44738c..f8ee4a29b1 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributeUpdateNodeCallback.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/AttributesUpdateNodeCallback.java @@ -22,12 +22,12 @@ import org.thingsboard.server.common.msg.TbMsg; import javax.annotation.Nullable; import java.util.List; -public class AttributeUpdateNodeCallback extends TelemetryNodeCallback { +public class AttributesUpdateNodeCallback extends TelemetryNodeCallback { private String scope; private List attributes; - public AttributeUpdateNodeCallback(TbContext ctx, TbMsg msg, String scope, List attributes) { + public AttributesUpdateNodeCallback(TbContext ctx, TbMsg msg, String scope, List attributes) { super(ctx, msg); this.scope = scope; this.attributes = attributes; @@ -37,7 +37,7 @@ public class AttributeUpdateNodeCallback extends TelemetryNodeCallback { public void onSuccess(@Nullable Void result) { TbContext ctx = this.getCtx(); TbMsg tbMsg = this.getMsg(); - ctx.enqueue(ctx.attributeUpdateActionMsg(tbMsg.getOriginator(), ctx.getSelfId(), scope, attributes), + ctx.enqueue(ctx.attributesUpdatedActionMsg(tbMsg.getOriginator(), ctx.getSelfId(), scope, attributes), () -> ctx.tellSuccess(tbMsg), throwable -> ctx.tellFailure(tbMsg, throwable)); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java index 0d9e9a1a75..9e8abbc6be 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java @@ -65,7 +65,7 @@ public class TbMsgDeleteAttributes implements TbNode { if (keysToDelete.isEmpty()) { ctx.tellSuccess(msg); } else { - ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), scope, keysToDelete, new AttributeDeleteNodeCallback(ctx, msg, scope, keysToDelete)); + ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), msg.getOriginator(), scope, keysToDelete, new AttributesDeleteNodeCallback(ctx, msg, scope, keysToDelete)); } } } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java index 22e745291f..8976a67318 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java @@ -20,7 +20,10 @@ import lombok.extern.slf4j.Slf4j; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.runner.RunWith; import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.junit.MockitoJUnitRunner; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNodeConfiguration; @@ -35,6 +38,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.function.Consumer; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; @@ -111,7 +115,15 @@ public class TbMsgDeleteAttributesTest { TbMsg msg = TbMsg.newMsg("POST_ATTRIBUTES_REQUEST", deviceId, metaData, data, callback); node.onMsg(ctx, msg); - verify(ctx, times(1)).attributeDeleteActionMsg(any(), any(), anyString(), anyList()); + ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); + ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + + verify(ctx, times(1)).enqueue(any(), successCaptor.capture(), failureCaptor.capture()); + successCaptor.getValue().run(); + 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()); } From 996aa0f5c27110cbdc1f577e23c229e6a50f9718 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Tue, 27 Sep 2022 10:52:12 +0300 Subject: [PATCH 13/24] refactor code --- .../rule/engine/telemetry/TbMsgAttributesNode.java | 12 +++++++----- .../rule/engine/telemetry/TbMsgDeleteAttributes.java | 4 +++- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java index 9f1c0d4e20..44cdc82c36 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java @@ -31,7 +31,7 @@ import org.thingsboard.server.common.msg.session.SessionMsgType; import org.thingsboard.server.common.transport.adaptor.JsonConverter; import java.util.ArrayList; -import java.util.Set; +import java.util.List; @Slf4j @RuleNode( @@ -39,7 +39,9 @@ import java.util.Set; name = "save attributes", configClazz = TbMsgAttributesNodeConfiguration.class, nodeDescription = "Saves attributes data", - nodeDetails = "Saves entity attributes based on configurable scope parameter. Expects messages with 'POST_ATTRIBUTES_REQUEST' message type", + 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 Success chain, otherwise, Failure chain is used.", uiResources = {"static/rulenode/rulenode-core-config.js"}, configDirective = "tbActionNodeAttributesConfig", icon = "file_upload" @@ -63,15 +65,15 @@ public class TbMsgAttributesNode implements TbNode { return; } String src = msg.getData(); - Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(src)); + List attributes = new ArrayList<>(JsonConverter.convertToAttributes(new JsonParser().parse(src))); String notifyDeviceStr = msg.getMetaData().getValue("notifyDevice"); ctx.getTelemetryService().saveAndNotify( ctx.getTenantId(), msg.getOriginator(), config.getScope(), - new ArrayList<>(attributes), + attributes, config.getNotifyDevice() || StringUtils.isEmpty(notifyDeviceStr) || Boolean.parseBoolean(notifyDeviceStr), - new TelemetryNodeCallback(ctx, msg) + new AttributesUpdateNodeCallback(ctx, msg, config.getScope(), attributes) ); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java index 9e8abbc6be..c5ead09150 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java @@ -37,7 +37,9 @@ import java.util.stream.Collectors; configClazz = TbMsgDeleteAttributesConfiguration.class, nodeDescription = "Delete attributes for Message Originator.", nodeDetails = "Attempt to remove attributes by selected keys. If msg originator doesn't have an attribute with " + - " a key selected in the configuration, it will be ignored.", + " a key selected in the configuration, it will be ignored. If delete operation is completed successfully, " + + " rule node will send the \"Attributes Deleted\" event to the root chain of the message originator and " + + " send the incoming message via Success chain, otherwise, Failure chain is used.", uiResources = {"static/rulenode/rulenode-core-config.js"}, configDirective = "tbActionNodeDeleteAttributesConfig", icon = "remove_circle" From c1a6addc2e8c10bf1c4eadaaddc036e63519c468 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Tue, 27 Sep 2022 15:01:14 +0300 Subject: [PATCH 14/24] private fields in node --- .../actors/ruleChain/DefaultTbContext.java | 20 +--------------- .../service/action/EntityActionService.java | 23 +++---------------- .../thingsboard/common/util/JacksonUtil.java | 18 +++++++++++++++ .../telemetry/TbMsgDeleteAttributes.java | 6 ++--- .../telemetry/TbMsgDeleteAttributesTest.java | 5 ---- 5 files changed, 25 insertions(+), 47 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 9130b51fd6..8c2c4a6e5e 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -53,8 +53,6 @@ import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; -import org.thingsboard.server.common.data.kv.DataType; -import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.rule.RuleNode; @@ -370,7 +368,7 @@ class DefaultTbContext implements TbContext { public TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes) { ObjectNode entityNode = JacksonUtil.newObjectNode(); if (attributes != null) { - attributes.forEach(attributeKvEntry -> addKvEntry(entityNode, attributeKvEntry)); + attributes.forEach(attributeKvEntry -> JacksonUtil.addKvEntry(entityNode, attributeKvEntry)); } return attributesActionMsg(originator, ruleNodeId, scope, DataConstants.ATTRIBUTES_UPDATED, JacksonUtil.toString(entityNode)); } @@ -402,22 +400,6 @@ class DefaultTbContext implements TbContext { return entityActionMsg(originator, tbMsgMetaData, msgData, action, queueName, ruleChainId); } - private void addKvEntry(ObjectNode entityNode, KvEntry kvEntry) { - if (kvEntry.getDataType() == DataType.BOOLEAN) { - kvEntry.getBooleanValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); - } else if (kvEntry.getDataType() == DataType.DOUBLE) { - kvEntry.getDoubleValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); - } else if (kvEntry.getDataType() == DataType.LONG) { - kvEntry.getLongValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); - } else if (kvEntry.getDataType() == DataType.JSON) { - if (kvEntry.getJsonValue().isPresent()) { - entityNode.set(kvEntry.getKey(), JacksonUtil.valueToTree(kvEntry.getJsonValue().get())); - } - } else { - entityNode.put(kvEntry.getKey(), kvEntry.getValueAsString()); - } - } - @Override public void onEdgeEventUpdate(TenantId tenantId, EdgeId edgeId) { mainCtx.getClusterService().onEdgeEventUpdate(tenantId, edgeId); diff --git a/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java b/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java index 79968187b6..e9f36768a1 100644 --- a/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java +++ b/application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java @@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityType; @@ -34,8 +35,6 @@ import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; -import org.thingsboard.server.common.data.kv.DataType; -import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; @@ -170,7 +169,7 @@ public class EntityActionService { metaData.putValue(DataConstants.SCOPE, scope); if (attributes != null) { for (AttributeKvEntry attr : attributes) { - addKvEntry(entityNode, attr); + JacksonUtil.addKvEntry(entityNode, attr); } } } else if (actionType == ActionType.ATTRIBUTES_DELETED) { @@ -246,27 +245,11 @@ public class EntityActionService { element.put("ts", entry.getKey()); ObjectNode values = element.putObject("values"); for (TsKvEntry tsKvEntry : entry.getValue()) { - addKvEntry(values, tsKvEntry); + JacksonUtil.addKvEntry(values, tsKvEntry); } result.add(element); } } } - private void addKvEntry(ObjectNode entityNode, KvEntry kvEntry) throws Exception { - if (kvEntry.getDataType() == DataType.BOOLEAN) { - kvEntry.getBooleanValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); - } else if (kvEntry.getDataType() == DataType.DOUBLE) { - kvEntry.getDoubleValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); - } else if (kvEntry.getDataType() == DataType.LONG) { - kvEntry.getLongValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); - } else if (kvEntry.getDataType() == DataType.JSON) { - if (kvEntry.getJsonValue().isPresent()) { - entityNode.set(kvEntry.getKey(), json.readTree(kvEntry.getJsonValue().get())); - } - } else { - entityNode.put(kvEntry.getKey(), kvEntry.getValueAsString()); - } - } - } diff --git a/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java b/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java index 9c4b68d3c0..6425f5cbea 100644 --- a/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java +++ b/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java @@ -24,6 +24,8 @@ import com.fasterxml.jackson.databind.SerializationFeature; import com.fasterxml.jackson.databind.json.JsonMapper; import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; +import org.thingsboard.server.common.data.kv.DataType; +import org.thingsboard.server.common.data.kv.KvEntry; import java.io.IOException; import java.util.ArrayList; @@ -212,4 +214,20 @@ public class JacksonUtil { } } } + + public static void addKvEntry(ObjectNode entityNode, KvEntry kvEntry) { + if (kvEntry.getDataType() == DataType.BOOLEAN) { + kvEntry.getBooleanValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); + } else if (kvEntry.getDataType() == DataType.DOUBLE) { + kvEntry.getDoubleValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); + } else if (kvEntry.getDataType() == DataType.LONG) { + kvEntry.getLongValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); + } else if (kvEntry.getDataType() == DataType.JSON) { + if (kvEntry.getJsonValue().isPresent()) { + entityNode.set(kvEntry.getKey(), JacksonUtil.valueToTree(kvEntry.getJsonValue().get())); + } + } else { + entityNode.put(kvEntry.getKey(), kvEntry.getValueAsString()); + } + } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java index c5ead09150..e8839291a0 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributes.java @@ -46,9 +46,9 @@ import java.util.stream.Collectors; ) public class TbMsgDeleteAttributes implements TbNode { - TbMsgDeleteAttributesConfiguration config; - String scope; - List keys; + private TbMsgDeleteAttributesConfiguration config; + private String scope; + private List keys; @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java index 8976a67318..8972a3aa1f 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgDeleteAttributesTest.java @@ -91,11 +91,6 @@ public class TbMsgDeleteAttributesTest { node.destroy(); } - @Test - void givenDefaultConfig_whenInit_thenOK() { - assertThat(node.config).isEqualTo(config); - } - @Test void givenDefaultConfig_whenVerify_thenOK() { TbMsgDeleteAttributesConfiguration defaultConfig = new TbMsgDeleteAttributesConfiguration().defaultConfiguration(); From ef875a97b49902d94f1e127fb3bc493ce6082a52 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Tue, 27 Sep 2022 15:36:14 +0300 Subject: [PATCH 15/24] refactor entity action methods in TbContext & fix device profile node exception --- .../actors/ruleChain/DefaultTbContext.java | 55 +++++++------------ .../engine/profile/TbDeviceProfileNode.java | 10 +++- 2 files changed, 28 insertions(+), 37 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 8c2c4a6e5e..0006940406 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -331,18 +331,11 @@ class DefaultTbContext implements TbContext { } public TbMsg deviceCreatedMsg(Device device, RuleNodeId ruleNodeId) { - RuleChainId ruleChainId = null; - String queueName = null; + DeviceProfile deviceProfile = null; if (device.getDeviceProfileId() != null) { - DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().find(device.getDeviceProfileId()); - if (deviceProfile == null) { - log.warn("[{}] Device profile is null!", device.getDeviceProfileId()); - } else { - ruleChainId = deviceProfile.getDefaultRuleChainId(); - queueName = deviceProfile.getDefaultQueueName(); - } + deviceProfile = mainCtx.getDeviceProfileCache().find(device.getDeviceProfileId()); } - return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, queueName, ruleChainId); + return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, deviceProfile); } public TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId) { @@ -350,19 +343,12 @@ class DefaultTbContext implements TbContext { } public TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action) { - RuleChainId ruleChainId = null; - String queueName = null; + DeviceProfile deviceProfile = null; if (EntityType.DEVICE.equals(alarm.getOriginator().getEntityType())) { DeviceId deviceId = new DeviceId(alarm.getOriginator().getId()); - DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); - if (deviceProfile == null) { - log.warn("[{}] Device profile is null!", deviceId); - } else { - ruleChainId = deviceProfile.getDefaultRuleChainId(); - queueName = deviceProfile.getDefaultQueueName(); - } + deviceProfile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); } - return entityActionMsg(alarm, alarm.getId(), ruleNodeId, action, queueName, ruleChainId); + return entityActionMsg(alarm, alarm.getOriginator(), ruleNodeId, action, deviceProfile); } public TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes) { @@ -383,21 +369,14 @@ class DefaultTbContext implements TbContext { } private TbMsg attributesActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, String action, String msgData) { - RuleChainId ruleChainId = null; - String queueName = null; + DeviceProfile deviceProfile = null; if (EntityType.DEVICE.equals(originator.getEntityType())) { DeviceId deviceId = new DeviceId(originator.getId()); - DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); - if (deviceProfile == null) { - log.warn("[{}] Device profile is null!", deviceId); - } else { - ruleChainId = deviceProfile.getDefaultRuleChainId(); - queueName = deviceProfile.getDefaultQueueName(); - } + deviceProfile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); } TbMsgMetaData tbMsgMetaData = getActionMetaData(ruleNodeId); tbMsgMetaData.putValue("scope", scope); - return entityActionMsg(originator, tbMsgMetaData, msgData, action, queueName, ruleChainId); + return entityActionMsg(originator, tbMsgMetaData, msgData, action, deviceProfile); } @Override @@ -406,18 +385,26 @@ class DefaultTbContext implements TbContext { } public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action) { - return entityActionMsg(entity, id, ruleNodeId, action, null, null); + return entityActionMsg(entity, id, ruleNodeId, action, null); } - public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, String queueName, RuleChainId ruleChainId) { + public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, DeviceProfile deviceProfile) { try { - return entityActionMsg(id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), action, queueName, ruleChainId); + return entityActionMsg(id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), action, deviceProfile); } catch (JsonProcessingException | IllegalArgumentException e) { throw new RuntimeException("Failed to process " + id.getEntityType().name().toLowerCase() + " " + action + " msg: " + e); } } - public TbMsg entityActionMsg(I id, TbMsgMetaData msgMetaData, String msgData, String action, String queueName, RuleChainId ruleChainId) { + public TbMsg entityActionMsg(I id, TbMsgMetaData msgMetaData, String msgData, String action, DeviceProfile deviceProfile) { + RuleChainId ruleChainId = null; + String queueName = null; + if (deviceProfile == null) { + log.warn("Device profile is null!"); + } else { + ruleChainId = deviceProfile.getDefaultRuleChainId(); + queueName = deviceProfile.getDefaultQueueName(); + } return TbMsg.newMsg(queueName, action, id, msgMetaData, msgData, ruleChainId, null); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java index 9b9974cbb9..5a8fd7983f 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java @@ -210,9 +210,13 @@ public class TbDeviceProfileNode implements TbNode { DeviceState deviceState = deviceStates.get(deviceId); if (deviceState != null) { DeviceProfileId currentProfileId = deviceState.getProfileId(); - Device device = JacksonUtil.fromString(deviceJson, Device.class); - if (!currentProfileId.equals(device.getDeviceProfileId())) { - removeDeviceState(deviceId); + try { + Device device = JacksonUtil.fromString(deviceJson, Device.class); + if (!currentProfileId.equals(device.getDeviceProfileId())) { + removeDeviceState(deviceId); + } + } catch (IllegalArgumentException ex) { + log.trace("[{}] Received device update notification with non-device msg body: [{}][{}]", ctx.getSelfId(), deviceId, ex.getMessage()); } } } From dbe7a2b9142ab9d02ee325a1369d7c7a6e34d5c4 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Wed, 28 Sep 2022 16:09:47 +0300 Subject: [PATCH 16/24] fixed logging issue in Device profile node --- .../thingsboard/rule/engine/profile/TbDeviceProfileNode.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java index 5a8fd7983f..d68b29734c 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java @@ -215,8 +215,8 @@ public class TbDeviceProfileNode implements TbNode { if (!currentProfileId.equals(device.getDeviceProfileId())) { removeDeviceState(deviceId); } - } catch (IllegalArgumentException ex) { - log.trace("[{}] Received device update notification with non-device msg body: [{}][{}]", ctx.getSelfId(), deviceId, ex.getMessage()); + } catch (IllegalArgumentException e) { + log.debug("[{}] Received device update notification with non-device msg body: [{}][{}]", ctx.getSelfId(), deviceId, e); } } } From c3b3a1ab3acd89a277c1eec9f3ba7947c6bcd0d5 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Wed, 5 Oct 2022 14:22:29 +0300 Subject: [PATCH 17/24] add INACTIVITY_EVENT --- .../rule/engine/action/TbCopyAttributesToEntityViewNode.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java index fd69303501..5f4322c62f 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java @@ -74,6 +74,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { if (DataConstants.ATTRIBUTES_UPDATED.equals(msg.getType()) || DataConstants.ATTRIBUTES_DELETED.equals(msg.getType()) || DataConstants.ACTIVITY_EVENT.equals(msg.getType()) || + DataConstants.INACTIVITY_EVENT.equals(msg.getType()) || SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msg.getType())) { if (!msg.getMetaData().getData().isEmpty()) { long now = System.currentTimeMillis(); @@ -91,6 +92,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { if ((endTime != 0 && endTime > now && startTime < now) || (endTime == 0 && startTime < now)) { if (DataConstants.ATTRIBUTES_UPDATED.equals(msg.getType()) || DataConstants.ACTIVITY_EVENT.equals(msg.getType()) || + DataConstants.INACTIVITY_EVENT.equals(msg.getType()) || SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msg.getType())) { Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); List filteredAttributes = From 1cfb0a03ab1216975e4a7130ab60ee903196499b Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Thu, 6 Oct 2022 12:34:27 +0300 Subject: [PATCH 18/24] refactor code --- .../TbCopyAttributesToEntityViewNode.java | 46 ++++++++++--------- 1 file changed, 24 insertions(+), 22 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java index 5f4322c62f..9921e4ebdb 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java @@ -21,6 +21,7 @@ import com.google.gson.JsonElement; import com.google.gson.JsonParser; import com.google.gson.JsonPrimitive; import lombok.extern.slf4j.Slf4j; +import org.jetbrains.annotations.NotNull; import org.thingsboard.common.util.CollectionsUtil; import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; @@ -90,26 +91,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { long startTime = entityView.getStartTimeMs(); long endTime = entityView.getEndTimeMs(); if ((endTime != 0 && endTime > now && startTime < now) || (endTime == 0 && startTime < now)) { - if (DataConstants.ATTRIBUTES_UPDATED.equals(msg.getType()) || - DataConstants.ACTIVITY_EVENT.equals(msg.getType()) || - DataConstants.INACTIVITY_EVENT.equals(msg.getType()) || - SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msg.getType())) { - Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); - List filteredAttributes = - attributes.stream().filter(attr -> attributeContainsInEntityView(scope, attr.getKey(), entityView)).collect(Collectors.toList()); - ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes, - new FutureCallback() { - @Override - public void onSuccess(@Nullable Void result) { - transformAndTellNext(ctx, msg, entityView); - } - - @Override - public void onFailure(Throwable t) { - ctx.tellFailure(msg, t); - } - }); - } else if (DataConstants.ATTRIBUTES_DELETED.equals(msg.getType())) { + if (DataConstants.ATTRIBUTES_DELETED.equals(msg.getType())) { List attributes = new ArrayList<>(); for (JsonElement element : new JsonParser().parse(msg.getData()).getAsJsonObject().get("attributes").getAsJsonArray()) { if (element.isJsonPrimitive()) { @@ -122,9 +104,14 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { List filteredAttributes = attributes.stream().filter(attr -> attributeContainsInEntityView(scope, attr, entityView)).collect(Collectors.toList()); if (!filteredAttributes.isEmpty()) { - ctx.getAttributesService().removeAll(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes); - transformAndTellNext(ctx, msg, entityView); + ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes, getFutureCallback(ctx, msg, entityView)); } + } else { + Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); + List filteredAttributes = + attributes.stream().filter(attr -> attributeContainsInEntityView(scope, attr.getKey(), entityView)).collect(Collectors.toList()); + ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes, + getFutureCallback(ctx, msg, entityView)); } } } @@ -139,6 +126,21 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { } } + @NotNull + private FutureCallback getFutureCallback(TbContext ctx, TbMsg msg, EntityView entityView) { + return new FutureCallback() { + @Override + public void onSuccess(@Nullable Void result) { + transformAndTellNext(ctx, msg, entityView); + } + + @Override + public void onFailure(Throwable t) { + ctx.tellFailure(msg, t); + } + }; + } + private void transformAndTellNext(TbContext ctx, TbMsg msg, EntityView entityView) { ctx.enqueueForTellNext(ctx.newMsg(msg.getQueueName(), msg.getType(), entityView.getId(), msg.getCustomerId(), msg.getMetaData(), msg.getData()), SUCCESS); } From aba19f171e3d3071c554c3e6ad8f06ff6f5973e2 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Thu, 6 Oct 2022 12:37:00 +0300 Subject: [PATCH 19/24] refactor code refactor deleteAndNotify callback --- .../rule/engine/action/TbCopyAttributesToEntityViewNode.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java index 9921e4ebdb..b9f4b7085e 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java @@ -104,7 +104,8 @@ public class TbCopyAttributesToEntityViewNode implements TbNode { List filteredAttributes = attributes.stream().filter(attr -> attributeContainsInEntityView(scope, attr, entityView)).collect(Collectors.toList()); if (!filteredAttributes.isEmpty()) { - ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes, getFutureCallback(ctx, msg, entityView)); + ctx.getTelemetryService().deleteAndNotify(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes, + getFutureCallback(ctx, msg, entityView)); } } else { Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData())); From 6e34747b959725beea67fb0b20ff9cc7bc231369 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Thu, 6 Oct 2022 19:42:30 +0300 Subject: [PATCH 20/24] added HasRuleEngineProfile interface & refactored DefaultTbContext --- .../actors/ruleChain/DefaultTbContext.java | 71 ++++++------------- .../server/common/data/DeviceProfile.java | 4 +- .../common/data/HasRuleEngineProfile.java | 26 +++++++ .../common/data/asset/AssetProfile.java | 3 +- 4 files changed, 52 insertions(+), 52 deletions(-) create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/HasRuleEngineProfile.java diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 283868a4d2..920f71346d 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -42,6 +42,7 @@ import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.HasRuleEngineProfile; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.alarm.Alarm; @@ -339,7 +340,7 @@ class DefaultTbContext implements TbContext { if (device.getDeviceProfileId() != null) { deviceProfile = mainCtx.getDeviceProfileCache().find(device.getDeviceProfileId()); } - return deviceActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, deviceProfile); + return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, deviceProfile); } public TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId) { @@ -347,20 +348,19 @@ class DefaultTbContext implements TbContext { if (asset.getAssetProfileId() != null) { assetProfile = mainCtx.getAssetProfileCache().find(asset.getAssetProfileId()); } - return assetActionMsg(asset, asset.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, assetProfile); + return entityActionMsg(asset, asset.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, assetProfile); } public TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action) { + HasRuleEngineProfile profile = null; if (EntityType.DEVICE.equals(alarm.getOriginator().getEntityType())) { DeviceId deviceId = new DeviceId(alarm.getOriginator().getId()); - DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); - return deviceActionMsg(alarm, alarm.getOriginator(), ruleNodeId, action, deviceProfile); + profile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); } else if (EntityType.ASSET.equals(alarm.getOriginator().getEntityType())) { AssetId assetId = new AssetId(alarm.getOriginator().getId()); - AssetProfile assetProfile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId); - return assetActionMsg(alarm, alarm.getOriginator(), ruleNodeId, action, assetProfile); + profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId); } - return entityActionMsg(alarm, alarm.getOriginator(), ruleNodeId, action); + return entityActionMsg(alarm, alarm.getId(), ruleNodeId, action, profile); } public TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes) { @@ -383,16 +383,15 @@ class DefaultTbContext implements TbContext { private TbMsg attributesActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, String action, String msgData) { TbMsgMetaData tbMsgMetaData = getActionMetaData(ruleNodeId); tbMsgMetaData.putValue("scope", scope); + HasRuleEngineProfile profile = null; if (EntityType.DEVICE.equals(originator.getEntityType())) { DeviceId deviceId = new DeviceId(originator.getId()); - DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); - return deviceActionMsg(originator, tbMsgMetaData, msgData, action, deviceProfile); + profile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId); } else if (EntityType.ASSET.equals(originator.getEntityType())) { AssetId assetId = new AssetId(originator.getId()); - AssetProfile assetProfile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId); - return assetActionMsg(originator, tbMsgMetaData, msgData, action, assetProfile); + profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId); } - return entityActionMsg(originator, tbMsgMetaData, msgData, action, null, null); + return entityActionMsg(originator, tbMsgMetaData, msgData, action, profile); } @Override @@ -400,52 +399,26 @@ class DefaultTbContext implements TbContext { mainCtx.getClusterService().onEdgeEventUpdate(tenantId, edgeId); } - public TbMsg deviceActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, DeviceProfile deviceProfile) { - try { - return deviceActionMsg(id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), action, deviceProfile); - } catch (JsonProcessingException | IllegalArgumentException e) { - throw new RuntimeException("Failed to process " + id.getEntityType().name().toLowerCase() + " " + action + " msg: " + e); - } + public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action) { + return entityActionMsg(entity, id, ruleNodeId, action, null); } - public TbMsg assetActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, AssetProfile assetProfile) { + public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, K profile) { try { - return assetActionMsg(id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), action, assetProfile); + return entityActionMsg(id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), action, profile); } catch (JsonProcessingException | IllegalArgumentException e) { throw new RuntimeException("Failed to process " + id.getEntityType().name().toLowerCase() + " " + action + " msg: " + e); } } - public TbMsg assetActionMsg(I id, TbMsgMetaData msgMetaData, String msgData, String action, AssetProfile assetProfile) { - RuleChainId ruleChainId = null; - String queueName = null; - if (assetProfile != null) { - ruleChainId = assetProfile.getDefaultRuleChainId(); - queueName = assetProfile.getDefaultQueueName(); + private TbMsg entityActionMsg(I id, TbMsgMetaData msgMetaData, String msgData, String action, K profile) { + String defaultQueueName = null; + RuleChainId defaultRuleChainId = null; + if (profile != null) { + defaultQueueName = profile.getDefaultQueueName(); + defaultRuleChainId = profile.getDefaultRuleChainId(); } - return entityActionMsg(id, msgMetaData, msgData, action, ruleChainId, queueName); - } - - public TbMsg deviceActionMsg(I id, TbMsgMetaData msgMetaData, String msgData, String action, DeviceProfile deviceProfile) { - RuleChainId ruleChainId = null; - String queueName = null; - if (deviceProfile != null) { - ruleChainId = deviceProfile.getDefaultRuleChainId(); - queueName = deviceProfile.getDefaultQueueName(); - } - return entityActionMsg(id, msgMetaData, msgData, action, ruleChainId, queueName); - } - - public TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action) { - try { - return entityActionMsg(id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), action, null, null); - } catch (JsonProcessingException | IllegalArgumentException e) { - throw new RuntimeException("Failed to process " + id.getEntityType().name().toLowerCase() + " " + action + " msg: " + e); - } - } - - private TbMsg entityActionMsg(I id, TbMsgMetaData msgMetaData, String msgData, String action, RuleChainId ruleChainId, String queueName) { - return TbMsg.newMsg(queueName, action, id, msgMetaData, msgData, ruleChainId, null); + return TbMsg.newMsg(defaultQueueName, action, id, msgMetaData, msgData, defaultRuleChainId, null); } @Override diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java index da8edc695e..a66f3a40d9 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java @@ -45,7 +45,7 @@ import static org.thingsboard.server.common.data.SearchTextBasedWithAdditionalIn @ToString(exclude = {"image", "profileDataBytes"}) @EqualsAndHashCode(callSuper = true) @Slf4j -public class DeviceProfile extends SearchTextBased implements HasName, HasTenantId, HasOtaPackage, ExportableEntity { +public class DeviceProfile extends SearchTextBased implements HasName, HasTenantId, HasOtaPackage, HasRuleEngineProfile, ExportableEntity { private static final long serialVersionUID = 6998485460273302018L; @@ -141,7 +141,7 @@ public class DeviceProfile extends SearchTextBased implements H } @ApiModelProperty(position = 5, value = "Used to mark the default profile. Default profile is used when the device profile is not specified during device creation.") - public boolean isDefault(){ + public boolean isDefault() { return isDefault; } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/HasRuleEngineProfile.java b/common/data/src/main/java/org/thingsboard/server/common/data/HasRuleEngineProfile.java new file mode 100644 index 0000000000..450a2235a7 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/HasRuleEngineProfile.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2022 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.server.common.data; + +import org.thingsboard.server.common.data.id.RuleChainId; + +public interface HasRuleEngineProfile { + + RuleChainId getDefaultRuleChainId(); + + String getDefaultQueueName(); + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/asset/AssetProfile.java b/common/data/src/main/java/org/thingsboard/server/common/data/asset/AssetProfile.java index 1bd9d9c6a5..01657feed6 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/asset/AssetProfile.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/asset/AssetProfile.java @@ -23,6 +23,7 @@ import lombok.ToString; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.ExportableEntity; import org.thingsboard.server.common.data.HasName; +import org.thingsboard.server.common.data.HasRuleEngineProfile; import org.thingsboard.server.common.data.HasTenantId; import org.thingsboard.server.common.data.SearchTextBased; import org.thingsboard.server.common.data.id.AssetProfileId; @@ -37,7 +38,7 @@ import org.thingsboard.server.common.data.validation.NoXss; @ToString(exclude = {"image"}) @EqualsAndHashCode(callSuper = true) @Slf4j -public class AssetProfile extends SearchTextBased implements HasName, HasTenantId, ExportableEntity { +public class AssetProfile extends SearchTextBased implements HasName, HasTenantId, HasRuleEngineProfile, ExportableEntity { private static final long serialVersionUID = 6998485460273302018L; From 673e2cb5be743c3dd8c164a1ed415036fe49ca88 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Thu, 6 Oct 2022 19:48:44 +0300 Subject: [PATCH 21/24] fix typo in alarmActionMsg --- .../thingsboard/server/actors/ruleChain/DefaultTbContext.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 920f71346d..62d4829085 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -360,7 +360,7 @@ class DefaultTbContext implements TbContext { AssetId assetId = new AssetId(alarm.getOriginator().getId()); profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId); } - return entityActionMsg(alarm, alarm.getId(), ruleNodeId, action, profile); + return entityActionMsg(alarm, alarm.getOriginator(), ruleNodeId, action, profile); } public TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes) { From 57eab8d73c86f704603ba7dc223eb76a6b614074 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Fri, 7 Oct 2022 12:36:10 +0300 Subject: [PATCH 22/24] add validation for empty list of attributes in save attributes node --- .../rule/engine/telemetry/TbMsgAttributesNode.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java index 44cdc82c36..b810774832 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java @@ -66,6 +66,10 @@ public class TbMsgAttributesNode implements TbNode { } String src = msg.getData(); List attributes = new ArrayList<>(JsonConverter.convertToAttributes(new JsonParser().parse(src))); + if (attributes.isEmpty()) { + ctx.tellSuccess(msg); + return; + } String notifyDeviceStr = msg.getMetaData().getValue("notifyDevice"); ctx.getTelemetryService().saveAndNotify( ctx.getTenantId(), From d62ee97213ed9cc336d937ce151aeb22f384ed23 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Fri, 7 Oct 2022 13:09:33 +0300 Subject: [PATCH 23/24] add JacksonUtil.toJsonNode --- .../src/main/java/org/thingsboard/common/util/JacksonUtil.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java b/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java index 6425f5cbea..98141e2a08 100644 --- a/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java +++ b/common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java @@ -224,7 +224,7 @@ public class JacksonUtil { kvEntry.getLongValue().ifPresent(value -> entityNode.put(kvEntry.getKey(), value)); } else if (kvEntry.getDataType() == DataType.JSON) { if (kvEntry.getJsonValue().isPresent()) { - entityNode.set(kvEntry.getKey(), JacksonUtil.valueToTree(kvEntry.getJsonValue().get())); + entityNode.set(kvEntry.getKey(), JacksonUtil.toJsonNode(kvEntry.getJsonValue().get())); } } else { entityNode.put(kvEntry.getKey(), kvEntry.getValueAsString()); From 35960d018e7df10b32d60acb60a625e173343bb6 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Wed, 19 Oct 2022 18:33:50 +0300 Subject: [PATCH 24/24] SQL batch sort enabled by default; additional deadlock logging --- .../src/main/resources/thingsboard.yml | 2 +- .../server/common/data/StringUtils.java | 10 +++++ .../server/dao/aspect/SqlDaoCallsAspect.java | 45 +++++++++++++++---- .../AttributeKvInsertRepository.java | 4 +- .../dao/sql/attributes/JpaAttributeDao.java | 2 +- .../SqlAttributesInsertRepository.java | 2 + .../dao/sql/event/EventInsertRepository.java | 4 +- .../server/dao/sql/event/JpaBaseEventDao.java | 2 +- .../dao/sqlts/AbstractSqlTimeseriesDao.java | 2 +- .../dao/sqlts/SqlTimeseriesLatestDao.java | 2 +- .../sql/SqlLatestInsertTsRepository.java | 2 + 11 files changed, 61 insertions(+), 16 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index f5ac40afd5..9f791cee9b 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -268,7 +268,7 @@ sql: audit_logs: partition_size: "${SQL_AUDIT_LOGS_PARTITION_SIZE_HOURS:168}" # Default value - 1 week # Specify whether to sort entities before batch update. Should be enabled for cluster mode to avoid deadlocks - batch_sort: "${SQL_BATCH_SORT:false}" + batch_sort: "${SQL_BATCH_SORT:true}" # Specify whether to remove null characters from strValue of attributes and timeseries before insert remove_null_chars: "${SQL_REMOVE_NULL_CHARS:true}" # Specify whether to log database queries and their parameters generated by entity query repository diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/StringUtils.java b/common/data/src/main/java/org/thingsboard/server/common/data/StringUtils.java index 18d3dfff1e..9acbb15646 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/StringUtils.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/StringUtils.java @@ -146,6 +146,16 @@ public class StringUtils { return org.apache.commons.lang3.StringUtils.substringAfterLast(str, sep); } + public static boolean containedByAny(String searchString, String... strings) { + if (searchString == null) return false; + for (String string : strings) { + if (string != null && string.contains(searchString)) { + return true; + } + } + return false; + } + public static String randomNumeric(int length) { return RandomStringUtils.randomNumeric(length); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/aspect/SqlDaoCallsAspect.java b/dao/src/main/java/org/thingsboard/server/dao/aspect/SqlDaoCallsAspect.java index ec8c97e19f..9c4ca8c536 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/aspect/SqlDaoCallsAspect.java +++ b/dao/src/main/java/org/thingsboard/server/dao/aspect/SqlDaoCallsAspect.java @@ -21,7 +21,6 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.ArrayUtils; -import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.exception.ExceptionUtils; import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.annotation.Around; @@ -29,9 +28,11 @@ import org.aspectj.lang.annotation.Aspect; import org.aspectj.lang.reflect.MethodSignature; import org.checkerframework.checker.nullness.qual.Nullable; import org.hibernate.exception.JDBCConnectionException; +import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.id.TenantId; import java.util.Arrays; @@ -46,6 +47,8 @@ import java.util.concurrent.ConcurrentMap; import java.util.function.Consumer; import java.util.stream.Collectors; +import static org.apache.commons.lang3.StringUtils.join; + @Aspect @ConditionalOnProperty(prefix = "sql", value = "log_tenant_stats", havingValue = "true") @Component @@ -55,6 +58,12 @@ public class SqlDaoCallsAspect { private final Set invalidTenantDbCallMethods = ConcurrentHashMap.newKeySet(); private final ConcurrentMap statsMap = new ConcurrentHashMap<>(); + @Value("${sql.batch_sort:true}") + private boolean batchSortEnabled; + + private static final String DEADLOCK_DETECTED_ERROR = "deadlock detected"; + + @Scheduled(initialDelayString = "${sql.log_tenant_stats_interval_ms:60000}", fixedDelayString = "${sql.log_tenant_stats_interval_ms:60000}") public void printStats() { @@ -135,7 +144,7 @@ public class SqlDaoCallsAspect { @SuppressWarnings({"rawtypes", "unchecked"}) @Around("@within(org.thingsboard.server.dao.util.SqlDao)") - public Object logExecutionTime(ProceedingJoinPoint joinPoint) throws Throwable { + public Object handleSqlCall(ProceedingJoinPoint joinPoint) throws Throwable { MethodSignature signature = (MethodSignature) joinPoint.getSignature(); var methodName = signature.toShortString(); if (invalidTenantDbCallMethods.contains(methodName)) { @@ -155,29 +164,47 @@ public class SqlDaoCallsAspect { new FutureCallback<>() { @Override public void onSuccess(@Nullable Object result) { - logTenantMethodExecution(tenantId, methodName, true, startTime, null); + reportSuccessfulMethodExecution(tenantId, methodName, startTime); } @Override public void onFailure(Throwable t) { - logTenantMethodExecution(tenantId, methodName, false, startTime, t); + reportFailedMethodExecution(tenantId, methodName, startTime, t, joinPoint); } }, MoreExecutors.directExecutor()); } else { - logTenantMethodExecution(tenantId, methodName, true, startTime, null); + reportSuccessfulMethodExecution(tenantId, methodName, startTime); } return result; } catch (Throwable t) { - logTenantMethodExecution(tenantId, methodName, false, startTime, t); + reportFailedMethodExecution(tenantId, methodName, startTime, t, joinPoint); throw t; } } - private void logTenantMethodExecution(TenantId tenantId, String method, boolean success, long startTime, Throwable t) { - if (!success && ExceptionUtils.indexOfThrowable(t, JDBCConnectionException.class) >= 0) { - return; + private void reportFailedMethodExecution(TenantId tenantId, String method, long startTime, Throwable t, ProceedingJoinPoint joinPoint) { + if (t != null) { + if (ExceptionUtils.indexOfThrowable(t, JDBCConnectionException.class) >= 0) { + return; + } + if (StringUtils.containedByAny(DEADLOCK_DETECTED_ERROR, ExceptionUtils.getRootCauseMessage(t), ExceptionUtils.getMessage(t))) { + if (!batchSortEnabled) { + log.warn("Deadlock was detected for method {} (tenant: {}). You might need to enable 'sql.batch_sort' option.", method, tenantId); + } else { + log.error("Deadlock was detected for method {} (tenant: {}). Arguments passed: \n{}\n The error: ", + method, tenantId, join(joinPoint.getArgs(), System.lineSeparator()), t); + } + } } + reportMethodExecution(tenantId, method, false, startTime); + } + + private void reportSuccessfulMethodExecution(TenantId tenantId, String method, long startTime) { + reportMethodExecution(tenantId, method, true, startTime); + } + + private void reportMethodExecution(TenantId tenantId, String method, boolean success, long startTime) { statsMap.computeIfAbsent(tenantId, DbCallStats::new) .onMethodCall(method, success, System.currentTimeMillis() - startTime); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/AttributeKvInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/AttributeKvInsertRepository.java index 472652e357..229d4f5740 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/AttributeKvInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/AttributeKvInsertRepository.java @@ -25,6 +25,7 @@ import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallbackWithoutResult; import org.springframework.transaction.support.TransactionTemplate; import org.thingsboard.server.dao.model.sql.AttributeKvEntity; +import org.thingsboard.server.dao.util.SqlDao; import java.sql.PreparedStatement; import java.sql.SQLException; @@ -35,6 +36,7 @@ import java.util.regex.Pattern; @Repository @Slf4j +@SqlDao public abstract class AttributeKvInsertRepository { private static final ThreadLocal PATTERN_THREAD_LOCAL = ThreadLocal.withInitial(() -> Pattern.compile(String.valueOf(Character.MIN_VALUE))); @@ -58,7 +60,7 @@ public abstract class AttributeKvInsertRepository { @Value("${sql.remove_null_chars:true}") private boolean removeNullChars; - protected void saveOrUpdate(List entities) { + public void saveOrUpdate(List entities) { transactionTemplate.execute(new TransactionCallbackWithoutResult() { @Override protected void doInTransactionWithoutResult(TransactionStatus status) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java index e040789f65..99ee1a9b00 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java @@ -78,7 +78,7 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl @Value("${sql.attributes.batch_threads:4}") private int batchThreads; - @Value("${sql.batch_sort:false}") + @Value("${sql.batch_sort:true}") private boolean batchSortEnabled; private TbSqlBlockingQueueWrapper queue; diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/SqlAttributesInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/SqlAttributesInsertRepository.java index e18f5178a8..adac4b4892 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/SqlAttributesInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/SqlAttributesInsertRepository.java @@ -17,9 +17,11 @@ package org.thingsboard.server.dao.sql.attributes; import org.springframework.stereotype.Repository; import org.springframework.transaction.annotation.Transactional; +import org.thingsboard.server.dao.util.SqlDao; @Repository @Transactional +@SqlDao public class SqlAttributesInsertRepository extends AttributeKvInsertRepository { } \ No newline at end of file diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java index 85220ddf9f..cb91a5e674 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java @@ -31,6 +31,7 @@ import org.thingsboard.server.common.data.event.LifecycleEvent; import org.thingsboard.server.common.data.event.RuleChainDebugEvent; import org.thingsboard.server.common.data.event.RuleNodeDebugEvent; import org.thingsboard.server.common.data.event.StatisticsEvent; +import org.thingsboard.server.dao.util.SqlDao; import javax.annotation.PostConstruct; import java.sql.PreparedStatement; @@ -45,6 +46,7 @@ import java.util.stream.Collectors; @Repository @Transactional +@SqlDao public class EventInsertRepository { private static final ThreadLocal PATTERN_THREAD_LOCAL = ThreadLocal.withInitial(() -> Pattern.compile(String.valueOf(Character.MIN_VALUE))); @@ -81,7 +83,7 @@ public class EventInsertRepository { "VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING;"); } - protected void save(List entities) { + public void save(List entities) { Map> eventsByType = entities.stream().collect(Collectors.groupingBy(Event::getType, Collectors.toList())); transactionTemplate.execute(new TransactionCallbackWithoutResult() { @Override diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java index 62ee11b828..4332319f0c 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java @@ -106,7 +106,7 @@ public class JpaBaseEventDao implements EventDao { @Value("${sql.events.batch_threads:3}") private int batchThreads; - @Value("${sql.batch_sort:false}") + @Value("${sql.batch_sort:true}") private boolean batchSortEnabled; private TbSqlBlockingQueueWrapper queue; diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java index 3de21e41bf..8439a4a063 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java @@ -63,7 +63,7 @@ public abstract class AbstractSqlTimeseriesDao extends BaseAbstractSqlTimeseries @Value("${sql.timescale.batch_threads:4}") protected int timescaleBatchThreads; - @Value("${sql.batch_sort:false}") + @Value("${sql.batch_sort:true}") protected boolean batchSortEnabled; @Value("${sql.ttl.ts.ts_key_value_ttl:0}") diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java index 00933140fb..214f7f0d92 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java @@ -94,7 +94,7 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme @Value("${sql.ts_latest.batch_threads:4}") private int tsLatestBatchThreads; - @Value("${sql.batch_sort:false}") + @Value("${sql.batch_sort:true}") protected boolean batchSortEnabled; @Autowired diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/latest/sql/SqlLatestInsertTsRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/latest/sql/SqlLatestInsertTsRepository.java index 451ba29987..087f5389da 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/latest/sql/SqlLatestInsertTsRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/latest/sql/SqlLatestInsertTsRepository.java @@ -24,6 +24,7 @@ import org.springframework.transaction.support.TransactionCallbackWithoutResult; import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity; import org.thingsboard.server.dao.sqlts.insert.AbstractInsertRepository; import org.thingsboard.server.dao.sqlts.insert.latest.InsertLatestTsRepository; +import org.thingsboard.server.dao.util.SqlDao; import org.thingsboard.server.dao.util.SqlTsLatestAnyDao; import java.sql.PreparedStatement; @@ -36,6 +37,7 @@ import java.util.List; @SqlTsLatestAnyDao @Repository @Transactional +@SqlDao public class SqlLatestInsertTsRepository extends AbstractInsertRepository implements InsertLatestTsRepository { @Value("${sql.ts_latest.update_by_latest_ts:true}")