|
|
|
@ -5,7 +5,7 @@ |
|
|
|
* 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
|
|
|
|
* 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, |
|
|
|
@ -130,6 +130,9 @@ class DefaultTbContext implements TbContext { |
|
|
|
.setTenantIdMSB(getTenantId().getId().getMostSignificantBits()) |
|
|
|
.setTenantIdLSB(getTenantId().getId().getLeastSignificantBits()) |
|
|
|
.setTbMsg(TbMsg.toByteString(tbMsg)).build(); |
|
|
|
if (nodeCtx.getSelf().isDebugMode()) { |
|
|
|
mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), tbMsg, "To Root Rule Chain"); |
|
|
|
} |
|
|
|
mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg, new SimpleTbQueueCallback(onSuccess, onFailure)); |
|
|
|
} |
|
|
|
|
|
|
|
@ -175,10 +178,10 @@ class DefaultTbContext implements TbContext { |
|
|
|
enqueueForTellNext(tpi, tbMsg, relationTypes, null, onSuccess, onFailure); |
|
|
|
} |
|
|
|
|
|
|
|
private void enqueueForTellNext(TopicPartitionInfo tpi, TbMsg tbMsg, 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) { |
|
|
|
RuleChainId ruleChainId = nodeCtx.getSelf().getRuleChainId(); |
|
|
|
RuleNodeId ruleNodeId = nodeCtx.getSelf().getId(); |
|
|
|
tbMsg = TbMsg.newMsg(tbMsg, ruleChainId, ruleNodeId); |
|
|
|
TbMsg tbMsg = TbMsg.newMsg(source, ruleChainId, ruleNodeId); |
|
|
|
TransportProtos.ToRuleEngineMsg.Builder msg = TransportProtos.ToRuleEngineMsg.newBuilder() |
|
|
|
.setTenantIdMSB(getTenantId().getId().getMostSignificantBits()) |
|
|
|
.setTenantIdLSB(getTenantId().getId().getLeastSignificantBits()) |
|
|
|
@ -187,6 +190,10 @@ class DefaultTbContext implements TbContext { |
|
|
|
if (failureMessage != null) { |
|
|
|
msg.setFailureMessage(failureMessage); |
|
|
|
} |
|
|
|
if (nodeCtx.getSelf().isDebugMode()) { |
|
|
|
relationTypes.forEach(relationType -> |
|
|
|
mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), tbMsg, relationType)); |
|
|
|
} |
|
|
|
mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg.build(), new SimpleTbQueueCallback(onSuccess, onFailure)); |
|
|
|
} |
|
|
|
|
|
|
|
|