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

Merge pull request #12756 from thingsboard/fix-rc/input-node

Browse Source 
used correct queueName for TPI in input node
This commit is contained in:
Viacheslav Klimov 2025-02-26 11:16:39 +02:00 committed by GitHub
commit bd52ea5355
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 3 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -124,6 +124,7 @@ import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import static org.thingsboard.server.common.data.DataConstants.MAIN_QUEUE_NAME;
import static org.thingsboard.server.common.data.msg.TbMsgType.ATTRIBUTES_DELETED;
import static org.thingsboard.server.common.data.msg.TbMsgType.ATTRIBUTES_UPDATED;
import static org.thingsboard.server.common.data.msg.TbMsgType.ENTITY_CREATED;
@ -175,7 +176,7 @@ public class DefaultTbContext implements TbContext {
}
TbMsg tbMsg = msg.copyWithRuleChainId(ruleChainId);
tbMsg.pushToStack(nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId());
TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getQueueName(), getTenantId(), tbMsg.getOriginator());
TopicPartitionInfo tpi = resolvePartition(msg);
doEnqueue(tpi, tbMsg, new SimpleTbQueueCallback(md -> ack(msg), t -> tellFailure(msg, t)));
}
@ -192,7 +193,7 @@ public class DefaultTbContext implements TbContext {
@Override
public void enqueue(TbMsg tbMsg, Runnable onSuccess, Consumer<Throwable> onFailure) {
TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getQueueName(), getTenantId(), tbMsg.getOriginator());
TopicPartitionInfo tpi = resolvePartition(tbMsg, MAIN_QUEUE_NAME);
enqueue(tpi, tbMsg, onFailure, onSuccess);
}