iot

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

+/**
+ * Copyright © 2016-2022 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 org.thingsboard.rule.engine.api.TbNode;
+import org.thingsboard.rule.engine.api.TbNodeConfiguration;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.TbActorCtx;
+import org.thingsboard.server.actors.TbActorRef;
+import org.thingsboard.server.actors.TbRuleNodeUpdateException;
+import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
+import org.thingsboard.server.common.data.ApiUsageRecordKey;
+import org.thingsboard.server.common.data.id.RuleNodeId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.plugin.ComponentLifecycleState;
+import org.thingsboard.server.common.data.rule.RuleNode;
+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.stats.TbApiUsageReportClient;
+
+/**
+ * @author Andrew Shvayka
+ */
+public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNodeId> {
+
+    private final String ruleChainName;
+    private final TbActorRef self;
+    private final TbApiUsageReportClient apiUsageClient;
+    private RuleNode ruleNode;
+    private TbNode tbNode;
+    private DefaultTbContext defaultCtx;
+    private RuleNodeInfo info;
+
+    RuleNodeActorMessageProcessor(TenantId tenantId, String ruleChainName, RuleNodeId ruleNodeId, ActorSystemContext systemContext
+            , TbActorRef parent, TbActorRef self) {
+        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");
+    }
+
+    @Override
+    public void start(TbActorCtx context) throws Exception {
+        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 (tbNode != null) {
+                tbNode.destroy();
+            }
+            try {
+                start(context);
+            } catch (Exception e) {
+                throw new TbRuleNodeUpdateException("Failed to update rule node", e);
+            }
+        }
+    }
+
+    @Override
+    public void stop(TbActorCtx context) {
+        if (tbNode != null) {
+            tbNode.destroy();
+            state = ComponentLifecycleState.SUSPENDED;
+        }
+    }
+
+    @Override
+    public void onPartitionChangeMsg(PartitionChangeMsg msg) {
+        if (tbNode != null) {
+            tbNode.onPartitionChangeMsg(defaultCtx, msg);
+        }
+    }
+
+    public void onRuleToSelfMsg(RuleNodeToSelfMsg msg) throws Exception {
+        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(), "Self");
+            }
+            try {
+                tbNode.onMsg(defaultCtx, msg.getMsg());
+            } catch (Exception e) {
+                defaultCtx.tellFailure(msg.getMsg(), e);
+            }
+        } else {
+            tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more then " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
+        }
+    }
+
+    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);
+            }
+        } else {
+            tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more then " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
+        }
+    }
+
+    @Override
+    public String getComponentName() {
+        return ruleNode.getName();
+    }
+
+    private TbNode initComponent(RuleNode ruleNode) throws Exception {
+        TbNode tbNode = null;
+        if (ruleNode != null) {
+            Class<?> componentClazz = Class.forName(ruleNode.getType());
+            tbNode = (TbNode) (componentClazz.getDeclaredConstructor().newInstance());
+            tbNode.init(defaultCtx, new TbNodeConfiguration(ruleNode.getConfiguration()));
+        }
+        return tbNode;
+    }
+
+    @Override
+    protected RuleNodeException getInactiveException() {
+        return new RuleNodeException("Rule Node is not active! Failed to initialize.", ruleChainName, ruleNode);
+    }
+}

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

+/**
+ * Copyright © 2016-2022 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.AllArgsConstructor;
+import lombok.Data;
+import org.thingsboard.server.actors.TbActorRef;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.rule.RuleNode;
+
+/**
+ * Created by ashvayka on 19.03.18.
+ */
+@Data
+@AllArgsConstructor
+final class RuleNodeCtx {
+    private final TenantId tenantId;
+    private final TbActorRef chainActor;
+    private final TbActorRef selfActor;
+    private RuleNode self;
+}

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

+/**
+ * Copyright © 2016-2022 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.Data;
+import org.thingsboard.server.common.data.id.EntityId;
+
+/**
+ * Created by ashvayka on 19.03.18.
+ */
+
+@Data
+final class RuleNodeRelation {
+
+    private final EntityId in;
+    private final EntityId out;
+    private final String type;
+
+}

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

+/**
+ * Copyright © 2016-2022 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.TbActorStopReason;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.TbRuleEngineActorMsg;
+import org.thingsboard.server.common.msg.queue.RuleEngineException;
+
+import java.io.Serializable;
+import java.util.Set;
+
+/**
+ * Created by ashvayka on 19.03.18.
+ */
+@EqualsAndHashCode(callSuper = true)
+@ToString
+class RuleNodeToRuleChainTellNextMsg extends TbRuleEngineActorMsg implements Serializable {
+
+    private static final long serialVersionUID = 4577026446412871820L;
+    @Getter
+    private final RuleChainId ruleChainId;
+    @Getter
+    private final RuleNodeId originator;
+    @Getter
+    private final Set<String> relationTypes;
+    @Getter
+    private final String failureMessage;
+
+    public RuleNodeToRuleChainTellNextMsg(RuleChainId ruleChainId, RuleNodeId originator, Set<String> relationTypes, TbMsg tbMsg, String failureMessage) {
+        super(tbMsg);
+        this.ruleChainId = ruleChainId;
+        this.originator = originator;
+        this.relationTypes = relationTypes;
+        this.failureMessage = failureMessage;
+    }
+
+    @Override
+    public void onTbActorStopped(TbActorStopReason reason) {
+        String message = reason == TbActorStopReason.STOPPED ? String.format("Rule chain [%s] stopped", ruleChainId.getId()) : String.format("Failed to initialize rule chain [%s]!", ruleChainId.getId());
+        msg.getCallback().onFailure(new RuleEngineException(message));
+    }
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.RULE_TO_RULE_CHAIN_TELL_NEXT_MSG;
+    }
+
+}

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

