From 3261cf4fe3f80a2d3ce62c44d00ede72b9bd81ca Mon Sep 17 00:00:00 2001
From: YevhenBondarenko <ybondarenko@thingsboard.io>
Date: Thu, 20 Apr 2023 16:07:14 +0200
Subject: [PATCH] implemented rule node singleton mode

---
 .../main/data/upgrade/3.4.4/schema_update.sql |  13 ++
 .../server/actors/ActorSystemContext.java     |   5 +
 .../RuleNodeActorMessageProcessor.java        | 111 ++++++++++++------
 .../AnnotationComponentDiscoveryService.java  |   1 +
 .../data/plugin/ComponentDescriptor.java      |   2 +
 .../plugin/ComponentSingletonSupport.java     |  22 ++++
 .../server/dao/model/ModelConstants.java      |   2 +
 .../model/sql/ComponentDescriptorEntity.java  |   7 ++
 .../server/dao/model/sql/RuleNodeEntity.java  |   5 +
 .../thingsboard/rule/engine/api/RuleNode.java |   3 +
 .../rule/engine/mqtt/TbMqttNode.java          |   2 +
 .../engine/mqtt/azure/TbAzureIotHubNode.java  |   5 +-
 .../import-export/import-export.service.ts    |   1 +
 .../rule-node-details.component.html          |   3 +
 .../rulechain/rule-node-details.component.ts  |   6 +
 .../rulechain/rulechain-page.component.ts     |  11 +-
 .../models/component-descriptor.models.ts     |   7 ++
 .../app/shared/models/rule-chain.models.ts    |   4 +-
 .../src/app/shared/models/rule-node.models.ts |   2 +
 .../assets/locale/locale.constant-en_US.json  |   1 +
 20 files changed, 170 insertions(+), 43 deletions(-)
 create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentSingletonSupport.java

diff --git a/application/src/main/data/upgrade/3.4.4/schema_update.sql b/application/src/main/data/upgrade/3.4.4/schema_update.sql
index 3776eebc12..e15a9fb9e9 100644
--- a/application/src/main/data/upgrade/3.4.4/schema_update.sql
+++ b/application/src/main/data/upgrade/3.4.4/schema_update.sql
@@ -610,3 +610,16 @@ END
 $$;
 
 -- TTL DROP PARTITIONS FUNCTIONS UPDATE END
+
+-- RULE NODE SINGLETON MODE SUPPORT
+
+ALTER TABLE rule_node ADD COLUMN IF NOT EXISTS singleton_mode bool DEFAULT false;
+
+UPDATE rule_node SET singleton_mode = true WHERE type IN ('org.thingsboard.rule.engine.mqtt.azure.TbAzureIotHubNode', 'org.thingsboard.rule.engine.mqtt.TbMqttNode');
+
+ALTER TABLE component_descriptor ADD COLUMN IF NOT EXISTS singleton varchar(255) DEFAULT 'NOT_SUPPORTED';
+
+UPDATE component_descriptor SET singleton = 'SUPPORTED' WHERE name = 'mqtt';
+
+UPDATE component_descriptor SET singleton = 'ONLY_SINGLETON' WHERE name = 'azure iot hub';
+
diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
index 26eab30d5d..1f0af4f1d1 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
@@ -87,6 +87,7 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.user.UserService;
 import org.thingsboard.server.dao.widget.WidgetTypeService;
 import org.thingsboard.server.dao.widget.WidgetsBundleService;
