tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

de-duplicate node: init commit

pull/7893/head
ShvaykaD 4 years ago
parent
9da5f3cee3
commit
d0cedf7e0a
5 changed files with 553 additions and 1 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 22
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/DeDuplicateStrategy.java
  2. 176
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDeDuplicateNode.java
  3. 43
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDeDuplicateNodeConfiguration.java
  4. 1
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
  5. 312
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbMsgDeDuplicateNodeTest.java

22
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/DeDuplicateStrategy.java
View File

@ -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
}

176
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDeDuplicateNode.java
View File

@ -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: <br></br>" +
"<b>FIRST</b> - return first message that arrived during de-duplication period.<br></br>" +
"<b>LAST</b> - return last message that arrived during de-duplication period.<br></br>" +
"<b>ALL</b> - return all messages as a single JSON array message. Where each element represents object with <b>msg</b> and <b>metadata</b> inner properties.<br></br>" +
"By default rule node <b>de-duplicates messages by message originator</b>, however, there is an option to de-duplicate messages independently from the incoming message's originator. " +
"In case of the <b>de-duplication strategy</b> set to <b>ALL</b> and <b>de-duplicates messages by message originator</b> checkbox set to <b>false</b>, you must configure the <b>Queue Name</b> and <b>Message Type</b> for the output array message. " +
"Also in this case the output message originator will be set to the current tenant id.<br></br>",
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<String, List<TbMsg>> 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<String> 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<TbMsg> 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<TbMsg> 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<TbMsg> 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;
}
}

43
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDeDuplicateNodeConfiguration.java
View File

@ -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<TbMsgDeDuplicateNodeConfiguration> {
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;
}
}

1
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
View File

@ -48,7 +48,6 @@ import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
uiResources = {"static/rulenode/rulenode-core-config.js"}, uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = "tbActionNodeMsgDelayConfig" configDirective = "tbActionNodeMsgDelayConfig"
) )
public class TbMsgDelayNode implements TbNode { public class TbMsgDelayNode implements TbNode {
private static final String TB_MSG_DELAY_NODE_MSG = "TbMsgDelayNodeMsg"; private static final String TB_MSG_DELAY_NODE_MSG = "TbMsgDelayNodeMsg";

312
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbMsgDeDuplicateNodeTest.java
View File

@ -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<TbMsg>) 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<Void>) 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<TbMsg> inputMsgs = createTbMsgs(deviceId);
for (TbMsg msg : inputMsgs) {
node.onMsg(ctx, msg);
}
awaitAllMsgsProcessedLatch.countDown();
awaitTellSelfLatch.await();
verify(ctx, times(inputMsgs.size())).ack(any());
ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class);
ArgumentCaptor<Consumer<Throwable>> 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<TbMsg> 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<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class);
ArgumentCaptor<Consumer<Throwable>> 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<TbMsg> inputMsgs = createTbMsgs(deviceId);
for (TbMsg msg : inputMsgs) {
node.onMsg(ctx, msg);
}
awaitAllMsgsProcessedLatch.countDown();
awaitTellSelfLatch.await();
verify(ctx, times(inputMsgs.size())).ack(any());
ArgumentCaptor<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class);
ArgumentCaptor<Consumer<Throwable>> 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<TbMsg> firstDeviceInputMsgs = createTbMsgs(firstDeviceId);
DeviceId secondDeviceId = new DeviceId(UUID.randomUUID());
List<TbMsg> secondDeviceInputMsgs = createTbMsgs(secondDeviceId);
List<TbMsg> 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<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class);
ArgumentCaptor<Consumer<Throwable>> failureCaptor = ArgumentCaptor.forClass(Consumer.class);
verify(ctx, times(2)).enqueueForTellNext(newMsgCaptor.capture(), eq(TbRelationTypes.SUCCESS), successCaptor.capture(), failureCaptor.capture());
List<TbMsg> 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<TbMsg> firstDeviceInputMsgs = createTbMsgs(firstDeviceId, 50);
DeviceId secondDeviceId = new DeviceId(UUID.randomUUID());
List<TbMsg> secondDeviceInputMsgs = createTbMsgs(secondDeviceId, 50);
List<TbMsg> 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<TbMsg> newMsgCaptor = ArgumentCaptor.forClass(TbMsg.class);
ArgumentCaptor<Runnable> successCaptor = ArgumentCaptor.forClass(Runnable.class);
ArgumentCaptor<Consumer<Throwable>> 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<TbMsg> createTbMsgs(DeviceId deviceId) {
return createTbMsgs(deviceId, 100);
}
private List<TbMsg> createTbMsgs(DeviceId deviceId, int msgCount) {
List<TbMsg> 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<TbMsg> 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);
}
}
Write Preview
Loading…
Cancel
Save