From 177c0f46ad0171bd2c40f3fcb87806fc94b6c1d6 Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Sat, 25 Sep 2021 20:36:06 +0300 Subject: [PATCH] enqueueForTellNext for specific queue produce new message with the new queue from parameter. Instead old queue from old Msg. It affects rule node statistics before. This fix is related to Checkpoint node --- .../server/actors/ruleChain/DefaultTbContext.java | 10 +++++++--- .../java/org/thingsboard/server/common/msg/TbMsg.java | 5 +++-- 2 files changed, 10 insertions(+), 5 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 78dbadc9bf..8971cf1a23 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 @@ -190,13 +190,13 @@ class DefaultTbContext implements TbContext { @Override public void enqueueForTellNext(TbMsg tbMsg, String queueName, String relationType, Runnable onSuccess, Consumer onFailure) { TopicPartitionInfo tpi = resolvePartition(tbMsg, queueName); - enqueueForTellNext(tpi, tbMsg, Collections.singleton(relationType), null, onSuccess, onFailure); + enqueueForTellNext(tpi, queueName, tbMsg, Collections.singleton(relationType), null, onSuccess, onFailure); } @Override public void enqueueForTellNext(TbMsg tbMsg, String queueName, Set relationTypes, Runnable onSuccess, Consumer onFailure) { TopicPartitionInfo tpi = resolvePartition(tbMsg, queueName); - enqueueForTellNext(tpi, tbMsg, relationTypes, null, onSuccess, onFailure); + enqueueForTellNext(tpi, queueName, tbMsg, relationTypes, null, onSuccess, onFailure); } private TopicPartitionInfo resolvePartition(TbMsg tbMsg, String queueName) { @@ -211,9 +211,13 @@ class DefaultTbContext implements TbContext { } private void enqueueForTellNext(TopicPartitionInfo tpi, TbMsg source, Set relationTypes, String failureMessage, Runnable onSuccess, Consumer onFailure) { + enqueueForTellNext(tpi, source.getQueueName(), source, relationTypes, failureMessage, onSuccess, onFailure); + } + + private void enqueueForTellNext(TopicPartitionInfo tpi, String queueName, TbMsg source, Set relationTypes, String failureMessage, Runnable onSuccess, Consumer onFailure) { RuleChainId ruleChainId = nodeCtx.getSelf().getRuleChainId(); RuleNodeId ruleNodeId = nodeCtx.getSelf().getId(); - TbMsg tbMsg = TbMsg.newMsg(source, ruleChainId, ruleNodeId); + TbMsg tbMsg = TbMsg.newMsg(source, queueName, ruleChainId, ruleNodeId); TransportProtos.ToRuleEngineMsg.Builder msg = TransportProtos.ToRuleEngineMsg.newBuilder() .setTenantIdMSB(getTenantId().getId().getMostSignificantBits()) .setTenantIdLSB(getTenantId().getId().getLeastSignificantBits()) diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java index 1207f4d634..83715994da 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java @@ -136,8 +136,9 @@ public final class TbMsg implements Serializable { tbMsg.data, ruleChainId, null, tbMsg.ruleNodeExecCounter.get(), tbMsg.getCallback()); } - public static TbMsg newMsg(TbMsg tbMsg, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { - return new TbMsg(tbMsg.getQueueName(), UUID.randomUUID(), tbMsg.getTs(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.customerId, tbMsg.getMetaData().copy(), + //used for enqueueForTellNext + public static TbMsg newMsg(TbMsg tbMsg, String queueName, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { + return new TbMsg(queueName, UUID.randomUUID(), tbMsg.getTs(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.customerId, tbMsg.getMetaData().copy(), tbMsg.getDataType(), tbMsg.getData(), ruleChainId, ruleNodeId, tbMsg.ruleNodeExecCounter.get(), TbMsgCallback.EMPTY); }