From 0fc4071d3e073aa8fa3d4b89f24424bb5345c2ba Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Wed, 27 Jul 2022 16:38:58 +0300 Subject: [PATCH 1/9] new RuleNode 'split array msg' --- .../engine/transform/TbSplitArrayMsgNode.java | 85 +++++++++++ .../transform/TbSplitArrayMsgNodeTest.java | 136 ++++++++++++++++++ 2 files changed, 221 insertions(+) create mode 100644 rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java create mode 100644 rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java new file mode 100644 index 0000000000..8050f9e492 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.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.transform; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ArrayNode; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; +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.plugin.ComponentType; +import org.thingsboard.server.common.msg.TbMsg; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutionException; + +@Slf4j +@RuleNode( + type = ComponentType.EXTERNAL, + name = "split array msg", + configClazz = EmptyNodeConfiguration.class, + nodeDescription = "Split array message into several msgs", + nodeDetails = "", + icon = "functions", + configDirective = "tbNodeEmptyConfig" +) +public class TbSplitArrayMsgNode implements TbNode { + + EmptyNodeConfiguration config; + + @Override + public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { + this.config = TbNodeUtils.convert(configuration, EmptyNodeConfiguration.class); + } + + @Override + public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException { + JsonNode jsonNode = JacksonUtil.toJsonNode(msg.getData()); + if (jsonNode.isArray()) { + ArrayNode data = (ArrayNode) jsonNode; + List messages = new ArrayList<>(); + data.forEach(msgNode -> { + messages.add(createNewMsg(msg, msgNode)); + }); + ctx.ack(msg); + if (messages.size() == 1) { + ctx.tellSuccess(messages.get(0)); + } else { + for (TbMsg newMsg : messages) { + ctx.tellSuccess(newMsg); + } + } + } else { + ctx.tellSuccess(msg); + } + } + + private TbMsg createNewMsg(TbMsg msg, JsonNode msgNode) { + return TbMsg.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(msgNode)); + } + + @Override + public void destroy() { + + } +} diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java new file mode 100644 index 0000000000..890f3d24dd --- /dev/null +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java @@ -0,0 +1,136 @@ +/** + * 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.transform; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +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.common.util.JacksonUtil; +import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; +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.id.DeviceId; +import org.thingsboard.server.common.data.id.EntityId; +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.Map; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +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; + +public class TbSplitArrayMsgNodeTest { + final ObjectMapper mapper = new ObjectMapper(); + + DeviceId deviceId; + TbSplitArrayMsgNode node; + EmptyNodeConfiguration config; + TbNodeConfiguration nodeConfiguration; + TbContext ctx; + TbMsgCallback callback; + + @BeforeEach + void setUp() throws TbNodeException { + deviceId = new DeviceId(UUID.randomUUID()); + callback = mock(TbMsgCallback.class); + ctx = mock(TbContext.class); + config = new EmptyNodeConfiguration(); + nodeConfiguration = new TbNodeConfiguration(mapper.valueToTree(config)); + node = spy(new TbSplitArrayMsgNode()); + node.init(ctx, nodeConfiguration); + } + + @AfterEach + void tearDown() { + node.destroy(); + } + + @Test + void givenDefaultConfig_whenInit_thenOK() { + assertThat(node.config).isEqualTo(config); + } + + @Test + void givenFewMsg_whenOnMsg_thenVerifyOutput() throws Exception { + String data = "[{\"Attribute_1\":22.5,\"Attribute_2\":10.3}, {\"Attribute_1\":1,\"Attribute_2\":2}]"; + VerifyOutputMsg(data); + } + + @Test + void givenOneMsg_whenOnMsg_thenVerifyOutput() throws Exception { + String data = "[{\"Attribute_1\":22.5,\"Attribute_2\":10.3}]"; + VerifyOutputMsg(data); + } + + @Test + void givenZeroMsg_whenOnMsg_thenVerifyOutput() throws Exception { + String data = "[]"; + VerifyOutputMsg(data); + } + + @Test + void givenNoArrayMsg_whenOnMsg_thenVerifyOutput() throws Exception { + String data = "{\"Attribute_1\":22.5,\"Attribute_2\":10.3}"; + JsonNode dataNode = JacksonUtil.toJsonNode(data); + TbMsg msg = getTbMsg(deviceId, dataNode.toString()); + node.onMsg(ctx, msg); + + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + verify(ctx, times(1)).tellSuccess(newMsgCaptor.capture()); + verify(ctx, never()).tellFailure(any(), any()); + + TbMsg newMsg = newMsgCaptor.getValue(); + assertThat(newMsg).isNotNull(); + + assertThat(newMsg).isSameAs(msg); + } + + private void VerifyOutputMsg(String data) throws Exception { + ArrayNode dataNode = (ArrayNode) JacksonUtil.toJsonNode(data); + node.onMsg(ctx, getTbMsg(deviceId, dataNode.toString())); + + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + verify(ctx, times(dataNode.size())).tellSuccess(newMsgCaptor.capture()); + verify(ctx, never()).tellFailure(any(), any()); + + if (!dataNode.isEmpty()) { + TbMsg newMsg = newMsgCaptor.getValue(); + assertThat(newMsg).isNotNull(); + + assertThat(newMsg.getData()).isEqualTo(dataNode.get(dataNode.size() - 1).toString()); + } + } + + private TbMsg getTbMsg(EntityId entityId, String data) { + Map mdMap = Map.of( + "country", "US", + "city", "NY" + ); + return TbMsg.newMsg("POST_ATTRIBUTES_REQUEST", entityId, new TbMsgMetaData(mdMap), data, callback); + } +} \ No newline at end of file From 9a8372ef4c6bd8f69ae096ae6b45c00d24f49129 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Thu, 28 Jul 2022 11:38:27 +0300 Subject: [PATCH 2/9] Test modification for RuleNode 'split array msg' --- .../rule/engine/transform/TbSplitArrayMsgNodeTest.java | 7 ------- 1 file changed, 7 deletions(-) diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java index 890f3d24dd..65c814cce0 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java @@ -117,13 +117,6 @@ public class TbSplitArrayMsgNodeTest { ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); verify(ctx, times(dataNode.size())).tellSuccess(newMsgCaptor.capture()); verify(ctx, never()).tellFailure(any(), any()); - - if (!dataNode.isEmpty()) { - TbMsg newMsg = newMsgCaptor.getValue(); - assertThat(newMsg).isNotNull(); - - assertThat(newMsg.getData()).isEqualTo(dataNode.get(dataNode.size() - 1).toString()); - } } private TbMsg getTbMsg(EntityId entityId, String data) { From 84c91a17cb06a7ac3e198444e0e129fcded7630a Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Wed, 17 Aug 2022 14:17:54 +0300 Subject: [PATCH 3/9] new ruleNode 'split array msg' --- .../rule/engine/transform/TbSplitArrayMsgNode.java | 6 ++++-- .../engine/transform/TbSplitArrayMsgNodeTest.java | 11 ++++++----- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java index 8050f9e492..d6dbc88776 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java @@ -39,8 +39,10 @@ import java.util.concurrent.ExecutionException; name = "split array msg", configClazz = EmptyNodeConfiguration.class, nodeDescription = "Split array message into several msgs", - nodeDetails = "", - icon = "functions", + nodeDetails = "Split the array fetched from the msg body. Returns inner objects of the extracted array as " + + "separate messages via Success chain. If msg data is not a JSON array, " + + "the incoming message is returned", + icon = "content_copy", configDirective = "tbNodeEmptyConfig" ) public class TbSplitArrayMsgNode implements TbNode { diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java index 65c814cce0..a54efad4aa 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java @@ -78,19 +78,19 @@ public class TbSplitArrayMsgNodeTest { @Test void givenFewMsg_whenOnMsg_thenVerifyOutput() throws Exception { String data = "[{\"Attribute_1\":22.5,\"Attribute_2\":10.3}, {\"Attribute_1\":1,\"Attribute_2\":2}]"; - VerifyOutputMsg(data); + VerifyOutputMsg(data, 2); } @Test void givenOneMsg_whenOnMsg_thenVerifyOutput() throws Exception { String data = "[{\"Attribute_1\":22.5,\"Attribute_2\":10.3}]"; - VerifyOutputMsg(data); + VerifyOutputMsg(data, 1); } @Test void givenZeroMsg_whenOnMsg_thenVerifyOutput() throws Exception { String data = "[]"; - VerifyOutputMsg(data); + VerifyOutputMsg(data, 0); } @Test @@ -110,12 +110,13 @@ public class TbSplitArrayMsgNodeTest { assertThat(newMsg).isSameAs(msg); } - private void VerifyOutputMsg(String data) throws Exception { - ArrayNode dataNode = (ArrayNode) JacksonUtil.toJsonNode(data); + private void VerifyOutputMsg(String data, int sizeArray) throws Exception { + JsonNode dataNode = JacksonUtil.toJsonNode(data); node.onMsg(ctx, getTbMsg(deviceId, dataNode.toString())); ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); verify(ctx, times(dataNode.size())).tellSuccess(newMsgCaptor.capture()); + verify(ctx, times(sizeArray)).tellSuccess(newMsgCaptor.capture()); verify(ctx, never()).tellFailure(any(), any()); } From 57a913aa004f76f7bb7b9ea07e836c174f58a558 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Wed, 17 Aug 2022 15:28:15 +0300 Subject: [PATCH 4/9] add node details --- .../rule/engine/transform/TbSplitArrayMsgNode.java | 8 ++++---- .../rule/engine/transform/TbSplitArrayMsgNodeTest.java | 9 ++++++--- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java index d6dbc88776..1c682fbf7c 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java @@ -39,9 +39,9 @@ import java.util.concurrent.ExecutionException; name = "split array msg", configClazz = EmptyNodeConfiguration.class, nodeDescription = "Split array message into several msgs", - nodeDetails = "Split the array fetched from the msg body. Returns inner objects of the extracted array as " - + "separate messages via Success chain. If msg data is not a JSON array, " - + "the incoming message is returned", + nodeDetails = "Split the array fetched from the msg body. If the msg data is not a JSON object returns the " + + "incoming message as outbound message with Failure chain, otherwise returns " + + "inner objects of the extracted array as separate messages via Success chain.", icon = "content_copy", configDirective = "tbNodeEmptyConfig" ) @@ -72,7 +72,7 @@ public class TbSplitArrayMsgNode implements TbNode { } } } else { - ctx.tellSuccess(msg); + ctx.tellFailure(msg, new RuntimeException("Msg data is not a JSON Array!")); } } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java index a54efad4aa..17bfd1663c 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java @@ -94,15 +94,18 @@ public class TbSplitArrayMsgNodeTest { } @Test - void givenNoArrayMsg_whenOnMsg_thenVerifyOutput() throws Exception { + void givenNoArrayMsg_whenOnMsg_thenFailure() throws Exception { String data = "{\"Attribute_1\":22.5,\"Attribute_2\":10.3}"; JsonNode dataNode = JacksonUtil.toJsonNode(data); TbMsg msg = getTbMsg(deviceId, dataNode.toString()); node.onMsg(ctx, msg); ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctx, times(1)).tellSuccess(newMsgCaptor.capture()); - verify(ctx, never()).tellFailure(any(), any()); + ArgumentCaptor exceptionCaptor = ArgumentCaptor.forClass(Exception.class); + verify(ctx, never()).tellSuccess(any()); + verify(ctx, times(1)).tellFailure(newMsgCaptor.capture(), exceptionCaptor.capture()); + + assertThat(exceptionCaptor.getValue()).isInstanceOf(RuntimeException.class); TbMsg newMsg = newMsgCaptor.getValue(); assertThat(newMsg).isNotNull(); From cf7ba9b390596b6e452ca135246f351299bd3d71 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Thu, 25 Aug 2022 11:47:06 +0300 Subject: [PATCH 5/9] refactor code --- .../rule/engine/transform/TbSplitArrayMsgNode.java | 13 +++++++++---- .../engine/transform/TbSplitArrayMsgNodeTest.java | 2 +- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java index 1c682fbf7c..4eaf19ec6f 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java @@ -61,12 +61,12 @@ public class TbSplitArrayMsgNode implements TbNode { ArrayNode data = (ArrayNode) jsonNode; List messages = new ArrayList<>(); data.forEach(msgNode -> { - messages.add(createNewMsg(msg, msgNode)); + messages.add(createMsg(msg, msgNode, data.size() > 1)); }); - ctx.ack(msg); if (messages.size() == 1) { ctx.tellSuccess(messages.get(0)); } else { + ctx.ack(msg); for (TbMsg newMsg : messages) { ctx.tellSuccess(newMsg); } @@ -76,8 +76,12 @@ public class TbSplitArrayMsgNode implements TbNode { } } - private TbMsg createNewMsg(TbMsg msg, JsonNode msgNode) { - return TbMsg.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(msgNode)); + private TbMsg createMsg(TbMsg msg, JsonNode msgNode, boolean newMessage) { + if (newMessage) { + return TbMsg.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(msgNode)); + } else { + return TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(msgNode)); + } } @Override @@ -85,3 +89,4 @@ public class TbSplitArrayMsgNode implements TbNode { } } + diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java index 17bfd1663c..adaf842d42 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java @@ -130,4 +130,4 @@ public class TbSplitArrayMsgNodeTest { ); return TbMsg.newMsg("POST_ATTRIBUTES_REQUEST", entityId, new TbMsgMetaData(mdMap), data, callback); } -} \ No newline at end of file +} From 6d6e7b212cd8287b5542a6d343036a0d6953edaa Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Tue, 6 Sep 2022 15:11:01 +0300 Subject: [PATCH 6/9] refactor code --- .../engine/transform/TbSplitArrayMsgNode.java | 25 +++++-------------- 1 file changed, 6 insertions(+), 19 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java index 4eaf19ec6f..86739284db 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java @@ -29,8 +29,6 @@ import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; -import java.util.ArrayList; -import java.util.List; import java.util.concurrent.ExecutionException; @Slf4j @@ -42,6 +40,7 @@ import java.util.concurrent.ExecutionException; nodeDetails = "Split the array fetched from the msg body. If the msg data is not a JSON object returns the " + "incoming message as outbound message with Failure chain, otherwise returns " + "inner objects of the extracted array as separate messages via Success chain.", + uiResources = {"static/rulenode/rulenode-core-config.js"}, icon = "content_copy", configDirective = "tbNodeEmptyConfig" ) @@ -59,31 +58,19 @@ public class TbSplitArrayMsgNode implements TbNode { JsonNode jsonNode = JacksonUtil.toJsonNode(msg.getData()); if (jsonNode.isArray()) { ArrayNode data = (ArrayNode) jsonNode; - List messages = new ArrayList<>(); - data.forEach(msgNode -> { - messages.add(createMsg(msg, msgNode, data.size() > 1)); - }); - if (messages.size() == 1) { - ctx.tellSuccess(messages.get(0)); + if (data.size() == 1) { + ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(data.get(0)))); } else { ctx.ack(msg); - for (TbMsg newMsg : messages) { - ctx.tellSuccess(newMsg); - } + data.forEach(msgNode -> { + ctx.tellSuccess(TbMsg.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(msgNode))); + }); } } else { ctx.tellFailure(msg, new RuntimeException("Msg data is not a JSON Array!")); } } - private TbMsg createMsg(TbMsg msg, JsonNode msgNode, boolean newMessage) { - if (newMessage) { - return TbMsg.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(msgNode)); - } else { - return TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(msgNode)); - } - } - @Override public void destroy() { From 8d50473a6e8a43db8b2ab4eded3dd76384fc0449 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Tue, 6 Sep 2022 15:13:04 +0300 Subject: [PATCH 7/9] add new line --- .../thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java | 1 - 1 file changed, 1 deletion(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java index 86739284db..c899a7a783 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java @@ -76,4 +76,3 @@ public class TbSplitArrayMsgNode implements TbNode { } } - From 576b232427bcda211b59da23fac5dc19495a9018 Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Fri, 23 Sep 2022 11:17:58 +0300 Subject: [PATCH 8/9] delete destroy --- .../rule/engine/transform/TbSplitArrayMsgNode.java | 5 ----- 1 file changed, 5 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java index c899a7a783..965cc3f058 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java @@ -70,9 +70,4 @@ public class TbSplitArrayMsgNode implements TbNode { ctx.tellFailure(msg, new RuntimeException("Msg data is not a JSON Array!")); } } - - @Override - public void destroy() { - - } } From d41a791dc312a0cb543df986ca13d3f8fd60171d Mon Sep 17 00:00:00 2001 From: Yuriy Lytvynchuk Date: Tue, 27 Sep 2022 17:12:01 +0300 Subject: [PATCH 9/9] add enqueueForTellNext --- .../transform/TbAbstractTransformNode.java | 3 +- .../engine/transform/TbSplitArrayMsgNode.java | 20 ++++++++--- .../transform/TbSplitArrayMsgNodeTest.java | 36 +++++++++++-------- 3 files changed, 39 insertions(+), 20 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbAbstractTransformNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbAbstractTransformNode.java index 29c61b81c7..e51a2d9205 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbAbstractTransformNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbAbstractTransformNode.java @@ -22,6 +22,7 @@ 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.TbRelationTypes; import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.queue.RuleEngineException; @@ -81,7 +82,7 @@ public abstract class TbAbstractTransformNode implements TbNode { ctx.tellFailure(msg, e); } }); - msgs.forEach(newMsg -> ctx.enqueueForTellNext(newMsg, "Success", wrapper::onSuccess, wrapper::onFailure)); + msgs.forEach(newMsg -> ctx.enqueueForTellNext(newMsg, TbRelationTypes.SUCCESS, wrapper::onSuccess, wrapper::onFailure)); } } else { ctx.tellNext(msg, FAILURE); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java index 965cc3f058..2dc27b6f92 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNode.java @@ -25,9 +25,12 @@ 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.TbRelationTypes; import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.queue.RuleEngineException; +import org.thingsboard.server.common.msg.queue.TbMsgCallback; import java.util.concurrent.ExecutionException; @@ -46,7 +49,7 @@ import java.util.concurrent.ExecutionException; ) public class TbSplitArrayMsgNode implements TbNode { - EmptyNodeConfiguration config; + private EmptyNodeConfiguration config; @Override public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { @@ -61,10 +64,19 @@ public class TbSplitArrayMsgNode implements TbNode { if (data.size() == 1) { ctx.tellSuccess(TbMsg.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(data.get(0)))); } else { - ctx.ack(msg); - data.forEach(msgNode -> { - ctx.tellSuccess(TbMsg.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(msgNode))); + TbMsgCallbackWrapper wrapper = new MultipleTbMsgsCallbackWrapper(data.size(), new TbMsgCallback() { + @Override + public void onSuccess() { + ctx.ack(msg); + } + + @Override + public void onFailure(RuleEngineException e) { + ctx.tellFailure(msg, e); + } }); + data.forEach(msgNode -> ctx.enqueueForTellNext(TbMsg.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getMetaData(), JacksonUtil.toString(msgNode)), + TbRelationTypes.SUCCESS, wrapper::onSuccess, wrapper::onFailure)); } } else { ctx.tellFailure(msg, new RuntimeException("Msg data is not a JSON Array!")); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java index adaf842d42..2925ab96ae 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbSplitArrayMsgNodeTest.java @@ -17,7 +17,6 @@ package org.thingsboard.rule.engine.transform; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.node.ArrayNode; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -35,9 +34,11 @@ import org.thingsboard.server.common.msg.queue.TbMsgCallback; 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; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; @@ -70,27 +71,22 @@ public class TbSplitArrayMsgNodeTest { node.destroy(); } - @Test - void givenDefaultConfig_whenInit_thenOK() { - assertThat(node.config).isEqualTo(config); - } - @Test void givenFewMsg_whenOnMsg_thenVerifyOutput() throws Exception { String data = "[{\"Attribute_1\":22.5,\"Attribute_2\":10.3}, {\"Attribute_1\":1,\"Attribute_2\":2}]"; - VerifyOutputMsg(data, 2); + VerifyOutputMsg(data); } @Test void givenOneMsg_whenOnMsg_thenVerifyOutput() throws Exception { String data = "[{\"Attribute_1\":22.5,\"Attribute_2\":10.3}]"; - VerifyOutputMsg(data, 1); + VerifyOutputMsg(data); } @Test void givenZeroMsg_whenOnMsg_thenVerifyOutput() throws Exception { String data = "[]"; - VerifyOutputMsg(data, 0); + VerifyOutputMsg(data); } @Test @@ -113,13 +109,23 @@ public class TbSplitArrayMsgNodeTest { assertThat(newMsg).isSameAs(msg); } - private void VerifyOutputMsg(String data, int sizeArray) throws Exception { + private void VerifyOutputMsg(String data) throws Exception { JsonNode dataNode = JacksonUtil.toJsonNode(data); - node.onMsg(ctx, getTbMsg(deviceId, dataNode.toString())); - - ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctx, times(dataNode.size())).tellSuccess(newMsgCaptor.capture()); - verify(ctx, times(sizeArray)).tellSuccess(newMsgCaptor.capture()); + TbMsg tbMsg = getTbMsg(deviceId, dataNode.toString()); + node.onMsg(ctx, tbMsg); + + if (dataNode.size() > 1) { + ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); + ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(ctx, times(dataNode.size())).enqueueForTellNext(any(), anyString(), successCaptor.capture(), failureCaptor.capture()); + for (Runnable valueCaptor : successCaptor.getAllValues()) { + valueCaptor.run(); + } + verify(ctx, times(1)).ack(tbMsg); + } else { + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + verify(ctx, times(dataNode.size())).tellSuccess(newMsgCaptor.capture()); + } verify(ctx, never()).tellFailure(any(), any()); }