Input and Output rule nodes and message routing

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

@@ -56,6 +56,7 @@ import org.thingsboard.server.common.data.rule.RuleNodeState;
 import org.thingsboard.server.common.msg.TbActorMsg;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.TbMsgProcessingStackItem;
 import org.thingsboard.server.common.msg.queue.ServiceQueue;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
@@ -134,6 +135,25 @@ class DefaultTbContext implements TbContext {
         scheduleMsgWithDelay(new RuleNodeToSelfMsg(this, msg), delayMs, nodeCtx.getSelfActor());
     }
 
+    @Override
+    public void input(TbMsg msg, RuleChainId ruleChainId) {
+        msg.pushToStack(nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId());
+        nodeCtx.getChainActor().tell(new RuleChainInputMsg(ruleChainId, msg));
+    }
+
+    @Override
+    public void output(TbMsg msg, String relationType) {
+        TbMsgProcessingStackItem item = msg.popFormStack();
+        if (item == null) {
+            ack(msg);
+        } else {
+            if (nodeCtx.getSelf().isDebugMode()) {
+                mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg, relationType);
+            }
+            nodeCtx.getChainActor().tell(new RuleChainOutputMsg(item.getRuleChainId(), item.getRuleNodeId(), relationType, msg));
+        }
+    }
+
     @Override
     public void enqueue(TbMsg tbMsg, Runnable onSuccess, Consumer<Throwable> onFailure) {
         TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), tbMsg.getOriginator());

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java

@@ -60,6 +60,12 @@ public class RuleChainActor extends ComponentActor<RuleChainId, RuleChainActorMe
             case RULE_CHAIN_TO_RULE_CHAIN_MSG:
                 processor.onRuleChainToRuleChainMsg((RuleChainToRuleChainMsg) msg);
                 break;
+            case RULE_CHAIN_INPUT_MSG:
+                processor.onRuleChainInputMsg((RuleChainInputMsg) msg);
+                break;
+            case RULE_CHAIN_OUTPUT_MSG:
+                processor.onRuleChainOutputMsg((RuleChainOutputMsg) msg);
+                break;
             case PARTITION_CHANGE_MSG:
                 processor.onPartitionChangeMsg((PartitionChangeMsg) msg);
                 break;

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java

@@ -66,6 +66,7 @@ import java.util.stream.Collectors;
 @Slf4j
 public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleChainId> {
 
+    private static final String NA_RELATION_TYPE = "";
     private final TbActorRef parent;
     private final TbActorRef self;
     private final Map<RuleNodeId, RuleNodeCtx> nodeActors;
@@ -201,33 +202,58 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
         TbMsg msg = envelope.getMsg();
         log.trace("[{}][{}] Processing message [{}]: {}", entityId, firstId, msg.getId(), msg);
         if (envelope.getRelationTypes() == null || envelope.getRelationTypes().isEmpty()) {
-            try {
+            onTellNext(msg, true);
-                checkActive(envelope.getMsg());
-                RuleNodeId targetId = msg.getRuleNodeId();
-                RuleNodeCtx targetCtx;
-                if (targetId == null) {
-                    targetCtx = firstNode;
-                    msg = msg.copyWithRuleChainId(entityId);
-                } else {
-                    targetCtx = nodeActors.get(targetId);
-                }
-                if (targetCtx != null) {
-                    log.trace("[{}][{}] Pushing message to target rule node", entityId, targetId);
-                    pushMsgToNode(targetCtx, msg, "");
-                } else {
-                    log.trace("[{}][{}] Rule node does not exist. Probably old message", entityId, targetId);
-                    msg.getCallback().onSuccess();
-                }
-            } catch (RuleNodeException rne) {
-                envelope.getMsg().getCallback().onFailure(rne);
-            } catch (Exception e) {
-                envelope.getMsg().getCallback().onFailure(new RuleEngineException(e.getMessage()));
-            }
         } else {
             onTellNext(envelope.getMsg(), envelope.getMsg().getRuleNodeId(), envelope.getRelationTypes(), envelope.getFailureMessage());
         }
     }
 
