dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Sergey Matvienko 2021-09-25 20:36:06 +03:00 committed by Andrew Shvayka
parent bbb7e89881
commit 177c0f46ad
2 changed files with 10 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
View File

@ -190,13 +190,13 @@ class DefaultTbContext implements TbContext {
@Override @Override
public void enqueueForTellNext(TbMsg tbMsg, String queueName, String relationType, Runnable onSuccess, Consumer<Throwable> onFailure) { public void enqueueForTellNext(TbMsg tbMsg, String queueName, String relationType, Runnable onSuccess, Consumer<Throwable> onFailure) {
TopicPartitionInfo tpi = resolvePartition(tbMsg, queueName); 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 @Override
public void enqueueForTellNext(TbMsg tbMsg, String queueName, Set<String> relationTypes, Runnable onSuccess, Consumer<Throwable> onFailure) { public void enqueueForTellNext(TbMsg tbMsg, String queueName, Set<String> relationTypes, Runnable onSuccess, Consumer<Throwable> onFailure) {
TopicPartitionInfo tpi = resolvePartition(tbMsg, queueName); 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) { private TopicPartitionInfo resolvePartition(TbMsg tbMsg, String queueName) {
@ -211,9 +211,13 @@ class DefaultTbContext implements TbContext {
} }
private void enqueueForTellNext(TopicPartitionInfo tpi, TbMsg source, Set<String> relationTypes, String failureMessage, Runnable onSuccess, Consumer<Throwable> onFailure) { private void enqueueForTellNext(TopicPartitionInfo tpi, TbMsg source, Set<String> relationTypes, String failureMessage, Runnable onSuccess, Consumer<Throwable> onFailure) {
enqueueForTellNext(tpi, source.getQueueName(), source, relationTypes, failureMessage, onSuccess, onFailure);
}
private void enqueueForTellNext(TopicPartitionInfo tpi, String queueName, TbMsg source, Set<String> relationTypes, String failureMessage, Runnable onSuccess, Consumer<Throwable> onFailure) {
RuleChainId ruleChainId = nodeCtx.getSelf().getRuleChainId(); RuleChainId ruleChainId = nodeCtx.getSelf().getRuleChainId();
RuleNodeId ruleNodeId = nodeCtx.getSelf().getId(); 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() TransportProtos.ToRuleEngineMsg.Builder msg = TransportProtos.ToRuleEngineMsg.newBuilder()
.setTenantIdMSB(getTenantId().getId().getMostSignificantBits()) .setTenantIdMSB(getTenantId().getId().getMostSignificantBits())
.setTenantIdLSB(getTenantId().getId().getLeastSignificantBits()) .setTenantIdLSB(getTenantId().getId().getLeastSignificantBits())

5
common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
View File

@ -136,8 +136,9 @@ public final class TbMsg implements Serializable {
tbMsg.data, ruleChainId, null, tbMsg.ruleNodeExecCounter.get(), tbMsg.getCallback()); tbMsg.data, ruleChainId, null, tbMsg.ruleNodeExecCounter.get(), tbMsg.getCallback());
} }
public static TbMsg newMsg(TbMsg tbMsg, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { //used for enqueueForTellNext
return new TbMsg(tbMsg.getQueueName(), UUID.randomUUID(), tbMsg.getTs(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.customerId, tbMsg.getMetaData().copy(), 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); tbMsg.getDataType(), tbMsg.getData(), ruleChainId, ruleNodeId, tbMsg.ruleNodeExecCounter.get(), TbMsgCallback.EMPTY);
} }