From b20057004be642ed2597eafbaf22793e32a9de99 Mon Sep 17 00:00:00 2001
From: Andrii Shvaika <ashvayka@thingsboard.io>
Date: Tue, 18 Apr 2023 17:09:34 +0300
Subject: [PATCH] Singleton rule node implementation

---
 .../actors/ruleChain/RuleChainActorMessageProcessor.java   | 1 +
 .../actors/ruleChain/RuleNodeActorMessageProcessor.java    | 7 ++++---
 .../org/thingsboard/server/common/data/rule/RuleNode.java  | 5 ++++-
 3 files changed, 9 insertions(+), 4 deletions(-)

diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
index 0e1e25bd27..12ea2fd651 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
@@ -388,6 +388,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
 
     private void pushMsgToNode(RuleNodeCtx nodeCtx, TbMsg msg, String fromRelationType) {
         if (nodeCtx != null) {
+            //TODO: analyze that singleton flag and execute putToQueue if needed.
             nodeCtx.getSelfActor().tell(new RuleChainToRuleNodeMsg(new DefaultTbContext(systemContext, ruleChainName, nodeCtx), msg, fromRelationType));
         } else {
             log.error("[{}][{}] RuleNodeCtx is empty", entityId, ruleChainName);
diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java
index a6049ce8f2..6174027d38 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java
@@ -39,11 +39,10 @@ import org.thingsboard.server.common.stats.TbApiUsageReportClient;
 public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNodeId> {
 
     private final String ruleChainName;
-    private final TbActorRef self;
     private final TbApiUsageReportClient apiUsageClient;
+    private final DefaultTbContext defaultCtx;
     private RuleNode ruleNode;
     private TbNode tbNode;
-    private DefaultTbContext defaultCtx;
     private RuleNodeInfo info;
 
     RuleNodeActorMessageProcessor(TenantId tenantId, String ruleChainName, RuleNodeId ruleNodeId, ActorSystemContext systemContext
@@ -51,7 +50,6 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
         super(systemContext, tenantId, ruleNodeId);
         this.apiUsageClient = systemContext.getApiUsageClient();
         this.ruleChainName = ruleChainName;
-        this.self = self;
         this.ruleNode = systemContext.getRuleChainService().findRuleNodeById(tenantId, entityId);
         this.defaultCtx = new DefaultTbContext(systemContext, ruleChainName, new RuleNodeCtx(tenantId, parent, self, ruleNode));
         this.info = new RuleNodeInfo(ruleNodeId, ruleChainName, ruleNode != null ? ruleNode.getName() : "Unknown");
@@ -59,6 +57,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
 
     @Override
     public void start(TbActorCtx context) throws Exception {
+        //TODO: do not start the node if singleton
         tbNode = initComponent(ruleNode);
         if (tbNode != null) {
             state = ComponentLifecycleState.ACTIVE;
@@ -95,12 +94,14 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
 
     @Override
     public void onPartitionChangeMsg(PartitionChangeMsg msg) {
+        //TODO: start the node
         if (tbNode != null) {
             tbNode.onPartitionChangeMsg(defaultCtx, msg);
         }
     }
 
     public void onRuleToSelfMsg(RuleNodeToSelfMsg msg) throws Exception {
+        // TODO: check that the rule node is singleton and use putToQueue
         checkComponentStateActive(msg.getMsg());
         TbMsg tbMsg = msg.getMsg();
         int ruleNodeCount = tbMsg.getAndIncrementRuleNodeCounter();
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/rule/RuleNode.java b/common/data/src/main/java/org/thingsboard/server/common/data/rule/RuleNode.java
index 32d396725b..e6041c94f7 100644
--- a/common/data/src/main/java/org/thingsboard/server/common/data/rule/RuleNode.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/rule/RuleNode.java
@@ -48,7 +48,9 @@ public class RuleNode extends SearchTextBasedWithAdditionalInfo<RuleNodeId> impl
     private String name;
     @ApiModelProperty(position = 6, value = "Enable/disable debug. ", example = "false")
     private boolean debugMode;
-    @ApiModelProperty(position = 7, value = "JSON with the rule node configuration. Structure depends on the rule node implementation.", dataType = "com.fasterxml.jackson.databind.JsonNode")
+    @ApiModelProperty(position = 7, value = "Enable/disable singleton mode. ", example = "false")
+    private boolean singletonMode;
+    @ApiModelProperty(position = 8, value = "JSON with the rule node configuration. Structure depends on the rule node implementation.", dataType = "com.fasterxml.jackson.databind.JsonNode")
     private transient JsonNode configuration;
     @JsonIgnore
     private byte[] configurationBytes;
@@ -69,6 +71,7 @@ public class RuleNode extends SearchTextBasedWithAdditionalInfo<RuleNodeId> impl
         this.type = ruleNode.getType();
         this.name = ruleNode.getName();
         this.debugMode = ruleNode.isDebugMode();
+        this.singletonMode = ruleNode.isSingletonMode();
         this.setConfiguration(ruleNode.getConfiguration());
         this.externalId = ruleNode.getExternalId();
     }