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 baae03761b..1e7d5d24e9 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 @@ -214,13 +214,19 @@ class DefaultTbContext implements TbContext { 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, t -> { - if (onFailure != null) { - onFailure.accept(t); - } else { - log.debug("[{}] Failed to put item into queue!", nodeCtx.getTenantId().getId(), t); - } - })); + mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg, new SimpleTbQueueCallback( + metadata -> { + if (onSuccess != null) { + onSuccess.run(); + } + }, + t -> { + if (onFailure != null) { + onFailure.accept(t); + } else { + log.debug("[{}] Failed to put item into queue!", nodeCtx.getTenantId().getId(), t); + } + })); } @Override @@ -306,13 +312,19 @@ class DefaultTbContext implements TbContext { relationTypes.forEach(relationType -> mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), tbMsg, relationType, null, failureMessage)); } - mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg.build(), new SimpleTbQueueCallback(onSuccess, t -> { - if (onFailure != null) { - onFailure.accept(t); - } else { - log.debug("[{}] Failed to put item into queue!", nodeCtx.getTenantId().getId(), t); - } - })); + mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg.build(), new SimpleTbQueueCallback( + metadata -> { + if (onSuccess != null) { + onSuccess.run(); + } + }, + t -> { + if (onFailure != null) { + onFailure.accept(t); + } else { + log.debug("[{}] Failed to put item into queue!", nodeCtx.getTenantId().getId(), t); + } + })); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/SimpleTbQueueCallback.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/SimpleTbQueueCallback.java index 5f86060d48..d153998ab4 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/SimpleTbQueueCallback.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/SimpleTbQueueCallback.java @@ -22,10 +22,10 @@ import java.util.function.Consumer; public class SimpleTbQueueCallback implements TbQueueCallback { - private final Runnable onSuccess; + private final Consumer onSuccess; private final Consumer onFailure; - public SimpleTbQueueCallback(Runnable onSuccess, Consumer onFailure) { + public SimpleTbQueueCallback(Consumer onSuccess, Consumer onFailure) { this.onSuccess = onSuccess; this.onFailure = onFailure; } @@ -33,7 +33,7 @@ public class SimpleTbQueueCallback implements TbQueueCallback { @Override public void onSuccess(TbQueueMsgMetadata metadata) { if (onSuccess != null) { - onSuccess.run(); + onSuccess.accept(metadata); } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeviceStateNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeviceStateNode.java index ccc252c912..65d7e0979e 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeviceStateNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeviceStateNode.java @@ -183,7 +183,7 @@ public class TbDeviceStateNode implements TbNode { } private TbQueueCallback getMsgEnqueuedCallback(TbContext ctx, TbMsg msg) { - return new SimpleTbQueueCallback(() -> ctx.tellSuccess(msg), t -> ctx.tellFailure(msg, t)); + return new SimpleTbQueueCallback(metadata -> ctx.tellSuccess(msg), t -> ctx.tellFailure(msg, t)); } }