Commit e724dcd0885380fa016a1a9373f4fe05af171d70
1 parent
f6b00b35
Added shutdown of all rule chains on tenant deletion
Showing
5 changed files
with
31 additions
and
7 deletions
... | ... | @@ -34,9 +34,11 @@ import org.thingsboard.server.actors.service.ContextBasedCreator; |
34 | 34 | import org.thingsboard.server.actors.service.DefaultActorService; |
35 | 35 | import org.thingsboard.server.actors.shared.rulechain.SystemRuleChainManager; |
36 | 36 | import org.thingsboard.server.actors.tenant.TenantActor; |
37 | +import org.thingsboard.server.common.data.EntityType; | |
37 | 38 | import org.thingsboard.server.common.data.Tenant; |
38 | 39 | import org.thingsboard.server.common.data.id.TenantId; |
39 | 40 | import org.thingsboard.server.common.data.page.PageDataIterable; |
41 | +import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; | |
40 | 42 | import org.thingsboard.server.common.msg.TbActorMsg; |
41 | 43 | import org.thingsboard.server.common.msg.aware.TenantAwareMsg; |
42 | 44 | import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; |
... | ... | @@ -129,7 +131,7 @@ public class AppActor extends RuleChainManagerActor { |
129 | 131 | |
130 | 132 | private void onServiceToRuleEngineMsg(ServiceToRuleEngineMsg msg) { |
131 | 133 | if (SYSTEM_TENANT.equals(msg.getTenantId())) { |
132 | - log.warn("[{}] Invalid service to rule engine msg called. System messages are not supported yet", SYSTEM_TENANT); | |
134 | + log.warn("[{}] Invalid service to rule engine msg called. System messages are not supported yet: {}", SYSTEM_TENANT, msg); | |
133 | 135 | } else { |
134 | 136 | getOrCreateTenantActor(msg.getTenantId()).tell(msg, self()); |
135 | 137 | } |
... | ... | @@ -142,11 +144,21 @@ public class AppActor extends RuleChainManagerActor { |
142 | 144 | } |
143 | 145 | |
144 | 146 | private void onComponentLifecycleMsg(ComponentLifecycleMsg msg) { |
145 | - ActorRef target; | |
147 | + ActorRef target = null; | |
146 | 148 | if (SYSTEM_TENANT.equals(msg.getTenantId())) { |
147 | 149 | target = getEntityActorRef(msg.getEntityId()); |
148 | 150 | } else { |
149 | - target = getOrCreateTenantActor(msg.getTenantId()); | |
151 | + if (msg.getEntityId().getEntityType() == EntityType.TENANT | |
152 | + && msg.getEvent() == ComponentLifecycleEvent.DELETED) { | |
153 | + log.debug("[{}] Handling tenant deleted notification: {}", msg.getTenantId(), msg); | |
154 | + ActorRef tenantActor = tenantActors.remove(new TenantId(msg.getEntityId().getId())); | |
155 | + if (tenantActor != null) { | |
156 | + log.debug("[{}] Deleting tenant actor: {}", msg.getTenantId(), tenantActor); | |
157 | + context().stop(tenantActor); | |
158 | + } | |
159 | + } else { | |
160 | + target = getOrCreateTenantActor(msg.getTenantId()); | |
161 | + } | |
150 | 162 | } |
151 | 163 | if (target != null) { |
152 | 164 | target.tell(msg, ActorRef.noSender()); | ... | ... |
... | ... | @@ -15,6 +15,7 @@ |
15 | 15 | */ |
16 | 16 | package org.thingsboard.server.actors.ruleChain; |
17 | 17 | |
18 | +import akka.actor.ActorInitializationException; | |
18 | 19 | import akka.actor.OneForOneStrategy; |
19 | 20 | import akka.actor.SupervisorStrategy; |
20 | 21 | import org.thingsboard.server.actors.ActorSystemContext; | ... | ... |
... | ... | @@ -96,7 +96,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh |
96 | 96 | log.trace("[{}][{}] Starting rule chain with {} nodes", tenantId, entityId, ruleNodeList.size()); |
97 | 97 | // Creating and starting the actors; |
98 | 98 | for (RuleNode ruleNode : ruleNodeList) { |
99 | - log.trace("[{}][{}] Creating rule node [{}]: {}", tenantId, entityId, ruleNode.getName(), ruleNode); | |
99 | + log.trace("[{}][{}] Creating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode); | |
100 | 100 | ActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode); |
101 | 101 | nodeActors.put(ruleNode.getId(), new RuleNodeCtx(tenantId, self, ruleNodeActor, ruleNode)); |
102 | 102 | } |
... | ... | @@ -116,11 +116,11 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh |
116 | 116 | for (RuleNode ruleNode : ruleNodeList) { |
117 | 117 | RuleNodeCtx existing = nodeActors.get(ruleNode.getId()); |
118 | 118 | if (existing == null) { |
119 | - log.trace("[{}][{}] Creating rule node [{}]: {}", tenantId, entityId, ruleNode.getName(), ruleNode); | |
119 | + log.trace("[{}][{}] Creating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode); | |
120 | 120 | ActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode); |
121 | 121 | nodeActors.put(ruleNode.getId(), new RuleNodeCtx(tenantId, self, ruleNodeActor, ruleNode)); |
122 | 122 | } else { |
123 | - log.trace("[{}][{}] Updating rule node [{}]: {}", tenantId, entityId, ruleNode.getName(), ruleNode); | |
123 | + log.trace("[{}][{}] Updating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode); | |
124 | 124 | existing.setSelf(ruleNode); |
125 | 125 | existing.getSelfActor().tell(new ComponentLifecycleMsg(tenantId, existing.getSelf().getId(), ComponentLifecycleEvent.UPDATED), self); |
126 | 126 | } |
... | ... | @@ -184,13 +184,15 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh |
184 | 184 | } |
185 | 185 | |
186 | 186 | firstId = ruleChain.getFirstRuleNodeId(); |
187 | - firstNode = nodeActors.get(ruleChain.getFirstRuleNodeId()); | |
187 | + firstNode = nodeActors.get(firstId); | |
188 | 188 | state = ComponentLifecycleState.ACTIVE; |
189 | 189 | } |
190 | 190 | |
191 | 191 | void onServiceToRuleEngineMsg(ServiceToRuleEngineMsg envelope) { |
192 | + log.trace("[{}][{}] Processing message [{}]: {}", entityId, firstId, envelope.getTbMsg().getId(), envelope.getTbMsg()); | |
192 | 193 | checkActive(); |
193 | 194 | if (firstNode != null) { |
195 | + log.trace("[{}][{}] Pushing message to first rule node", entityId, firstId); | |
194 | 196 | pushMsgToNode(firstNode, enrichWithRuleChainId(envelope.getTbMsg()), ""); |
195 | 197 | } |
196 | 198 | } | ... | ... |
... | ... | @@ -37,6 +37,7 @@ import org.thingsboard.server.common.data.EntityType; |
37 | 37 | import org.thingsboard.server.common.data.id.DeviceId; |
38 | 38 | import org.thingsboard.server.common.data.id.RuleChainId; |
39 | 39 | import org.thingsboard.server.common.data.id.TenantId; |
40 | +import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; | |
40 | 41 | import org.thingsboard.server.common.data.rule.RuleChain; |
41 | 42 | import org.thingsboard.server.common.msg.TbActorMsg; |
42 | 43 | import org.thingsboard.server.common.msg.aware.DeviceAwareMsg; |
... | ... | @@ -76,6 +77,11 @@ public class TenantActor extends RuleChainManagerActor { |
76 | 77 | } |
77 | 78 | |
78 | 79 | @Override |
80 | + public void postStop() { | |
81 | + log.info("[{}] Stopping tenant actor.", tenantId); | |
82 | + } | |
83 | + | |
84 | + @Override | |
79 | 85 | protected boolean process(TbActorMsg msg) { |
80 | 86 | switch (msg.getMsgType()) { |
81 | 87 | case CLUSTER_EVENT_MSG: | ... | ... |
... | ... | @@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.exception.ThingsboardException; |
32 | 32 | import org.thingsboard.server.common.data.id.TenantId; |
33 | 33 | import org.thingsboard.server.common.data.page.TextPageData; |
34 | 34 | import org.thingsboard.server.common.data.page.TextPageLink; |
35 | +import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; | |
35 | 36 | import org.thingsboard.server.dao.tenant.TenantService; |
36 | 37 | import org.thingsboard.server.service.install.InstallScripts; |
37 | 38 | |
... | ... | @@ -84,6 +85,8 @@ public class TenantController extends BaseController { |
84 | 85 | try { |
85 | 86 | TenantId tenantId = new TenantId(toUUID(strTenantId)); |
86 | 87 | tenantService.deleteTenant(tenantId); |
88 | + | |
89 | + actorService.onEntityStateChange(tenantId, tenantId, ComponentLifecycleEvent.DELETED); | |
87 | 90 | } catch (Exception e) { |
88 | 91 | throw handleException(e); |
89 | 92 | } | ... | ... |