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 42993273e2..4fef96c000 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 @@ -66,7 +66,6 @@ import org.thingsboard.server.service.script.RuleNodeJsScriptEngine; import java.util.Collections; import java.util.Set; -import java.util.concurrent.TimeUnit; import java.util.function.Consumer; /** @@ -121,7 +120,7 @@ class DefaultTbContext implements TbContext { @Override public void enqueue(TbMsg tbMsg, String queueName, Runnable onSuccess, Consumer onFailure) { - TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, queueName, getTenantId(), tbMsg.getOriginator()); + TopicPartitionInfo tpi = resolvePartition(tbMsg, queueName); enqueue(tpi, tbMsg, onFailure, onSuccess); } @@ -138,46 +137,54 @@ class DefaultTbContext implements TbContext { @Override public void enqueueForTellFailure(TbMsg tbMsg, String failureMessage) { - TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), getTenantId(), tbMsg.getOriginator()); + TopicPartitionInfo tpi = resolvePartition(tbMsg); enqueueForTellNext(tpi, tbMsg, Collections.singleton(TbRelationTypes.FAILURE), failureMessage, null, null); } @Override public void enqueueForTellNext(TbMsg tbMsg, String relationType) { - TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), getTenantId(), tbMsg.getOriginator()); + TopicPartitionInfo tpi = resolvePartition(tbMsg); enqueueForTellNext(tpi, tbMsg, Collections.singleton(relationType), null, null, null); } @Override public void enqueueForTellNext(TbMsg tbMsg, Set relationTypes) { - TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), getTenantId(), tbMsg.getOriginator()); + TopicPartitionInfo tpi = resolvePartition(tbMsg); enqueueForTellNext(tpi, tbMsg, relationTypes, null, null, null); } @Override public void enqueueForTellNext(TbMsg tbMsg, String relationType, Runnable onSuccess, Consumer onFailure) { - TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), getTenantId(), tbMsg.getOriginator()); + TopicPartitionInfo tpi = resolvePartition(tbMsg); enqueueForTellNext(tpi, tbMsg, Collections.singleton(relationType), null, onSuccess, onFailure); } @Override public void enqueueForTellNext(TbMsg tbMsg, Set relationTypes, Runnable onSuccess, Consumer onFailure) { - TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), getTenantId(), tbMsg.getOriginator()); + TopicPartitionInfo tpi = resolvePartition(tbMsg); enqueueForTellNext(tpi, tbMsg, relationTypes, null, onSuccess, onFailure); } @Override public void enqueueForTellNext(TbMsg tbMsg, String queueName, String relationType, Runnable onSuccess, Consumer onFailure) { - TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, queueName, getTenantId(), tbMsg.getOriginator()); + TopicPartitionInfo tpi = resolvePartition(tbMsg, queueName); enqueueForTellNext(tpi, tbMsg, Collections.singleton(relationType), null, onSuccess, onFailure); } @Override public void enqueueForTellNext(TbMsg tbMsg, String queueName, Set relationTypes, Runnable onSuccess, Consumer onFailure) { - TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, queueName, getTenantId(), tbMsg.getOriginator()); + TopicPartitionInfo tpi = resolvePartition(tbMsg, queueName); enqueueForTellNext(tpi, tbMsg, relationTypes, null, onSuccess, onFailure); } + private TopicPartitionInfo resolvePartition(TbMsg tbMsg, String queueName) { + return mainCtx.resolve(ServiceType.TB_RULE_ENGINE, queueName, getTenantId(), tbMsg.getOriginator()); + } + + private TopicPartitionInfo resolvePartition(TbMsg tbMsg) { + return resolvePartition(tbMsg, tbMsg.getQueueName()); + } + private void enqueueForTellNext(TopicPartitionInfo tpi, TbMsg source, Set relationTypes, String failureMessage, Runnable onSuccess, Consumer onFailure) { RuleChainId ruleChainId = nodeCtx.getSelf().getRuleChainId(); RuleNodeId ruleNodeId = nodeCtx.getSelf().getId();