Use consumer instead of runnable in SimpleTbQueueCallback

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

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<TbQueueMsgMetadata> onSuccess;
     private final Consumer<Throwable> onFailure;
 
-    public SimpleTbQueueCallback(Runnable onSuccess, Consumer<Throwable> onFailure) {
+    public SimpleTbQueueCallback(Consumer<TbQueueMsgMetadata> onSuccess, Consumer<Throwable> 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);
         }
     }
 

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));
     }
 
 }