diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/DeDuplicateStrategy.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/DeDuplicateStrategy.java new file mode 100644 index 0000000000..69056de9fd --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/DeDuplicateStrategy.java @@ -0,0 +1,22 @@ +/** + * 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.delay; + +public enum DeDuplicateStrategy { + + FIRST, LAST, ALL + +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDeDuplicateNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDeDuplicateNode.java new file mode 100644 index 0000000000..ca93be30e1 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDeDuplicateNode.java @@ -0,0 +1,176 @@ +/** + * 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.delay; + +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.JacksonUtil; +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.TbRelationTypes; +import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.server.common.data.EntityType; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityIdFactory; +import org.thingsboard.server.common.data.plugin.ComponentType; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbMsgMetaData; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +@RuleNode( + type = ComponentType.ACTION, + name = "de-duplicate", + configClazz = TbMsgDeDuplicateNodeConfiguration.class, + nodeDescription = "De-duplicate messages for a configurable period based on a specified de-duplication strategy.", + nodeDetails = "Rule node allows you to select one of the following strategy to de-duplicate messages:

" + + "FIRST - return first message that arrived during de-duplication period.

" + + "LAST - return last message that arrived during de-duplication period.

" + + "ALL - return all messages as a single JSON array message. Where each element represents object with msg and metadata inner properties.

" + + "By default rule node de-duplicates messages by message originator, however, there is an option to de-duplicate messages independently from the incoming message's originator. " + + "In case of the de-duplication strategy set to ALL and de-duplicates messages by message originator checkbox set to false, you must configure the Queue Name and Message Type for the output array message. " + + "Also in this case the output message originator will be set to the current tenant id.

", + icon = "pause", + uiResources = {"static/rulenode/rulenode-core-config.js"}, + configDirective = "" +) +@Slf4j +public class TbMsgDeDuplicateNode implements TbNode { + + private static final String TB_MSG_DEDUPLICATION_TIMEOUT_MSG = "TbMsgDeDuplicateNodeMsg"; + + private TbMsgDeDuplicateNodeConfiguration config; + + private Map> deDuplicationMap; + private long delay; + private long lastScheduledTs; + private boolean deDuplicateByOriginator; + private UUID nextTickId; + + + @Override + public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { + this.config = TbNodeUtils.convert(configuration, TbMsgDeDuplicateNodeConfiguration.class); + this.delay = TimeUnit.SECONDS.toMillis(config.getDelay()); + this.deDuplicateByOriginator = config.isDeDuplicateByOriginator(); + this.deDuplicationMap = new HashMap<>(); + scheduleTickMsg(ctx); + } + + @Override + public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException { + if (TB_MSG_DEDUPLICATION_TIMEOUT_MSG.equals(msg.getType())) { + if (msg.getId().equals(nextTickId)) { + long deDuplicationTimeoutMs = System.currentTimeMillis(); + List keysToRemove = deDuplicationMap + .keySet() + .stream() + .filter(key -> key.startsWith(String.valueOf(lastScheduledTs)) || Long.parseLong(key.split("_")[0]) < deDuplicationTimeoutMs) + .collect(Collectors.toList()); + scheduleTickMsg(ctx); + keysToRemove.forEach(key -> { + List msgList = deDuplicationMap.get(key); + TbMsg outMsg; + switch (config.getStrategy()) { + case FIRST: + outMsg = msgList.get(0); + break; + case LAST: + outMsg = msgList.get(msgList.size() - 1); + break; + default: + EntityId originator; + String queueName; + String outMsgType; + if (deDuplicateByOriginator) { + String[] keyElements = key.split("_"); + originator = EntityIdFactory.getByTypeAndId(keyElements[1], keyElements[2]); + queueName = msgList.get(0).getQueueName(); + outMsgType = msgList.get(0).getType(); + } else { + originator = ctx.getTenantId(); + queueName = config.getQueueName(); + outMsgType = config.getOutMsgType(); + } + outMsg = TbMsg.newMsg(queueName, outMsgType, originator, getMetadata(), getMergedData(msgList)); + break; + } + ctx.enqueueForTellNext(outMsg, TbRelationTypes.SUCCESS, + () -> deDuplicationMap.remove(key), + throwable -> log.trace("Failed to enqueue de-duplication output message due to: ", throwable)); + }); + } else { + log.trace("[{}][{}] Received outdated tick msg: {}", ctx.getTenantId(), ctx.getSelfId(), msg.getId()); + } + } else { + String key = String.format("%d_%s", lastScheduledTs, getDeDuplicationId(ctx, msg)); + List deDuplicateMsgList = deDuplicationMap.computeIfAbsent(key, k -> new ArrayList<>()); + if (deDuplicateMsgList.size() < config.getMaxPendingMsgs()) { + deDuplicateMsgList.add(msg); + ctx.ack(msg); + } else { + ctx.tellFailure(msg, new RuntimeException("Max limit of pending messages reached for key: " + key)); + } + } + } + + private String getDeDuplicationId(TbContext ctx, TbMsg msg) { + return deDuplicateByOriginator ? + msg.getOriginator().getEntityType().name() + "_" + msg.getOriginator().getId() : + EntityType.RULE_NODE.name() + "_" + ctx.getSelfId().getId(); + } + + private void scheduleTickMsg(TbContext ctx) { + long curTs = System.currentTimeMillis(); + if (lastScheduledTs == 0L) { + lastScheduledTs = curTs; + } + lastScheduledTs = lastScheduledTs + delay; + long curDelay = Math.max(0L, (lastScheduledTs - curTs)); + TbMsg tickMsg = ctx.newMsg(null, TB_MSG_DEDUPLICATION_TIMEOUT_MSG, ctx.getSelfId(), new TbMsgMetaData(), ""); + nextTickId = tickMsg.getId(); + ctx.tellSelf(tickMsg, curDelay); + } + + private String getMergedData(List msgs) { + ArrayNode mergedData = JacksonUtil.OBJECT_MAPPER.createArrayNode(); + msgs.forEach(msg -> { + ObjectNode msgNode = JacksonUtil.newObjectNode(); + msgNode.set("msg", JacksonUtil.toJsonNode(msg.getData())); + msgNode.set("metadata", JacksonUtil.valueToTree(msg.getMetaData())); + mergedData.add(msgNode); + }); + return JacksonUtil.toString(mergedData); + } + + private TbMsgMetaData getMetadata() { + TbMsgMetaData metaData = new TbMsgMetaData(); + metaData.putValue("ts", String.valueOf(System.currentTimeMillis())); + return metaData; + } + +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDeDuplicateNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDeDuplicateNodeConfiguration.java new file mode 100644 index 0000000000..6153591abe --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDeDuplicateNodeConfiguration.java @@ -0,0 +1,43 @@ +/** + * 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.delay; + +import lombok.Data; +import org.thingsboard.rule.engine.api.NodeConfiguration; + +@Data +public class TbMsgDeDuplicateNodeConfiguration implements NodeConfiguration { + + private int delay; + private int maxPendingMsgs; + private boolean deDuplicateByOriginator; + private DeDuplicateStrategy strategy; + + // only for DeDuplicateStrategy.ALL when deDuplicateByOriginator set to false, + // otherwise these parameters will be used from first msg of deDuplication msgs list. + private String outMsgType; + private String queueName; + + @Override + public TbMsgDeDuplicateNodeConfiguration defaultConfiguration() { + TbMsgDeDuplicateNodeConfiguration configuration = new TbMsgDeDuplicateNodeConfiguration(); + configuration.setDelay(60); + configuration.setMaxPendingMsgs(100); + configuration.setDeDuplicateByOriginator(true); + configuration.setStrategy(DeDuplicateStrategy.FIRST); + return configuration; + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java index 5a313da469..5c2efb7e5b 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java @@ -48,7 +48,6 @@ import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS; uiResources = {"static/rulenode/rulenode-core-config.js"}, configDirective = "tbActionNodeMsgDelayConfig" ) - public class TbMsgDelayNode implements TbNode { private static final String TB_MSG_DELAY_NODE_MSG = "TbMsgDelayNodeMsg"; diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbMsgDeDuplicateNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbMsgDeDuplicateNodeTest.java new file mode 100644 index 0000000000..5411dda07b --- /dev/null +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbMsgDeDuplicateNodeTest.java @@ -0,0 +1,312 @@ +/** + * 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.action; + +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.ArgumentMatchers; +import org.mockito.stubbing.Answer; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.rule.engine.api.TbContext; +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.delay.DeDuplicateStrategy; +import org.thingsboard.rule.engine.delay.TbMsgDeDuplicateNode; +import org.thingsboard.rule.engine.delay.TbMsgDeDuplicateNodeConfiguration; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.RuleNodeId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.session.SessionMsgType; + +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.function.Consumer; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.ArgumentMatchers.nullable; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@Slf4j +public class TbMsgDeDuplicateNodeTest { + + private static final String MAIN_QUEUE_NAME = "Main"; + private static final String HIGH_PRIORITY_QUEUE_NAME = "HighPriority"; + private static final String TB_MSG_DEDUPLICATION_TIMEOUT_MSG = "TbMsgDeDuplicateNodeMsg"; + + private TbContext ctx; + + private ExecutorService executorService = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("de-duplication-node-test")); + + private TenantId tenantId; + private RuleNodeId ruleNodeId; + + private TbMsgDeDuplicateNode node; + private TbMsgDeDuplicateNodeConfiguration config; + private TbNodeConfiguration nodeConfiguration; + + private CountDownLatch awaitTellSelfLatch; + private CountDownLatch awaitAllMsgsProcessedLatch; + private int scheduleCount; + + @BeforeEach + @SuppressWarnings("unchecked") + public void init() throws TbNodeException { + ctx = mock(TbContext.class); + awaitTellSelfLatch = new CountDownLatch(1); + awaitAllMsgsProcessedLatch = new CountDownLatch(1); + + tenantId = TenantId.fromUUID(UUID.randomUUID()); + ruleNodeId = new RuleNodeId(UUID.randomUUID()); + + when(ctx.getSelfId()).thenReturn(ruleNodeId); + when(ctx.getTenantId()).thenReturn(tenantId); + + doAnswer((Answer) invocationOnMock -> { + String type = (String) (invocationOnMock.getArguments())[1]; + EntityId originator = (EntityId) (invocationOnMock.getArguments())[2]; + TbMsgMetaData metaData = (TbMsgMetaData) (invocationOnMock.getArguments())[3]; + String data = (String) (invocationOnMock.getArguments())[4]; + return TbMsg.newMsg(type, originator, metaData.copy(), data); + }).when(ctx).newMsg(isNull(), eq(TB_MSG_DEDUPLICATION_TIMEOUT_MSG), nullable(EntityId.class), any(TbMsgMetaData.class), any(String.class)); + + scheduleCount = 0; + + doAnswer((Answer) invocationOnMock -> { + scheduleCount++; + if (scheduleCount == 1) { + TbMsg msg = (TbMsg) (invocationOnMock.getArguments())[0]; + executorService.submit(() -> { + try { + awaitAllMsgsProcessedLatch.await(); + node.onMsg(ctx, msg); + awaitTellSelfLatch.countDown(); + } catch (ExecutionException | InterruptedException | TbNodeException e) { + log.error("Failed to execute tellSelf method call due to: ", e); + } + }); + } + return null; + }).when(ctx).tellSelf(ArgumentMatchers.any(TbMsg.class), ArgumentMatchers.anyLong()); + + node = new TbMsgDeDuplicateNode(); + config = new TbMsgDeDuplicateNodeConfiguration().defaultConfiguration(); + } + + @AfterEach + public void destroy() { + if (executorService != null) { + executorService.shutdown(); + } + node.destroy(); + } + + @Test + public void given_multipleMessages_thenVerifyOutputFirst() throws TbNodeException, ExecutionException, InterruptedException { + nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); + node.init(ctx, nodeConfiguration); + + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + List inputMsgs = createTbMsgs(deviceId); + for (TbMsg msg : inputMsgs) { + node.onMsg(ctx, msg); + } + awaitAllMsgsProcessedLatch.countDown(); + awaitTellSelfLatch.await(); + + verify(ctx, times(inputMsgs.size())).ack(any()); + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); + ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); + Assertions.assertEquals(inputMsgs.get(0), newMsgCaptor.getValue()); + } + + @Test + public void given_moreMessagesThenAllowedInConfiguration_thenVerifyTellFailure() throws TbNodeException, ExecutionException, InterruptedException { + nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); + node.init(ctx, nodeConfiguration); + + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + List inputMsgs = createTbMsgs(deviceId, 101); + for (TbMsg msg : inputMsgs) { + node.onMsg(ctx, msg); + } + awaitAllMsgsProcessedLatch.countDown(); + awaitTellSelfLatch.await(); + + verify(ctx, times(inputMsgs.size() -1)).ack(any()); + verify(ctx, times(1)).tellFailure(eq(inputMsgs.get(inputMsgs.size() -1)), any()); + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); + ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); + Assertions.assertEquals(inputMsgs.get(0), newMsgCaptor.getValue()); + } + + @Test + public void given_multipleMessages_thenVerifyOutputLast() throws TbNodeException, ExecutionException, InterruptedException { + config.setStrategy(DeDuplicateStrategy.LAST); + nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); + node.init(ctx, nodeConfiguration); + + DeviceId deviceId = new DeviceId(UUID.randomUUID()); + List inputMsgs = createTbMsgs(deviceId); + for (TbMsg msg : inputMsgs) { + node.onMsg(ctx, msg); + } + awaitAllMsgsProcessedLatch.countDown(); + awaitTellSelfLatch.await(); + + verify(ctx, times(inputMsgs.size())).ack(any()); + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); + ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); + Assertions.assertEquals(inputMsgs.get(inputMsgs.size() - 1), newMsgCaptor.getValue()); + } + + @Test + public void given_multipleMessagesFromTwoOriginators_thenVerifyOutputAllForEachOriginator() throws TbNodeException, ExecutionException, InterruptedException { + config.setStrategy(DeDuplicateStrategy.ALL); + nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); + node.init(ctx, nodeConfiguration); + + DeviceId firstDeviceId = new DeviceId(UUID.randomUUID()); + List firstDeviceInputMsgs = createTbMsgs(firstDeviceId); + + DeviceId secondDeviceId = new DeviceId(UUID.randomUUID()); + List secondDeviceInputMsgs = createTbMsgs(secondDeviceId); + + List inputMsgs = new ArrayList<>(); + inputMsgs.addAll(firstDeviceInputMsgs); + inputMsgs.addAll(secondDeviceInputMsgs); + + for (TbMsg msg : inputMsgs) { + node.onMsg(ctx, msg); + } + awaitAllMsgsProcessedLatch.countDown(); + awaitTellSelfLatch.await(); + + verify(ctx, times(inputMsgs.size())).ack(any()); + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); + ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(ctx, times(2)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); + + List outMessages = newMsgCaptor.getAllValues(); + Assertions.assertEquals(2, outMessages.size()); + for (TbMsg tbMsg : outMessages) { + if (tbMsg.getOriginator().equals(firstDeviceId)) { + Assertions.assertEquals(getMergedData(firstDeviceInputMsgs), tbMsg.getData()); + } else { + Assertions.assertEquals(getMergedData(secondDeviceInputMsgs), tbMsg.getData()); + Assertions.assertEquals(secondDeviceId, tbMsg.getOriginator()); + } + } + } + + @Test + public void given_multipleMessagesFromTwoOriginators_thenVerifyOutputAll() throws TbNodeException, ExecutionException, InterruptedException { + config.setStrategy(DeDuplicateStrategy.ALL); + config.setDeDuplicateByOriginator(false); + config.setQueueName(HIGH_PRIORITY_QUEUE_NAME); + config.setOutMsgType(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()); + nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); + node.init(ctx, nodeConfiguration); + + DeviceId firstDeviceId = new DeviceId(UUID.randomUUID()); + List firstDeviceInputMsgs = createTbMsgs(firstDeviceId, 50); + + DeviceId secondDeviceId = new DeviceId(UUID.randomUUID()); + List secondDeviceInputMsgs = createTbMsgs(secondDeviceId, 50); + + List inputMsgs = new ArrayList<>(); + inputMsgs.addAll(firstDeviceInputMsgs); + inputMsgs.addAll(secondDeviceInputMsgs); + + for (TbMsg msg : inputMsgs) { + node.onMsg(ctx, msg); + } + awaitAllMsgsProcessedLatch.countDown(); + awaitTellSelfLatch.await(); + + verify(ctx, times(inputMsgs.size())).ack(any()); + ArgumentCaptor newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class); + ArgumentCaptor successCaptor = ArgumentCaptor.forClass(Runnable.class); + ArgumentCaptor> failureCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(ctx, times(1)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture()); + + Assertions.assertEquals(1, newMsgCaptor.getAllValues().size()); + TbMsg outMessage = newMsgCaptor.getAllValues().get(0); + Assertions.assertEquals(getMergedData(inputMsgs), outMessage.getData()); + Assertions.assertEquals(tenantId, outMessage.getOriginator()); + } + + private List createTbMsgs(DeviceId deviceId) { + return createTbMsgs(deviceId, 100); + } + + private List createTbMsgs(DeviceId deviceId, int msgCount) { + List inputMsgs = new ArrayList<>(); + for (int i = 0; i < msgCount; i++) { + ObjectNode dataNode = JacksonUtil.newObjectNode(); + dataNode.put("msgId", i); + dataNode.put("deviceId", deviceId.getId().toString()); + TbMsg tbMsg = TbMsg.newMsg( + MAIN_QUEUE_NAME, + SessionMsgType.POST_TELEMETRY_REQUEST.name(), + deviceId, + new TbMsgMetaData(), + JacksonUtil.toString(dataNode)); + inputMsgs.add(tbMsg); + } + return inputMsgs; + } + + private String getMergedData(List msgs) { + ArrayNode mergedData = JacksonUtil.OBJECT_MAPPER.createArrayNode(); + msgs.forEach(msg -> { + ObjectNode msgNode = JacksonUtil.newObjectNode(); + msgNode.set("msg", JacksonUtil.toJsonNode(msg.getData())); + msgNode.set("metadata", JacksonUtil.valueToTree(msg.getMetaData())); + mergedData.add(msgNode); + }); + return JacksonUtil.toString(mergedData); + } + +}