+/**
+ * Copyright © 2016-2022 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.ToString;
+import org.thingsboard.rule.engine.api.TbContext;
+import org.thingsboard.server.common.msg.MsgType;
+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.queue.RuleNodeException;
+
+/**
+ * Created by ashvayka on 19.03.18.
+ */
+@EqualsAndHashCode(callSuper = true)
+@ToString
+final class RuleNodeToSelfMsg extends TbToRuleNodeActorMsg {
+
+    public RuleNodeToSelfMsg(TbContext ctx, TbMsg tbMsg) {
+        super(ctx, tbMsg);
+    }
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.RULE_TO_SELF_MSG;
+    }
+
+}

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

+/**
+ * Copyright © 2016-2022 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/ruleChain/TbToRuleNodeActorMsg.java

+/**
+ * Copyright © 2016-2022 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 org.thingsboard.rule.engine.api.TbContext;
+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.queue.RuleNodeException;
+
+@EqualsAndHashCode(callSuper = true)
+public abstract class TbToRuleNodeActorMsg extends TbRuleEngineActorMsg {
+
+    @Getter
+    private final TbContext ctx;
+
+    public TbToRuleNodeActorMsg(TbContext ctx, TbMsg tbMsg) {
+        super(tbMsg);
+        this.ctx = ctx;
+    }
+
+    @Override
+    public void onTbActorStopped(TbActorStopReason reason) {
+        String message = reason == TbActorStopReason.STOPPED ? "Rule node stopped" : "Failed to initialize rule node!";
+        msg.getCallback().onFailure(new RuleNodeException(message, ctx.getRuleChainName(), ctx.getSelf()));
+    }
+}

application/src/main/java/org/thingsboard/server/actors/service/ActorService.java

+/**
+ * Copyright © 2016-2022 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.service;
+
+public interface ActorService {
+
+
+}

application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java

+/**
+ * Copyright © 2016-2022 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.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.TbActor;
+import org.thingsboard.server.actors.TbActorCtx;
+import org.thingsboard.server.actors.TbActorException;
+import org.thingsboard.server.actors.TbRuleNodeUpdateException;
+import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
+import org.thingsboard.server.actors.stats.StatsPersistMsg;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
+import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
+import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
+
+/**
+ * @author Andrew Shvayka
+ */
+@Slf4j
+public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgProcessor<T>> extends ContextAwareActor {
+
+    private long lastPersistedErrorTs = 0L;
+    protected final TenantId tenantId;
+    protected final T id;
+    protected P processor;
+    private long messagesProcessed;
+    private long errorsOccurred;
+
+    public ComponentActor(ActorSystemContext systemContext, TenantId tenantId, T id) {
+        super(systemContext);
+        this.tenantId = tenantId;
+        this.id = id;
+    }
+
+    abstract protected P createProcessor(TbActorCtx ctx);
+
+    @Override
+    public void init(TbActorCtx ctx) throws TbActorException {
+        super.init(ctx);
+        this.processor = createProcessor(ctx);
+        initProcessor(ctx);
+    }
+
+    protected void initProcessor(TbActorCtx ctx) throws TbActorException {
+        try {
+            log.debug("[{}][{}][{}] Starting processor.", tenantId, id, id.getEntityType());
+            processor.start(ctx);
+            logLifecycleEvent(ComponentLifecycleEvent.STARTED);
+            if (systemContext.isStatisticsEnabled()) {
+                scheduleStatsPersistTick();
+            }
+        } catch (Exception e) {
+            log.debug("[{}][{}] Failed to start {} processor.", tenantId, id, id.getEntityType(), e);
+            logAndPersist("OnStart", e, true);
+            logLifecycleEvent(ComponentLifecycleEvent.STARTED, e);
+            throw new TbActorException("Failed to init actor", e);
+        }
+    }
+
+    private void scheduleStatsPersistTick() {
+        try {
+            processor.scheduleStatsPersistTick(ctx, systemContext.getStatisticsPersistFrequency());
+        } catch (Exception e) {
+            log.error("[{}][{}] Failed to schedule statistics store message. No statistics is going to be stored: {}", tenantId, id, e.getMessage());
+            logAndPersist("onScheduleStatsPersistMsg", e);
+        }
+    }
+
+    @Override
+    public void destroy() {
+        try {
+            log.debug("[{}][{}][{}] Stopping processor.", tenantId, id, id.getEntityType());
+            if (processor != null) {
+                processor.stop(ctx);
+            }
+            logLifecycleEvent(ComponentLifecycleEvent.STOPPED);
+        } catch (Exception e) {
+            log.warn("[{}][{}] Failed to stop {} processor: {}", tenantId, id, id.getEntityType(), e.getMessage());
+            logAndPersist("OnStop", e, true);
+            logLifecycleEvent(ComponentLifecycleEvent.STOPPED, e);
+        }
+    }
+
+    protected void onComponentLifecycleMsg(ComponentLifecycleMsg msg) {
+        log.debug("[{}][{}][{}] onComponentLifecycleMsg: [{}]", tenantId, id, id.getEntityType(), msg.getEvent());
+        try {
+            switch (msg.getEvent()) {
+                case CREATED:
+                    processor.onCreated(ctx);
+                    break;
+                case UPDATED:
+                    processor.onUpdate(ctx);
+                    break;
+                case ACTIVATED:
+                    processor.onActivate(ctx);
+                    break;
+                case SUSPENDED:
+                    processor.onSuspend(ctx);
+                    break;
+                case DELETED:
+                    processor.onStop(ctx);
+                    ctx.stop(ctx.getSelf());
+                    break;
+                default:
+                    break;
+            }
+            logLifecycleEvent(msg.getEvent());
+        } catch (Exception e) {
+            logAndPersist("onLifecycleMsg", e, true);
+            logLifecycleEvent(msg.getEvent(), e);
+            if (e instanceof TbRuleNodeUpdateException) {
+                throw (TbRuleNodeUpdateException) e;
+            }
+        }
+    }
+
+    protected void onClusterEventMsg(PartitionChangeMsg msg) {
+        try {
+            processor.onPartitionChangeMsg(msg);
+        } catch (Exception e) {
+            logAndPersist("onClusterEventMsg", e);
+        }
+    }
+
+    protected void onStatsPersistTick(EntityId entityId) {
+        try {
+            systemContext.getStatsActor().tell(new StatsPersistMsg(messagesProcessed, errorsOccurred, tenantId, entityId));
+            resetStatsCounters();
+        } catch (Exception e) {
+            logAndPersist("onStatsPersistTick", e);
+        }
+    }
+
+    private void resetStatsCounters() {
+        messagesProcessed = 0;
+        errorsOccurred = 0;
+    }
+
+    protected void increaseMessagesProcessedCount() {
+        messagesProcessed++;
+    }
+
+    protected void logAndPersist(String method, Exception e) {
+        logAndPersist(method, e, false);
+    }
+
+    private void logAndPersist(String method, Exception e, boolean critical) {
+        errorsOccurred++;
+        String componentName = processor != null ? processor.getComponentName() : "Unknown";
+        if (critical) {
+            log.debug("[{}][{}][{}] Failed to process method: {}", id, tenantId, componentName, method);
+            log.debug("Critical Error: ", e);
+        } else {
+            log.trace("[{}][{}][{}] Failed to process method: {}", id, tenantId, componentName, method);
+            log.trace("Debug Error: ", e);
+        }
+        long ts = System.currentTimeMillis();
+        if (ts - lastPersistedErrorTs > getErrorPersistFrequency()) {
+            systemContext.persistError(tenantId, id, method, e);
+            lastPersistedErrorTs = ts;
+        }
+    }
+
+    private void logLifecycleEvent(ComponentLifecycleEvent event) {
+        logLifecycleEvent(event, null);
+    }
+
+    private void logLifecycleEvent(ComponentLifecycleEvent event, Exception e) {
+        systemContext.persistLifecycleEvent(tenantId, id, event, e);
+    }
+
+    protected abstract long getErrorPersistFrequency();
+}