+import org.thingsboard.server.queue.discovery.DiscoveryService;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
@@ -179,6 +180,10 @@ public class ActorSystemContext {
     @Setter
     private ComponentDiscoveryService componentService;
 
+    @Autowired
+    @Getter
+    private DiscoveryService discoveryService;
+
     @Autowired
     @Getter
     private DataDecodingEncodingService encodingService;
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 6174027d38..223dfa2de2 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
@@ -31,7 +31,10 @@ import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
 import org.thingsboard.server.common.msg.queue.RuleNodeException;
 import org.thingsboard.server.common.msg.queue.RuleNodeInfo;
+import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.common.stats.TbApiUsageReportClient;
+import org.thingsboard.server.gen.transport.TransportProtos;
 
 /**
  * @author Andrew Shvayka
@@ -57,29 +60,37 @@ 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;
+        if (isMyNode()) {
+            tbNode = initComponent(ruleNode);
+            if (tbNode != null) {
+                state = ComponentLifecycleState.ACTIVE;
+            }
         }
     }
 
     @Override
     public void onUpdate(TbActorCtx context) throws Exception {
-        RuleNode newRuleNode = systemContext.getRuleChainService().findRuleNodeById(tenantId, entityId);
-        this.info = new RuleNodeInfo(entityId, ruleChainName, newRuleNode != null ? newRuleNode.getName() : "Unknown");
-        boolean restartRequired = state != ComponentLifecycleState.ACTIVE ||
-                !(ruleNode.getType().equals(newRuleNode.getType()) && ruleNode.getConfiguration().equals(newRuleNode.getConfiguration()));
-        this.ruleNode = newRuleNode;
-        this.defaultCtx.updateSelf(newRuleNode);
-        if (restartRequired) {
+        if (isMyNode()) {
+            RuleNode newRuleNode = systemContext.getRuleChainService().findRuleNodeById(tenantId, entityId);
+            this.info = new RuleNodeInfo(entityId, ruleChainName, newRuleNode != null ? newRuleNode.getName() : "Unknown");
+            boolean restartRequired = state != ComponentLifecycleState.ACTIVE ||
+                    !(ruleNode.getType().equals(newRuleNode.getType()) && ruleNode.getConfiguration().equals(newRuleNode.getConfiguration()));
+            this.ruleNode = newRuleNode;
+            this.defaultCtx.updateSelf(newRuleNode);
+            if (restartRequired) {
+                if (tbNode != null) {
+                    tbNode.destroy();
+                }
+                try {
+                    start(context);
+                } catch (Exception e) {
+                    throw new TbRuleNodeUpdateException("Failed to update rule node", e);
+                }
+            }
+        } else {
             if (tbNode != null) {
                 tbNode.destroy();
-            }
-            try {
-                start(context);
-            } catch (Exception e) {
-                throw new TbRuleNodeUpdateException("Failed to update rule node", e);
+                tbNode = null;
             }
         }
     }
@@ -93,15 +104,20 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     }
 
     @Override
-    public void onPartitionChangeMsg(PartitionChangeMsg msg) {
-        //TODO: start the node
+    public void onPartitionChangeMsg(PartitionChangeMsg msg) throws Exception {
         if (tbNode != null) {
-            tbNode.onPartitionChangeMsg(defaultCtx, msg);
+            if (!isMyNode()) {
+                tbNode.destroy();
+                tbNode = null;
+            } else {
+                tbNode.onPartitionChangeMsg(defaultCtx, msg);
+            }
+        } else if (isMyNode()) {
+            start(null);
         }
     }
 
     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();
@@ -122,23 +138,27 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     }
 
     void onRuleChainToRuleNodeMsg(RuleChainToRuleNodeMsg msg) throws Exception {
-        msg.getMsg().getCallback().onProcessingStart(info);
-        checkComponentStateActive(msg.getMsg());
-        TbMsg tbMsg = msg.getMsg();
-        int ruleNodeCount = tbMsg.getAndIncrementRuleNodeCounter();
-        int maxRuleNodeExecutionsPerMessage = getTenantProfileConfiguration().getMaxRuleNodeExecsPerMessage();
-        if (maxRuleNodeExecutionsPerMessage == 0 || ruleNodeCount < maxRuleNodeExecutionsPerMessage) {
-            apiUsageClient.report(tenantId, tbMsg.getCustomerId(), ApiUsageRecordKey.RE_EXEC_COUNT);
-            if (ruleNode.isDebugMode()) {
-                systemContext.persistDebugInput(tenantId, entityId, msg.getMsg(), msg.getFromRelationType());
-            }
-            try {
-                tbNode.onMsg(msg.getCtx(), msg.getMsg());
-            } catch (Exception e) {
-                msg.getCtx().tellFailure(msg.getMsg(), e);
-            }
+        if (!isMyNode()) {
+            putToQueue(msg.getMsg());
         } else {
-            tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more then " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
+            msg.getMsg().getCallback().onProcessingStart(info);
+            checkComponentStateActive(msg.getMsg());
+            TbMsg tbMsg = msg.getMsg();
+            int ruleNodeCount = tbMsg.getAndIncrementRuleNodeCounter();
+            int maxRuleNodeExecutionsPerMessage = getTenantProfileConfiguration().getMaxRuleNodeExecsPerMessage();
+            if (maxRuleNodeExecutionsPerMessage == 0 || ruleNodeCount < maxRuleNodeExecutionsPerMessage) {
+                apiUsageClient.report(tenantId, tbMsg.getCustomerId(), ApiUsageRecordKey.RE_EXEC_COUNT);
+                if (ruleNode.isDebugMode()) {
+                    systemContext.persistDebugInput(tenantId, entityId, msg.getMsg(), msg.getFromRelationType());
+                }
+                try {
+                    tbNode.onMsg(msg.getCtx(), msg.getMsg());
+                } catch (Exception e) {
+                    msg.getCtx().tellFailure(msg.getMsg(), e);
+                }
+            } else {
+                tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more then " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
+            }
         }
     }
 
@@ -161,4 +181,23 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     protected RuleNodeException getInactiveException() {
         return new RuleNodeException("Rule Node is not active! Failed to initialize.", ruleChainName, ruleNode);
     }
+
+    private boolean isMyNode() {
+        return systemContext.getDiscoveryService().isMonolith()
+                || !ruleNode.isSingletonMode()
+//                || !ruleNode.getName().equals("singleton")
+                || defaultCtx.isLocalEntity(ruleNode.getId());
+    }
+
+    private void putToQueue(TbMsg source) {
+        TbMsg tbMsg = TbMsg.newMsg(source, source.getQueueName(), source.getRuleChainId(), entityId);
+        TopicPartitionInfo tpi = systemContext.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), tenantId, entityId);
+        TransportProtos.ToRuleEngineMsg toQueueMsg = TransportProtos.ToRuleEngineMsg.newBuilder()
+                .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
+                .setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
+                .setTbMsg(TbMsg.toByteString(tbMsg))
+                .build();
+        systemContext.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), toQueueMsg, null);
+        defaultCtx.ack(source);
+    }
 }
diff --git a/application/src/main/java/org/thingsboard/server/service/component/AnnotationComponentDiscoveryService.java b/application/src/main/java/org/thingsboard/server/service/component/AnnotationComponentDiscoveryService.java
index 732d8ce409..bb63f5da4c 100644
--- a/application/src/main/java/org/thingsboard/server/service/component/AnnotationComponentDiscoveryService.java
+++ b/application/src/main/java/org/thingsboard/server/service/component/AnnotationComponentDiscoveryService.java
@@ -155,6 +155,7 @@ public class AnnotationComponentDiscoveryService implements ComponentDiscoverySe
             RuleNode ruleNodeAnnotation = clazz.getAnnotation(RuleNode.class);
             scannedComponent.setName(ruleNodeAnnotation.name());
             scannedComponent.setScope(ruleNodeAnnotation.scope());
+            scannedComponent.setSingleton(ruleNodeAnnotation.singleton());
             NodeDefinition nodeDefinition = prepareNodeDefinition(ruleNodeAnnotation);
             ObjectNode configurationDescriptor = mapper.createObjectNode();
             JsonNode node = mapper.valueToTree(nodeDefinition);
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentDescriptor.java b/common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentDescriptor.java
index 288801b8d6..9839e34482 100644
--- a/common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentDescriptor.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentDescriptor.java
@@ -36,6 +36,8 @@ public class ComponentDescriptor extends SearchTextBased<ComponentDescriptorId>
     @Getter @Setter private ComponentType type;
     @ApiModelProperty(position = 4, value = "Scope of the Rule Node. Always set to 'TENANT', since no rule chains on the 'SYSTEM' level yet.", accessMode = ApiModelProperty.AccessMode.READ_ONLY, allowableValues = "TENANT", example = "TENANT")
     @Getter @Setter private ComponentScope scope;
+    @ApiModelProperty(position = 4, value = "", accessMode = ApiModelProperty.AccessMode.READ_ONLY, allowableValues = "SUPPORTED, NOT_SUPPORTEd, ONLY_SINGLETON", example = "SUPPORTED")
+    @Getter @Setter private ComponentSingletonSupport singleton;
     @Length(fieldName = "name")
     @ApiModelProperty(position = 5, value = "Name of the Rule Node. Taken from the @RuleNode annotation.", accessMode = ApiModelProperty.AccessMode.READ_ONLY, example = "Custom Rule Node")
     @Getter @Setter private String name;
diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentSingletonSupport.java b/common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentSingletonSupport.java
new file mode 100644
index 0000000000..0472f26808
--- /dev/null
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentSingletonSupport.java
@@ -0,0 +1,22 @@
+/**
+ * Copyright © 2016-2023 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.data.plugin;
+
+public enum ComponentSingletonSupport {
+    SUPPORTED,
+    NOT_SUPPORTED,
+    ONLY_SINGLETON
+}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java
index 6197d8f908..7b1df3c4a9 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java
@@ -403,6 +403,7 @@ public class ModelConstants {
     public static final String COMPONENT_DESCRIPTOR_COLUMN_FAMILY_NAME = "component_descriptor";
     public static final String COMPONENT_DESCRIPTOR_TYPE_PROPERTY = "type";
     public static final String COMPONENT_DESCRIPTOR_SCOPE_PROPERTY = "scope";
+    public static final String COMPONENT_DESCRIPTOR_SINGLETON_PROPERTY = "singleton";
     public static final String COMPONENT_DESCRIPTOR_NAME_PROPERTY = "name";
     public static final String COMPONENT_DESCRIPTOR_CLASS_PROPERTY = "clazz";
     public static final String COMPONENT_DESCRIPTOR_CONFIGURATION_DESCRIPTOR_PROPERTY = "configuration_descriptor";
@@ -446,6 +447,7 @@ public class ModelConstants {
     public static final String EVENT_MESSAGE_COLUMN_NAME = "e_message";
 
     public static final String DEBUG_MODE = "debug_mode";
+    public static final String SINGLETON_MODE = "singleton_mode";
 
     /**
      * Cassandra rule chain constants.
diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/ComponentDescriptorEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/ComponentDescriptorEntity.java
index 10e8bb59d5..47c34e0d58 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/ComponentDescriptorEntity.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/ComponentDescriptorEntity.java
@@ -23,6 +23,7 @@ import org.hibernate.annotations.TypeDef;
 import org.thingsboard.server.common.data.id.ComponentDescriptorId;
 import org.thingsboard.server.common.data.plugin.ComponentDescriptor;
 import org.thingsboard.server.common.data.plugin.ComponentScope;
+import org.thingsboard.server.common.data.plugin.ComponentSingletonSupport;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.dao.model.BaseSqlEntity;
 import org.thingsboard.server.dao.model.ModelConstants;
@@ -50,6 +51,10 @@ public class ComponentDescriptorEntity extends BaseSqlEntity<ComponentDescriptor
     @Column(name = ModelConstants.COMPONENT_DESCRIPTOR_SCOPE_PROPERTY)
     private ComponentScope scope;
 
+    @Enumerated(EnumType.STRING)
+    @Column(name = ModelConstants.COMPONENT_DESCRIPTOR_SINGLETON_PROPERTY)
+    private ComponentSingletonSupport singleton;
+
     @Column(name = ModelConstants.COMPONENT_DESCRIPTOR_NAME_PROPERTY)
     private String name;
 
@@ -77,6 +82,7 @@ public class ComponentDescriptorEntity extends BaseSqlEntity<ComponentDescriptor
         this.actions = component.getActions();
         this.type = component.getType();
         this.scope = component.getScope();
+        this.singleton = component.getSingleton();
         this.name = component.getName();
         this.clazz = component.getClazz();
         this.configurationDescriptor = component.getConfigurationDescriptor();
@@ -89,6 +95,7 @@ public class ComponentDescriptorEntity extends BaseSqlEntity<ComponentDescriptor
         data.setCreatedTime(createdTime);
         data.setType(type);
         data.setScope(scope);
+        data.setSingleton(singleton);
         data.setName(this.getName());
         data.setClazz(this.getClazz());
         data.setActions(this.getActions());
diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/RuleNodeEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/RuleNodeEntity.java
index 6ce0e45e21..2dce9fb305 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/RuleNodeEntity.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/RuleNodeEntity.java
@@ -64,6 +64,9 @@ public class RuleNodeEntity extends BaseSqlEntity<RuleNode> implements SearchTex
     @Column(name = ModelConstants.DEBUG_MODE)
     private boolean debugMode;
 
+    @Column(name = ModelConstants.SINGLETON_MODE)
+    private boolean singletonMode;
+
     @Column(name = ModelConstants.EXTERNAL_ID_PROPERTY)
     private UUID externalId;
 
@@ -81,6 +84,7 @@ public class RuleNodeEntity extends BaseSqlEntity<RuleNode> implements SearchTex
         this.type = ruleNode.getType();
         this.name = ruleNode.getName();
         this.debugMode = ruleNode.isDebugMode();
+        this.singletonMode = ruleNode.isSingletonMode();
         this.searchText = ruleNode.getName();
         this.configuration = ruleNode.getConfiguration();
         this.additionalInfo = ruleNode.getAdditionalInfo();
@@ -109,6 +113,7 @@ public class RuleNodeEntity extends BaseSqlEntity<RuleNode> implements SearchTex
         ruleNode.setType(type);
         ruleNode.setName(name);
         ruleNode.setDebugMode(debugMode);
+        ruleNode.setSingletonMode(singletonMode);
         ruleNode.setConfiguration(configuration);
         ruleNode.setAdditionalInfo(additionalInfo);
         if (externalId != null) {
diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNode.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNode.java
index 5fbd65e711..9fb8bb66c0 100644
--- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNode.java
+++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNode.java
@@ -16,6 +16,7 @@
 package org.thingsboard.rule.engine.api;
 
 import org.thingsboard.server.common.data.plugin.ComponentScope;
+import org.thingsboard.server.common.data.plugin.ComponentSingletonSupport;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.data.rule.RuleChainType;
 
@@ -38,6 +39,8 @@ public @interface RuleNode {
 
     Class<? extends NodeConfiguration> configClazz();
 
+    ComponentSingletonSupport singleton() default ComponentSingletonSupport.NOT_SUPPORTED;
+
     boolean inEnabled() default true;
 
     boolean outEnabled() default true;
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java
index d1feae6271..95ba04d687 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java
@@ -33,6 +33,7 @@ import org.thingsboard.rule.engine.api.util.TbNodeUtils;
 import org.thingsboard.rule.engine.credentials.BasicCredentials;
 import org.thingsboard.rule.engine.credentials.ClientCredentials;
 import org.thingsboard.rule.engine.credentials.CredentialsType;
+import org.thingsboard.server.common.data.plugin.ComponentSingletonSupport;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
@@ -47,6 +48,7 @@ import java.util.concurrent.TimeoutException;
         type = ComponentType.EXTERNAL,
         name = "mqtt",
         configClazz = TbMqttNodeConfiguration.class,
+        singleton = ComponentSingletonSupport.SUPPORTED,
         nodeDescription = "Publish messages to the MQTT broker",
         nodeDetails = "Will publish message payload to the MQTT broker with QoS <b>AT_LEAST_ONCE</b>.",
         uiResources = {"static/rulenode/rulenode-core-config.js"},
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/azure/TbAzureIotHubNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/azure/TbAzureIotHubNode.java
index 34c26d820d..0b3d8f36ad 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/azure/TbAzureIotHubNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/azure/TbAzureIotHubNode.java
@@ -16,9 +16,7 @@
 package org.thingsboard.rule.engine.mqtt.azure;
 
 import io.netty.handler.codec.mqtt.MqttVersion;
-import io.netty.handler.ssl.SslContext;
 import lombok.extern.slf4j.Slf4j;
-import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.common.util.AzureIotHubUtil;
 import org.thingsboard.mqtt.MqttClientConfig;
 import org.thingsboard.rule.engine.api.RuleNode;
@@ -26,12 +24,12 @@ import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
 import org.thingsboard.rule.engine.api.TbNodeException;
 import org.thingsboard.rule.engine.api.util.TbNodeUtils;
-import org.thingsboard.rule.engine.credentials.BasicCredentials;
 import org.thingsboard.rule.engine.credentials.CertPemCredentials;
 import org.thingsboard.rule.engine.credentials.ClientCredentials;
 import org.thingsboard.rule.engine.credentials.CredentialsType;
 import org.thingsboard.rule.engine.mqtt.TbMqttNode;
 import org.thingsboard.rule.engine.mqtt.TbMqttNodeConfiguration;
+import org.thingsboard.server.common.data.plugin.ComponentSingletonSupport;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 
 import javax.net.ssl.SSLException;
@@ -41,6 +39,7 @@ import javax.net.ssl.SSLException;
         type = ComponentType.EXTERNAL,
         name = "azure iot hub",
         configClazz = TbAzureIotHubNodeConfiguration.class,
+        singleton = ComponentSingletonSupport.ONLY_SINGLETON,
         nodeDescription = "Publish messages to the Azure IoT Hub",
         nodeDetails = "Will publish message payload to the Azure IoT Hub with QoS <b>AT_LEAST_ONCE</b>.",
         uiResources = {"static/rulenode/rulenode-core-config.js"},
diff --git a/ui-ngx/src/app/modules/home/components/import-export/import-export.service.ts b/ui-ngx/src/app/modules/home/components/import-export/import-export.service.ts
index a4b3a1d4de..a6e7e7f89e 100644
--- a/ui-ngx/src/app/modules/home/components/import-export/import-export.service.ts
+++ b/ui-ngx/src/app/modules/home/components/import-export/import-export.service.ts
@@ -467,6 +467,7 @@ export class ImportExportService {
           const ruleChainNode: RuleNode = {
             name: '',
             debugMode: false,
+            singletonMode: false,
             type: 'org.thingsboard.rule.engine.flow.TbRuleChainInputNode',
             configuration: {
               ruleChainId: ruleChainConnection.targetRuleChainId.id
diff --git a/ui-ngx/src/app/modules/home/pages/rulechain/rule-node-details.component.html b/ui-ngx/src/app/modules/home/pages/rulechain/rule-node-details.component.html
index b6912f0b00..4a237663b9 100644
--- a/ui-ngx/src/app/modules/home/pages/rulechain/rule-node-details.component.html
+++ b/ui-ngx/src/app/modules/home/pages/rulechain/rule-node-details.component.html
@@ -39,6 +39,9 @@
         <mat-checkbox formControlName="debugMode">
           {{ 'rulenode.debug-mode' | translate }}
         </mat-checkbox>
+        <mat-checkbox *ngIf="isSingletonEditAllowed()" formControlName="singletonMode">
+          {{ 'rulenode.singleton-mode' | translate }}
+        </mat-checkbox>
       </section>
       <tb-rule-node-config #ruleNodeConfigComponent
         formControlName="configuration"
diff --git a/ui-ngx/src/app/modules/home/pages/rulechain/rule-node-details.component.ts b/ui-ngx/src/app/modules/home/pages/rulechain/rule-node-details.component.ts
index a0df64cd0e..dd21990c1e 100644
--- a/ui-ngx/src/app/modules/home/pages/rulechain/rule-node-details.component.ts
+++ b/ui-ngx/src/app/modules/home/pages/rulechain/rule-node-details.component.ts
@@ -26,6 +26,7 @@ import { RuleChainService } from '@core/http/rule-chain.service';
 import { RuleNodeConfigComponent } from './rule-node-config.component';
 import { Router } from '@angular/router';
 import { RuleChainType } from '@app/shared/models/rule-chain.models';
+import { ComponentSingletonSupport } from "@shared/models/component-descriptor.models";
 
 @Component({
   selector: 'tb-rule-node',
@@ -78,6 +79,7 @@ export class RuleNodeDetailsComponent extends PageComponent implements OnInit, O
       this.ruleNodeFormGroup = this.fb.group({
         name: [this.ruleNode.name, [Validators.required, Validators.pattern('(.|\\s)*\\S(.|\\s)*'), Validators.maxLength(255)]],
         debugMode: [this.ruleNode.debugMode, []],
+        singletonMode: [this.ruleNode.singletonMode, []],
         configuration: [this.ruleNode.configuration, [Validators.required]],
         additionalInfo: this.fb.group(
           {
@@ -130,4 +132,8 @@ export class RuleNodeDetailsComponent extends PageComponent implements OnInit, O
       }
     }
   }
+
+  isSingletonEditAllowed() {
+    return this.ruleNode.component.singleton === ComponentSingletonSupport.SUPPORTED;
+  }
 }
diff --git a/ui-ngx/src/app/modules/home/pages/rulechain/rulechain-page.component.ts b/ui-ngx/src/app/modules/home/pages/rulechain/rulechain-page.component.ts
index ab553793fc..72dd1415ee 100644
--- a/ui-ngx/src/app/modules/home/pages/rulechain/rulechain-page.component.ts
+++ b/ui-ngx/src/app/modules/home/pages/rulechain/rulechain-page.component.ts
@@ -554,6 +554,7 @@ export class RuleChainPageComponent extends PageComponent
         additionalInfo: ruleNode.additionalInfo,
         configuration: ruleNode.configuration,
         debugMode: ruleNode.debugMode,
+        singletonMode: ruleNode.singletonMode,
         x: Math.round(ruleNode.additionalInfo.layoutX),
         y: Math.round(ruleNode.additionalInfo.layoutY),
         component,
@@ -912,7 +913,8 @@ export class RuleChainPageComponent extends PageComponent
             name: node.name,
             configuration: deepClone(node.configuration),
             additionalInfo: node.additionalInfo ? deepClone(node.additionalInfo) : {},
-            debugMode: node.debugMode
+            debugMode: node.debugMode,
+            singletonMode: node.singletonMode
           };
           if (minX === null) {
             minX = node.x;
@@ -983,7 +985,8 @@ export class RuleChainPageComponent extends PageComponent
             name: outputEdge.label,
             configuration: {},
             additionalInfo: {},
-            debugMode: false
+            debugMode: false,
+            singletonMode: false
           };
           outputNode.additionalInfo.layoutX = Math.round(destNode.x);
           outputNode.additionalInfo.layoutY = Math.round(destNode.y);
@@ -1029,6 +1032,7 @@ export class RuleChainPageComponent extends PageComponent
               ruleChainId: ruleChain.id.id
             },
             debugMode: false,
+            singletonMode: false,
             x: Math.round(ruleChainNodeX),
             y: Math.round(ruleChainNodeY),
             nodeClass: descriptor.nodeClass,
@@ -1420,7 +1424,8 @@ export class RuleChainPageComponent extends PageComponent
             name: node.name,
             configuration: node.configuration,
             additionalInfo: node.additionalInfo ? node.additionalInfo : {},
-            debugMode: node.debugMode
+            debugMode: node.debugMode,
+            singletonMode: node.singletonMode
           };
           ruleNode.additionalInfo.layoutX = Math.round(node.x);
           ruleNode.additionalInfo.layoutY = Math.round(node.y);
diff --git a/ui-ngx/src/app/shared/models/component-descriptor.models.ts b/ui-ngx/src/app/shared/models/component-descriptor.models.ts
index 56dd29cb4e..d8e4aba2aa 100644
--- a/ui-ngx/src/app/shared/models/component-descriptor.models.ts
+++ b/ui-ngx/src/app/shared/models/component-descriptor.models.ts
@@ -30,9 +30,16 @@ export enum ComponentScope {
   TENANT = 'TENANT'
 }
 
+export enum ComponentSingletonSupport {
+  SUPPORTED = 'SUPPORTED',
+  NOT_SUPPORTED = 'NOT_SUPPORTED',
+  ONLY_SINGLETON = 'ONLY_SINGLETON'
+}
+
 export interface ComponentDescriptor {
   type: ComponentType | RuleNodeType;
   scope?: ComponentScope;
+  singleton: ComponentSingletonSupport;
   name: string;
   clazz: string;
   configurationDescriptor?: any;
diff --git a/ui-ngx/src/app/shared/models/rule-chain.models.ts b/ui-ngx/src/app/shared/models/rule-chain.models.ts
index 20086ca21b..57a58ba110 100644
--- a/ui-ngx/src/app/shared/models/rule-chain.models.ts
+++ b/ui-ngx/src/app/shared/models/rule-chain.models.ts
@@ -19,7 +19,7 @@ import { TenantId } from '@shared/models/id/tenant-id';
 import { RuleChainId } from '@shared/models/id/rule-chain-id';
 import { RuleNodeId } from '@shared/models/id/rule-node-id';
 import { RuleNode, RuleNodeComponentDescriptor, RuleNodeType } from '@shared/models/rule-node.models';
-import { ComponentType } from '@shared/models/component-descriptor.models';
+import { ComponentSingletonSupport, ComponentType } from '@shared/models/component-descriptor.models';
 
 export interface RuleChain extends BaseData<RuleChainId>, ExportableEntity<RuleChainId> {
   tenantId: TenantId;
@@ -64,6 +64,7 @@ export const ruleNodeTypeComponentTypes: ComponentType[] =
 export const unknownNodeComponent: RuleNodeComponentDescriptor = {
   type: RuleNodeType.UNKNOWN,
   name: 'unknown',
+  singleton: ComponentSingletonSupport.NOT_SUPPORTED,
   clazz: 'tb.internal.Unknown',
   configurationDescriptor: {
     nodeDefinition: {
@@ -80,6 +81,7 @@ export const unknownNodeComponent: RuleNodeComponentDescriptor = {
 
 export const inputNodeComponent: RuleNodeComponentDescriptor = {
   type: RuleNodeType.INPUT,
+  singleton: ComponentSingletonSupport.NOT_SUPPORTED,
   name: 'Input',
   clazz: 'tb.internal.Input'
 };
diff --git a/ui-ngx/src/app/shared/models/rule-node.models.ts b/ui-ngx/src/app/shared/models/rule-node.models.ts
index 9d78a5ced8..8ad24b795b 100644
--- a/ui-ngx/src/app/shared/models/rule-node.models.ts
+++ b/ui-ngx/src/app/shared/models/rule-node.models.ts
@@ -36,6 +36,7 @@ export interface RuleNode extends BaseData<RuleNodeId> {
   type: string;
   name: string;
   debugMode: boolean;
+  singletonMode: boolean;
   configuration: RuleNodeConfiguration;
   additionalInfo?: any;
 }
@@ -318,6 +319,7 @@ export interface FcRuleNode extends FcRuleNodeType {
   additionalInfo?: any;
   configuration?: RuleNodeConfiguration;
   debugMode?: boolean;
+  singletonMode?: boolean;
   error?: string;
   highlighted?: boolean;
   componentClazz?: string;
diff --git a/ui-ngx/src/assets/locale/locale.constant-en_US.json b/ui-ngx/src/assets/locale/locale.constant-en_US.json
index 8c664e48f5..73687635db 100644
--- a/ui-ngx/src/assets/locale/locale.constant-en_US.json
+++ b/ui-ngx/src/assets/locale/locale.constant-en_US.json
@@ -3304,6 +3304,7 @@
         "deselect-all": "Deselect all",
         "rulenode-details": "Rule node details",
         "debug-mode": "Debug mode",
+        "singleton-mode": "Singleton mode",
         "configuration": "Configuration",
         "link": "Link",
         "link-details": "Rule node link details",