+    private void onTellNext(TbMsg msg, boolean useRuleNodeIdFromMsg) {
+        try {
+            checkActive(msg);
+            RuleNodeId targetId = useRuleNodeIdFromMsg ? msg.getRuleNodeId() : null;
+            RuleNodeCtx targetCtx;
+            if (targetId == null) {
+                targetCtx = firstNode;
+                msg = msg.copyWithRuleChainId(entityId);
+            } else {
+                targetCtx = nodeActors.get(targetId);
+            }
+            if (targetCtx != null) {
+                log.trace("[{}][{}] Pushing message to target rule node", entityId, targetId);
+                pushMsgToNode(targetCtx, msg, NA_RELATION_TYPE);
+            } else {
+                log.trace("[{}][{}] Rule node does not exist. Probably old message", entityId, targetId);
+                msg.getCallback().onSuccess();
+            }
+        } catch (RuleNodeException rne) {
+            msg.getCallback().onFailure(rne);
+        } catch (Exception e) {
+            msg.getCallback().onFailure(new RuleEngineException(e.getMessage()));
+        }
+    }
+
+    public void onRuleChainInputMsg(RuleChainInputMsg envelope) {
+        if (entityId.equals(envelope.getRuleChainId())) {
+            onTellNext(envelope.getMsg(), false);
+        } else {
+            parent.tell(envelope);
+        }
+    }
+
+    public void onRuleChainOutputMsg(RuleChainOutputMsg envelope) {
+        if (entityId.equals(envelope.getRuleChainId())) {
+            var originatorNodeId = envelope.getTargetRuleNodeId();
+            RuleNodeCtx ruleNodeCtx = nodeActors.get(originatorNodeId);
+            if (ruleNodeCtx != null && ruleNodeCtx.getSelf().isDebugMode()) {
+                systemContext.persistDebugOutput(tenantId, originatorNodeId, envelope.getMsg(), envelope.getRelationType());
+            }
+            onTellNext(envelope.getMsg(), originatorNodeId, Collections.singleton(envelope.getRelationType()), RuleNodeException.UNKNOWN);
+        } else {
+            parent.tell(envelope);
+        }
+    }
+
     void onRuleChainToRuleChainMsg(RuleChainToRuleChainMsg envelope) {
         try {
             checkActive(envelope.getMsg());
@@ -356,4 +382,5 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
         RuleNode firstRuleNode = firstNode != null ? firstNode.getSelf() : null;
         return new RuleNodeException("Rule Chain is not active!  Failed to initialize.", ruleChainName, firstRuleNode);
     }
+
 }

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainInputMsg.java

@@ -0,0 +1,41 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.actors.ruleChain;
+
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import org.thingsboard.server.common.data.id.RuleChainId;
+import org.thingsboard.server.common.data.id.RuleNodeId;
+import org.thingsboard.server.common.msg.MsgType;
+import org.thingsboard.server.common.msg.TbMsg;
+
+/**
+ * Created by ashvayka on 19.03.18.
+ */
+@EqualsAndHashCode(callSuper = true)
+@ToString
+public final class RuleChainInputMsg extends TbToRuleChainActorMsg {
+
+    public RuleChainInputMsg(RuleChainId target, TbMsg tbMsg) {
+        super(tbMsg, target);
+    }
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.RULE_CHAIN_INPUT_MSG;
+    }
+}

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainOutputMsg.java

@@ -0,0 +1,49 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.actors.ruleChain;
+
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import org.thingsboard.server.common.data.id.RuleChainId;
+import org.thingsboard.server.common.data.id.RuleNodeId;
+import org.thingsboard.server.common.msg.MsgType;
+import org.thingsboard.server.common.msg.TbMsg;
+
+/**
+ * Created by ashvayka on 19.03.18.
+ */
+@EqualsAndHashCode(callSuper = true)
+@ToString
+public final class RuleChainOutputMsg extends TbToRuleChainActorMsg {
+
+    @Getter
+    private final RuleNodeId targetRuleNodeId;
+
+    @Getter
+    private final String relationType;
+
+    public RuleChainOutputMsg(RuleChainId target, RuleNodeId targetRuleNodeId, String relationType, TbMsg tbMsg) {
+        super(tbMsg, target);
+        this.targetRuleNodeId = targetRuleNodeId;
+        this.relationType = relationType;
+    }
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.RULE_CHAIN_OUTPUT_MSG;
+    }
+}

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainToRuleChainMsg.java

@@ -31,33 +31,19 @@ import org.thingsboard.server.common.msg.queue.RuleEngineException;
  */
 @EqualsAndHashCode(callSuper = true)
 @ToString