application/src/main/java/org/thingsboard/server/actors/service/ContextAwareActor.java

+/**
+ * Copyright © 2016-2022 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.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.thingsboard.server.actors.AbstractTbActor;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.ProcessFailureStrategy;
+import org.thingsboard.server.actors.TbActor;
+import org.thingsboard.server.actors.TbActorCtx;
+import org.thingsboard.server.common.msg.TbActorMsg;
+
+@Slf4j
+public abstract class ContextAwareActor extends AbstractTbActor {
+
+    public static final int ENTITY_PACK_LIMIT = 1024;
+
+    protected final ActorSystemContext systemContext;
+
+    public ContextAwareActor(ActorSystemContext systemContext) {
+        super();
+        this.systemContext = systemContext;
+    }
+
+    @Override
+    public boolean process(TbActorMsg msg) {
+        if (log.isDebugEnabled()) {
+            log.debug("Processing msg: {}", msg);
+        }
+        if (!doProcess(msg)) {
+            log.warn("Unprocessed message: {}!", msg);
+        }
+        return false;
+    }
+
+    protected abstract boolean doProcess(TbActorMsg msg);
+
+    @Override
+    public ProcessFailureStrategy onProcessFailure(Throwable t) {
+        log.debug("[{}] Processing failure: ", getActorRef().getActorId(), t);
+        return doProcessFailure(t);
+    }
+
+    protected ProcessFailureStrategy doProcessFailure(Throwable t) {
+        if (t instanceof Error) {
+            return ProcessFailureStrategy.stop();
+        } else {
+            return ProcessFailureStrategy.resume();
+        }
+    }
+}

application/src/main/java/org/thingsboard/server/actors/service/ContextBasedCreator.java

+/**
+ * Copyright © 2016-2022 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.service;
+
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.TbActorCreator;
+
+public abstract class ContextBasedCreator implements TbActorCreator {
+
+    protected final transient ActorSystemContext context;
+
+    public ContextBasedCreator(ActorSystemContext context) {
+        super();
+        this.context = context;
+    }
+}

application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java

+/**
+ * Copyright © 2016-2022 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.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.core.annotation.Order;
+import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.ThingsBoardExecutors;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.DefaultTbActorSystem;
+import org.thingsboard.server.actors.TbActorRef;
+import org.thingsboard.server.actors.TbActorSystem;
+import org.thingsboard.server.actors.TbActorSystemSettings;
+import org.thingsboard.server.actors.app.AppActor;
+import org.thingsboard.server.actors.app.AppInitMsg;
+import org.thingsboard.server.actors.stats.StatsActor;
+import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
+import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
+import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
+import org.thingsboard.server.queue.util.AfterStartUp;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@Service
+@Slf4j
+public class DefaultActorService extends TbApplicationEventListener<PartitionChangeEvent> implements ActorService {
+
+    public static final String APP_DISPATCHER_NAME = "app-dispatcher";
+    public static final String TENANT_DISPATCHER_NAME = "tenant-dispatcher";
+    public static final String DEVICE_DISPATCHER_NAME = "device-dispatcher";
+    public static final String RULE_DISPATCHER_NAME = "rule-dispatcher";
+
+    @Autowired
+    private ActorSystemContext actorContext;
+
+    private TbActorSystem system;
+
+    private TbActorRef appActor;
+
+    @Value("${actors.system.throughput:5}")
+    private int actorThroughput;
+
+    @Value("${actors.system.max_actor_init_attempts:10}")
+    private int maxActorInitAttempts;
+
+    @Value("${actors.system.scheduler_pool_size:1}")
+    private int schedulerPoolSize;
+
+    @Value("${actors.system.app_dispatcher_pool_size:1}")
+    private int appDispatcherSize;
+
+    @Value("${actors.system.tenant_dispatcher_pool_size:2}")
+    private int tenantDispatcherSize;
+
+    @Value("${actors.system.device_dispatcher_pool_size:4}")
+    private int deviceDispatcherSize;
+
+    @Value("${actors.system.rule_dispatcher_pool_size:4}")
+    private int ruleDispatcherSize;
+
+    @PostConstruct
+    public void initActorSystem() {
+        log.info("Initializing actor system.");
+        actorContext.setActorService(this);
+        TbActorSystemSettings settings = new TbActorSystemSettings(actorThroughput, schedulerPoolSize, maxActorInitAttempts);
+        system = new DefaultTbActorSystem(settings);
+
+        system.createDispatcher(APP_DISPATCHER_NAME, initDispatcherExecutor(APP_DISPATCHER_NAME, appDispatcherSize));
+        system.createDispatcher(TENANT_DISPATCHER_NAME, initDispatcherExecutor(TENANT_DISPATCHER_NAME, tenantDispatcherSize));
+        system.createDispatcher(DEVICE_DISPATCHER_NAME, initDispatcherExecutor(DEVICE_DISPATCHER_NAME, deviceDispatcherSize));
+        system.createDispatcher(RULE_DISPATCHER_NAME, initDispatcherExecutor(RULE_DISPATCHER_NAME, ruleDispatcherSize));
+
+        actorContext.setActorSystem(system);
+
+        appActor = system.createRootActor(APP_DISPATCHER_NAME, new AppActor.ActorCreator(actorContext));
+        actorContext.setAppActor(appActor);
+
+        TbActorRef statsActor = system.createRootActor(TENANT_DISPATCHER_NAME, new StatsActor.ActorCreator(actorContext, "StatsActor"));
+        actorContext.setStatsActor(statsActor);
+
+        log.info("Actor system initialized.");
+    }
+
+    private ExecutorService initDispatcherExecutor(String dispatcherName, int poolSize) {
+        if (poolSize == 0) {
+            int cores = Runtime.getRuntime().availableProcessors();
+            poolSize = Math.max(1, cores / 2);
+        }
+        if (poolSize == 1) {
+            return Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName(dispatcherName));
+        } else {
+            return ThingsBoardExecutors.newWorkStealingPool(poolSize, dispatcherName);
+        }
+    }
+
+    @AfterStartUp(order = AfterStartUp.ACTOR_SYSTEM)
+    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
+        log.info("Received application ready event. Sending application init message to actor system");
+        appActor.tellWithHighPriority(new AppInitMsg());
+    }
+
+    @Override
+    protected void onTbApplicationEvent(PartitionChangeEvent event) {
+        log.info("Received partition change event.");
+        this.appActor.tellWithHighPriority(new PartitionChangeMsg(event.getQueueKey().getType(), event.getPartitions()));
+    }
+
+    @PreDestroy
+    public void stopActorSystem() {
+        if (system != null) {
+            log.info("Stopping actor system.");
+            system.stop();
+            log.info("Actor system stopped.");
+        }
+    }
+
+}

application/src/main/java/org/thingsboard/server/actors/shared/AbstractContextAwareMsgProcessor.java

+/**
+ * Copyright © 2016-2022 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.shared;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.TbActorCtx;
+import org.thingsboard.server.common.msg.TbActorMsg;
+
+import java.util.concurrent.ScheduledExecutorService;
+
+@Slf4j
+public abstract class AbstractContextAwareMsgProcessor {
+
+    protected final static ObjectMapper mapper = new ObjectMapper();
+
+    protected final ActorSystemContext systemContext;
+
+    protected AbstractContextAwareMsgProcessor(ActorSystemContext systemContext) {
+        super();
+        this.systemContext = systemContext;
+    }
+
+    private ScheduledExecutorService getScheduler() {
+        return systemContext.getScheduler();
+    }
+
+    protected void schedulePeriodicMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs, long periodInMs) {
+        systemContext.schedulePeriodicMsgWithDelay(ctx, msg, delayInMs, periodInMs);
+    }
+
+    protected void scheduleMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs) {
+        systemContext.scheduleMsgWithDelay(ctx, msg, delayInMs);
+    }
+
+}

application/src/main/java/org/thingsboard/server/actors/shared/ActorTerminationMsg.java

+/**
+ * Copyright © 2016-2022 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.shared;
+
+public abstract class ActorTerminationMsg<T> {
+
+    private final T id;
+
+    public ActorTerminationMsg(T id) {
+        super();
+        this.id = id;
+    }
+
+    public T getId() {
+        return id;
+    }
+
+}

application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java

+/**
+ * Copyright © 2016-2022 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.shared;
+
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.TbActorCtx;
+import org.thingsboard.server.actors.stats.StatsPersistTick;
+import org.thingsboard.server.common.data.TenantProfile;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.plugin.ComponentLifecycleState;
+import org.thingsboard.server.common.data.tenant.profile.TenantProfileConfiguration;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
+import org.thingsboard.server.common.msg.queue.RuleNodeException;
+
+@Slf4j
+public abstract class ComponentMsgProcessor<T extends EntityId> extends AbstractContextAwareMsgProcessor {
+
+    protected final TenantId tenantId;
+    protected final T entityId;
+    protected ComponentLifecycleState state;
+
+    protected ComponentMsgProcessor(ActorSystemContext systemContext, TenantId tenantId, T id) {
+        super(systemContext);
+        this.tenantId = tenantId;
+        this.entityId = id;
+    }
+
+    protected TenantProfileConfiguration getTenantProfileConfiguration() {
+        return systemContext.getTenantProfileCache().get(tenantId).getProfileData().getConfiguration();
+    }
+
+    public abstract String getComponentName();
+
+    public abstract void start(TbActorCtx context) throws Exception;
+
+    public abstract void stop(TbActorCtx context) throws Exception;
+
+    public abstract void onPartitionChangeMsg(PartitionChangeMsg msg) throws Exception;
+
+    public void onCreated(TbActorCtx context) throws Exception {
+        start(context);
+    }
+
+    public void onUpdate(TbActorCtx context) throws Exception {
+        restart(context);
+    }
+
+    public void onActivate(TbActorCtx context) throws Exception {
+        restart(context);
+    }
+
+    public void onSuspend(TbActorCtx context) throws Exception {
+        stop(context);
+    }
+
+    public void onStop(TbActorCtx context) throws Exception {
+        stop(context);
+    }
+
+    private void restart(TbActorCtx context) throws Exception {
+        stop(context);
+        start(context);
+    }
+
+    public void scheduleStatsPersistTick(TbActorCtx context, long statsPersistFrequency) {
+        schedulePeriodicMsgWithDelay(context, new StatsPersistTick(), statsPersistFrequency, statsPersistFrequency);
+    }
+
+    protected boolean checkMsgValid(TbMsg tbMsg) {
+        var valid = tbMsg.isValid();
+        if (!valid) {
+            if (log.isTraceEnabled()) {
+                log.trace("Skip processing of message: {} because it is no longer valid!", tbMsg);
+            }
+        }
+        return valid;
+    }
+
+    protected void checkComponentStateActive(TbMsg tbMsg) throws RuleNodeException {
+        if (state != ComponentLifecycleState.ACTIVE) {
+            log.debug("Component is not active. Current state [{}] for processor [{}][{}] tenant [{}]", state, entityId.getEntityType(), entityId, tenantId);
+            RuleNodeException ruleNodeException = getInactiveException();
+            if (tbMsg != null) {
+                tbMsg.getCallback().onFailure(ruleNodeException);
+            }
+            throw ruleNodeException;
+        }
+    }
+
+    abstract protected RuleNodeException getInactiveException();
+
+}

application/src/main/java/org/thingsboard/server/actors/stats/StatsActor.java

+/**
+ * Copyright © 2016-2022 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.stats;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.TbActor;
+import org.thingsboard.server.actors.TbActorId;
+import org.thingsboard.server.actors.TbStringActorId;
+import org.thingsboard.server.actors.service.ContextAwareActor;
+import org.thingsboard.server.actors.service.ContextBasedCreator;
+import org.thingsboard.server.common.data.DataConstants;
+import org.thingsboard.server.common.data.EventInfo;
+import org.thingsboard.server.common.data.event.StatisticsEvent;
+import org.thingsboard.server.common.msg.MsgType;
+import org.thingsboard.server.common.msg.TbActorMsg;
+
+@Slf4j
+public class StatsActor extends ContextAwareActor {
+
+    private final ObjectMapper mapper = new ObjectMapper();
+
+    public StatsActor(ActorSystemContext context) {
+        super(context);
+    }
+
+    @Override
+    protected boolean doProcess(TbActorMsg msg) {
+        log.debug("Received message: {}", msg);
+        if (msg.getMsgType().equals(MsgType.STATS_PERSIST_MSG)) {
+            onStatsPersistMsg((StatsPersistMsg) msg);
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    public void onStatsPersistMsg(StatsPersistMsg msg) {
+        if (msg.isEmpty()) {
+            return;
+        }
+        systemContext.getEventService().saveAsync(StatisticsEvent.builder()
+                .tenantId(msg.getTenantId())
+                .entityId(msg.getEntityId().getId())
+                .serviceId(systemContext.getServiceInfoProvider().getServiceId())
+                .messagesProcessed(msg.getMessagesProcessed())
+                .errorsOccurred(msg.getErrorsOccurred())
+                .build()
+        );
+    }
+
+    private JsonNode toBodyJson(String serviceId, long messagesProcessed, long errorsOccurred) {
+        return mapper.createObjectNode().put("server", serviceId).put("messagesProcessed", messagesProcessed).put("errorsOccurred", errorsOccurred);
+    }
+
+    public static class ActorCreator extends ContextBasedCreator {
+        private final String actorId;
+
+        public ActorCreator(ActorSystemContext context, String actorId) {
+            super(context);
+            this.actorId = actorId;
+        }
+
+        @Override
+        public TbActorId createActorId() {
+            return new TbStringActorId(actorId);
+        }
+
+        @Override
+        public TbActor createActor() {
+            return new StatsActor(context);
+        }
+    }
+}

application/src/main/java/org/thingsboard/server/actors/stats/StatsPersistMsg.java

+/**
+ * Copyright © 2016-2022 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.stats;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.ToString;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.MsgType;
+import org.thingsboard.server.common.msg.TbActorMsg;
+
+@AllArgsConstructor
+@Getter
+@ToString
+public final class StatsPersistMsg implements TbActorMsg {
+
+    private final long messagesProcessed;
+    private final long errorsOccurred;
+    private final TenantId tenantId;
+    private final EntityId entityId;
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.STATS_PERSIST_MSG;
+    }
+
+    public boolean isEmpty() {
+        return messagesProcessed == 0 && errorsOccurred == 0;
+    }
+
+}

application/src/main/java/org/thingsboard/server/actors/stats/StatsPersistTick.java

+/**
+ * Copyright © 2016-2022 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.stats;
+
+import org.thingsboard.server.common.msg.MsgType;
+import org.thingsboard.server.common.msg.TbActorMsg;
+
+public final class StatsPersistTick implements TbActorMsg {
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.STATS_PERSIST_TICK_MSG;
+    }
+}

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

+/**
+ * Copyright © 2016-2022 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.tenant;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import org.thingsboard.server.common.msg.tools.TbRateLimits;
+
+@Data
+@AllArgsConstructor
+public class DebugTbRateLimits {
+
+    private TbRateLimits tbRateLimits;
+    private boolean ruleChainEventSaved;
+
+}

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

+/**
+ * Copyright © 2016-2022 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.tenant;
+
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.actors.TbActor;
+import org.thingsboard.server.actors.TbActorCtx;
+import org.thingsboard.server.actors.TbActorException;
+import org.thingsboard.server.actors.TbActorId;
+import org.thingsboard.server.actors.TbActorNotRegisteredException;
+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.RuleChainManagerActor;
+import org.thingsboard.server.actors.service.ContextBasedCreator;
+import org.thingsboard.server.actors.service.DefaultActorService;
+import org.thingsboard.server.common.data.ApiUsageState;
+import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.Tenant;
+import org.thingsboard.server.common.data.TenantProfile;
+import org.thingsboard.server.common.data.edge.Edge;
+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.TenantId;
+import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
+import org.thingsboard.server.common.data.rule.RuleChain;
+import org.thingsboard.server.common.data.rule.RuleChainType;
+import org.thingsboard.server.common.msg.MsgType;
+import org.thingsboard.server.common.msg.TbActorMsg;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.aware.DeviceAwareMsg;
+import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg;
+import org.thingsboard.server.common.msg.edge.EdgeSessionMsg;
+import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
+import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
+import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
+import org.thingsboard.server.common.msg.queue.RuleEngineException;
+import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.service.edge.rpc.EdgeRpcService;
+import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
+
+import java.util.List;
+
+@Slf4j
+public class TenantActor extends RuleChainManagerActor {
+
+    private boolean isRuleEngine;
+    private boolean isCore;
+    private ApiUsageState apiUsageState;
+
+    private TenantActor(ActorSystemContext systemContext, TenantId tenantId) {
+        super(systemContext, tenantId);
+    }
+
+    boolean cantFindTenant = false;
+
+    @Override
+    public void init(TbActorCtx ctx) throws TbActorException {
+        super.init(ctx);
+        log.debug("[{}] Starting tenant actor.", tenantId);
+        try {
+            Tenant tenant = systemContext.getTenantService().findTenantById(tenantId);
+            if (tenant == null) {
+                cantFindTenant = true;
+                log.info("[{}] Started tenant actor for missing tenant.", tenantId);
+            } else {
+                TenantProfile tenantProfile = systemContext.getTenantProfileCache().get(tenant.getTenantProfileId());
+
+                isCore = systemContext.getServiceInfoProvider().isService(ServiceType.TB_CORE);
+                isRuleEngine = systemContext.getServiceInfoProvider().isService(ServiceType.TB_RULE_ENGINE);
+                if (isRuleEngine) {
+                    try {
+                        if (getApiUsageState().isReExecEnabled()) {
+                            log.debug("[{}] Going to init rule chains", tenantId);
+                            initRuleChains();
+                        } else {
+                            log.info("[{}] Skip init of the rule chains due to API limits", tenantId);
+                        }
+                    } catch (Exception e) {
+                        cantFindTenant = true;
+                    }
+                }
+                log.debug("[{}] Tenant actor started.", tenantId);
+            }
+        } catch (Exception e) {
+            log.warn("[{}] Unknown failure", tenantId, e);
+        }
+    }
+
+    @Override
+    public void destroy() {
+        log.info("[{}] Stopping tenant actor.", tenantId);
+    }
+
+    @Override
+    protected boolean doProcess(TbActorMsg msg) {
+        if (cantFindTenant) {
+            log.info("[{}] Processing missing Tenant msg: {}", tenantId, msg);
+            if (msg.getMsgType().equals(MsgType.QUEUE_TO_RULE_ENGINE_MSG)) {
+                QueueToRuleEngineMsg queueMsg = (QueueToRuleEngineMsg) msg;
+                queueMsg.getMsg().getCallback().onSuccess();
+            } else if (msg.getMsgType().equals(MsgType.TRANSPORT_TO_DEVICE_ACTOR_MSG)) {
+                TransportToDeviceActorMsgWrapper transportMsg = (TransportToDeviceActorMsgWrapper) msg;
+                transportMsg.getCallback().onSuccess();
+            }
+            return true;
+        }
+        switch (msg.getMsgType()) {
+            case PARTITION_CHANGE_MSG:
+                PartitionChangeMsg partitionChangeMsg = (PartitionChangeMsg) msg;
+                ServiceType serviceType = partitionChangeMsg.getServiceType();
+                if (ServiceType.TB_RULE_ENGINE.equals(serviceType)) {
+                    //To Rule Chain Actors
+                    broadcast(msg);
+                } else if (ServiceType.TB_CORE.equals(serviceType)) {
+                    List<TbActorId> deviceActorIds = ctx.filterChildren(new TbEntityTypeActorIdPredicate(EntityType.DEVICE) {
+                        @Override
+                        protected boolean testEntityId(EntityId entityId) {
+                            return super.testEntityId(entityId) && !isMyPartition(entityId);
+                        }
+                    });
+                    deviceActorIds.forEach(id -> ctx.stop(id));
+                }
+                break;
+            case COMPONENT_LIFE_CYCLE_MSG:
+                onComponentLifecycleMsg((ComponentLifecycleMsg) msg);
+                break;
+            case QUEUE_TO_RULE_ENGINE_MSG:
+                onQueueToRuleEngineMsg((QueueToRuleEngineMsg) msg);
+                break;
+            case TRANSPORT_TO_DEVICE_ACTOR_MSG:
+                onToDeviceActorMsg((DeviceAwareMsg) msg, false);
+                break;
+            case DEVICE_ATTRIBUTES_UPDATE_TO_DEVICE_ACTOR_MSG:
+            case DEVICE_CREDENTIALS_UPDATE_TO_DEVICE_ACTOR_MSG:
+            case DEVICE_NAME_OR_TYPE_UPDATE_TO_DEVICE_ACTOR_MSG:
+            case DEVICE_EDGE_UPDATE_TO_DEVICE_ACTOR_MSG:
+            case DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG:
+            case DEVICE_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG:
+            case SERVER_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG:
+            case REMOVE_RPC_TO_DEVICE_ACTOR_MSG:
+                onToDeviceActorMsg((DeviceAwareMsg) msg, true);
+                break;
+            case SESSION_TIMEOUT_MSG:
+                ctx.broadcastToChildrenByType(msg, EntityType.DEVICE);
+                break;
+            case RULE_CHAIN_INPUT_MSG:
+            case RULE_CHAIN_OUTPUT_MSG:
+            case RULE_CHAIN_TO_RULE_CHAIN_MSG:
+                onRuleChainMsg((RuleChainAwareMsg) msg);
+                break;
+            case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG:
+            case EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG:
+            case EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG:
+                onToEdgeSessionMsg((EdgeSessionMsg) msg);
+                break;
+            default:
+                return false;
+        }
+        return true;
+    }
+
+    private boolean isMyPartition(EntityId entityId) {
+        return systemContext.resolve(ServiceType.TB_CORE, tenantId, entityId).isMyPartition();
+    }
+
+    private void onQueueToRuleEngineMsg(QueueToRuleEngineMsg msg) {
+        if (!isRuleEngine) {
+            log.warn("RECEIVED INVALID MESSAGE: {}", msg);
+            return;
+        }
+        TbMsg tbMsg = msg.getMsg();
+        if (getApiUsageState().isReExecEnabled()) {
+            if (tbMsg.getRuleChainId() == null) {
+                if (getRootChainActor() != null) {
+                    getRootChainActor().tell(msg);
+                } else {
+                    tbMsg.getCallback().onFailure(new RuleEngineException("No Root Rule Chain available!"));
+                    log.info("[{}] No Root Chain: {}", tenantId, msg);
+                }
+            } else {
+                try {
+                    ctx.tell(new TbEntityActorId(tbMsg.getRuleChainId()), msg);
+                } catch (TbActorNotRegisteredException ex) {
+                    log.trace("Received message for non-existing rule chain: [{}]", tbMsg.getRuleChainId());
+                    //TODO: 3.1 Log it to dead letters queue;
+                    tbMsg.getCallback().onSuccess();
+                }
+            }
+        } else {
+            log.trace("[{}] Ack message because Rule Engine is disabled", tenantId);
+            tbMsg.getCallback().onSuccess();
+        }
+    }
+
+    private void onRuleChainMsg(RuleChainAwareMsg msg) {
+        if (getApiUsageState().isReExecEnabled()) {
+            getOrCreateActor(msg.getRuleChainId()).tell(msg);
+        }
+    }
+
+    private void onToDeviceActorMsg(DeviceAwareMsg msg, boolean priority) {
+        if (!isCore) {
+            log.warn("RECEIVED INVALID MESSAGE: {}", msg);
+        }
+        TbActorRef deviceActor = getOrCreateDeviceActor(msg.getDeviceId());
+        if (priority) {
+            deviceActor.tellWithHighPriority(msg);
+        } else {
+            deviceActor.tell(msg);
+        }
+    }
+
+    private void onComponentLifecycleMsg(ComponentLifecycleMsg msg) {
+        if (msg.getEntityId().getEntityType().equals(EntityType.API_USAGE_STATE)) {
+            ApiUsageState old = getApiUsageState();
+            apiUsageState = new ApiUsageState(systemContext.getApiUsageStateService().getApiUsageState(tenantId));
+            if (old.isReExecEnabled() && !apiUsageState.isReExecEnabled()) {
+                log.info("[{}] Received API state update. Going to DISABLE Rule Engine execution.", tenantId);
+                destroyRuleChains();
+            } else if (!old.isReExecEnabled() && apiUsageState.isReExecEnabled()) {
+                log.info("[{}] Received API state update. Going to ENABLE Rule Engine execution.", tenantId);
+                initRuleChains();
+            }
+        } else if (msg.getEntityId().getEntityType() == EntityType.EDGE) {
+            EdgeId edgeId = new EdgeId(msg.getEntityId().getId());
+            EdgeRpcService edgeRpcService = systemContext.getEdgeRpcService();
+            if (msg.getEvent() == ComponentLifecycleEvent.DELETED) {
+                edgeRpcService.deleteEdge(tenantId, edgeId);
+            } else if (msg.getEvent() == ComponentLifecycleEvent.UPDATED) {
+                Edge edge = systemContext.getEdgeService().findEdgeById(tenantId, edgeId);
+                edgeRpcService.updateEdge(tenantId, edge);
+            }
+        } else if (isRuleEngine) {
+            TbActorRef target = getEntityActorRef(msg.getEntityId());
+            if (target != null) {
+                if (msg.getEntityId().getEntityType() == EntityType.RULE_CHAIN) {
+                    RuleChain ruleChain = systemContext.getRuleChainService().
+                            findRuleChainById(tenantId, new RuleChainId(msg.getEntityId().getId()));
+                    if (ruleChain != null && RuleChainType.CORE.equals(ruleChain.getType())) {
+                        visit(ruleChain, target);
+                    }
+                }
+                target.tellWithHighPriority(msg);
+            } else {
+                log.debug("[{}] Invalid component lifecycle msg: {}", tenantId, msg);
+            }
+        }
+    }
+
+    private TbActorRef getOrCreateDeviceActor(DeviceId deviceId) {
+        return ctx.getOrCreateChildActor(new TbEntityActorId(deviceId),
+                () -> DefaultActorService.DEVICE_DISPATCHER_NAME,
+                () -> new DeviceActorCreator(systemContext, tenantId, deviceId));
+    }
+
+    private void onToEdgeSessionMsg(EdgeSessionMsg msg) {
+        systemContext.getEdgeRpcService().onToEdgeSessionMsg(tenantId, msg);
+    }
+
+    private ApiUsageState getApiUsageState() {
+        if (apiUsageState == null) {
+            apiUsageState = new ApiUsageState(systemContext.getApiUsageStateService().getApiUsageState(tenantId));
+        }
+        return apiUsageState;
+    }
+
+    public static class ActorCreator extends ContextBasedCreator {
+
+        private final TenantId tenantId;
+
+        public ActorCreator(ActorSystemContext context, TenantId tenantId) {
+            super(context);
+            this.tenantId = tenantId;
+        }
+
+        @Override
+        public TbActorId createActorId() {
+            return new TbEntityActorId(tenantId);
+        }
+
+        @Override
+        public TbActor createActor() {
+            return new TenantActor(context, tenantId);
+        }
+    }
+
+}