Merge remote-tracking branch 'upstream/master'

vparomskiy
2 parents 48e2b1a2 362c2750
Showing 6 changed files with 32 additions and 8 deletions
application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java
application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
application/src/main/java/org/thingsboard/server/controller/TenantController.java
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java
--- a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
View file @b1eb65d
+++ b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
View file @b1eb65d
@@ -34,9 +34,11 @@ import org.thingsboard.server.actors.service.ContextBasedCreator;
 import org.thingsboard.server.actors.service.DefaultActorService;
 import org.thingsboard.server.actors.shared.rulechain.SystemRuleChainManager;
 import org.thingsboard.server.actors.tenant.TenantActor;
+import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.Tenant;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageDataIterable;
+import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
 import org.thingsboard.server.common.msg.TbActorMsg;
 import org.thingsboard.server.common.msg.aware.TenantAwareMsg;
 import org.thingsboard.server.common.msg.cluster.SendToClusterMsg;
@@ -129,7 +131,7 @@ public class AppActor extends RuleChainManagerActor {
     private void onServiceToRuleEngineMsg(ServiceToRuleEngineMsg msg) {
         if (SYSTEM_TENANT.equals(msg.getTenantId())) {
-            log.warn("[{}] Invalid service to rule engine msg called. System messages are not supported yet", SYSTEM_TENANT);
+            log.warn("[{}] Invalid service to rule engine msg called. System messages are not supported yet: {}", SYSTEM_TENANT, msg);
         } else {
             getOrCreateTenantActor(msg.getTenantId()).tell(msg, self());
         }
@@ -142,11 +144,21 @@ public class AppActor extends RuleChainManagerActor {
     }
     private void onComponentLifecycleMsg(ComponentLifecycleMsg msg) {
-        ActorRef target;
+        ActorRef target = null;
         if (SYSTEM_TENANT.equals(msg.getTenantId())) {
             target = getEntityActorRef(msg.getEntityId());
         } else {
-            target = getOrCreateTenantActor(msg.getTenantId());
+            if (msg.getEntityId().getEntityType() == EntityType.TENANT
+                    && msg.getEvent() == ComponentLifecycleEvent.DELETED) {
+                log.debug("[{}] Handling tenant deleted notification: {}", msg.getTenantId(), msg);
+                ActorRef tenantActor = tenantActors.remove(new TenantId(msg.getEntityId().getId()));
+                if (tenantActor != null) {
+                    log.debug("[{}] Deleting tenant actor: {}", msg.getTenantId(), tenantActor);
+                    context().stop(tenantActor);
+                }
+            } else {
+                target = getOrCreateTenantActor(msg.getTenantId());
+            }
         }
         if (target != null) {
             target.tell(msg, ActorRef.noSender());
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java
View file @b1eb65d
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java
View file @b1eb65d
@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.actors.ruleChain;
+import akka.actor.ActorInitializationException;
 import akka.actor.OneForOneStrategy;
 import akka.actor.SupervisorStrategy;
 import org.thingsboard.server.actors.ActorSystemContext;
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
View file @b1eb65d
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
View file @b1eb65d
@@ -96,7 +96,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
             log.trace("[{}][{}] Starting rule chain with {} nodes", tenantId, entityId, ruleNodeList.size());
             // Creating and starting the actors;
             for (RuleNode ruleNode : ruleNodeList) {
-                log.trace("[{}][{}] Creating rule node [{}]: {}", tenantId, entityId, ruleNode.getName(), ruleNode);
+                log.trace("[{}][{}] Creating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode);
                 ActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode);
                 nodeActors.put(ruleNode.getId(), new RuleNodeCtx(tenantId, self, ruleNodeActor, ruleNode));
             }
@@ -116,11 +116,11 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
         for (RuleNode ruleNode : ruleNodeList) {
             RuleNodeCtx existing = nodeActors.get(ruleNode.getId());
             if (existing == null) {
-                log.trace("[{}][{}] Creating rule node [{}]: {}", tenantId, entityId, ruleNode.getName(), ruleNode);
+                log.trace("[{}][{}] Creating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode);
                 ActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode);
                 nodeActors.put(ruleNode.getId(), new RuleNodeCtx(tenantId, self, ruleNodeActor, ruleNode));
             } else {
-                log.trace("[{}][{}] Updating rule node [{}]: {}", tenantId, entityId, ruleNode.getName(), ruleNode);
+                log.trace("[{}][{}] Updating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode);
                 existing.setSelf(ruleNode);
                 existing.getSelfActor().tell(new ComponentLifecycleMsg(tenantId, existing.getSelf().getId(), ComponentLifecycleEvent.UPDATED), self);
             }
@@ -184,13 +184,15 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
         }
         firstId = ruleChain.getFirstRuleNodeId();
-        firstNode = nodeActors.get(ruleChain.getFirstRuleNodeId());
+        firstNode = nodeActors.get(firstId);
         state = ComponentLifecycleState.ACTIVE;
     }
     void onServiceToRuleEngineMsg(ServiceToRuleEngineMsg envelope) {
+        log.trace("[{}][{}] Processing message [{}]: {}", entityId, firstId, envelope.getTbMsg().getId(), envelope.getTbMsg());
         checkActive();
         if (firstNode != null) {
+            log.trace("[{}][{}] Pushing message to first rule node", entityId, firstId);
             pushMsgToNode(firstNode, enrichWithRuleChainId(envelope.getTbMsg()), "");
         }
     }
--- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
View file @b1eb65d
+++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
View file @b1eb65d
@@ -37,6 +37,7 @@ import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.id.DeviceId;
 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.msg.TbActorMsg;
 import org.thingsboard.server.common.msg.aware.DeviceAwareMsg;
@@ -76,6 +77,11 @@ public class TenantActor extends RuleChainManagerActor {
     }
     @Override
+    public void postStop() {
+        log.info("[{}] Stopping tenant actor.", tenantId);
+    }
+
+    @Override
     protected boolean process(TbActorMsg msg) {
         switch (msg.getMsgType()) {
             case CLUSTER_EVENT_MSG:
--- a/application/src/main/java/org/thingsboard/server/controller/TenantController.java
View file @b1eb65d
+++ b/application/src/main/java/org/thingsboard/server/controller/TenantController.java
View file @b1eb65d
@@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.exception.ThingsboardException;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.TextPageData;
 import org.thingsboard.server.common.data.page.TextPageLink;
+import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
 import org.thingsboard.server.dao.tenant.TenantService;
 import org.thingsboard.server.service.install.InstallScripts;
@@ -84,6 +85,8 @@ public class TenantController extends BaseController {
         try {
             TenantId tenantId = new TenantId(toUUID(strTenantId));
             tenantService.deleteTenant(tenantId);
+
+            actorService.onEntityStateChange(tenantId, tenantId, ComponentLifecycleEvent.DELETED);
         } catch (Exception e) {
             throw handleException(e);
         }
--- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java
View file @b1eb65d
+++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java
View file @b1eb65d
@@ -119,10 +119,10 @@ public class GatewaySessionHandler {
                             GatewayDeviceSessionCtx deviceSessionCtx = new GatewayDeviceSessionCtx(GatewaySessionHandler.this, msg.getDeviceInfo(), mqttQoSMap);
                             if (devices.putIfAbsent(deviceName, deviceSessionCtx) == null) {
                                 SessionInfoProto deviceSessionInfo = deviceSessionCtx.getSessionInfo();
+                                transportService.registerAsyncSession(deviceSessionInfo, deviceSessionCtx);
                                 transportService.process(deviceSessionInfo, AbstractTransportService.getSessionEventMsg(TransportProtos.SessionEvent.OPEN), null);
                                 transportService.process(deviceSessionInfo, TransportProtos.SubscribeToRPCMsg.getDefaultInstance(), null);
                                 transportService.process(deviceSessionInfo, TransportProtos.SubscribeToAttributeUpdatesMsg.getDefaultInstance(), null);
-                                transportService.registerAsyncSession(deviceSessionInfo, deviceSessionCtx);
                             }
                             future.set(devices.get(deviceName));
                         }