-public final class RuleChainToRuleChainMsg extends TbRuleEngineActorMsg implements RuleChainAwareMsg {
+public final class RuleChainToRuleChainMsg extends TbToRuleChainActorMsg  {
 
-    @Getter
-    private final RuleChainId target;
     @Getter
     private final RuleChainId source;
     @Getter
     private final String fromRelationType;
 
     public RuleChainToRuleChainMsg(RuleChainId target, RuleChainId source, TbMsg tbMsg, String fromRelationType) {
-        super(tbMsg);
+        super(tbMsg, target);
-        this.target = target;
         this.source = source;
         this.fromRelationType = fromRelationType;
     }
 
-    @Override
-    public void onTbActorStopped(TbActorStopReason reason) {
-        String message = reason == TbActorStopReason.STOPPED ? String.format("Rule chain [%s] stopped", target.getId()) : String.format("Failed to initialize rule chain [%s]!", target.getId());
-        msg.getCallback().onFailure(new RuleEngineException(message));
-    }
-
-    @Override
-    public RuleChainId getRuleChainId() {
-        return target;
-    }
-
     @Override
     public MsgType getMsgType() {
         return MsgType.RULE_CHAIN_TO_RULE_CHAIN_MSG;

application/src/main/java/org/thingsboard/server/actors/ruleChain/TbToRuleChainActorMsg.java

@@ -0,0 +1,50 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.actors.ruleChain;
+
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import org.thingsboard.server.common.data.id.RuleChainId;
+import org.thingsboard.server.common.msg.TbActorStopReason;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbRuleEngineActorMsg;
+import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg;
+import org.thingsboard.server.common.msg.queue.RuleEngineException;
+
+@EqualsAndHashCode(callSuper = true)
+@ToString
+public abstract class TbToRuleChainActorMsg extends TbRuleEngineActorMsg implements RuleChainAwareMsg {
+
+    @Getter
+    private final RuleChainId target;
+
+    public TbToRuleChainActorMsg(TbMsg msg, RuleChainId target) {
+        super(msg);
+        this.target = target;
+    }
+
+    @Override
+    public RuleChainId getRuleChainId() {
+        return target;
+    }
+
+    @Override
+    public void onTbActorStopped(TbActorStopReason reason) {
+        String message = reason == TbActorStopReason.STOPPED ? String.format("Rule chain [%s] stopped", target.getId()) : String.format("Failed to initialize rule chain [%s]!", target.getId());
+        msg.getCallback().onFailure(new RuleEngineException(message));
+    }
+}

application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java

@@ -26,7 +26,9 @@ import org.thingsboard.server.actors.TbActorRef;
 import org.thingsboard.server.actors.TbEntityActorId;
 import org.thingsboard.server.actors.TbEntityTypeActorIdPredicate;
 import org.thingsboard.server.actors.device.DeviceActorCreator;
+import org.thingsboard.server.actors.ruleChain.RuleChainInputMsg;
 import org.thingsboard.server.actors.ruleChain.RuleChainManagerActor;
+import org.thingsboard.server.actors.ruleChain.RuleChainOutputMsg;
 import org.thingsboard.server.actors.service.ContextBasedCreator;
 import org.thingsboard.server.actors.service.DefaultActorService;
 import org.thingsboard.server.common.data.ApiUsageState;
@@ -168,6 +170,8 @@ public class TenantActor extends RuleChainManagerActor {
             case REMOVE_RPC_TO_DEVICE_ACTOR_MSG:
                 onToDeviceActorMsg((DeviceAwareMsg) msg, true);
                 break;
+            case RULE_CHAIN_INPUT_MSG:
+            case RULE_CHAIN_OUTPUT_MSG:
             case RULE_CHAIN_TO_RULE_CHAIN_MSG:
                 onRuleChainMsg((RuleChainAwareMsg) msg);
                 break;

common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java

@@ -61,6 +61,16 @@ public enum MsgType {
      */
     RULE_CHAIN_TO_RULE_CHAIN_MSG,
 
+    /**
+     * Message that is sent by RuleNodeActor as input to other RuleChain with command to process TbMsg.
+     */
+    RULE_CHAIN_INPUT_MSG,
+
+    /**
+     * Message that is sent by RuleNodeActor as output to RuleNode in other RuleChain with command to process TbMsg.
+     */
+    RULE_CHAIN_OUTPUT_MSG,
+
     /**
      * Message that is sent by RuleActor to RuleChainActor with command to process TbMsg by next nodes in chain.
      */

common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java

@@ -55,24 +55,25 @@ public final class TbMsg implements Serializable {
     private final RuleChainId ruleChainId;
     private final RuleNodeId ruleNodeId;
     @Getter(value = AccessLevel.NONE)
-    private final AtomicInteger ruleNodeExecCounter;
+    @JsonIgnore
-
+    //This field is not serialized because we use queues and there is no need to do it
-
+    private final TbMsgProcessingCtx ctx;
-    public int getAndIncrementRuleNodeCounter() {
-        return ruleNodeExecCounter.getAndIncrement();
-    }
 
     //This field is not serialized because we use queues and there is no need to do it
     @JsonIgnore
     transient private final TbMsgCallback callback;
 
+    public int getAndIncrementRuleNodeCounter() {
+        return ctx.getAndIncrementRuleNodeCounter();
+    }
+
     public static TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
         return newMsg(queueName, type, originator, null, metaData, data, ruleChainId, ruleNodeId);
     }
 
     public static TbMsg newMsg(String queueName, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
         return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
-                metaData.copy(), TbMsgDataType.JSON, data, ruleChainId, ruleNodeId, 0, TbMsgCallback.EMPTY);
+                metaData.copy(), TbMsgDataType.JSON, data, ruleChainId, ruleNodeId, null, TbMsgCallback.EMPTY);
     }
 
     public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data) {
@@ -80,7 +81,8 @@ public final class TbMsg implements Serializable {
     }
 
     public static TbMsg newMsg(String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
-        return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId, metaData.copy(), TbMsgDataType.JSON, data, null, null, 0, TbMsgCallback.EMPTY);
+        return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
+                metaData.copy(), TbMsgDataType.JSON, data, null, null, null, TbMsgCallback.EMPTY);
     }
 
     // REALLY NEW MSG
@@ -90,11 +92,13 @@ public final class TbMsg implements Serializable {
     }
 
     public static TbMsg newMsg(String queueName, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
-        return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId, metaData.copy(), TbMsgDataType.JSON, data, null, null, 0, TbMsgCallback.EMPTY);
+        return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
+                metaData.copy(), TbMsgDataType.JSON, data, null, null, null, TbMsgCallback.EMPTY);
     }
 
     public static TbMsg newMsg(String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data) {
-        return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId, metaData.copy(), dataType, data, null, null, 0, TbMsgCallback.EMPTY);
+        return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
+                metaData.copy(), dataType, data, null, null, null, TbMsgCallback.EMPTY);
     }
 
     public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data) {
@@ -104,46 +108,48 @@ public final class TbMsg implements Serializable {
     // For Tests only
 
     public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
-        return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null, metaData.copy(), dataType, data, ruleChainId, ruleNodeId, 0, TbMsgCallback.EMPTY);
+        return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null,
+                metaData.copy(), dataType, data, ruleChainId, ruleNodeId, null, TbMsgCallback.EMPTY);
     }
 
     public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data, TbMsgCallback callback) {
-        return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null, metaData.copy(), TbMsgDataType.JSON, data, null, null, 0, callback);
+        return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null,
+                metaData.copy(), TbMsgDataType.JSON, data, null, null, null, callback);
     }
 
     public static TbMsg transformMsg(TbMsg tbMsg, String type, EntityId originator, TbMsgMetaData metaData, String data) {
         return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, type, originator, tbMsg.customerId, metaData.copy(), tbMsg.dataType,
-                data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ruleNodeExecCounter.get(), tbMsg.callback);
+                data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.callback);
     }
 
     public static TbMsg transformMsg(TbMsg tbMsg, CustomerId customerId) {
         return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, customerId, tbMsg.metaData, tbMsg.dataType,
-                tbMsg.data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ruleNodeExecCounter.get(), tbMsg.getCallback());
+                tbMsg.data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     public static TbMsg transformMsg(TbMsg tbMsg, RuleChainId ruleChainId) {
         return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
-                tbMsg.data, ruleChainId, null, tbMsg.ruleNodeExecCounter.get(), tbMsg.getCallback());
+                tbMsg.data, ruleChainId, null, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     public static TbMsg transformMsg(TbMsg tbMsg, String queueName) {
         return new TbMsg(queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
-                tbMsg.data, tbMsg.getRuleChainId(), null, tbMsg.ruleNodeExecCounter.get(), tbMsg.getCallback());
+                tbMsg.data, tbMsg.getRuleChainId(), null, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     public static TbMsg transformMsg(TbMsg tbMsg, RuleChainId ruleChainId, String queueName) {
         return new TbMsg(queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
-                tbMsg.data, ruleChainId, null, tbMsg.ruleNodeExecCounter.get(), tbMsg.getCallback());
+                tbMsg.data, ruleChainId, null, tbMsg.ctx.copy(), tbMsg.getCallback());
     }
 
     //used for enqueueForTellNext
     public static TbMsg newMsg(TbMsg tbMsg, String queueName, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
         return new TbMsg(queueName, UUID.randomUUID(), tbMsg.getTs(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.customerId, tbMsg.getMetaData().copy(),
-                tbMsg.getDataType(), tbMsg.getData(), ruleChainId, ruleNodeId, tbMsg.ruleNodeExecCounter.get(), TbMsgCallback.EMPTY);
+                tbMsg.getDataType(), tbMsg.getData(), ruleChainId, ruleNodeId, tbMsg.ctx.copy(), TbMsgCallback.EMPTY);
     }
 
     private TbMsg(String queueName, UUID id, long ts, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data,
-                  RuleChainId ruleChainId, RuleNodeId ruleNodeId, int ruleNodeExecCounter, TbMsgCallback callback) {
+                  RuleChainId ruleChainId, RuleNodeId ruleNodeId, TbMsgProcessingCtx ctx, TbMsgCallback callback) {
         this.id = id;
         this.queueName = queueName != null ? queueName : ServiceQueue.MAIN;
         if (ts > 0) {
@@ -167,7 +173,7 @@ public final class TbMsg implements Serializable {
         this.data = data;
         this.ruleChainId = ruleChainId;
         this.ruleNodeId = ruleNodeId;
-        this.ruleNodeExecCounter = new AtomicInteger(ruleNodeExecCounter);
+        this.ctx = ctx != null ? ctx : new TbMsgProcessingCtx();
         if (callback != null) {
             this.callback = callback;
         } else {
@@ -209,7 +215,8 @@ public final class TbMsg implements Serializable {
 
         builder.setDataType(msg.getDataType().ordinal());
         builder.setData(msg.getData());
-        builder.setRuleNodeExecCounter(msg.ruleNodeExecCounter.get());
+
+        builder.setCtx(msg.ctx.toProto());
         return builder.build().toByteArray();
     }
 
@@ -231,8 +238,17 @@ public final class TbMsg implements Serializable {
                 ruleNodeId = new RuleNodeId(new UUID(proto.getRuleNodeIdMSB(), proto.getRuleNodeIdLSB()));
             }
 
+            TbMsgProcessingCtx ctx;
+            if (proto.hasCtx()) {
+                ctx = TbMsgProcessingCtx.fromProto(proto.getCtx());
+            } else {
+                // Backward compatibility with unprocessed messages fetched from queue after update.
+                ctx = new TbMsgProcessingCtx(proto.getRuleNodeExecCounter());
+            }
+
             TbMsgDataType dataType = TbMsgDataType.values()[proto.getDataType()];
-            return new TbMsg(queueName, UUID.fromString(proto.getId()), proto.getTs(), proto.getType(), entityId, customerId, metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, proto.getRuleNodeExecCounter(), callback);
+            return new TbMsg(queueName, UUID.fromString(proto.getId()), proto.getTs(), proto.getType(), entityId, customerId,
+                    metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, ctx, callback);
         } catch (InvalidProtocolBufferException e) {
             throw new IllegalStateException("Could not parse protobuf for TbMsg", e);
         }
@@ -243,11 +259,13 @@ public final class TbMsg implements Serializable {
     }
 
     public TbMsg copyWithRuleChainId(RuleChainId ruleChainId, UUID msgId) {
-        return new TbMsg(this.queueName, msgId, this.ts, this.type, this.originator, this.customerId, this.metaData, this.dataType, this.data, ruleChainId, null, this.ruleNodeExecCounter.get(), callback);
+        return new TbMsg(this.queueName, msgId, this.ts, this.type, this.originator, this.customerId,
+                this.metaData, this.dataType, this.data, ruleChainId, null, this.ctx, callback);
     }
 
     public TbMsg copyWithRuleNodeId(RuleChainId ruleChainId, RuleNodeId ruleNodeId, UUID msgId) {
-        return new TbMsg(this.queueName, msgId, this.ts, this.type, this.originator, this.customerId, this.metaData, this.dataType, this.data, ruleChainId, ruleNodeId, this.ruleNodeExecCounter.get(), callback);
+        return new TbMsg(this.queueName, msgId, this.ts, this.type, this.originator, this.customerId,
+                this.metaData, this.dataType, this.data, ruleChainId, ruleNodeId, this.ctx, callback);
     }
 
     public TbMsgCallback getCallback() {
@@ -262,4 +280,12 @@ public final class TbMsg implements Serializable {
     public String getQueueName() {
         return queueName != null ? queueName : ServiceQueue.MAIN;
     }
+
+    public void pushToStack(RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
+        ctx.push(ruleChainId, ruleNodeId);
+    }
+
+    public TbMsgProcessingStackItem popFormStack() {
+        return ctx.pop();
+    }
 }

common/message/src/main/java/org/thingsboard/server/common/msg/TbMsgProcessingCtx.java

@@ -0,0 +1,94 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.msg;
+
+import org.thingsboard.server.common.data.id.RuleChainId;
+import org.thingsboard.server.common.data.id.RuleNodeId;
+import org.thingsboard.server.common.msg.gen.MsgProtos;
+
+import java.io.Serializable;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Created by ashvayka on 13.01.18.
+ */
+public final class TbMsgProcessingCtx implements Serializable {
+
+    private final AtomicInteger ruleNodeExecCounter;
+    private volatile LinkedList<TbMsgProcessingStackItem> stack;
+
+    public TbMsgProcessingCtx() {
+        this(0);
+    }
+
+    public TbMsgProcessingCtx(int ruleNodeExecCounter) {
+        this(ruleNodeExecCounter, null);
+    }
+
+    protected TbMsgProcessingCtx(int ruleNodeExecCounter, LinkedList<TbMsgProcessingStackItem> stack) {
+        this.ruleNodeExecCounter = new AtomicInteger(ruleNodeExecCounter);
+        this.stack = stack;
+    }
+
+    public int getAndIncrementRuleNodeCounter() {
+        return ruleNodeExecCounter.getAndIncrement();
+    }
+
+    public TbMsgProcessingCtx copy() {
+        return new TbMsgProcessingCtx(ruleNodeExecCounter.get());
+    }
+
+    public void push(RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
+        if (stack == null) {
+            stack = new LinkedList<>();
+        }
+        stack.add(new TbMsgProcessingStackItem(ruleChainId, ruleNodeId));
+    }
+
+    public TbMsgProcessingStackItem pop() {
+        return !stack.isEmpty() ? stack.removeLast() : null;
+    }
+
+    public static TbMsgProcessingCtx fromProto(MsgProtos.TbMsgProcessingCtxProto ctx) {
+        int ruleNodeExecCounter = ctx.getRuleNodeExecCounter();
+        if (ctx.getStackCount() > 0) {
+            LinkedList<TbMsgProcessingStackItem> stack = new LinkedList<>();
+            for (MsgProtos.TbMsgProcessingStackItemProto item : ctx.getStackList()) {
+                stack.add(new TbMsgProcessingStackItem(
+                        new RuleChainId(new UUID(item.getRuleChainIdMSB(), item.getRuleChainIdLSB())),
+                        new RuleNodeId(new UUID(item.getRuleNodeIdMSB(), item.getRuleNodeIdLSB()))
+                ));
+            }
+            return new TbMsgProcessingCtx(ruleNodeExecCounter, stack);
+        } else {
+            return new TbMsgProcessingCtx(ruleNodeExecCounter);
+        }
+    }
+
+    public MsgProtos.TbMsgProcessingCtxProto toProto() {
+        var ctxBuilder = MsgProtos.TbMsgProcessingCtxProto.newBuilder();
+        ctxBuilder.setRuleNodeExecCounter(ruleNodeExecCounter.get());
+        if (stack != null) {
+            for (TbMsgProcessingStackItem item : stack) {
+                ctxBuilder.addStack(item.toProto());
+            }
+        }
+        return ctxBuilder.build();
+    }
+}

common/message/src/main/java/org/thingsboard/server/common/msg/TbMsgProcessingStackItem.java

@@ -0,0 +1,38 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.msg;
+
+import lombok.Data;
+import org.thingsboard.server.common.data.id.RuleChainId;
+import org.thingsboard.server.common.data.id.RuleNodeId;
+import org.thingsboard.server.common.msg.gen.MsgProtos;
+
+@Data
+public class TbMsgProcessingStackItem {
+
+    private final RuleChainId ruleChainId;
+    private final RuleNodeId ruleNodeId;
+
+    MsgProtos.TbMsgProcessingStackItemProto toProto() {
+        return MsgProtos.TbMsgProcessingStackItemProto.newBuilder()
+                .setRuleChainIdMSB(ruleChainId.getId().getMostSignificantBits())
+                .setRuleChainIdLSB(ruleChainId.getId().getLeastSignificantBits())
+                .setRuleNodeIdMSB(ruleNodeId.getId().getMostSignificantBits())
+                .setRuleNodeIdLSB(ruleNodeId.getId().getLeastSignificantBits())
+                .build();
+    }
+
+}

common/message/src/main/java/org/thingsboard/server/common/msg/queue/RuleNodeException.java

@@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.rule.RuleNode;
 public class RuleNodeException extends RuleEngineException {
 
     private static final long serialVersionUID = -1776681087370749776L;
+    public static final String UNKNOWN = "Unknown";
 
     @Getter
     private final String ruleChainName;
@@ -45,7 +46,7 @@ public class RuleNodeException extends RuleEngineException {
             this.ruleChainId = ruleNode.getRuleChainId();
             this.ruleNodeId = ruleNode.getId();
         } else {
-            ruleNodeName = "Unknown";
+            ruleNodeName = UNKNOWN;
             ruleChainId = new RuleChainId(RuleChainId.NULL_UUID);
             ruleNodeId = new RuleNodeId(RuleNodeId.NULL_UUID);
         }

common/message/src/main/proto/tbmsg.proto

@@ -19,10 +19,24 @@ package msgqueue;
 option java_package = "org.thingsboard.server.common.msg.gen";
 option java_outer_classname = "MsgProtos";
 
+// Stores message metadata as map of strings
 message TbMsgMetaDataProto {
     map<string, string> data = 1;
 }
 
+// Stores stack of nested (caller) rule chains
+message TbMsgProcessingStackItemProto {
+    int64 ruleChainIdMSB = 1;
+    int64 ruleChainIdLSB = 2;
+    int64 ruleNodeIdMSB = 3;
+    int64 ruleNodeIdLSB = 4;
+}
+
+message TbMsgProcessingCtxProto {
+    int32 ruleNodeExecCounter = 1;
+    repeated TbMsgProcessingStackItemProto stack = 2;
+}
+
 message TbMsgProto {
     string id = 1;
     string type = 2;
@@ -39,14 +53,17 @@ message TbMsgProto {
 
     TbMsgMetaDataProto metaData = 11;
 
-    //Transaction Data (12) was removed in 2.5
+    // Transaction Data (12) was removed in 2.5
 
     int32 dataType = 13;
     string data = 14;
 
     int64 ts = 15;
+    // Will be removed in 3.4. Moved to processing context
     int32 ruleNodeExecCounter = 16;
 
     int64 customerIdMSB = 17;
     int64 customerIdLSB = 18;
+
+    TbMsgProcessingCtxProto ctx = 19;
 }

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java

@@ -29,6 +29,7 @@ import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.EdgeId;
 import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.RuleChainId;
 import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
@@ -121,6 +122,24 @@ public interface TbContext {
      */
     void enqueue(TbMsg msg, Runnable onSuccess, Consumer<Throwable> onFailure);
 
+    /**
+     * Sends message to the nested rule chain.
+     * Fails processing of the message if the nested rule chain is not found.
+     *
+     * @param msg - the message
+     * @param ruleChainId - the id of a nested rule chain
+     */
+    void input(TbMsg msg, RuleChainId ruleChainId);
+
+    /**
+     * Sends message to the caller rule chain.
+     * Acknowledge the message if no caller rule chain is present in processing stack
+     *
+     * @param msg - the message
+     * @param relationType - the relation type that will be used to route messages in the caller rule chain
+     */
+    void output(TbMsg msg, String relationType);
+
     /**
      * Puts new message to custom queue for processing
      *

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNode.java

@@ -0,0 +1,65 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.flow;
+
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.rule.engine.api.RuleNode;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNode;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.rule.engine.api.TbRelationTypes;
+import org.thingsboard.rule.engine.api.util.TbNodeUtils;
+import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.id.EntityIdFactory;
+import org.thingsboard.server.common.data.id.RuleChainId;
+import org.thingsboard.server.common.data.plugin.ComponentType;
+import org.thingsboard.server.common.msg.TbMsg;
+
+import java.util.UUID;
+
+@Slf4j
+@RuleNode(
+        type = ComponentType.ACTION,
+        name = "rule chain input",
+        configClazz = TbRuleChainInputNodeConfiguration.class,
+        nodeDescription = "transfers the message to another rule chain",
+        nodeDetails = "Allows to nest the rule chain similar to single rule node. " +
+                "The incoming message is forwarded to the input node of the specified target rule chain. " +
+                "The target rule chain may produce multiple labeled outputs. " +
+                "You may use the outputs to forward the results of processing to other rule nodes.",
+        customRelations = true
+)
+public class TbRuleChainInputNode implements TbNode {
+
+    private TbRuleChainInputNodeConfiguration config;
+    private RuleChainId ruleChainId;
+
+    @Override
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
+        this.config = TbNodeUtils.convert(configuration, TbRuleChainInputNodeConfiguration.class);
+        this.ruleChainId = new RuleChainId(UUID.fromString(config.getRuleChainId()));
+    }
+
+    @Override
+    public void onMsg(TbContext ctx, TbMsg msg) {
+        ctx.input(msg, ruleChainId);
+    }
+
+    @Override
+    public void destroy() {
+    }
+}

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbRuleChainInputNodeConfiguration.java

@@ -0,0 +1,32 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.flow;
+
+import lombok.Data;
+import org.thingsboard.rule.engine.api.NodeConfiguration;
+import org.thingsboard.server.common.data.id.RuleChainId;
+
+@Data
+public class TbRuleChainInputNodeConfiguration implements NodeConfiguration<TbRuleChainInputNodeConfiguration> {
+
+    private String ruleChainId;
+
+    @Override
+    public TbRuleChainInputNodeConfiguration defaultConfiguration() {
+        return new TbRuleChainInputNodeConfiguration();
+    }
+
+}

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbRuleChainOutputNode.java

@@ -0,0 +1,58 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.flow;
+
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.rule.engine.api.RuleNode;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.rule.engine.api.TbNode;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.rule.engine.api.TbRelationTypes;
+import org.thingsboard.rule.engine.api.util.TbNodeUtils;
+import org.thingsboard.server.common.data.plugin.ComponentType;
+import org.thingsboard.server.common.msg.TbMsg;
+
+@Slf4j
+@RuleNode(
+        type = ComponentType.ACTION,
+        name = "rule chain output",
+        configClazz = TbRuleChainOutputNodeConfiguration.class,
+        nodeDescription = "transfers the message to the caller rule chain",
+        nodeDetails = "Produces output of the rule chain processing. " +
+                "The output is forwarded to the caller rule chain, as an output of the corresponding \"input\" rule node. " +
+                "The rule node configuration contains the \"label\" parameter. " +
+                "This parameter corresponds to the relation type of the output message, and it is used to forward messages to other rule nodes in the caller rule chain. ",
+        outEnabled = false
+)
+public class TbRuleChainOutputNode implements TbNode {
+
+    private TbRuleChainOutputNodeConfiguration config;
+
+    @Override
+    public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
+        this.config = TbNodeUtils.convert(configuration, TbRuleChainOutputNodeConfiguration.class);
+    }
+
+    @Override
+    public void onMsg(TbContext ctx, TbMsg msg) {
+        ctx.output(msg, config.getLabel());
+    }
+
+    @Override
+    public void destroy() {
+    }
+}

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbRuleChainOutputNodeConfiguration.java

@@ -0,0 +1,35 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.flow;
+
+import lombok.Data;
+import org.thingsboard.rule.engine.api.NodeConfiguration;
+import org.thingsboard.rule.engine.api.TbRelationTypes;
+import org.thingsboard.server.common.data.DataConstants;
+
+@Data
+public class TbRuleChainOutputNodeConfiguration implements NodeConfiguration<TbRuleChainOutputNodeConfiguration> {
+
+    private String label;
+
+    @Override
+    public TbRuleChainOutputNodeConfiguration defaultConfiguration() {
+        var result = new TbRuleChainOutputNodeConfiguration();
+        result.setLabel(TbRelationTypes.SUCCESS);
+        return result;
+    }
+
+}