简柏林 / thingskit · Commits

Commit bb1585592badaa1ae00a888d55f34c2eebb3ba69

Authored by Andrii Shvaika
1 parent f572f473

Actor System refactoring

Inline Side-by-side
Showing 48 changed files with 796 additions and 683 deletions
... ... @@ -59,6 +59,10 @@
59 59 </dependency>
60 60 <dependency>
61 61 <groupId>org.thingsboard.common</groupId>
  62 + <artifactId>actor</artifactId>
  63 + </dependency>
  64 + <dependency>
  65 + <groupId>org.thingsboard.common</groupId>
62 66 <artifactId>util</artifactId>
63 67 </dependency>
64 68 <dependency>
... ... @@ -174,14 +178,6 @@
174 178 <artifactId>spring-context-support</artifactId>
175 179 </dependency>
176 180 <dependency>
177   - <groupId>com.typesafe.akka</groupId>
178   - <artifactId>akka-actor_${scala.version}</artifactId>
179   - </dependency>
180   - <dependency>
181   - <groupId>com.typesafe.akka</groupId>
182   - <artifactId>akka-slf4j_${scala.version}</artifactId>
183   - </dependency>
184   - <dependency>
185 181 <groupId>org.slf4j</groupId>
186 182 <artifactId>slf4j-api</artifactId>
187 183 </dependency>
... ...
... ... @@ -15,9 +15,6 @@
15 15 */
16 16 package org.thingsboard.server.actors;
17 17
18   -import akka.actor.ActorRef;
19   -import akka.actor.ActorSystem;
20   -import akka.actor.Scheduler;
21 18 import com.fasterxml.jackson.databind.JsonNode;
22 19 import com.fasterxml.jackson.databind.ObjectMapper;
23 20 import com.fasterxml.jackson.databind.node.ObjectNode;
... ... @@ -25,8 +22,6 @@ import com.google.common.util.concurrent.FutureCallback;
25 22 import com.google.common.util.concurrent.Futures;
26 23 import com.google.common.util.concurrent.ListenableFuture;
27 24 import com.google.common.util.concurrent.MoreExecutors;
28   -import com.typesafe.config.Config;
29   -import com.typesafe.config.ConfigFactory;
30 25 import lombok.Getter;
31 26 import lombok.Setter;
32 27 import lombok.extern.slf4j.Slf4j;
... ... @@ -91,12 +86,13 @@ import java.io.StringWriter;
91 86 import java.util.Optional;
92 87 import java.util.concurrent.ConcurrentHashMap;
93 88 import java.util.concurrent.ConcurrentMap;
  89 +import java.util.concurrent.ScheduledExecutorService;
  90 +import java.util.concurrent.TimeUnit;
94 91 import java.util.concurrent.atomic.AtomicInteger;
95 92
96 93 @Slf4j
97 94 @Component
98 95 public class ActorSystemContext {
99   - private static final String AKKA_CONF_FILE_NAME = "actor-system.conf";
100 96
101 97 protected final ObjectMapper mapper = new ObjectMapper();
102 98
... ... @@ -260,14 +256,6 @@ public class ActorSystemContext {
260 256 @Getter
261 257 private long syncSessionTimeout;
262 258
263   - @Value("${actors.queue.enabled}")
264   - @Getter
265   - private boolean queuePersistenceEnabled;
266   -
267   - @Value("${actors.queue.timeout}")
268   - @Getter
269   - private long queuePersistenceTimeout;
270   -
271 259 @Value("${actors.rule.chain.error_persist_frequency}")
272 260 @Getter
273 261 private long ruleChainErrorPersistFrequency;
... ... @@ -327,17 +315,14 @@ public class ActorSystemContext {
327 315
328 316 @Getter
329 317 @Setter
330   - private ActorSystem actorSystem;
  318 + private TbActorSystem actorSystem;
331 319
332 320 @Setter
333   - private ActorRef appActor;
  321 + private TbActorRef appActor;
334 322
335 323 @Getter
336 324 @Setter
337   - private ActorRef statsActor;
338   -
339   - @Getter
340   - private final Config config;
  325 + private TbActorRef statsActor;
341 326
342 327 @Autowired(required = false)
343 328 @Getter
... ... @@ -351,14 +336,8 @@ public class ActorSystemContext {
351 336 @Getter
352 337 private RedisTemplate<String, Object> redisTemplate;
353 338
354   - public ActorSystemContext() {
355   - config = ConfigFactory.parseResources(AKKA_CONF_FILE_NAME).withFallback(ConfigFactory.load());
356   - }
357   -
358   -
359   -
360   - public Scheduler getScheduler() {
361   - return actorSystem.scheduler();
  339 + public ScheduledExecutorService getScheduler() {
  340 + return actorSystem.getScheduler();
362 341 }
363 342
364 343 public void persistError(TenantId tenantId, EntityId entityId, String method, Exception e) {
... ... @@ -531,7 +510,18 @@ public class ActorSystemContext {
531 510 return Exception.class.isInstance(error) ? (Exception) error : new Exception(error);
532 511 }
533 512
534   - public void tell(TbActorMsg tbActorMsg, ActorRef sender) {
535   - appActor.tell(tbActorMsg, sender);
  513 + public void tell(TbActorMsg tbActorMsg) {
  514 + appActor.tell(tbActorMsg);
  515 + }
  516 +
  517 +
  518 + public void schedulePeriodicMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs, long periodInMs) {
  519 + log.debug("Scheduling periodic msg {} every {} ms with delay {} ms", msg, periodInMs, delayInMs);
  520 + getScheduler().scheduleWithFixedDelay(() -> ctx.tell(msg), delayInMs, periodInMs, TimeUnit.MILLISECONDS);
  521 + }
  522 +
  523 + public void scheduleMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs) {
  524 + log.debug("Scheduling msg {} with delay {} ms", msg, delayInMs);
  525 + getScheduler().schedule(() -> ctx.tell(msg), delayInMs, TimeUnit.MILLISECONDS);
536 526 }
537 527 }
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.actors;
  17 +
  18 +import lombok.RequiredArgsConstructor;
  19 +import org.thingsboard.server.common.data.EntityType;
  20 +import org.thingsboard.server.common.data.id.EntityId;
  21 +
  22 +import java.util.function.Predicate;
  23 +
  24 +@RequiredArgsConstructor
  25 +public class TbEntityTypeActorIdPredicate implements Predicate<TbActorId> {
  26 +
  27 + private final EntityType entityType;
  28 +
  29 + @Override
  30 + public boolean test(TbActorId actorId) {
  31 + return actorId instanceof TbEntityActorId && testEntityId(((TbEntityActorId) actorId).getEntityId());
  32 + }
  33 +
  34 + protected boolean testEntityId(EntityId entityId) {
  35 + return entityId.getEntityType().equals(entityType);
  36 + }
  37 +}
... ...
... ... @@ -15,21 +15,19 @@
15 15 */
16 16 package org.thingsboard.server.actors.app;
17 17
18   -import akka.actor.ActorRef;
19   -import akka.actor.LocalActorRef;
20   -import akka.actor.OneForOneStrategy;
21   -import akka.actor.Props;
22   -import akka.actor.SupervisorStrategy;
23   -import akka.actor.Terminated;
24   -import com.google.common.collect.BiMap;
25   -import com.google.common.collect.HashBiMap;
  18 +import lombok.extern.slf4j.Slf4j;
26 19 import org.thingsboard.server.actors.ActorSystemContext;
  20 +import org.thingsboard.server.actors.TbActor;
  21 +import org.thingsboard.server.actors.TbActorId;
  22 +import org.thingsboard.server.actors.TbActorRef;
  23 +import org.thingsboard.server.actors.TbEntityActorId;
27 24 import org.thingsboard.server.actors.service.ContextAwareActor;
28 25 import org.thingsboard.server.actors.service.ContextBasedCreator;
29 26 import org.thingsboard.server.actors.service.DefaultActorService;
30 27 import org.thingsboard.server.actors.tenant.TenantActor;
31 28 import org.thingsboard.server.common.data.EntityType;
32 29 import org.thingsboard.server.common.data.Tenant;
  30 +import org.thingsboard.server.common.data.id.EntityId;
33 31 import org.thingsboard.server.common.data.id.TenantId;
34 32 import org.thingsboard.server.common.data.page.PageDataIterable;
35 33 import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
... ... @@ -43,38 +41,27 @@ import org.thingsboard.server.common.msg.queue.ServiceType;
43 41 import org.thingsboard.server.dao.model.ModelConstants;
44 42 import org.thingsboard.server.dao.tenant.TenantService;
45 43 import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
46   -import scala.concurrent.duration.Duration;
47 44
48 45 import java.util.HashSet;
49 46 import java.util.Optional;
50 47 import java.util.Set;
51 48
  49 +@Slf4j
52 50 public class AppActor extends ContextAwareActor {
53 51
54 52 private static final TenantId SYSTEM_TENANT = new TenantId(ModelConstants.NULL_UUID);
55 53 private final TenantService tenantService;
56   - private final BiMap<TenantId, ActorRef> tenantActors;
57 54 private final Set<TenantId> deletedTenants;
58 55 private boolean ruleChainsInitialized;
59 56
60 57 private AppActor(ActorSystemContext systemContext) {
61 58 super(systemContext);
62 59 this.tenantService = systemContext.getTenantService();
63   - this.tenantActors = HashBiMap.create();
64 60 this.deletedTenants = new HashSet<>();
65 61 }
66 62
67 63 @Override
68   - public SupervisorStrategy supervisorStrategy() {
69   - return strategy;
70   - }
71   -
72   - @Override
73   - public void preStart() {
74   - }
75   -
76   - @Override
77   - protected boolean process(TbActorMsg msg) {
  64 + protected boolean doProcess(TbActorMsg msg) {
78 65 if (!ruleChainsInitialized) {
79 66 initTenantActors();
80 67 ruleChainsInitialized = true;
... ... @@ -86,7 +73,7 @@ public class AppActor extends ContextAwareActor {
86 73 case APP_INIT_MSG:
87 74 break;
88 75 case PARTITION_CHANGE_MSG:
89   - broadcast(msg);
  76 + ctx.broadcastToChildren(msg);
90 77 break;
91 78 case COMPONENT_LIFE_CYCLE_MSG:
92 79 onComponentLifecycleMsg((ComponentLifecycleMsg) msg);
... ... @@ -145,19 +132,15 @@ public class AppActor extends ContextAwareActor {
145 132 msg.getTbMsg().getCallback().onFailure(new RuleEngineException("Message has system tenant id!"));
146 133 } else {
147 134 if (!deletedTenants.contains(msg.getTenantId())) {
148   - getOrCreateTenantActor(msg.getTenantId()).tell(msg, self());
  135 + getOrCreateTenantActor(msg.getTenantId()).tell(msg);
149 136 } else {
150 137 msg.getTbMsg().getCallback().onSuccess();
151 138 }
152 139 }
153 140 }
154 141
155   - protected void broadcast(Object msg) {
156   - tenantActors.values().forEach(actorRef -> actorRef.tell(msg, ActorRef.noSender()));
157   - }
158   -
159 142 private void onComponentLifecycleMsg(ComponentLifecycleMsg msg) {
160   - ActorRef target = null;
  143 + TbActorRef target = null;
161 144 if (SYSTEM_TENANT.equals(msg.getTenantId())) {
162 145 log.warn("Message has system tenant id: {}", msg);
163 146 } else {
... ... @@ -166,17 +149,13 @@ public class AppActor extends ContextAwareActor {
166 149 log.info("[{}] Handling tenant deleted notification: {}", msg.getTenantId(), msg);
167 150 TenantId tenantId = new TenantId(msg.getEntityId().getId());
168 151 deletedTenants.add(tenantId);
169   - ActorRef tenantActor = tenantActors.get(tenantId);
170   - if (tenantActor != null) {
171   - log.debug("[{}] Deleting tenant actor: {}", msg.getTenantId(), tenantActor);
172   - context().stop(tenantActor);
173   - }
  152 + ctx.stop(new TbEntityActorId(tenantId));
174 153 } else {
175 154 target = getOrCreateTenantActor(msg.getTenantId());
176 155 }
177 156 }
178 157 if (target != null) {
179   - target.tell(msg, ActorRef.noSender());
  158 + target.tell(msg);
180 159 } else {
181 160 log.debug("[{}] Invalid component lifecycle msg: {}", msg.getTenantId(), msg);
182 161 }
... ... @@ -184,7 +163,7 @@ public class AppActor extends ContextAwareActor {
184 163
185 164 private void onToDeviceActorMsg(TenantAwareMsg msg) {
186 165 if (!deletedTenants.contains(msg.getTenantId())) {
187   - getOrCreateTenantActor(msg.getTenantId()).tell(msg, ActorRef.noSender());
  166 + getOrCreateTenantActor(msg.getTenantId()).tell(msg);
188 167 } else {
189 168 if (msg instanceof TransportToDeviceActorMsgWrapper) {
190 169 ((TransportToDeviceActorMsgWrapper) msg).getCallback().onSuccess();
... ... @@ -192,49 +171,27 @@ public class AppActor extends ContextAwareActor {
192 171 }
193 172 }
194 173
195   - private ActorRef getOrCreateTenantActor(TenantId tenantId) {
196   - return tenantActors.computeIfAbsent(tenantId, k -> {
197   - log.info("[{}] Creating tenant actor.", tenantId);
198   - ActorRef tenantActor = context().actorOf(Props.create(new TenantActor.ActorCreator(systemContext, tenantId))
199   - .withDispatcher(DefaultActorService.CORE_DISPATCHER_NAME), tenantId.toString());
200   - context().watch(tenantActor);
201   - log.info("[{}] Created tenant actor: {}.", tenantId, tenantActor);
202   - return tenantActor;
203   - });
204   - }
205   -
206   - @Override
207   - protected void processTermination(Terminated message) {
208   - ActorRef terminated = message.actor();
209   - if (terminated instanceof LocalActorRef) {
210   - boolean removed = tenantActors.inverse().remove(terminated) != null;
211   - if (removed) {
212   - log.debug("[{}] Removed actor:", terminated);
213   - }
214   - } else {
215   - throw new IllegalStateException("Remote actors are not supported!");
216   - }
  174 + private TbActorRef getOrCreateTenantActor(TenantId tenantId) {
  175 + return ctx.getOrCreateChildActor(new TbEntityActorId(tenantId),
  176 + () -> DefaultActorService.TENANT_DISPATCHER_NAME,
  177 + () -> new TenantActor.ActorCreator(systemContext, tenantId));
217 178 }
218 179
219   - public static class ActorCreator extends ContextBasedCreator<AppActor> {
220   - private static final long serialVersionUID = 1L;
  180 + public static class ActorCreator extends ContextBasedCreator {
221 181
222 182 public ActorCreator(ActorSystemContext context) {
223 183 super(context);
224 184 }
225 185
226 186 @Override
227   - public AppActor create() {
  187 + public TbActorId createActorId() {
  188 + return new TbEntityActorId(new TenantId(EntityId.NULL_UUID));
  189 + }
  190 +
  191 + @Override
  192 + public TbActor createActor() {
228 193 return new AppActor(context);
229 194 }
230 195 }
231 196
232   - private final SupervisorStrategy strategy = new OneForOneStrategy(3, Duration.create("1 minute"), t -> {
233   - log.warn("Unknown failure", t);
234   - if (t instanceof RuntimeException) {
235   - return SupervisorStrategy.restart();
236   - } else {
237   - return SupervisorStrategy.stop();
238   - }
239   - });
240 197 }
... ...
... ... @@ -15,9 +15,11 @@
15 15 */
16 16 package org.thingsboard.server.actors.device;
17 17
  18 +import lombok.extern.slf4j.Slf4j;
18 19 import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg;
19 20 import org.thingsboard.rule.engine.api.msg.DeviceNameOrTypeUpdateMsg;
20 21 import org.thingsboard.server.actors.ActorSystemContext;
  22 +import org.thingsboard.server.actors.TbActorCtx;
21 23 import org.thingsboard.server.actors.service.ContextAwareActor;
22 24 import org.thingsboard.server.common.data.id.DeviceId;
23 25 import org.thingsboard.server.common.data.id.TenantId;
... ... @@ -26,6 +28,7 @@ import org.thingsboard.server.common.msg.timeout.DeviceActorServerSideRpcTimeout
26 28 import org.thingsboard.server.service.rpc.ToDeviceRpcRequestActorMsg;
27 29 import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
28 30
  31 +@Slf4j
29 32 public class DeviceActor extends ContextAwareActor {
30 33
31 34 private final DeviceActorMessageProcessor processor;
... ... @@ -36,10 +39,11 @@ public class DeviceActor extends ContextAwareActor {
36 39 }
37 40
38 41 @Override
39   - public void preStart() {
  42 + public void init(TbActorCtx ctx) {
  43 + super.init(ctx);
40 44 log.debug("[{}][{}] Starting device actor.", processor.tenantId, processor.deviceId);
41 45 try {
42   - processor.initSessionTimeout(context());
  46 + processor.initSessionTimeout(ctx);
43 47 log.debug("[{}][{}] Device actor started.", processor.tenantId, processor.deviceId);
44 48 } catch (Exception e) {
45 49 log.warn("[{}][{}] Unknown failure", processor.tenantId, processor.deviceId, e);
... ... @@ -47,18 +51,13 @@ public class DeviceActor extends ContextAwareActor {
47 51 }
48 52
49 53 @Override
50   - public void postStop() {
51   -
52   - }
53   -
54   - @Override
55   - protected boolean process(TbActorMsg msg) {
  54 + protected boolean doProcess(TbActorMsg msg) {
56 55 switch (msg.getMsgType()) {
57 56 case TRANSPORT_TO_DEVICE_ACTOR_MSG:
58   - processor.process(context(), (TransportToDeviceActorMsgWrapper) msg);
  57 + processor.process(ctx, (TransportToDeviceActorMsgWrapper) msg);
59 58 break;
60 59 case DEVICE_ATTRIBUTES_UPDATE_TO_DEVICE_ACTOR_MSG:
61   - processor.processAttributesUpdate(context(), (DeviceAttributesEventNotificationMsg) msg);
  60 + processor.processAttributesUpdate(ctx, (DeviceAttributesEventNotificationMsg) msg);
62 61 break;
63 62 case DEVICE_CREDENTIALS_UPDATE_TO_DEVICE_ACTOR_MSG:
64 63 processor.processCredentialsUpdate();
... ... @@ -67,10 +66,10 @@ public class DeviceActor extends ContextAwareActor {
67 66 processor.processNameOrTypeUpdate((DeviceNameOrTypeUpdateMsg) msg);
68 67 break;
69 68 case DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG:
70   - processor.processRpcRequest(context(), (ToDeviceRpcRequestActorMsg) msg);
  69 + processor.processRpcRequest(ctx, (ToDeviceRpcRequestActorMsg) msg);
71 70 break;
72 71 case DEVICE_ACTOR_SERVER_SIDE_RPC_TIMEOUT_MSG:
73   - processor.processServerSideRpcTimeout(context(), (DeviceActorServerSideRpcTimeoutMsg) msg);
  72 + processor.processServerSideRpcTimeout(ctx, (DeviceActorServerSideRpcTimeoutMsg) msg);
74 73 break;
75 74 case SESSION_TIMEOUT_MSG:
76 75 processor.checkSessionsTimeout();
... ...
... ... @@ -16,12 +16,14 @@
16 16 package org.thingsboard.server.actors.device;
17 17
18 18 import org.thingsboard.server.actors.ActorSystemContext;
  19 +import org.thingsboard.server.actors.TbActor;
  20 +import org.thingsboard.server.actors.TbActorId;
  21 +import org.thingsboard.server.actors.TbEntityActorId;
19 22 import org.thingsboard.server.actors.service.ContextBasedCreator;
20 23 import org.thingsboard.server.common.data.id.DeviceId;
21 24 import org.thingsboard.server.common.data.id.TenantId;
22 25
23   -public class DeviceActorCreator extends ContextBasedCreator<DeviceActor> {
24   - private static final long serialVersionUID = 1L;
  26 +public class DeviceActorCreator extends ContextBasedCreator {
25 27
26 28 private final TenantId tenantId;
27 29 private final DeviceId deviceId;
... ... @@ -33,7 +35,13 @@ public class DeviceActorCreator extends ContextBasedCreator<DeviceActor> {
33 35 }
34 36
35 37 @Override
36   - public DeviceActor create() {
  38 + public TbActorId createActorId() {
  39 + return new TbEntityActorId(deviceId);
  40 + }
  41 +
  42 + @Override
  43 + public TbActor createActor() {
37 44 return new DeviceActor(context, tenantId, deviceId);
38 45 }
  46 +
39 47 }
... ...
... ... @@ -15,7 +15,6 @@
15 15 */
16 16 package org.thingsboard.server.actors.device;
17 17
18   -import akka.actor.ActorContext;
19 18 import com.google.common.util.concurrent.FutureCallback;
20 19 import com.google.common.util.concurrent.Futures;
21 20 import com.google.common.util.concurrent.ListenableFuture;
... ... @@ -27,6 +26,7 @@ import org.thingsboard.rule.engine.api.RpcError;
27 26 import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg;
28 27 import org.thingsboard.rule.engine.api.msg.DeviceNameOrTypeUpdateMsg;
29 28 import org.thingsboard.server.actors.ActorSystemContext;
  29 +import org.thingsboard.server.actors.TbActorCtx;
30 30 import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor;
31 31 import org.thingsboard.server.common.data.Device;
32 32 import org.thingsboard.server.common.data.id.DeviceId;
... ... @@ -127,7 +127,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
127 127 }
128 128 }
129 129
130   - void processRpcRequest(ActorContext context, ToDeviceRpcRequestActorMsg msg) {
  130 + void processRpcRequest(TbActorCtx context, ToDeviceRpcRequestActorMsg msg) {
131 131 ToDeviceRpcRequest request = msg.getMsg();
132 132 ToDeviceRpcRequestBody body = request.getBody();
133 133 ToDeviceRpcRequestMsg rpcRequest = ToDeviceRpcRequestMsg.newBuilder().setRequestId(
... ... @@ -162,13 +162,13 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
162 162 }
163 163 }
164 164
165   - private void registerPendingRpcRequest(ActorContext context, ToDeviceRpcRequestActorMsg msg, boolean sent, ToDeviceRpcRequestMsg rpcRequest, long timeout) {
  165 + private void registerPendingRpcRequest(TbActorCtx context, ToDeviceRpcRequestActorMsg msg, boolean sent, ToDeviceRpcRequestMsg rpcRequest, long timeout) {
166 166 toDeviceRpcPendingMap.put(rpcRequest.getRequestId(), new ToDeviceRpcRequestMetadata(msg, sent));
167 167 DeviceActorServerSideRpcTimeoutMsg timeoutMsg = new DeviceActorServerSideRpcTimeoutMsg(rpcRequest.getRequestId(), timeout);
168 168 scheduleMsgWithDelay(context, timeoutMsg, timeoutMsg.getTimeout());
169 169 }
170 170
171   - void processServerSideRpcTimeout(ActorContext context, DeviceActorServerSideRpcTimeoutMsg msg) {
  171 + void processServerSideRpcTimeout(TbActorCtx context, DeviceActorServerSideRpcTimeoutMsg msg) {
172 172 ToDeviceRpcRequestMetadata requestMd = toDeviceRpcPendingMap.remove(msg.getId());
173 173 if (requestMd != null) {
174 174 log.debug("[{}] RPC request [{}] timeout detected!", deviceId, msg.getId());
... ... @@ -177,7 +177,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
177 177 }
178 178 }
179 179
180   - private void sendPendingRequests(ActorContext context, UUID sessionId, SessionInfoProto sessionInfo) {
  180 + private void sendPendingRequests(TbActorCtx context, UUID sessionId, SessionInfoProto sessionInfo) {
181 181 SessionType sessionType = getSessionType(sessionId);
182 182 if (!toDeviceRpcPendingMap.isEmpty()) {
183 183 log.debug("[{}] Pushing {} pending RPC messages to new async session [{}]", deviceId, toDeviceRpcPendingMap.size(), sessionId);
... ... @@ -198,7 +198,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
198 198 sentOneWayIds.forEach(toDeviceRpcPendingMap::remove);
199 199 }
200 200
201   - private Consumer<Map.Entry<Integer, ToDeviceRpcRequestMetadata>> processPendingRpc(ActorContext context, UUID sessionId, String nodeId, Set<Integer> sentOneWayIds) {
  201 + private Consumer<Map.Entry<Integer, ToDeviceRpcRequestMetadata>> processPendingRpc(TbActorCtx context, UUID sessionId, String nodeId, Set<Integer> sentOneWayIds) {
202 202 return entry -> {
203 203 ToDeviceRpcRequest request = entry.getValue().getMsg().getMsg();
204 204 ToDeviceRpcRequestBody body = request.getBody();
... ... @@ -212,7 +212,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
212 212 };
213 213 }
214 214
215   - void process(ActorContext context, TransportToDeviceActorMsgWrapper wrapper) {
  215 + void process(TbActorCtx context, TransportToDeviceActorMsgWrapper wrapper) {
216 216 TransportToDeviceActorMsg msg = wrapper.getMsg();
217 217 TbCallback callback = wrapper.getCallback();
218 218 if (msg.hasSessionEvent()) {
... ... @@ -239,7 +239,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
239 239 callback.onSuccess();
240 240 }
241 241
242   - private void handleClaimDeviceMsg(ActorContext context, SessionInfoProto sessionInfo, TransportProtos.ClaimDeviceMsg msg) {
  242 + private void handleClaimDeviceMsg(TbActorCtx context, SessionInfoProto sessionInfo, TransportProtos.ClaimDeviceMsg msg) {
243 243 DeviceId deviceId = new DeviceId(new UUID(msg.getDeviceIdMSB(), msg.getDeviceIdLSB()));
244 244 systemContext.getClaimDevicesService().registerClaimingInfo(tenantId, deviceId, msg.getSecretKey(), msg.getDurationMs());
245 245 }
... ... @@ -252,7 +252,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
252 252 systemContext.getDeviceStateService().onDeviceDisconnect(deviceId);
253 253 }
254 254
255   - private void handleGetAttributesRequest(ActorContext context, SessionInfoProto sessionInfo, GetAttributeRequestMsg request) {
  255 + private void handleGetAttributesRequest(TbActorCtx context, SessionInfoProto sessionInfo, GetAttributeRequestMsg request) {
256 256 int requestId = request.getRequestId();
257 257 Futures.addCallback(getAttributesKvEntries(request), new FutureCallback<List<List<AttributeKvEntry>>>() {
258 258 @Override
... ... @@ -310,7 +310,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
310 310 return sessions.containsKey(sessionId) ? SessionType.ASYNC : SessionType.SYNC;
311 311 }
312 312
313   - void processAttributesUpdate(ActorContext context, DeviceAttributesEventNotificationMsg msg) {
  313 + void processAttributesUpdate(TbActorCtx context, DeviceAttributesEventNotificationMsg msg) {
314 314 if (attributeSubscriptions.size() > 0) {
315 315 boolean hasNotificationData = false;
316 316 AttributeUpdateNotificationMsg.Builder notification = AttributeUpdateNotificationMsg.newBuilder();
... ... @@ -349,7 +349,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
349 349 }
350 350 }
351 351
352   - private void processRpcResponses(ActorContext context, SessionInfoProto sessionInfo, ToDeviceRpcResponseMsg responseMsg) {
  352 + private void processRpcResponses(TbActorCtx context, SessionInfoProto sessionInfo, ToDeviceRpcResponseMsg responseMsg) {
353 353 UUID sessionId = getSessionId(sessionInfo);
354 354 log.debug("[{}] Processing rpc command response [{}]", deviceId, sessionId);
355 355 ToDeviceRpcRequestMetadata requestMd = toDeviceRpcPendingMap.remove(responseMsg.getRequestId());
... ... @@ -362,7 +362,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
362 362 }
363 363 }
364 364
365   - private void processSubscriptionCommands(ActorContext context, SessionInfoProto sessionInfo, SubscribeToAttributeUpdatesMsg subscribeCmd) {
  365 + private void processSubscriptionCommands(TbActorCtx context, SessionInfoProto sessionInfo, SubscribeToAttributeUpdatesMsg subscribeCmd) {
366 366 UUID sessionId = getSessionId(sessionInfo);
367 367 if (subscribeCmd.getUnsubscribe()) {
368 368 log.debug("[{}] Canceling attributes subscription for session [{}]", deviceId, sessionId);
... ... @@ -383,7 +383,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
383 383 return new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB());
384 384 }
385 385
386   - private void processSubscriptionCommands(ActorContext context, SessionInfoProto sessionInfo, SubscribeToRPCMsg subscribeCmd) {
  386 + private void processSubscriptionCommands(TbActorCtx context, SessionInfoProto sessionInfo, SubscribeToRPCMsg subscribeCmd) {
387 387 UUID sessionId = getSessionId(sessionInfo);
388 388 if (subscribeCmd.getUnsubscribe()) {
389 389 log.debug("[{}] Canceling rpc subscription for session [{}]", deviceId, sessionId);
... ... @@ -433,7 +433,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
433 433 }
434 434 }
435 435
436   - private void handleSessionActivity(ActorContext context, SessionInfoProto sessionInfoProto, SubscriptionInfoProto subscriptionInfo) {
  436 + private void handleSessionActivity(TbActorCtx context, SessionInfoProto sessionInfoProto, SubscriptionInfoProto subscriptionInfo) {
437 437 UUID sessionId = getSessionId(sessionInfoProto);
438 438 SessionInfoMetaData sessionMD = sessions.computeIfAbsent(sessionId,
439 439 id -> new SessionInfoMetaData(new SessionInfo(SessionType.ASYNC, sessionInfoProto.getNodeId()), 0L));
... ... @@ -612,8 +612,8 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
612 612 .addAllSessions(sessionsList).build().toByteArray());
613 613 }
614 614
615   - void initSessionTimeout(ActorContext context) {
616   - schedulePeriodicMsgWithDelay(context, SessionTimeoutCheckMsg.instance(), systemContext.getSessionInactivityTimeout(), systemContext.getSessionInactivityTimeout());
  615 + void initSessionTimeout(TbActorCtx ctx) {
  616 + schedulePeriodicMsgWithDelay(ctx, SessionTimeoutCheckMsg.instance(), systemContext.getSessionInactivityTimeout(), systemContext.getSessionInactivityTimeout());
617 617 }
618 618
619 619 void checkSessionsTimeout() {
... ...
... ... @@ -15,7 +15,6 @@
15 15 */
16 16 package org.thingsboard.server.actors.ruleChain;
17 17
18   -import akka.actor.ActorRef;
19 18 import com.datastax.driver.core.ResultSetFuture;
20 19 import com.fasterxml.jackson.core.JsonProcessingException;
21 20 import com.fasterxml.jackson.databind.ObjectMapper;
... ... @@ -30,6 +29,7 @@ import org.thingsboard.rule.engine.api.ScriptEngine;
30 29 import org.thingsboard.rule.engine.api.TbContext;
31 30 import org.thingsboard.rule.engine.api.TbRelationTypes;
32 31 import org.thingsboard.server.actors.ActorSystemContext;
  32 +import org.thingsboard.server.actors.TbActorRef;
33 33 import org.thingsboard.server.common.data.Customer;
34 34 import org.thingsboard.server.common.data.DataConstants;
35 35 import org.thingsboard.server.common.data.Device;
... ... @@ -40,6 +40,7 @@ import org.thingsboard.server.common.data.id.RuleChainId;
40 40 import org.thingsboard.server.common.data.id.RuleNodeId;
41 41 import org.thingsboard.server.common.data.id.TenantId;
42 42 import org.thingsboard.server.common.data.rule.RuleNode;
  43 +import org.thingsboard.server.common.msg.TbActorMsg;
43 44 import org.thingsboard.server.common.msg.TbMsg;
44 45 import org.thingsboard.server.common.msg.TbMsgMetaData;
45 46 import org.thingsboard.server.common.msg.queue.ServiceType;
... ... @@ -62,7 +63,6 @@ import org.thingsboard.server.gen.transport.TransportProtos;
62 63 import org.thingsboard.server.queue.TbQueueCallback;
63 64 import org.thingsboard.server.queue.TbQueueMsgMetadata;
64 65 import org.thingsboard.server.service.script.RuleNodeJsScriptEngine;
65   -import scala.concurrent.duration.Duration;
66 66
67 67 import java.util.Collections;
68 68 import java.util.Set;
... ... @@ -104,7 +104,7 @@ class DefaultTbContext implements TbContext {
104 104 if (nodeCtx.getSelf().isDebugMode()) {
105 105 relationTypes.forEach(relationType -> mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg, relationType, th));
106 106 }
107   - nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getId(), relationTypes, msg, th != null ? th.getMessage() : null), nodeCtx.getSelfActor());
  107 + nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getId(), relationTypes, msg, th != null ? th.getMessage() : null));
108 108 }
109 109
110 110 @Override
... ... @@ -130,7 +130,7 @@ class DefaultTbContext implements TbContext {
130 130 .setTenantIdMSB(getTenantId().getId().getMostSignificantBits())
131 131 .setTenantIdLSB(getTenantId().getId().getLeastSignificantBits())
132 132 .setTbMsg(TbMsg.toByteString(tbMsg)).build();
133   - mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg, new SimpleTbQueueCallback(onSuccess, onFailure));
  133 + mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg, new SimpleTbQueueCallback(onSuccess, onFailure));
134 134 }
135 135
136 136 @Override
... ... @@ -187,7 +187,7 @@ class DefaultTbContext implements TbContext {
187 187 if (failureMessage != null) {
188 188 msg.setFailureMessage(failureMessage);
189 189 }
190   - mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg.build(), new SimpleTbQueueCallback(onSuccess, onFailure));
  190 + mainCtx.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), msg.build(), new SimpleTbQueueCallback(onSuccess, onFailure));
191 191 }
192 192
193 193 @Override
... ... @@ -203,8 +203,8 @@ class DefaultTbContext implements TbContext {
203 203 return mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), entityId).isMyPartition();
204 204 }
205 205
206   - private void scheduleMsgWithDelay(Object msg, long delayInMs, ActorRef target) {
207   - mainCtx.getScheduler().scheduleOnce(Duration.create(delayInMs, TimeUnit.MILLISECONDS), target, msg, mainCtx.getActorSystem().dispatcher(), nodeCtx.getSelfActor());
  206 + private void scheduleMsgWithDelay(TbActorMsg msg, long delayInMs, TbActorRef target) {
  207 + mainCtx.scheduleMsgWithDelay(target, msg, delayInMs);
208 208 }
209 209
210 210 @Override
... ... @@ -213,7 +213,7 @@ class DefaultTbContext implements TbContext {
213 213 mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg, TbRelationTypes.FAILURE, th);
214 214 }
215 215 nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getId(), Collections.singleton(TbRelationTypes.FAILURE),
216   - msg, th != null ? th.getMessage() : null), nodeCtx.getSelfActor());
  216 + msg, th != null ? th.getMessage() : null));
217 217 }
218 218
219 219 public void updateSelf(RuleNode self) {
... ...
... ... @@ -15,9 +15,11 @@
15 15 */
16 16 package org.thingsboard.server.actors.ruleChain;
17 17
18   -import akka.actor.OneForOneStrategy;
19   -import akka.actor.SupervisorStrategy;
20 18 import org.thingsboard.server.actors.ActorSystemContext;
  19 +import org.thingsboard.server.actors.TbActor;
  20 +import org.thingsboard.server.actors.TbActorCtx;
  21 +import org.thingsboard.server.actors.TbActorId;
  22 +import org.thingsboard.server.actors.TbEntityActorId;
21 23 import org.thingsboard.server.actors.service.ComponentActor;
22 24 import org.thingsboard.server.actors.service.ContextBasedCreator;
23 25 import org.thingsboard.server.common.data.id.RuleChainId;
... ... @@ -27,18 +29,24 @@ import org.thingsboard.server.common.msg.TbActorMsg;
27 29 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
28 30 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
29 31 import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
30   -import scala.concurrent.duration.Duration;
31 32
32 33 public class RuleChainActor extends ComponentActor<RuleChainId, RuleChainActorMessageProcessor> {
33 34
  35 + private final RuleChain ruleChain;
  36 +
34 37 private RuleChainActor(ActorSystemContext systemContext, TenantId tenantId, RuleChain ruleChain) {
35 38 super(systemContext, tenantId, ruleChain.getId());
36   - setProcessor(new RuleChainActorMessageProcessor(tenantId, ruleChain, systemContext,
37   - context().parent(), context().self()));
  39 + this.ruleChain = ruleChain;
38 40 }
39 41
40 42 @Override
41   - protected boolean process(TbActorMsg msg) {
  43 + protected RuleChainActorMessageProcessor createProcessor(TbActorCtx ctx) {
  44 + return new RuleChainActorMessageProcessor(tenantId, ruleChain, systemContext,
  45 + ctx.getParentRef(), ctx);
  46 + }
  47 +
  48 + @Override
  49 + protected boolean doProcess(TbActorMsg msg) {
42 50 switch (msg.getMsgType()) {
43 51 case COMPONENT_LIFE_CYCLE_MSG:
44 52 onComponentLifecycleMsg((ComponentLifecycleMsg) msg);
... ... @@ -64,7 +72,7 @@ public class RuleChainActor extends ComponentActor<RuleChainId, RuleChainActorMe
64 72 return true;
65 73 }
66 74
67   - public static class ActorCreator extends ContextBasedCreator<RuleChainActor> {
  75 + public static class ActorCreator extends ContextBasedCreator {
68 76 private static final long serialVersionUID = 1L;
69 77
70 78 private final TenantId tenantId;
... ... @@ -77,7 +85,12 @@ public class RuleChainActor extends ComponentActor<RuleChainId, RuleChainActorMe
77 85 }
78 86
79 87 @Override
80   - public RuleChainActor create() {
  88 + public TbActorId createActorId() {
  89 + return new TbEntityActorId(ruleChain.getId());
  90 + }
  91 +
  92 + @Override
  93 + public TbActor createActor() {
81 94 return new RuleChainActor(context, tenantId, ruleChain);
82 95 }
83 96 }
... ... @@ -87,13 +100,4 @@ public class RuleChainActor extends ComponentActor<RuleChainId, RuleChainActorMe
87 100 return systemContext.getRuleChainErrorPersistFrequency();
88 101 }
89 102
90   - @Override
91   - public SupervisorStrategy supervisorStrategy() {
92   - return strategy;
93   - }
94   -
95   - private final SupervisorStrategy strategy = new OneForOneStrategy(3, Duration.create("1 minute"), t -> {
96   - logAndPersist("Unknown Failure", ActorSystemContext.toException(t));
97   - return SupervisorStrategy.resume();
98   - });
99 103 }
... ...
... ... @@ -15,12 +15,12 @@
15 15 */
16 16 package org.thingsboard.server.actors.ruleChain;
17 17
18   -import akka.actor.ActorContext;
19   -import akka.actor.ActorRef;
20   -import akka.actor.Props;
21 18 import lombok.extern.slf4j.Slf4j;
22 19 import org.thingsboard.rule.engine.api.TbRelationTypes;
23 20 import org.thingsboard.server.actors.ActorSystemContext;
  21 +import org.thingsboard.server.actors.TbActorCtx;
  22 +import org.thingsboard.server.actors.TbActorRef;
  23 +import org.thingsboard.server.actors.TbEntityActorId;
24 24 import org.thingsboard.server.actors.service.DefaultActorService;
25 25 import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
26 26 import org.thingsboard.server.common.data.EntityType;
... ... @@ -62,8 +62,8 @@ import java.util.stream.Collectors;
62 62 @Slf4j
63 63 public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleChainId> {
64 64
65   - private final ActorRef parent;
66   - private final ActorRef self;
  65 + private final TbActorRef parent;
  66 + private final TbActorRef self;
67 67 private final Map<RuleNodeId, RuleNodeCtx> nodeActors;
68 68 private final Map<RuleNodeId, List<RuleNodeRelation>> nodeRoutes;
69 69 private final RuleChainService service;
... ... @@ -75,7 +75,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
75 75 private boolean started;
76 76
77 77 RuleChainActorMessageProcessor(TenantId tenantId, RuleChain ruleChain, ActorSystemContext systemContext
78   - , ActorRef parent, ActorRef self) {
  78 + , TbActorRef parent, TbActorRef self) {
79 79 super(systemContext, tenantId, ruleChain.getId());
80 80 this.ruleChainName = ruleChain.getName();
81 81 this.parent = parent;
... ... @@ -92,7 +92,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
92 92 }
93 93
94 94 @Override
95   - public void start(ActorContext context) {
  95 + public void start(TbActorCtx context) {
96 96 if (!started) {
97 97 RuleChain ruleChain = service.findRuleChainById(tenantId, entityId);
98 98 if (ruleChain != null) {
... ... @@ -101,7 +101,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
101 101 // Creating and starting the actors;
102 102 for (RuleNode ruleNode : ruleNodeList) {
103 103 log.trace("[{}][{}] Creating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode);
104   - ActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode);
  104 + TbActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode);
105 105 nodeActors.put(ruleNode.getId(), new RuleNodeCtx(tenantId, self, ruleNodeActor, ruleNode));
106 106 }
107 107 initRoutes(ruleChain, ruleNodeList);
... ... @@ -113,7 +113,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
113 113 }
114 114
115 115 @Override
116   - public void onUpdate(ActorContext context) {
  116 + public void onUpdate(TbActorCtx context) {
117 117 RuleChain ruleChain = service.findRuleChainById(tenantId, entityId);
118 118 if (ruleChain != null) {
119 119 ruleChainName = ruleChain.getName();
... ... @@ -123,12 +123,12 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
123 123 RuleNodeCtx existing = nodeActors.get(ruleNode.getId());
124 124 if (existing == null) {
125 125 log.trace("[{}][{}] Creating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode);
126   - ActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode);
  126 + TbActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode);
127 127 nodeActors.put(ruleNode.getId(), new RuleNodeCtx(tenantId, self, ruleNodeActor, ruleNode));
128 128 } else {
129 129 log.trace("[{}][{}] Updating rule node [{}]: {}", entityId, ruleNode.getId(), ruleNode.getName(), ruleNode);
130 130 existing.setSelf(ruleNode);
131   - existing.getSelfActor().tell(new ComponentLifecycleMsg(tenantId, existing.getSelf().getId(), ComponentLifecycleEvent.UPDATED), self);
  131 + existing.getSelfActor().tell(new ComponentLifecycleMsg(tenantId, existing.getSelf().getId(), ComponentLifecycleEvent.UPDATED));
132 132 }
133 133 }
134 134
... ... @@ -137,7 +137,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
137 137 removedRules.forEach(ruleNodeId -> {
138 138 log.trace("[{}][{}] Removing rule node [{}]", tenantId, entityId, ruleNodeId);
139 139 RuleNodeCtx removed = nodeActors.remove(ruleNodeId);
140   - removed.getSelfActor().tell(new ComponentLifecycleMsg(tenantId, removed.getSelf().getId(), ComponentLifecycleEvent.DELETED), self);
  140 + removed.getSelfActor().tell(new ComponentLifecycleMsg(tenantId, removed.getSelf().getId(), ComponentLifecycleEvent.DELETED));
141 141 });
142 142
143 143 initRoutes(ruleChain, ruleNodeList);
... ... @@ -145,26 +145,23 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
145 145 }
146 146
147 147 @Override
148   - public void stop(ActorContext context) {
  148 + public void stop(TbActorCtx ctx) {
149 149 log.trace("[{}][{}] Stopping rule chain with {} nodes", tenantId, entityId, nodeActors.size());
150   - nodeActors.values().stream().map(RuleNodeCtx::getSelfActor).forEach(context::stop);
  150 + nodeActors.values().stream().map(RuleNodeCtx::getSelfActor).map(TbActorRef::getActorId).forEach(ctx::stop);
151 151 nodeActors.clear();
152 152 nodeRoutes.clear();
153   - context.stop(self);
154 153 started = false;
155 154 }
156 155
157 156 @Override
158 157 public void onPartitionChangeMsg(PartitionChangeMsg msg) {
159   - nodeActors.values().stream().map(RuleNodeCtx::getSelfActor).forEach(actorRef -> actorRef.tell(msg, self));
  158 + nodeActors.values().stream().map(RuleNodeCtx::getSelfActor).forEach(actorRef -> actorRef.tell(msg));
160 159 }
161 160
162   - private ActorRef createRuleNodeActor(ActorContext context, RuleNode ruleNode) {
163   - String dispatcherName = tenantId.getId().equals(EntityId.NULL_UUID) ?
164   - DefaultActorService.SYSTEM_RULE_DISPATCHER_NAME : DefaultActorService.TENANT_RULE_DISPATCHER_NAME;
165   - return context.actorOf(
166   - Props.create(new RuleNodeActor.ActorCreator(systemContext, tenantId, entityId, ruleNode.getName(), ruleNode.getId()))
167   - .withDispatcher(dispatcherName), ruleNode.getId().toString());
  161 + private TbActorRef createRuleNodeActor(TbActorCtx ctx, RuleNode ruleNode) {
  162 + return ctx.getOrCreateChildActor(new TbEntityActorId(ruleNode.getId()),
  163 + () -> DefaultActorService.RULE_DISPATCHER_NAME,
  164 + () -> new RuleNodeActor.ActorCreator(systemContext, tenantId, entityId, ruleNode.getName(), ruleNode.getId()));
168 165 }
169 166
170 167 private void initRoutes(RuleChain ruleChain, List<RuleNode> ruleNodeList) {
... ... @@ -303,7 +300,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
303 300 pushMsgToNode(nodeActors.get(new RuleNodeId(target.getId())), msg, fromRelationType);
304 301 break;
305 302 case RULE_CHAIN:
306   - parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, msg, fromRelationType), self);
  303 + parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, msg, fromRelationType));
307 304 break;
308 305 }
309 306 } else {
... ... @@ -334,7 +331,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
334 331
335 332 private void pushMsgToNode(RuleNodeCtx nodeCtx, TbMsg msg, String fromRelationType) {
336 333 if (nodeCtx != null) {
337   - nodeCtx.getSelfActor().tell(new RuleChainToRuleNodeMsg(new DefaultTbContext(systemContext, nodeCtx), msg, fromRelationType), self);
  334 + nodeCtx.getSelfActor().tell(new RuleChainToRuleNodeMsg(new DefaultTbContext(systemContext, nodeCtx), msg, fromRelationType));
338 335 } else {
339 336 log.error("[{}][{}] RuleNodeCtx is empty", entityId, ruleChainName);
340 337 msg.getCallback().onFailure(new RuleEngineException("Rule Node CTX is empty"));
... ...
... ... @@ -15,21 +15,22 @@
15 15 */
16 16 package org.thingsboard.server.actors.ruleChain;
17 17
18   -import akka.actor.ActorContext;
19   -import akka.actor.ActorRef;
20   -import akka.actor.Props;
21   -import com.google.common.collect.BiMap;
22   -import com.google.common.collect.HashBiMap;
23 18 import lombok.Getter;
  19 +import lombok.extern.slf4j.Slf4j;
24 20 import org.thingsboard.server.actors.ActorSystemContext;
  21 +import org.thingsboard.server.actors.TbActorRef;
  22 +import org.thingsboard.server.actors.TbEntityActorId;
  23 +import org.thingsboard.server.actors.TbEntityTypeActorIdPredicate;
25 24 import org.thingsboard.server.actors.service.ContextAwareActor;
26 25 import org.thingsboard.server.actors.service.DefaultActorService;
  26 +import org.thingsboard.server.actors.tenant.TenantActor;
27 27 import org.thingsboard.server.common.data.EntityType;
28 28 import org.thingsboard.server.common.data.id.EntityId;
29 29 import org.thingsboard.server.common.data.id.RuleChainId;
30 30 import org.thingsboard.server.common.data.id.TenantId;
31 31 import org.thingsboard.server.common.data.page.PageDataIterable;
32 32 import org.thingsboard.server.common.data.rule.RuleChain;
  33 +import org.thingsboard.server.common.msg.TbActorMsg;
33 34 import org.thingsboard.server.dao.rule.RuleChainService;
34 35
35 36 import java.util.function.Function;
... ... @@ -37,20 +38,19 @@ import java.util.function.Function;
37 38 /**
38 39 * Created by ashvayka on 15.03.18.
39 40 */
  41 +@Slf4j
40 42 public abstract class RuleChainManagerActor extends ContextAwareActor {
41 43
42 44 protected final TenantId tenantId;
43 45 private final RuleChainService ruleChainService;
44   - private final BiMap<RuleChainId, ActorRef> actors;
45 46 @Getter
46 47 protected RuleChain rootChain;
47 48 @Getter
48   - protected ActorRef rootChainActor;
  49 + protected TbActorRef rootChainActor;
49 50
50 51 public RuleChainManagerActor(ActorSystemContext systemContext, TenantId tenantId) {
51 52 super(systemContext);
52 53 this.tenantId = tenantId;
53   - this.actors = HashBiMap.create();
54 54 this.ruleChainService = systemContext.getRuleChainService();
55 55 }
56 56
... ... @@ -58,46 +58,41 @@ public abstract class RuleChainManagerActor extends ContextAwareActor {
58 58 for (RuleChain ruleChain : new PageDataIterable<>(link -> ruleChainService.findTenantRuleChains(tenantId, link), ContextAwareActor.ENTITY_PACK_LIMIT)) {
59 59 RuleChainId ruleChainId = ruleChain.getId();
60 60 log.debug("[{}|{}] Creating rule chain actor", ruleChainId.getEntityType(), ruleChain.getId());
61   - //TODO: remove this cast making UUIDBased subclass of EntityId an interface and vice versa.
62   - ActorRef actorRef = getOrCreateActor(this.context(), ruleChainId, id -> ruleChain);
  61 + TbActorRef actorRef = getOrCreateActor(ruleChainId, id -> ruleChain);
63 62 visit(ruleChain, actorRef);
64 63 log.debug("[{}|{}] Rule Chain actor created.", ruleChainId.getEntityType(), ruleChainId.getId());
65 64 }
66 65 }
67 66
68   - protected void visit(RuleChain entity, ActorRef actorRef) {
  67 + protected void visit(RuleChain entity, TbActorRef actorRef) {
69 68 if (entity != null && entity.isRoot()) {
70 69 rootChain = entity;
71 70 rootChainActor = actorRef;
72 71 }
73 72 }
74 73
75   - public ActorRef getOrCreateActor(akka.actor.ActorContext context, RuleChainId ruleChainId) {
76   - return getOrCreateActor(context, ruleChainId, eId -> ruleChainService.findRuleChainById(TenantId.SYS_TENANT_ID, eId));
  74 + protected TbActorRef getOrCreateActor(RuleChainId ruleChainId) {
  75 + return getOrCreateActor(ruleChainId, eId -> ruleChainService.findRuleChainById(TenantId.SYS_TENANT_ID, eId));
77 76 }
78 77
79   - public ActorRef getOrCreateActor(akka.actor.ActorContext context, RuleChainId ruleChainId, Function<RuleChainId, RuleChain> provider) {
80   - return actors.computeIfAbsent(ruleChainId, eId -> {
81   - RuleChain ruleChain = provider.apply(eId);
82   - return context.actorOf(Props.create(new RuleChainActor.ActorCreator(systemContext, tenantId, ruleChain))
83   - .withDispatcher(DefaultActorService.TENANT_RULE_DISPATCHER_NAME), eId.toString());
84   - });
  78 + protected TbActorRef getOrCreateActor(RuleChainId ruleChainId, Function<RuleChainId, RuleChain> provider) {
  79 + return ctx.getOrCreateChildActor(new TbEntityActorId(ruleChainId),
  80 + () -> DefaultActorService.RULE_DISPATCHER_NAME,
  81 + () -> {
  82 + RuleChain ruleChain = provider.apply(ruleChainId);
  83 + return new RuleChainActor.ActorCreator(systemContext, tenantId, ruleChain);
  84 + });
85 85 }
86 86
87   - protected ActorRef getEntityActorRef(EntityId entityId) {
88   - ActorRef target = null;
  87 + protected TbActorRef getEntityActorRef(EntityId entityId) {
  88 + TbActorRef target = null;
89 89 if (entityId.getEntityType() == EntityType.RULE_CHAIN) {
90   - target = getOrCreateActor(this.context(), (RuleChainId) entityId);
  90 + target = getOrCreateActor((RuleChainId) entityId);
91 91 }
92 92 return target;
93 93 }
94 94
95   - protected void broadcast(Object msg) {
96   - actors.values().forEach(actorRef -> actorRef.tell(msg, ActorRef.noSender()));
  95 + protected void broadcast(TbActorMsg msg) {
  96 + ctx.broadcastToChildren(msg, new TbEntityTypeActorIdPredicate(EntityType.RULE_CHAIN));
97 97 }
98   -
99   - public ActorRef get(RuleChainId id) {
100   - return actors.get(id);
101   - }
102   -
103 98 }
... ...
... ... @@ -15,31 +15,43 @@
15 15 */
16 16 package org.thingsboard.server.actors.ruleChain;
17 17
  18 +import lombok.extern.slf4j.Slf4j;
18 19 import org.thingsboard.server.actors.ActorSystemContext;
  20 +import org.thingsboard.server.actors.TbActor;
  21 +import org.thingsboard.server.actors.TbActorCtx;
  22 +import org.thingsboard.server.actors.TbActorId;
  23 +import org.thingsboard.server.actors.TbEntityActorId;
19 24 import org.thingsboard.server.actors.service.ComponentActor;
20 25 import org.thingsboard.server.actors.service.ContextBasedCreator;
21 26 import org.thingsboard.server.common.data.id.RuleChainId;
22 27 import org.thingsboard.server.common.data.id.RuleNodeId;
23 28 import org.thingsboard.server.common.data.id.TenantId;
  29 +import org.thingsboard.server.common.data.rule.RuleChain;
24 30 import org.thingsboard.server.common.msg.TbActorMsg;
25 31 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
26 32 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
27 33
  34 +@Slf4j
28 35 public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessageProcessor> {
29 36
30 37 private final String ruleChainName;
31 38 private final RuleChainId ruleChainId;
  39 + private final RuleNodeId ruleNodeId;
32 40
33 41 private RuleNodeActor(ActorSystemContext systemContext, TenantId tenantId, RuleChainId ruleChainId, String ruleChainName, RuleNodeId ruleNodeId) {
34 42 super(systemContext, tenantId, ruleNodeId);
35 43 this.ruleChainName = ruleChainName;
36 44 this.ruleChainId = ruleChainId;
37   - setProcessor(new RuleNodeActorMessageProcessor(tenantId, this.ruleChainName, ruleNodeId, systemContext,
38   - context().parent(), context().self()));
  45 + this.ruleNodeId = ruleNodeId;
39 46 }
40 47
41 48 @Override
42   - protected boolean process(TbActorMsg msg) {
  49 + protected RuleNodeActorMessageProcessor createProcessor(TbActorCtx ctx) {
  50 + return new RuleNodeActorMessageProcessor(tenantId, this.ruleChainName, ruleNodeId, systemContext, ctx.getParentRef(), ctx);
  51 + }
  52 +
  53 + @Override
  54 + protected boolean doProcess(TbActorMsg msg) {
43 55 switch (msg.getMsgType()) {
44 56 case COMPONENT_LIFE_CYCLE_MSG:
45 57 onComponentLifecycleMsg((ComponentLifecycleMsg) msg);
... ... @@ -93,8 +105,7 @@ public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessa
93 105 logAndPersist("onRuleMsg", ActorSystemContext.toException(msg.getError()));
94 106 }
95 107
96   - public static class ActorCreator extends ContextBasedCreator<RuleNodeActor> {
97   - private static final long serialVersionUID = 1L;
  108 + public static class ActorCreator extends ContextBasedCreator {
98 109
99 110 private final TenantId tenantId;
100 111 private final RuleChainId ruleChainId;
... ... @@ -111,7 +122,12 @@ public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessa
111 122 }
112 123
113 124 @Override
114   - public RuleNodeActor create() throws Exception {
  125 + public TbActorId createActorId() {
  126 + return new TbEntityActorId(ruleNodeId);
  127 + }
  128 +
  129 + @Override
  130 + public TbActor createActor() {
115 131 return new RuleNodeActor(context, tenantId, ruleChainId, ruleChainName, ruleNodeId);
116 132 }
117 133 }
... ...
... ... @@ -15,11 +15,11 @@
15 15 */
16 16 package org.thingsboard.server.actors.ruleChain;
17 17
18   -import akka.actor.ActorContext;
19   -import akka.actor.ActorRef;
20 18 import org.thingsboard.rule.engine.api.TbNode;
21 19 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
22 20 import org.thingsboard.server.actors.ActorSystemContext;
  21 +import org.thingsboard.server.actors.TbActorCtx;
  22 +import org.thingsboard.server.actors.TbActorRef;
23 23 import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
24 24 import org.thingsboard.server.common.data.id.RuleNodeId;
25 25 import org.thingsboard.server.common.data.id.TenantId;
... ... @@ -34,13 +34,13 @@ import org.thingsboard.server.common.msg.queue.RuleNodeException;
34 34 public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNodeId> {
35 35
36 36 private final String ruleChainName;
37   - private final ActorRef self;
  37 + private final TbActorRef self;
38 38 private RuleNode ruleNode;
39 39 private TbNode tbNode;
40 40 private DefaultTbContext defaultCtx;
41 41
42 42 RuleNodeActorMessageProcessor(TenantId tenantId, String ruleChainName, RuleNodeId ruleNodeId, ActorSystemContext systemContext
43   - , ActorRef parent, ActorRef self) {
  43 + , TbActorRef parent, TbActorRef self) {
44 44 super(systemContext, tenantId, ruleNodeId);
45 45 this.ruleChainName = ruleChainName;
46 46 this.self = self;
... ... @@ -49,7 +49,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
49 49 }
50 50
51 51 @Override
52   - public void start(ActorContext context) throws Exception {
  52 + public void start(TbActorCtx context) throws Exception {
53 53 tbNode = initComponent(ruleNode);
54 54 if (tbNode != null) {
55 55 state = ComponentLifecycleState.ACTIVE;
... ... @@ -57,7 +57,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
57 57 }
58 58
59 59 @Override
60   - public void onUpdate(ActorContext context) throws Exception {
  60 + public void onUpdate(TbActorCtx context) throws Exception {
61 61 RuleNode newRuleNode = systemContext.getRuleChainService().findRuleNodeById(tenantId, entityId);
62 62 boolean restartRequired = state != ComponentLifecycleState.ACTIVE ||
63 63 !(ruleNode.getType().equals(newRuleNode.getType()) && ruleNode.getConfiguration().equals(newRuleNode.getConfiguration()));
... ... @@ -72,11 +72,11 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
72 72 }
73 73
74 74 @Override
75   - public void stop(ActorContext context) {
  75 + public void stop(TbActorCtx context) {
76 76 if (tbNode != null) {
77 77 tbNode.destroy();
  78 + state = ComponentLifecycleState.SUSPENDED;
78 79 }
79   - context.stop(self);
80 80 }
81 81
82 82 @Override
... ...
... ... @@ -15,9 +15,9 @@
15 15 */
16 16 package org.thingsboard.server.actors.ruleChain;
17 17
18   -import akka.actor.ActorRef;
19 18 import lombok.AllArgsConstructor;
20 19 import lombok.Data;
  20 +import org.thingsboard.server.actors.TbActorRef;
21 21 import org.thingsboard.server.common.data.id.TenantId;
22 22 import org.thingsboard.server.common.data.rule.RuleNode;
23 23
... ... @@ -28,7 +28,7 @@ import org.thingsboard.server.common.data.rule.RuleNode;
28 28 @AllArgsConstructor
29 29 final class RuleNodeCtx {
30 30 private final TenantId tenantId;
31   - private final ActorRef chainActor;
32   - private final ActorRef selfActor;
  31 + private final TbActorRef chainActor;
  32 + private final TbActorRef selfActor;
33 33 private RuleNode self;
34 34 }
... ...
... ... @@ -15,19 +15,21 @@
15 15 */
16 16 package org.thingsboard.server.actors.service;
17 17
18   -import akka.actor.ActorRef;
  18 +import lombok.extern.slf4j.Slf4j;
19 19 import org.thingsboard.server.actors.ActorSystemContext;
  20 +import org.thingsboard.server.actors.TbActorCtx;
20 21 import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
21 22 import org.thingsboard.server.actors.stats.StatsPersistMsg;
22 23 import org.thingsboard.server.common.data.id.EntityId;
23 24 import org.thingsboard.server.common.data.id.TenantId;
24 25 import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
25   -import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
26 26 import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
  27 +import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
27 28
28 29 /**
29 30 * @author Andrew Shvayka
30 31 */
  32 +@Slf4j
31 33 public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgProcessor<T>> extends ContextAwareActor {
32 34
33 35 private long lastPersistedErrorTs = 0L;
... ... @@ -43,15 +45,19 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
43 45 this.id = id;
44 46 }
45 47
46   - protected void setProcessor(P processor) {
47   - this.processor = processor;
48   - }
  48 + abstract protected P createProcessor(TbActorCtx ctx);
49 49
50 50 @Override
51   - public void preStart() {
  51 + public void init(TbActorCtx ctx) {
  52 + super.init(ctx);
  53 + this.processor = createProcessor(ctx);
  54 + initProcessor(ctx);
  55 + }
  56 +
  57 + protected void initProcessor(TbActorCtx ctx) {
52 58 try {
53 59 log.debug("[{}][{}][{}] Starting processor.", tenantId, id, id.getEntityType());
54   - processor.start(context());
  60 + processor.start(ctx);
55 61 logLifecycleEvent(ComponentLifecycleEvent.STARTED);
56 62 if (systemContext.isStatisticsEnabled()) {
57 63 scheduleStatsPersistTick();
... ... @@ -66,7 +72,7 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
66 72
67 73 private void scheduleStatsPersistTick() {
68 74 try {
69   - processor.scheduleStatsPersistTick(context(), systemContext.getStatisticsPersistFrequency());
  75 + processor.scheduleStatsPersistTick(ctx, systemContext.getStatisticsPersistFrequency());
70 76 } catch (Exception e) {
71 77 log.error("[{}][{}] Failed to schedule statistics store message. No statistics is going to be stored: {}", tenantId, id, e.getMessage());
72 78 logAndPersist("onScheduleStatsPersistMsg", e);
... ... @@ -74,10 +80,10 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
74 80 }
75 81
76 82 @Override
77   - public void postStop() {
  83 + public void destroy() {
78 84 try {
79   - log.debug("[{}][{}] Stopping processor.", tenantId, id, id.getEntityType());
80   - processor.stop(context());
  85 + log.debug("[{}][{}][{}] Stopping processor.", tenantId, id, id.getEntityType());
  86 + processor.stop(ctx);
81 87 logLifecycleEvent(ComponentLifecycleEvent.STOPPED);
82 88 } catch (Exception e) {
83 89 log.warn("[{}][{}] Failed to stop {} processor: {}", tenantId, id, id.getEntityType(), e.getMessage());
... ... @@ -91,19 +97,20 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
91 97 try {
92 98 switch (msg.getEvent()) {
93 99 case CREATED:
94   - processor.onCreated(context());
  100 + processor.onCreated(ctx);
95 101 break;
96 102 case UPDATED:
97   - processor.onUpdate(context());
  103 + processor.onUpdate(ctx);
98 104 break;
99 105 case ACTIVATED:
100   - processor.onActivate(context());
  106 + processor.onActivate(ctx);
101 107 break;
102 108 case SUSPENDED:
103   - processor.onSuspend(context());
  109 + processor.onSuspend(ctx);
104 110 break;
105 111 case DELETED:
106   - processor.onStop(context());
  112 + processor.onStop(ctx);
  113 + ctx.stop(ctx.getSelf());
107 114 break;
108 115 default:
109 116 break;
... ... @@ -125,7 +132,7 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
125 132
126 133 protected void onStatsPersistTick(EntityId entityId) {
127 134 try {
128   - systemContext.getStatsActor().tell(new StatsPersistMsg(messagesProcessed, errorsOccurred, tenantId, entityId), ActorRef.noSender());
  135 + systemContext.getStatsActor().tell(new StatsPersistMsg(messagesProcessed, errorsOccurred, tenantId, entityId));
129 136 resetStatsCounters();
130 137 } catch (Exception e) {
131 138 logAndPersist("onStatsPersistTick", e);
... ... @@ -149,10 +156,10 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
149 156 errorsOccurred++;
150 157 String componentName = processor != null ? processor.getComponentName() : "Unknown";
151 158 if (critical) {
152   - log.warn("[{}][{}][{}] Failed to process {} msg: {}", id, tenantId, componentName, method);
  159 + log.warn("[{}][{}][{}] Failed to process method: {}", id, tenantId, componentName, method);
153 160 log.warn("Critical Error: ", e);
154 161 } else {
155   - log.debug("[{}][{}][{}] Failed to process {} msg: {}", id, tenantId, componentName, method);
  162 + log.debug("[{}][{}][{}] Failed to process method: {}", id, tenantId, componentName, method);
156 163 log.debug("Debug Error: ", e);
157 164 }
158 165 long ts = System.currentTimeMillis();
... ...
... ... @@ -15,17 +15,18 @@
15 15 */
16 16 package org.thingsboard.server.actors.service;
17 17
18   -import akka.actor.Terminated;
19   -import akka.actor.UntypedAbstractActor;
  18 +import lombok.extern.slf4j.Slf4j;
20 19 import org.slf4j.Logger;
21 20 import org.slf4j.LoggerFactory;
  21 +import org.thingsboard.server.actors.AbstractTbActor;
22 22 import org.thingsboard.server.actors.ActorSystemContext;
  23 +import org.thingsboard.server.actors.ProcessFailureStrategy;
  24 +import org.thingsboard.server.actors.TbActor;
  25 +import org.thingsboard.server.actors.TbActorCtx;
23 26 import org.thingsboard.server.common.msg.TbActorMsg;
24 27
25   -
26   -public abstract class ContextAwareActor extends UntypedAbstractActor {
27   -
28   - protected final Logger log = LoggerFactory.getLogger(getClass());
  28 +@Slf4j
  29 +public abstract class ContextAwareActor extends AbstractTbActor {
29 30
30 31 public static final int ENTITY_PACK_LIMIT = 1024;
31 32
... ... @@ -37,27 +38,29 @@ public abstract class ContextAwareActor extends UntypedAbstractActor {
37 38 }
38 39
39 40 @Override
40   - public void onReceive(Object msg) {
  41 + public boolean process(TbActorMsg msg) {
41 42 if (log.isDebugEnabled()) {
42 43 log.debug("Processing msg: {}", msg);
43 44 }
44   - if (msg instanceof TbActorMsg) {
45   - try {
46   - if (!process((TbActorMsg) msg)) {
47   - log.warn("Unknown message: {}!", msg);
48   - }
49   - } catch (Exception e) {
50   - throw e;
51   - }
52   - } else if (msg instanceof Terminated) {
53   - processTermination((Terminated) msg);
54   - } else {
55   - log.warn("Unknown message: {}!", msg);
  45 + if (!doProcess(msg)) {
  46 + log.warn("Unprocessed message: {}!", msg);
56 47 }
  48 + return false;
57 49 }
58 50
59   - protected void processTermination(Terminated msg) {
  51 + protected abstract boolean doProcess(TbActorMsg msg);
  52 +
  53 + @Override
  54 + public ProcessFailureStrategy onProcessFailure(Throwable t) {
  55 + log.debug("[{}] Processing failure: ", getActorRef().getActorId(), t);
  56 + return doProcessFailure(t);
60 57 }
61 58
62   - protected abstract boolean process(TbActorMsg msg);
  59 + protected ProcessFailureStrategy doProcessFailure(Throwable t) {
  60 + if (t instanceof Error) {
  61 + return ProcessFailureStrategy.stop();
  62 + } else {
  63 + return ProcessFailureStrategy.resume();
  64 + }
  65 + }
63 66 }
... ...
... ... @@ -15,12 +15,10 @@
15 15 */
16 16 package org.thingsboard.server.actors.service;
17 17
18   -import akka.japi.Creator;
19 18 import org.thingsboard.server.actors.ActorSystemContext;
  19 +import org.thingsboard.server.actors.TbActorCreator;
20 20
21   -public abstract class ContextBasedCreator<T> implements Creator<T> {
22   -
23   - private static final long serialVersionUID = 1L;
  21 +public abstract class ContextBasedCreator implements TbActorCreator {
24 22
25 23 protected final transient ActorSystemContext context;
26 24
... ...
... ... @@ -15,85 +15,116 @@
15 15 */
16 16 package org.thingsboard.server.actors.service;
17 17
18   -import akka.actor.ActorRef;
19   -import akka.actor.ActorSystem;
20   -import akka.actor.Props;
21   -import akka.actor.Terminated;
22 18 import lombok.extern.slf4j.Slf4j;
23 19 import org.springframework.beans.factory.annotation.Autowired;
24 20 import org.springframework.beans.factory.annotation.Value;
25 21 import org.springframework.boot.context.event.ApplicationReadyEvent;
26 22 import org.springframework.context.event.EventListener;
27   -import org.springframework.scheduling.annotation.Scheduled;
28 23 import org.springframework.stereotype.Service;
29 24 import org.thingsboard.server.actors.ActorSystemContext;
  25 +import org.thingsboard.server.actors.DefaultTbActorSystem;
  26 +import org.thingsboard.server.actors.TbActorId;
  27 +import org.thingsboard.server.actors.TbActorRef;
  28 +import org.thingsboard.server.actors.TbActorSystem;
  29 +import org.thingsboard.server.actors.TbActorSystemSettings;
30 30 import org.thingsboard.server.actors.app.AppActor;
31 31 import org.thingsboard.server.actors.app.AppInitMsg;
32 32 import org.thingsboard.server.actors.stats.StatsActor;
33 33 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
34 34 import org.thingsboard.server.queue.discovery.PartitionChangeEvent;
35   -import scala.concurrent.Await;
36   -import scala.concurrent.Future;
37   -import scala.concurrent.duration.Duration;
38 35
39 36 import javax.annotation.PostConstruct;
40 37 import javax.annotation.PreDestroy;
41   -import java.util.concurrent.atomic.AtomicInteger;
  38 +import java.util.concurrent.ExecutorService;
  39 +import java.util.concurrent.Executors;
  40 +import java.util.concurrent.ScheduledExecutorService;
42 41
43 42 @Service
44 43 @Slf4j
45 44 public class DefaultActorService implements ActorService {
46 45
47   - private static final String ACTOR_SYSTEM_NAME = "Akka";
48   -
49 46 public static final String APP_DISPATCHER_NAME = "app-dispatcher";
50   - public static final String CORE_DISPATCHER_NAME = "core-dispatcher";
51   - public static final String SYSTEM_RULE_DISPATCHER_NAME = "system-rule-dispatcher";
52   - public static final String TENANT_RULE_DISPATCHER_NAME = "rule-dispatcher";
  47 + public static final String TENANT_DISPATCHER_NAME = "tenant-dispatcher";
  48 + public static final String DEVICE_DISPATCHER_NAME = "device-dispatcher";
  49 + public static final String RULE_DISPATCHER_NAME = "rule-dispatcher";
53 50
54 51 @Autowired
55 52 private ActorSystemContext actorContext;
56 53
57   - private ActorSystem system;
  54 + private TbActorSystem system;
  55 +
  56 + private TbActorRef appActor;
  57 +
  58 + @Value("${actors.system.throughput:5}")
  59 + private int actorThroughput;
  60 +
  61 + @Value("${actors.system.max_actor_init_attempts:10}")
  62 + private int maxActorInitAttempts;
  63 +
  64 + @Value("${actors.system.scheduler_pool_size:1}")
  65 + private int schedulerPoolSize;
  66 +
  67 + @Value("${actors.system.app_dispatcher_pool_size:1}")
  68 + private int appDispatcherSize;
  69 +
  70 + @Value("${actors.system.tenant_dispatcher_pool_size:2}")
  71 + private int tenantDispatcherSize;
58 72
59   - private ActorRef appActor;
  73 + @Value("${actors.system.device_dispatcher_pool_size:4}")
  74 + private int deviceDispatcherSize;
  75 +
  76 + @Value("${actors.system.rule_dispatcher_pool_size:4}")
  77 + private int ruleDispatcherSize;
60 78
61 79 @PostConstruct
62 80 public void initActorSystem() {
63   - log.info("Initializing Actor system.");
  81 + log.info("Initializing actor system.");
64 82 actorContext.setActorService(this);
65   - system = ActorSystem.create(ACTOR_SYSTEM_NAME, actorContext.getConfig());
  83 + TbActorSystemSettings settings = new TbActorSystemSettings(actorThroughput, schedulerPoolSize, maxActorInitAttempts);
  84 + system = new DefaultTbActorSystem(settings);
  85 +
  86 + system.createDispatcher(APP_DISPATCHER_NAME, initDispatcherExecutor(appDispatcherSize));
  87 + system.createDispatcher(TENANT_DISPATCHER_NAME, initDispatcherExecutor(tenantDispatcherSize));
  88 + system.createDispatcher(DEVICE_DISPATCHER_NAME, initDispatcherExecutor(deviceDispatcherSize));
  89 + system.createDispatcher(RULE_DISPATCHER_NAME, initDispatcherExecutor(ruleDispatcherSize));
  90 +
66 91 actorContext.setActorSystem(system);
67 92
68   - appActor = system.actorOf(Props.create(new AppActor.ActorCreator(actorContext)).withDispatcher(APP_DISPATCHER_NAME), "appActor");
  93 + appActor = system.createRootActor(APP_DISPATCHER_NAME, new AppActor.ActorCreator(actorContext));
69 94 actorContext.setAppActor(appActor);
70 95
71   - ActorRef statsActor = system.actorOf(Props.create(new StatsActor.ActorCreator(actorContext)).withDispatcher(CORE_DISPATCHER_NAME), "statsActor");
  96 + TbActorRef statsActor = system.createRootActor(TENANT_DISPATCHER_NAME, new StatsActor.ActorCreator(actorContext, "StatsActor"));
72 97 actorContext.setStatsActor(statsActor);
73 98
74 99 log.info("Actor system initialized.");
75 100 }
76 101
  102 + private ExecutorService initDispatcherExecutor(int poolSize) {
  103 + if (poolSize == 0) {
  104 + int cores = Runtime.getRuntime().availableProcessors();
  105 + poolSize = Math.max(1, cores / 2);
  106 + }
  107 + return Executors.newWorkStealingPool(poolSize);
  108 + }
  109 +
77 110 @EventListener(ApplicationReadyEvent.class)
78 111 public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
79 112 log.info("Received application ready event. Sending application init message to actor system");
80   - appActor.tell(new AppInitMsg(), ActorRef.noSender());
  113 + appActor.tell(new AppInitMsg());
81 114 }
82 115
83 116 @EventListener(PartitionChangeEvent.class)
84 117 public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
85 118 log.info("Received partition change event.");
86   - this.appActor.tell(new PartitionChangeMsg(partitionChangeEvent.getServiceQueueKey(), partitionChangeEvent.getPartitions()), ActorRef.noSender());
  119 + this.appActor.tell(new PartitionChangeMsg(partitionChangeEvent.getServiceQueueKey(), partitionChangeEvent.getPartitions()));
87 120 }
88 121
89 122 @PreDestroy
90 123 public void stopActorSystem() {
91   - Future<Terminated> status = system.terminate();
92   - try {
93   - Terminated terminated = Await.result(status, Duration.Inf());
94   - log.info("Actor system terminated: {}", terminated);
95   - } catch (Exception e) {
96   - log.error("Failed to terminate actor system.", e);
  124 + if (system != null) {
  125 + log.info("Stopping actor system.");
  126 + system.stop();
  127 + log.info("Actor system stopped.");
97 128 }
98 129 }
99 130
... ...
... ... @@ -15,18 +15,13 @@
15 15 */
16 16 package org.thingsboard.server.actors.shared;
17 17
18   -import akka.actor.ActorContext;
19   -import akka.actor.ActorRef;
20   -import akka.actor.Scheduler;
21   -import akka.event.LoggingAdapter;
22 18 import com.fasterxml.jackson.databind.ObjectMapper;
23   -import lombok.AllArgsConstructor;
24   -import lombok.Data;
25 19 import lombok.extern.slf4j.Slf4j;
26 20 import org.thingsboard.server.actors.ActorSystemContext;
27   -import scala.concurrent.ExecutionContextExecutor;
28   -import scala.concurrent.duration.Duration;
  21 +import org.thingsboard.server.actors.TbActorCtx;
  22 +import org.thingsboard.server.common.msg.TbActorMsg;
29 23
  24 +import java.util.concurrent.ScheduledExecutorService;
30 25 import java.util.concurrent.TimeUnit;
31 26
32 27 @Slf4j
... ... @@ -40,31 +35,16 @@ public abstract class AbstractContextAwareMsgProcessor {
40 35 this.systemContext = systemContext;
41 36 }
42 37
43   - private Scheduler getScheduler() {
  38 + private ScheduledExecutorService getScheduler() {
44 39 return systemContext.getScheduler();
45 40 }
46 41
47   - private ExecutionContextExecutor getSystemDispatcher() {
48   - return systemContext.getActorSystem().dispatcher();
  42 + protected void schedulePeriodicMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs, long periodInMs) {
  43 + systemContext.schedulePeriodicMsgWithDelay(ctx, msg, delayInMs, periodInMs);
49 44 }
50 45
51   - protected void schedulePeriodicMsgWithDelay(ActorContext ctx, Object msg, long delayInMs, long periodInMs) {
52   - schedulePeriodicMsgWithDelay(msg, delayInMs, periodInMs, ctx.self());
  46 + protected void scheduleMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs) {
  47 + systemContext.scheduleMsgWithDelay(ctx, msg, delayInMs);
53 48 }
54 49
55   - private void schedulePeriodicMsgWithDelay(Object msg, long delayInMs, long periodInMs, ActorRef target) {
56   - log.debug("Scheduling periodic msg {} every {} ms with delay {} ms", msg, periodInMs, delayInMs);
57   - getScheduler().schedule(Duration.create(delayInMs, TimeUnit.MILLISECONDS), Duration.create(periodInMs, TimeUnit.MILLISECONDS), target, msg, getSystemDispatcher(), null);
58   - }
59   -
60   - protected void scheduleMsgWithDelay(ActorContext ctx, Object msg, long delayInMs) {
61   - scheduleMsgWithDelay(msg, delayInMs, ctx.self());
62   - }
63   -
64   - private void scheduleMsgWithDelay(Object msg, long delayInMs, ActorRef target) {
65   - log.debug("Scheduling msg {} with delay {} ms", msg, delayInMs);
66   - getScheduler().scheduleOnce(Duration.create(delayInMs, TimeUnit.MILLISECONDS), target, msg, getSystemDispatcher(), null);
67   - }
68   -
69   -
70 50 }
... ...
... ... @@ -15,16 +15,15 @@
15 15 */
16 16 package org.thingsboard.server.actors.shared;
17 17
18   -import akka.actor.ActorContext;
19 18 import lombok.extern.slf4j.Slf4j;
20 19 import org.thingsboard.server.actors.ActorSystemContext;
  20 +import org.thingsboard.server.actors.TbActorCtx;
21 21 import org.thingsboard.server.actors.stats.StatsPersistTick;
22 22 import org.thingsboard.server.common.data.id.EntityId;
23 23 import org.thingsboard.server.common.data.id.TenantId;
24 24 import org.thingsboard.server.common.data.plugin.ComponentLifecycleState;
25 25 import org.thingsboard.server.common.msg.TbMsg;
26 26 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
27   -import org.thingsboard.server.common.msg.queue.RuleEngineException;
28 27 import org.thingsboard.server.common.msg.queue.RuleNodeException;
29 28
30 29 @Slf4j
... ... @@ -42,38 +41,38 @@ public abstract class ComponentMsgProcessor<T extends EntityId> extends Abstract
42 41
43 42 public abstract String getComponentName();
44 43
45   - public abstract void start(ActorContext context) throws Exception;
  44 + public abstract void start(TbActorCtx context) throws Exception;
46 45
47   - public abstract void stop(ActorContext context) throws Exception;
  46 + public abstract void stop(TbActorCtx context) throws Exception;
48 47
49 48 public abstract void onPartitionChangeMsg(PartitionChangeMsg msg) throws Exception;
50 49
51   - public void onCreated(ActorContext context) throws Exception {
  50 + public void onCreated(TbActorCtx context) throws Exception {
52 51 start(context);
53 52 }
54 53
55   - public void onUpdate(ActorContext context) throws Exception {
  54 + public void onUpdate(TbActorCtx context) throws Exception {
56 55 restart(context);
57 56 }
58 57
59   - public void onActivate(ActorContext context) throws Exception {
  58 + public void onActivate(TbActorCtx context) throws Exception {
60 59 restart(context);
61 60 }
62 61
63   - public void onSuspend(ActorContext context) throws Exception {
  62 + public void onSuspend(TbActorCtx context) throws Exception {
64 63 stop(context);
65 64 }
66 65
67   - public void onStop(ActorContext context) throws Exception {
  66 + public void onStop(TbActorCtx context) throws Exception {
68 67 stop(context);
69 68 }
70 69
71   - private void restart(ActorContext context) throws Exception {
  70 + private void restart(TbActorCtx context) throws Exception {
72 71 stop(context);
73 72 start(context);
74 73 }
75 74
76   - public void scheduleStatsPersistTick(ActorContext context, long statsPersistFrequency) {
  75 + public void scheduleStatsPersistTick(TbActorCtx context, long statsPersistFrequency) {
77 76 schedulePeriodicMsgWithDelay(context, new StatsPersistTick(), statsPersistFrequency, statsPersistFrequency);
78 77 }
79 78
... ...
... ... @@ -19,10 +19,15 @@ import com.fasterxml.jackson.databind.JsonNode;
19 19 import com.fasterxml.jackson.databind.ObjectMapper;
20 20 import lombok.extern.slf4j.Slf4j;
21 21 import org.thingsboard.server.actors.ActorSystemContext;
  22 +import org.thingsboard.server.actors.TbActor;
  23 +import org.thingsboard.server.actors.TbActorCtx;
  24 +import org.thingsboard.server.actors.TbActorId;
  25 +import org.thingsboard.server.actors.TbStringActorId;
22 26 import org.thingsboard.server.actors.service.ContextAwareActor;
23 27 import org.thingsboard.server.actors.service.ContextBasedCreator;
24 28 import org.thingsboard.server.common.data.DataConstants;
25 29 import org.thingsboard.server.common.data.Event;
  30 +import org.thingsboard.server.common.msg.MsgType;
26 31 import org.thingsboard.server.common.msg.TbActorMsg;
27 32
28 33 @Slf4j
... ... @@ -35,24 +40,17 @@ public class StatsActor extends ContextAwareActor {
35 40 }
36 41
37 42 @Override
38   - protected boolean process(TbActorMsg msg) {
39   - //TODO Move everything here, to work with TbActorMsg\
40   - return false;
41   - }
42   -
43   - @Override
44   - public void onReceive(Object msg) {
  43 + protected boolean doProcess(TbActorMsg msg) {
45 44 log.debug("Received message: {}", msg);
46   - if (msg instanceof StatsPersistMsg) {
47   - try {
48   - onStatsPersistMsg((StatsPersistMsg) msg);
49   - } catch (Exception e) {
50   - log.warn("Failed to persist statistics: {}", msg, e);
51   - }
  45 + if (msg.getMsgType().equals(MsgType.STATS_PERSIST_MSG)) {
  46 + onStatsPersistMsg((StatsPersistMsg) msg);
  47 + return true;
  48 + } else {
  49 + return false;
52 50 }
53 51 }
54 52
55   - public void onStatsPersistMsg(StatsPersistMsg msg) throws Exception {
  53 + public void onStatsPersistMsg(StatsPersistMsg msg) {
56 54 Event event = new Event();
57 55 event.setEntityId(msg.getEntityId());
58 56 event.setTenantId(msg.getTenantId());
... ... @@ -65,15 +63,21 @@ public class StatsActor extends ContextAwareActor {
65 63 return mapper.createObjectNode().put("server", serviceId).put("messagesProcessed", messagesProcessed).put("errorsOccurred", errorsOccurred);
66 64 }
67 65
68   - public static class ActorCreator extends ContextBasedCreator<StatsActor> {
69   - private static final long serialVersionUID = 1L;
  66 + public static class ActorCreator extends ContextBasedCreator {
  67 + private final String actorId;
70 68
71   - public ActorCreator(ActorSystemContext context) {
  69 + public ActorCreator(ActorSystemContext context, String actorId) {
72 70 super(context);
  71 + this.actorId = actorId;
  72 + }
  73 +
  74 + @Override
  75 + public TbActorId createActorId() {
  76 + return new TbStringActorId(actorId);
73 77 }
74 78
75 79 @Override
76   - public StatsActor create() {
  80 + public TbActor createActor() {
77 81 return new StatsActor(context);
78 82 }
79 83 }
... ...
... ... @@ -20,13 +20,21 @@ import lombok.Getter;
20 20 import lombok.ToString;
21 21 import org.thingsboard.server.common.data.id.EntityId;
22 22 import org.thingsboard.server.common.data.id.TenantId;
  23 +import org.thingsboard.server.common.msg.MsgType;
  24 +import org.thingsboard.server.common.msg.TbActorMsg;
23 25
24 26 @AllArgsConstructor
25 27 @Getter
26 28 @ToString
27   -public final class StatsPersistMsg {
  29 +public final class StatsPersistMsg implements TbActorMsg {
  30 +
28 31 private long messagesProcessed;
29 32 private long errorsOccurred;
30 33 private TenantId tenantId;
31 34 private EntityId entityId;
  35 +
  36 + @Override
  37 + public MsgType getMsgType() {
  38 + return MsgType.STATS_PERSIST_MSG;
  39 + }
32 40 }
... ...
... ... @@ -15,16 +15,15 @@
15 15 */
16 16 package org.thingsboard.server.actors.tenant;
17 17
18   -import akka.actor.ActorInitializationException;
19   -import akka.actor.ActorRef;
20   -import akka.actor.LocalActorRef;
21   -import akka.actor.OneForOneStrategy;
22   -import akka.actor.Props;
23   -import akka.actor.SupervisorStrategy;
24   -import akka.actor.Terminated;
25   -import com.google.common.collect.BiMap;
26   -import com.google.common.collect.HashBiMap;
  18 +import lombok.extern.slf4j.Slf4j;
27 19 import org.thingsboard.server.actors.ActorSystemContext;
  20 +import org.thingsboard.server.actors.TbActor;
  21 +import org.thingsboard.server.actors.TbActorCtx;
  22 +import org.thingsboard.server.actors.TbActorId;
  23 +import org.thingsboard.server.actors.TbActorNotRegisteredException;
  24 +import org.thingsboard.server.actors.TbActorRef;
  25 +import org.thingsboard.server.actors.TbEntityActorId;
  26 +import org.thingsboard.server.actors.TbEntityTypeActorIdPredicate;
28 27 import org.thingsboard.server.actors.device.DeviceActorCreator;
29 28 import org.thingsboard.server.actors.ruleChain.RuleChainManagerActor;
30 29 import org.thingsboard.server.actors.service.ContextBasedCreator;
... ... @@ -32,6 +31,7 @@ import org.thingsboard.server.actors.service.DefaultActorService;
32 31 import org.thingsboard.server.common.data.EntityType;
33 32 import org.thingsboard.server.common.data.Tenant;
34 33 import org.thingsboard.server.common.data.id.DeviceId;
  34 +import org.thingsboard.server.common.data.id.EntityId;
35 35 import org.thingsboard.server.common.data.id.RuleChainId;
36 36 import org.thingsboard.server.common.data.id.TenantId;
37 37 import org.thingsboard.server.common.data.rule.RuleChain;
... ... @@ -45,32 +45,25 @@ import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
45 45 import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
46 46 import org.thingsboard.server.common.msg.queue.RuleEngineException;
47 47 import org.thingsboard.server.common.msg.queue.ServiceType;
48   -import scala.concurrent.duration.Duration;
49 48
50 49 import java.util.List;
51 50 import java.util.Optional;
52   -import java.util.stream.Collectors;
53 51
  52 +@Slf4j
54 53 public class TenantActor extends RuleChainManagerActor {
55 54
56   - private final BiMap<DeviceId, ActorRef> deviceActors;
57 55 private boolean isRuleEngineForCurrentTenant;
58 56 private boolean isCore;
59 57
60 58 private TenantActor(ActorSystemContext systemContext, TenantId tenantId) {
61 59 super(systemContext, tenantId);
62   - this.deviceActors = HashBiMap.create();
63   - }
64   -
65   - @Override
66   - public SupervisorStrategy supervisorStrategy() {
67   - return strategy;
68 60 }
69 61
70 62 boolean cantFindTenant = false;
71 63
72 64 @Override
73   - public void preStart() {
  65 + public void init(TbActorCtx ctx) {
  66 + super.init(ctx);
74 67 log.info("[{}] Starting tenant actor.", tenantId);
75 68 try {
76 69 Tenant tenant = systemContext.getTenantService().findTenantById(tenantId);
... ... @@ -104,12 +97,12 @@ public class TenantActor extends RuleChainManagerActor {
104 97 }
105 98
106 99 @Override
107   - public void postStop() {
  100 + public void destroy() {
108 101 log.info("[{}] Stopping tenant actor.", tenantId);
109 102 }
110 103
111 104 @Override
112   - protected boolean process(TbActorMsg msg) {
  105 + protected boolean doProcess(TbActorMsg msg) {
113 106 if (cantFindTenant) {
114 107 log.info("[{}] Processing missing Tenant msg: {}", tenantId, msg);
115 108 if (msg.getMsgType().equals(MsgType.QUEUE_TO_RULE_ENGINE_MSG)) {
... ... @@ -126,13 +119,13 @@ public class TenantActor extends RuleChainManagerActor {
126 119 //To Rule Chain Actors
127 120 broadcast(msg);
128 121 } else if (ServiceType.TB_CORE.equals(serviceType)) {
129   - //To Device Actors
130   - List<DeviceId> repartitionedDevices =
131   - deviceActors.keySet().stream().filter(deviceId -> !isMyPartition(deviceId)).collect(Collectors.toList());
132   - for (DeviceId deviceId : repartitionedDevices) {
133   - ActorRef deviceActor = deviceActors.remove(deviceId);
134   - context().stop(deviceActor);
135   - }
  122 + List<TbActorId> deviceActorIds = ctx.filterChildren(new TbEntityTypeActorIdPredicate(EntityType.DEVICE) {
  123 + @Override
  124 + protected boolean testEntityId(EntityId entityId) {
  125 + return super.testEntityId(entityId) && !isMyPartition(entityId);
  126 + }
  127 + });
  128 + deviceActorIds.forEach(id -> ctx.stop(id));
136 129 }
137 130 break;
138 131 case COMPONENT_LIFE_CYCLE_MSG:
... ... @@ -158,8 +151,8 @@ public class TenantActor extends RuleChainManagerActor {
158 151 return true;
159 152 }
160 153
161   - private boolean isMyPartition(DeviceId deviceId) {
162   - return systemContext.resolve(ServiceType.TB_CORE, tenantId, deviceId).isMyPartition();
  154 + private boolean isMyPartition(EntityId entityId) {
  155 + return systemContext.resolve(ServiceType.TB_CORE, tenantId, entityId).isMyPartition();
163 156 }
164 157
165 158 private void onQueueToRuleEngineMsg(QueueToRuleEngineMsg msg) {
... ... @@ -170,16 +163,15 @@ public class TenantActor extends RuleChainManagerActor {
170 163 TbMsg tbMsg = msg.getTbMsg();
171 164 if (tbMsg.getRuleChainId() == null) {
172 165 if (getRootChainActor() != null) {
173   - getRootChainActor().tell(msg, self());
  166 + getRootChainActor().tell(msg);
174 167 } else {
175 168 tbMsg.getCallback().onFailure(new RuleEngineException("No Root Rule Chain available!"));
176 169 log.info("[{}] No Root Chain: {}", tenantId, msg);
177 170 }
178 171 } else {
179   - ActorRef ruleChainActor = get(tbMsg.getRuleChainId());
180   - if (ruleChainActor != null) {
181   - ruleChainActor.tell(msg, self());
182   - } else {
  172 + try {
  173 + ctx.tell(new TbEntityActorId(tbMsg.getRuleChainId()), msg);
  174 + } catch (TbActorNotRegisteredException ex) {
183 175 log.trace("Received message for non-existing rule chain: [{}]", tbMsg.getRuleChainId());
184 176 //TODO: 3.1 Log it to dead letters queue;
185 177 tbMsg.getCallback().onSuccess();
... ... @@ -188,61 +180,39 @@ public class TenantActor extends RuleChainManagerActor {
188 180 }
189 181
190 182 private void onRuleChainMsg(RuleChainAwareMsg msg) {
191   - getOrCreateActor(context(), msg.getRuleChainId()).tell(msg, self());
  183 + getOrCreateActor(msg.getRuleChainId()).tell(msg);
192 184 }
193 185
194 186 private void onToDeviceActorMsg(DeviceAwareMsg msg) {
195 187 if (!isCore) {
196 188 log.warn("RECEIVED INVALID MESSAGE: {}", msg);
197 189 }
198   - getOrCreateDeviceActor(msg.getDeviceId()).tell(msg, ActorRef.noSender());
  190 + getOrCreateDeviceActor(msg.getDeviceId()).tell(msg);
199 191 }
200 192
201 193 private void onComponentLifecycleMsg(ComponentLifecycleMsg msg) {
202 194 if (isRuleEngineForCurrentTenant) {
203   - ActorRef target = getEntityActorRef(msg.getEntityId());
  195 + TbActorRef target = getEntityActorRef(msg.getEntityId());
204 196 if (target != null) {
205 197 if (msg.getEntityId().getEntityType() == EntityType.RULE_CHAIN) {
206 198 RuleChain ruleChain = systemContext.getRuleChainService().
207 199 findRuleChainById(tenantId, new RuleChainId(msg.getEntityId().getId()));
208 200 visit(ruleChain, target);
209 201 }
210   - target.tell(msg, ActorRef.noSender());
  202 + target.tell(msg);
211 203 } else {
212 204 log.debug("[{}] Invalid component lifecycle msg: {}", tenantId, msg);
213 205 }
214 206 }
215 207 }
216 208
217   - private ActorRef getOrCreateDeviceActor(DeviceId deviceId) {
218   - return deviceActors.computeIfAbsent(deviceId, k -> {
219   - log.debug("[{}][{}] Creating device actor.", tenantId, deviceId);
220   - ActorRef deviceActor = context().actorOf(Props.create(new DeviceActorCreator(systemContext, tenantId, deviceId))
221   - .withDispatcher(DefaultActorService.CORE_DISPATCHER_NAME)
222   - , deviceId.toString());
223   - context().watch(deviceActor);
224   - log.debug("[{}][{}] Created device actor: {}.", tenantId, deviceId, deviceActor);
225   - return deviceActor;
226   - });
  209 + private TbActorRef getOrCreateDeviceActor(DeviceId deviceId) {
  210 + return ctx.getOrCreateChildActor(new TbEntityActorId(deviceId),
  211 + () -> DefaultActorService.DEVICE_DISPATCHER_NAME,
  212 + () -> new DeviceActorCreator(systemContext, tenantId, deviceId));
227 213 }
228 214
229   - @Override
230   - protected void processTermination(Terminated message) {
231   - ActorRef terminated = message.actor();
232   - if (terminated instanceof LocalActorRef) {
233   - boolean removed = deviceActors.inverse().remove(terminated) != null;
234   - if (removed) {
235   - log.debug("[{}] Removed actor:", terminated);
236   - } else {
237   - log.debug("Removed actor was not found in the device map!");
238   - }
239   - } else {
240   - throw new IllegalStateException("Remote actors are not supported!");
241   - }
242   - }
243   -
244   - public static class ActorCreator extends ContextBasedCreator<TenantActor> {
245   - private static final long serialVersionUID = 1L;
  215 + public static class ActorCreator extends ContextBasedCreator {
246 216
247 217 private final TenantId tenantId;
248 218
... ... @@ -252,18 +222,14 @@ public class TenantActor extends RuleChainManagerActor {
252 222 }
253 223
254 224 @Override
255   - public TenantActor create() {
256   - return new TenantActor(context, tenantId);
  225 + public TbActorId createActorId() {
  226 + return new TbEntityActorId(tenantId);
257 227 }
258   - }
259 228
260   - private final SupervisorStrategy strategy = new OneForOneStrategy(3, Duration.create("1 minute"), t -> {
261   - log.warn("[{}] Unknown failure", tenantId, t);
262   - if (t instanceof ActorInitializationException) {
263   - return SupervisorStrategy.stop();
264   - } else {
265   - return SupervisorStrategy.resume();
  229 + @Override
  230 + public TbActor createActor() {
  231 + return new TenantActor(context, tenantId);
266 232 }
267   - });
  233 + }
268 234
269 235 }
... ...
... ... @@ -15,7 +15,6 @@
15 15 */
16 16 package org.thingsboard.server.service.queue;
17 17
18   -import akka.actor.ActorRef;
19 18 import lombok.extern.slf4j.Slf4j;
20 19 import org.springframework.beans.factory.annotation.Value;
21 20 import org.springframework.scheduling.annotation.Scheduled;
... ... @@ -150,7 +149,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
150 149 tbCoreDeviceRpcService.forwardRpcRequestToDeviceActor((ToDeviceRpcRequestActorMsg) tbActorMsg);
151 150 } else {
152 151 log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get());
153   - actorContext.tell(actorMsg.get(), ActorRef.noSender());
  152 + actorContext.tell(actorMsg.get());
154 153 }
155 154 }
156 155 callback.onSuccess();
... ... @@ -208,7 +207,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
208 207 Optional<TbActorMsg> actorMsg = encodingService.decode(toCoreNotification.getComponentLifecycleMsg().toByteArray());
209 208 if (actorMsg.isPresent()) {
210 209 log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get());
211   - actorContext.tell(actorMsg.get(), ActorRef.noSender());
  210 + actorContext.tell(actorMsg.get());
212 211 }
213 212 callback.onSuccess();
214 213 }
... ... @@ -279,7 +278,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
279 278 if (statsEnabled) {
280 279 stats.log(toDeviceActorMsg);
281 280 }
282   - actorContext.tell(new TransportToDeviceActorMsgWrapper(toDeviceActorMsg, callback), ActorRef.noSender());
  281 + actorContext.tell(new TransportToDeviceActorMsgWrapper(toDeviceActorMsg, callback));
283 282 }
284 283
285 284 private void throwNotHandled(Object msg, TbCallback callback) {
... ...
... ... @@ -15,7 +15,6 @@
15 15 */
16 16 package org.thingsboard.server.service.queue;
17 17
18   -import akka.actor.ActorRef;
19 18 import com.google.protobuf.ProtocolStringList;
20 19 import lombok.extern.slf4j.Slf4j;
21 20 import org.springframework.beans.factory.annotation.Value;
... ... @@ -51,7 +50,6 @@ import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStr
51 50 import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy;
52 51 import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory;
53 52 import org.thingsboard.server.service.rpc.FromDeviceRpcResponse;
54   -import org.thingsboard.server.service.rpc.TbCoreDeviceRpcService;
55 53 import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService;
56 54 import org.thingsboard.server.service.stats.RuleEngineStatisticsService;
57 55
... ... @@ -232,7 +230,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
232 230 Optional<TbActorMsg> actorMsg = encodingService.decode(nfMsg.getComponentLifecycleMsg().toByteArray());
233 231 if (actorMsg.isPresent()) {
234 232 log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get());
235   - actorContext.tell(actorMsg.get(), ActorRef.noSender());
  233 + actorContext.tell(actorMsg.get());
236 234 }
237 235 callback.onSuccess();
238 236 } else if (nfMsg.hasFromDeviceRpcResponse()) {
... ... @@ -261,7 +259,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
261 259 }
262 260 }
263 261 msg = new QueueToRuleEngineMsg(tenantId, tbMsg, relationTypes, toRuleEngineMsg.getFailureMessage());
264   - actorContext.tell(msg, ActorRef.noSender());
  262 + actorContext.tell(msg);
265 263 }
266 264
267 265 @Scheduled(fixedDelayString = "${queue.rule-engine.stats.print-interval-ms}")
... ...
... ... @@ -15,7 +15,6 @@
15 15 */
16 16 package org.thingsboard.server.service.rpc;
17 17
18   -import akka.actor.ActorRef;
19 18 import com.fasterxml.jackson.core.JsonProcessingException;
20 19 import com.fasterxml.jackson.databind.ObjectMapper;
21 20 import com.fasterxml.jackson.databind.node.ObjectNode;
... ... @@ -122,7 +121,7 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService {
122 121 log.trace("[{}][{}] Processing local rpc call to device actor [{}]", request.getTenantId(), request.getId(), request.getDeviceId());
123 122 UUID requestId = request.getId();
124 123 localToDeviceRpcRequests.put(requestId, rpcMsg);
125   - actorContext.tell(rpcMsg, ActorRef.noSender());
  124 + actorContext.tell(rpcMsg);
126 125 scheduleToDeviceTimeout(request, requestId);
127 126 }
128 127
... ...
1   -#
2   -# Copyright © 2016-2020 The Thingsboard Authors
3   -#
4   -# Licensed under the Apache License, Version 2.0 (the "License");
5   -# you may not use this file except in compliance with the License.
6   -# You may obtain a copy of the License at
7   -#
8   -# http://www.apache.org/licenses/LICENSE-2.0
9   -#
10   -# Unless required by applicable law or agreed to in writing, software
11   -# distributed under the License is distributed on an "AS IS" BASIS,
12   -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13   -# See the License for the specific language governing permissions and
14   -# limitations under the License.
15   -#
16   -
17   -
18   -akka {
19   - # JVM shutdown, System.exit(-1), in case of a fatal error,
20   - # such as OutOfMemoryError
21   - jvm-exit-on-fatal-error = off
22   - loglevel = "INFO"
23   - loggers = ["akka.event.slf4j.Slf4jLogger"]
24   -}
25   -
26   -# This dispatcher is used for app
27   -app-dispatcher {
28   - type = Dispatcher
29   - executor = "fork-join-executor"
30   - fork-join-executor {
31   - # Min number of threads to cap factor-based parallelism number to
32   - parallelism-min = 1
33   - # Max number of threads to cap factor-based parallelism number to
34   - parallelism-max = 1
35   -
36   - # The parallelism factor is used to determine thread pool size using the
37   - # following formula: ceil(available processors * factor). Resulting size
38   - # is then bounded by the parallelism-min and parallelism-max values.
39   - parallelism-factor = 1.0
40   - }
41   - # How long time the dispatcher will wait for new actors until it shuts down
42   - shutdown-timeout = 1s
43   -
44   - # Throughput defines the number of messages that are processed in a batch
45   - # before the thread is returned to the pool. Set to 1 for as fair as possible.
46   - throughput = 5
47   -}
48   -
49   -# This dispatcher is used for rpc actors
50   -rpc-dispatcher {
51   - type = Dispatcher
52   - executor = "fork-join-executor"
53   - fork-join-executor {
54   - # Min number of threads to cap factor-based parallelism number to
55   - parallelism-min = 2
56   - # Max number of threads to cap factor-based parallelism number to
57   - parallelism-max = 8
58   -
59   - # The parallelism factor is used to determine thread pool size using the
60   - # following formula: ceil(available processors * factor). Resulting size
61   - # is then bounded by the parallelism-min and parallelism-max values.
62   - parallelism-factor = 0.5
63   - }
64   - # How long time the dispatcher will wait for new actors until it shuts down
65   - shutdown-timeout = 1s
66   -
67   - # Throughput defines the number of messages that are processed in a batch
68   - # before the thread is returned to the pool. Set to 1 for as fair as possible.
69   - throughput = 5
70   -}
71   -
72   -# This dispatcher is used for auth
73   -core-dispatcher {
74   - type = Dispatcher
75   - executor = "fork-join-executor"
76   - fork-join-executor {
77   - # Min number of threads to cap factor-based parallelism number to
78   - parallelism-min = 2
79   - # Max number of threads to cap factor-based parallelism number to
80   - parallelism-max = 12
81   -
82   - # The parallelism factor is used to determine thread pool size using the
83   - # following formula: ceil(available processors * factor). Resulting size
84   - # is then bounded by the parallelism-min and parallelism-max values.
85   - parallelism-factor = 0.25
86   - }
87   - # How long time the dispatcher will wait for new actors until it shuts down
88   - shutdown-timeout = 1s
89   -
90   - # Throughput defines the number of messages that are processed in a batch
91   - # before the thread is returned to the pool. Set to 1 for as fair as possible.
92   - throughput = 5
93   -}
94   -
95   -# This dispatcher is used for system rule chains and rule node actors
96   -system-rule-dispatcher {
97   - type = Dispatcher
98   - executor = "fork-join-executor"
99   - fork-join-executor {
100   - # Min number of threads to cap factor-based parallelism number to
101   - parallelism-min = 2
102   - # Max number of threads to cap factor-based parallelism number to
103   - parallelism-max = 12
104   -
105   - # The parallelism factor is used to determine thread pool size using the
106   - # following formula: ceil(available processors * factor). Resulting size
107   - # is then bounded by the parallelism-min and parallelism-max values.
108   - parallelism-factor = 0.25
109   - }
110   - # How long time the dispatcher will wait for new actors until it shuts down
111   - shutdown-timeout = 1s
112   -
113   - # Throughput defines the number of messages that are processed in a batch
114   - # before the thread is returned to the pool. Set to 1 for as fair as possible.
115   - throughput = 5
116   -}
117   -
118   -# This dispatcher is used for tenant rule chains and rule node actors
119   -rule-dispatcher {
120   - type = Dispatcher
121   - executor = "fork-join-executor"
122   - fork-join-executor {
123   - # Min number of threads to cap factor-based parallelism number to
124   - parallelism-min = 2
125   - # Max number of threads to cap factor-based parallelism number to
126   - parallelism-max = 12
127   -
128   - # The parallelism factor is used to determine thread pool size using the
129   - # following formula: ceil(available processors * factor). Resulting size
130   - # is then bounded by the parallelism-min and parallelism-max values.
131   - parallelism-factor = 0.25
132   - }
133   - # How long time the dispatcher will wait for new actors until it shuts down
134   - shutdown-timeout = 1s
135   -
136   - # Throughput defines the number of messages that are processed in a batch
137   - # before the thread is returned to the pool. Set to 1 for as fair as possible.
138   - throughput = 5
139   -}
... ... @@ -281,6 +281,14 @@ sql:
281 281
282 282 # Actor system parameters
283 283 actors:
  284 + system:
  285 + throughput: "${ACTORS_SYSTEM_THROUGHPUT:5}"
  286 + scheduler_pool_size: "${ACTORS_SYSTEM_SCHEDULER_POOL_SIZE:1}"
  287 + max_actor_init_attempts: "${ACTORS_SYSTEM_MAX_ACTOR_INIT_ATTEMPTS:10}"
  288 + app_dispatcher_pool_size: "${ACTORS_SYSTEM_APP_DISPATCHER_POOL_SIZE:1}"
  289 + tenant_dispatcher_pool_size: "${ACTORS_SYSTEM_TENANT_DISPATCHER_POOL_SIZE:2}"
  290 + device_dispatcher_pool_size: "${ACTORS_SYSTEM_DEVICE_DISPATCHER_POOL_SIZE:4}"
  291 + rule_dispatcher_pool_size: "${ACTORS_SYSTEM_RULE_DISPATCHER_POOL_SIZE:4}"
284 292 tenant:
285 293 create_components_on_init: "${ACTORS_TENANT_CREATE_COMPONENTS_ON_INIT:true}"
286 294 session:
... ... @@ -318,11 +326,6 @@ actors:
318 326 enabled: "${ACTORS_STATISTICS_ENABLED:true}"
319 327 js_print_interval_ms: "${ACTORS_JS_STATISTICS_PRINT_INTERVAL_MS:10000}"
320 328 persist_frequency: "${ACTORS_STATISTICS_PERSIST_FREQUENCY:3600000}"
321   - queue:
322   - # Enable/disable persistence of un-processed messages to the queue
323   - enabled: "${ACTORS_QUEUE_ENABLED:true}"
324   - # Maximum allowed timeout for persistence into the queue
325   - timeout: "${ACTORS_QUEUE_PERSISTENCE_TIMEOUT:30000}"
326 329
327 330 cache:
328 331 # caffeine or redis
... ...
... ... @@ -15,8 +15,6 @@
15 15 */
16 16 package org.thingsboard.server.rules.flow;
17 17
18   -import akka.actor.ActorRef;
19   -import com.datastax.driver.core.utils.UUIDs;
20 18 import lombok.extern.slf4j.Slf4j;
21 19 import org.junit.After;
22 20 import org.junit.Assert;
... ... @@ -26,7 +24,6 @@ import org.mockito.Mockito;
26 24 import org.springframework.beans.factory.annotation.Autowired;
27 25 import org.thingsboard.rule.engine.metadata.TbGetAttributesNodeConfiguration;
28 26 import org.thingsboard.server.actors.ActorSystemContext;
29   -import org.thingsboard.server.actors.service.ActorService;
30 27 import org.thingsboard.server.common.data.*;
31 28 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
32 29 import org.thingsboard.server.common.data.kv.StringDataEntry;
... ... @@ -36,7 +33,6 @@ import org.thingsboard.server.common.data.rule.RuleChainMetaData;
36 33 import org.thingsboard.server.common.data.rule.RuleNode;
37 34 import org.thingsboard.server.common.data.security.Authority;
38 35 import org.thingsboard.server.common.msg.TbMsg;
39   -import org.thingsboard.server.common.msg.TbMsgDataType;
40 36 import org.thingsboard.server.common.msg.TbMsgMetaData;
41 37 import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
42 38 import org.thingsboard.server.common.msg.queue.TbMsgCallback;
... ... @@ -151,7 +147,7 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
151 147 TbMsg tbMsg = TbMsg.newMsg("CUSTOM", device.getId(), new TbMsgMetaData(), "{}", tbMsgCallback);
152 148 QueueToRuleEngineMsg qMsg = new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg, null, null);
153 149 // Pushing Message to the system
154   - actorSystem.tell(qMsg, ActorRef.noSender());
  150 + actorSystem.tell(qMsg);
155 151 Mockito.verify(tbMsgCallback, Mockito.timeout(10000)).onSuccess();
156 152
157 153 TimePageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);
... ... @@ -263,7 +259,7 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
263 259 TbMsg tbMsg = TbMsg.newMsg("CUSTOM", device.getId(), new TbMsgMetaData(), "{}", tbMsgCallback);
264 260 QueueToRuleEngineMsg qMsg = new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg, null, null);
265 261 // Pushing Message to the system
266   - actorSystem.tell(qMsg, ActorRef.noSender());
  262 + actorSystem.tell(qMsg);
267 263
268 264 Mockito.verify(tbMsgCallback, Mockito.timeout(10000)).onSuccess();
269 265
... ...
... ... @@ -15,8 +15,6 @@
15 15 */
16 16 package org.thingsboard.server.rules.lifecycle;
17 17
18   -import akka.actor.ActorRef;
19   -import com.datastax.driver.core.utils.UUIDs;
20 18 import lombok.extern.slf4j.Slf4j;
21 19 import org.junit.After;
22 20 import org.junit.Assert;
... ... @@ -26,7 +24,6 @@ import org.mockito.Mockito;
26 24 import org.springframework.beans.factory.annotation.Autowired;
27 25 import org.thingsboard.rule.engine.metadata.TbGetAttributesNodeConfiguration;
28 26 import org.thingsboard.server.actors.ActorSystemContext;
29   -import org.thingsboard.server.actors.service.ActorService;
30 27 import org.thingsboard.server.common.data.DataConstants;
31 28 import org.thingsboard.server.common.data.Device;
32 29 import org.thingsboard.server.common.data.Event;
... ... @@ -40,7 +37,6 @@ import org.thingsboard.server.common.data.rule.RuleChainMetaData;
40 37 import org.thingsboard.server.common.data.rule.RuleNode;
41 38 import org.thingsboard.server.common.data.security.Authority;
42 39 import org.thingsboard.server.common.msg.TbMsg;
43   -import org.thingsboard.server.common.msg.TbMsgDataType;
44 40 import org.thingsboard.server.common.msg.TbMsgMetaData;
45 41 import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
46 42 import org.thingsboard.server.common.msg.queue.TbMsgCallback;
... ... @@ -142,7 +138,7 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
142 138 TbMsg tbMsg = TbMsg.newMsg("CUSTOM", device.getId(), new TbMsgMetaData(), "{}", tbMsgCallback);
143 139 QueueToRuleEngineMsg qMsg = new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg, null, null);
144 140 // Pushing Message to the system
145   - actorSystem.tell(qMsg, ActorRef.noSender());
  141 + actorSystem.tell(qMsg);
146 142 Mockito.verify(tbMsgCallback, Mockito.timeout(3000)).onSuccess();
147 143
148 144
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.actors;
  17 +
  18 +import lombok.Getter;
  19 +
  20 +public abstract class AbstractTbActor implements TbActor {
  21 +
  22 + @Getter
  23 + protected TbActorCtx ctx;
  24 +
  25 + @Override
  26 + public void init(TbActorCtx ctx) {
  27 + this.ctx = ctx;
  28 + }
  29 +
  30 + @Override
  31 + public TbActorRef getActorRef() {
  32 + return ctx;
  33 + }
  34 +}
... ...
... ... @@ -21,13 +21,19 @@ import lombok.extern.slf4j.Slf4j;
21 21 import org.thingsboard.common.util.ThingsBoardThreadFactory;
22 22 import org.thingsboard.server.common.msg.TbActorMsg;
23 23
  24 +import java.util.Collections;
  25 +import java.util.List;
  26 +import java.util.Set;
24 27 import java.util.concurrent.ConcurrentHashMap;
25 28 import java.util.concurrent.ConcurrentMap;
26 29 import java.util.concurrent.ExecutorService;
27 30 import java.util.concurrent.Executors;
28 31 import java.util.concurrent.ScheduledExecutorService;
  32 +import java.util.concurrent.TimeUnit;
29 33 import java.util.concurrent.locks.Lock;
30 34 import java.util.concurrent.locks.ReentrantLock;
  35 +import java.util.function.Predicate;
  36 +import java.util.stream.Collectors;
31 37
32 38 @Slf4j
33 39 @Data
... ... @@ -36,6 +42,8 @@ public class DefaultTbActorSystem implements TbActorSystem {
36 42 private final ConcurrentMap<String, Dispatcher> dispatchers = new ConcurrentHashMap<>();
37 43 private final ConcurrentMap<TbActorId, TbActorMailbox> actors = new ConcurrentHashMap<>();
38 44 private final ConcurrentMap<TbActorId, ReentrantLock> actorCreationLocks = new ConcurrentHashMap<>();
  45 + private final ConcurrentMap<TbActorId, Set<TbActorId>> parentChildMap = new ConcurrentHashMap<>();
  46 +
39 47 @Getter
40 48 private final TbActorSystemSettings settings;
41 49 @Getter
... ... @@ -65,16 +73,21 @@ public class DefaultTbActorSystem implements TbActorSystem {
65 73 }
66 74
67 75 @Override
68   - public TbActorId createRootActor(String dispatcherId, TbActorCreator creator) {
  76 + public TbActorRef getActor(TbActorId actorId) {
  77 + return actors.get(actorId);
  78 + }
  79 +
  80 + @Override
  81 + public TbActorRef createRootActor(String dispatcherId, TbActorCreator creator) {
69 82 return createActor(dispatcherId, creator, null);
70 83 }
71 84
72 85 @Override
73   - public TbActorId createChildActor(String dispatcherId, TbActorCreator creator, TbActorId parent) {
  86 + public TbActorRef createChildActor(String dispatcherId, TbActorCreator creator, TbActorId parent) {
74 87 return createActor(dispatcherId, creator, parent);
75 88 }
76 89
77   - private TbActorId createActor(String dispatcherId, TbActorCreator creator, TbActorId parent) {
  90 + private TbActorRef createActor(String dispatcherId, TbActorCreator creator, TbActorId parent) {
78 91 Dispatcher dispatcher = dispatchers.get(dispatcherId);
79 92 if (dispatcher == null) {
80 93 log.warn("Dispatcher with id [{}] is not registered!", dispatcherId);
... ... @@ -93,9 +106,20 @@ public class DefaultTbActorSystem implements TbActorSystem {
93 106 if (actorMailbox == null) {
94 107 log.debug("Creating actor with id [{}]!", actorId);
95 108 TbActor actor = creator.createActor();
96   - TbActorMailbox mailbox = new TbActorMailbox(this, settings, actorId, parent, actor, dispatcher);
  109 + TbActorRef parentRef = null;
  110 + if (parent != null) {
  111 + parentRef = getActor(parent);
  112 + if (parentRef == null) {
  113 + throw new TbActorNotRegisteredException(parent, "Parent Actor with id [" + parent + "] is not registered!");
  114 + }
  115 + }
  116 + TbActorMailbox mailbox = new TbActorMailbox(this, settings, actorId, parentRef, actor, dispatcher);
97 117 actors.put(actorId, mailbox);
98 118 mailbox.initActor();
  119 + actorMailbox = mailbox;
  120 + if (parent != null) {
  121 + parentChildMap.computeIfAbsent(parent, id -> ConcurrentHashMap.newKeySet()).add(actorId);
  122 + }
99 123 } else {
100 124 log.debug("Actor with id [{}] is already registered!", actorId);
101 125 }
... ... @@ -104,7 +128,12 @@ public class DefaultTbActorSystem implements TbActorSystem {
104 128 actorCreationLocks.remove(actorId);
105 129 }
106 130 }
107   - return actorId;
  131 + return actorMailbox;
  132 + }
  133 +
  134 + @Override
  135 + public void tell(TbActorRef target, TbActorMsg actorMsg) {
  136 + target.tell(actorMsg);
108 137 }
109 138
110 139 @Override
... ... @@ -117,7 +146,41 @@ public class DefaultTbActorSystem implements TbActorSystem {
117 146 }
118 147
119 148 @Override
  149 + public void broadcastToChildren(TbActorId parent, TbActorMsg msg) {
  150 + broadcastToChildren(parent, id -> true, msg);
  151 + }
  152 +
  153 + @Override
  154 + public void broadcastToChildren(TbActorId parent, Predicate<TbActorId> childFilter, TbActorMsg msg) {
  155 + Set<TbActorId> children = parentChildMap.get(parent);
  156 + if (children != null) {
  157 + children.stream().filter(childFilter).forEach(id -> tell(id, msg));
  158 + }
  159 + }
  160 +
  161 + @Override
  162 + public List<TbActorId> filterChildren(TbActorId parent, Predicate<TbActorId> childFilter) {
  163 + Set<TbActorId> children = parentChildMap.get(parent);
  164 + if (children != null) {
  165 + return children.stream().filter(childFilter).collect(Collectors.toList());
  166 + } else {
  167 + return Collections.emptyList();
  168 + }
  169 + }
  170 +
  171 + @Override
  172 + public void stop(TbActorRef actorRef) {
  173 + stop(actorRef.getActorId());
  174 + }
  175 +
  176 + @Override
120 177 public void stop(TbActorId actorId) {
  178 + Set<TbActorId> children = parentChildMap.remove(actorId);
  179 + if (children != null) {
  180 + for (TbActorId child : children) {
  181 + stop(child);
  182 + }
  183 + }
121 184 TbActorMailbox mailbox = actors.remove(actorId);
122 185 if (mailbox != null) {
123 186 mailbox.destroy();
... ... @@ -126,7 +189,17 @@ public class DefaultTbActorSystem implements TbActorSystem {
126 189
127 190 @Override
128 191 public void stop() {
129   - dispatchers.values().forEach(dispatcher -> dispatcher.getExecutor().shutdownNow());
  192 + dispatchers.values().forEach(dispatcher -> {
  193 + dispatcher.getExecutor().shutdown();
  194 + try {
  195 + dispatcher.getExecutor().awaitTermination(3, TimeUnit.SECONDS);
  196 + } catch (InterruptedException e) {
  197 + log.warn("[{}] Failed to stop dispatcher", dispatcher.getDispatcherId(), e);
  198 + }
  199 + });
  200 + if (scheduler != null) {
  201 + scheduler.shutdownNow();
  202 + }
130 203 actors.clear();
131 204 }
132 205
... ...
... ... @@ -19,11 +19,15 @@ import org.thingsboard.server.common.msg.TbActorMsg;
19 19
20 20 public interface TbActor {
21 21
22   - void init();
  22 + boolean process(TbActorMsg msg);
23 23
24   - boolean process(TbActorCtx ctx, TbActorMsg msg);
  24 + TbActorRef getActorRef();
25 25
26   - void destroy();
  26 + default void init(TbActorCtx ctx) {
  27 + }
  28 +
  29 + default void destroy() {
  30 + }
27 31
28 32 default InitFailureStrategy onInitFailure(int attempt, Throwable t) {
29 33 return InitFailureStrategy.retryWithDelay(5000);
... ...
... ... @@ -17,12 +17,25 @@ package org.thingsboard.server.actors;
17 17
18 18 import org.thingsboard.server.common.msg.TbActorMsg;
19 19
20   -public interface TbActorCtx {
  20 +import java.util.List;
  21 +import java.util.function.Predicate;
  22 +import java.util.function.Supplier;
  23 +
  24 +public interface TbActorCtx extends TbActorRef {
21 25
22 26 TbActorId getSelf();
23 27
24   - TbActorId getParent();
  28 + TbActorRef getParentRef();
  29 +
  30 + void tell(TbActorId target, TbActorMsg msg);
  31 +
  32 + void stop(TbActorId target);
  33 +
  34 + TbActorRef getOrCreateChildActor(TbActorId actorId, Supplier<String> dispatcher, Supplier<TbActorCreator> creator);
  35 +
  36 + void broadcastToChildren(TbActorMsg msg);
25 37
26   - void tell(TbActorId target, TbActorMsg actorMsg);
  38 + void broadcastToChildren(TbActorMsg msg, Predicate<TbActorId> childFilter);
27 39
  40 + List<TbActorId> filterChildren(Predicate<TbActorId> childFilter);
28 41 }
... ...
... ... @@ -15,33 +15,6 @@
15 15 */
16 16 package org.thingsboard.server.actors;
17 17
18   -import org.thingsboard.server.common.data.id.EntityId;
  18 +public interface TbActorId {
19 19
20   -import java.util.Objects;
21   -
22   -public class TbActorId {
23   -
24   - private final EntityId entityId;
25   -
26   - public TbActorId(EntityId entityId) {
27   - this.entityId = entityId;
28   - }
29   -
30   - @Override
31   - public String toString() {
32   - return entityId.toString();
33   - }
34   -
35   - @Override
36   - public boolean equals(Object o) {
37   - if (this == o) return true;
38   - if (o == null || getClass() != o.getClass()) return false;
39   - TbActorId actorId = (TbActorId) o;
40   - return entityId.equals(actorId.entityId);
41   - }
42   -
43   - @Override
44   - public int hashCode() {
45   - return Objects.hash(entityId);
46   - }
47 20 }
... ...
... ... @@ -19,9 +19,12 @@ import lombok.Data;
19 19 import lombok.extern.slf4j.Slf4j;
20 20 import org.thingsboard.server.common.msg.TbActorMsg;
21 21
  22 +import java.util.List;
22 23 import java.util.concurrent.ConcurrentLinkedQueue;
23 24 import java.util.concurrent.TimeUnit;
24 25 import java.util.concurrent.atomic.AtomicBoolean;
  26 +import java.util.function.Predicate;
  27 +import java.util.function.Supplier;
25 28
26 29 @Slf4j
27 30 @Data
... ... @@ -35,7 +38,7 @@ public final class TbActorMailbox implements TbActorCtx {
35 38 private final TbActorSystem system;
36 39 private final TbActorSystemSettings settings;
37 40 private final TbActorId selfId;
38   - private final TbActorId parentId;
  41 + private final TbActorRef parentRef;
39 42 private final TbActor actor;
40 43 private final Dispatcher dispatcher;
41 44 private final ConcurrentLinkedQueue<TbActorMsg> msgs = new ConcurrentLinkedQueue<>();
... ... @@ -47,11 +50,12 @@ public final class TbActorMailbox implements TbActorCtx {
47 50 dispatcher.getExecutor().execute(() -> tryInit(1));
48 51 }
49 52
  53 +
50 54 private void tryInit(int attempt) {
51 55 try {
52 56 log.debug("[{}] Trying to init actor, attempt: {}", selfId, attempt);
53 57 if (!destroyInProgress.get()) {
54   - actor.init();
  58 + actor.init(this);
55 59 if (!destroyInProgress.get()) {
56 60 ready.set(READY);
57 61 tryProcessQueue(false);
... ... @@ -94,7 +98,7 @@ public final class TbActorMailbox implements TbActorCtx {
94 98 if (msg != null) {
95 99 try {
96 100 log.debug("[{}] Going to process message: {}", selfId, msg);
97   - actor.process(this, msg);
  101 + actor.process(msg);
98 102 } catch (Throwable t) {
99 103 log.debug("[{}] Failed to process message: {}", selfId, msg, t);
100 104 ProcessFailureStrategy strategy = actor.onProcessFailure(t);
... ... @@ -121,13 +125,38 @@ public final class TbActorMailbox implements TbActorCtx {
121 125 }
122 126
123 127 @Override
124   - public TbActorId getParent() {
125   - return parentId;
  128 + public void tell(TbActorId target, TbActorMsg actorMsg) {
  129 + system.tell(target, actorMsg);
126 130 }
127 131
128 132 @Override
129   - public void tell(TbActorId target, TbActorMsg actorMsg) {
130   - system.tell(target, actorMsg);
  133 + public void broadcastToChildren(TbActorMsg msg) {
  134 + system.broadcastToChildren(selfId, msg);
  135 + }
  136 +
  137 + @Override
  138 + public void broadcastToChildren(TbActorMsg msg, Predicate<TbActorId> childFilter) {
  139 + system.broadcastToChildren(selfId, childFilter, msg);
  140 + }
  141 +
  142 + @Override
  143 + public List<TbActorId> filterChildren(Predicate<TbActorId> childFilter) {
  144 + return system.filterChildren(selfId, childFilter);
  145 + }
  146 +
  147 + @Override
  148 + public void stop(TbActorId target) {
  149 + system.stop(target);
  150 + }
  151 +
  152 + @Override
  153 + public TbActorRef getOrCreateChildActor(TbActorId actorId, Supplier<String> dispatcher, Supplier<TbActorCreator> creator) {
  154 + TbActorRef actorRef = system.getActor(actorId);
  155 + if (actorRef == null) {
  156 + return system.createChildActor(dispatcher.get(), creator.get(), selfId);
  157 + } else {
  158 + return actorRef;
  159 + }
131 160 }
132 161
133 162 public void destroy() {
... ... @@ -141,4 +170,14 @@ public final class TbActorMailbox implements TbActorCtx {
141 170 }
142 171 });
143 172 }
  173 +
  174 + @Override
  175 + public TbActorId getActorId() {
  176 + return selfId;
  177 + }
  178 +
  179 + @Override
  180 + public void tell(TbActorMsg actorMsg) {
  181 + enqueue(actorMsg);
  182 + }
144 183 }
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.actors;
  17 +
  18 +import org.thingsboard.server.common.msg.TbActorMsg;
  19 +
  20 +public interface TbActorRef {
  21 +
  22 + TbActorId getActorId();
  23 +
  24 + void tell(TbActorMsg actorMsg);
  25 +
  26 +}
... ...
... ... @@ -17,8 +17,10 @@ package org.thingsboard.server.actors;
17 17
18 18 import org.thingsboard.server.common.msg.TbActorMsg;
19 19
  20 +import java.util.List;
20 21 import java.util.concurrent.ExecutorService;
21 22 import java.util.concurrent.ScheduledExecutorService;
  23 +import java.util.function.Predicate;
22 24
23 25 public interface TbActorSystem {
24 26
... ... @@ -28,14 +30,25 @@ public interface TbActorSystem {
28 30
29 31 void destroyDispatcher(String dispatcherId);
30 32
31   - TbActorId createRootActor(String dispatcherId, TbActorCreator creator);
  33 + TbActorRef getActor(TbActorId actorId);
32 34
33   - TbActorId createChildActor(String dispatcherId, TbActorCreator creator, TbActorId parent);
  35 + TbActorRef createRootActor(String dispatcherId, TbActorCreator creator);
  36 +
  37 + TbActorRef createChildActor(String dispatcherId, TbActorCreator creator, TbActorId parent);
  38 +
  39 + void tell(TbActorRef target, TbActorMsg actorMsg);
34 40
35 41 void tell(TbActorId target, TbActorMsg actorMsg);
36 42
  43 + void stop(TbActorRef actorRef);
  44 +
37 45 void stop(TbActorId actorId);
38 46
39 47 void stop();
40 48
  49 + void broadcastToChildren(TbActorId parent, TbActorMsg msg);
  50 +
  51 + void broadcastToChildren(TbActorId parent, Predicate<TbActorId> childFilter, TbActorMsg msg);
  52 +
  53 + List<TbActorId> filterChildren(TbActorId parent, Predicate<TbActorId> childFilter);
41 54 }
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.actors;
  17 +
  18 +import lombok.Getter;
  19 +import org.thingsboard.server.common.data.id.EntityId;
  20 +
  21 +import java.util.Objects;
  22 +
  23 +public class TbEntityActorId implements TbActorId {
  24 +
  25 + @Getter
  26 + private final EntityId entityId;
  27 +
  28 + public TbEntityActorId(EntityId entityId) {
  29 + this.entityId = entityId;
  30 + }
  31 +
  32 + @Override
  33 + public String toString() {
  34 + return entityId.toString();
  35 + }
  36 +
  37 + @Override
  38 + public boolean equals(Object o) {
  39 + if (this == o) return true;
  40 + if (o == null || getClass() != o.getClass()) return false;
  41 + TbEntityActorId that = (TbEntityActorId) o;
  42 + return entityId.equals(that.entityId);
  43 + }
  44 +
  45 + @Override
  46 + public int hashCode() {
  47 + return Objects.hash(entityId);
  48 + }
  49 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2020 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.actors;
  17 +
  18 +import java.util.Objects;
  19 +
  20 +public class TbStringActorId implements TbActorId {
  21 +
  22 + private final String id;
  23 +
  24 + public TbStringActorId(String id) {
  25 + this.id = id;
  26 + }
  27 +
  28 + @Override
  29 + public String toString() {
  30 + return id;
  31 + }
  32 +
  33 + @Override
  34 + public boolean equals(Object o) {
  35 + if (this == o) return true;
  36 + if (o == null || getClass() != o.getClass()) return false;
  37 + TbStringActorId that = (TbStringActorId) o;
  38 + return id.equals(that.id);
  39 + }
  40 +
  41 + @Override
  42 + public int hashCode() {
  43 + return Objects.hash(id);
  44 + }
  45 +}
... ...
... ... @@ -81,10 +81,10 @@ public class ActorSystemTest {
81 81 ActorTestCtx testCtx1 = getActorTestCtx(1);
82 82 ActorTestCtx testCtx2 = getActorTestCtx(1);
83 83
84   - TbActorId actorId1 = actorSystem.createRootActor(ROOT_DISPATCHER, new SlowInitActor.SlowInitActorCreator(
85   - new TbActorId(new DeviceId(UUID.randomUUID())), testCtx1));
86   - TbActorId actorId2 = actorSystem.createRootActor(ROOT_DISPATCHER, new SlowInitActor.SlowInitActorCreator(
87   - new TbActorId(new DeviceId(UUID.randomUUID())), testCtx2));
  84 + TbActorRef actorId1 = actorSystem.createRootActor(ROOT_DISPATCHER, new SlowInitActor.SlowInitActorCreator(
  85 + new TbEntityActorId(new DeviceId(UUID.randomUUID())), testCtx1));
  86 + TbActorRef actorId2 = actorSystem.createRootActor(ROOT_DISPATCHER, new SlowInitActor.SlowInitActorCreator(
  87 + new TbEntityActorId(new DeviceId(UUID.randomUUID())), testCtx2));
88 88
89 89 actorSystem.tell(actorId1, new IntTbActorMsg(42));
90 90 actorSystem.tell(actorId2, new IntTbActorMsg(42));
... ... @@ -98,7 +98,7 @@ public class ActorSystemTest {
98 98 public void testOneActorCreated() throws InterruptedException {
99 99 ActorTestCtx testCtx1 = getActorTestCtx(1);
100 100 ActorTestCtx testCtx2 = getActorTestCtx(1);
101   - TbActorId actorId = new TbActorId(new DeviceId(UUID.randomUUID()));
  101 + TbActorId actorId = new TbEntityActorId(new DeviceId(UUID.randomUUID()));
102 102 submitPool.submit(() -> actorSystem.createRootActor(ROOT_DISPATCHER, new SlowCreateActor.SlowCreateActorCreator(actorId, testCtx1)));
103 103 submitPool.submit(() -> actorSystem.createRootActor(ROOT_DISPATCHER, new SlowCreateActor.SlowCreateActorCreator(actorId, testCtx2)));
104 104
... ... @@ -112,7 +112,7 @@ public class ActorSystemTest {
112 112 @Test
113 113 public void testActorCreatorCalledOnce() throws InterruptedException {
114 114 ActorTestCtx testCtx = getActorTestCtx(1);
115   - TbActorId actorId = new TbActorId(new DeviceId(UUID.randomUUID()));
  115 + TbActorId actorId = new TbEntityActorId(new DeviceId(UUID.randomUUID()));
116 116 for(int i =0; i < 1000; i++) {
117 117 submitPool.submit(() -> actorSystem.createRootActor(ROOT_DISPATCHER, new SlowCreateActor.SlowCreateActorCreator(actorId, testCtx)));
118 118 }
... ... @@ -138,12 +138,12 @@ public class ActorSystemTest {
138 138
139 139 List<ActorTestCtx> testCtxes = new ArrayList<>();
140 140
141   - List<TbActorId> actorIds = new ArrayList<>();
  141 + List<TbActorRef> actorRefs = new ArrayList<>();
142 142 for (int actorIdx = 0; actorIdx < actorsCount; actorIdx++) {
143 143 ActorTestCtx testCtx = getActorTestCtx(msgNumber);
144 144
145   - actorIds.add(actorSystem.createRootActor(ROOT_DISPATCHER, new TestRootActor.TestRootActorCreator(
146   - new TbActorId(new DeviceId(UUID.randomUUID())), testCtx)));
  145 + actorRefs.add(actorSystem.createRootActor(ROOT_DISPATCHER, new TestRootActor.TestRootActorCreator(
  146 + new TbEntityActorId(new DeviceId(UUID.randomUUID())), testCtx)));
147 147 testCtxes.add(testCtx);
148 148 }
149 149
... ... @@ -151,7 +151,7 @@ public class ActorSystemTest {
151 151
152 152 for (int i = 0; i < msgNumber; i++) {
153 153 int tmp = randomIntegers[i];
154   - submitPool.execute(() -> actorIds.forEach(actorId -> actorSystem.tell(actorId, new IntTbActorMsg(tmp))));
  154 + submitPool.execute(() -> actorRefs.forEach(actorId -> actorSystem.tell(actorId, new IntTbActorMsg(tmp))));
155 155 }
156 156 log.info("Submitted all messages");
157 157
... ...
... ... @@ -25,13 +25,13 @@ public class SlowInitActor extends TestRootActor {
25 25 }
26 26
27 27 @Override
28   - public void init() {
  28 + public void init(TbActorCtx ctx) {
29 29 try {
30 30 Thread.sleep(500);
31 31 } catch (InterruptedException e) {
32 32 e.printStackTrace();
33 33 }
34   - super.init();
  34 + super.init(ctx);
35 35 }
36 36
37 37 public static class SlowInitActorCreator implements TbActorCreator {
... ...
... ... @@ -20,7 +20,7 @@ import lombok.extern.slf4j.Slf4j;
20 20 import org.thingsboard.server.common.msg.TbActorMsg;
21 21
22 22 @Slf4j
23   -public class TestRootActor implements TbActor {
  23 +public class TestRootActor extends AbstractTbActor {
24 24
25 25 @Getter
26 26 private final TbActorId actorId;
... ... @@ -37,12 +37,13 @@ public class TestRootActor implements TbActor {
37 37 }
38 38
39 39 @Override
40   - public void init() {
  40 + public void init(TbActorCtx ctx) {
  41 + super.init(ctx);
41 42 initialized = true;
42 43 }
43 44
44 45 @Override
45   - public boolean process(TbActorCtx ctx, TbActorMsg msg) {
  46 + public boolean process(TbActorMsg msg) {
46 47 if (initialized) {
47 48 int value = ((IntTbActorMsg) msg).getValue();
48 49 sum += value;
... ...
... ... @@ -21,7 +21,6 @@ import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
21 21 /**
22 22 * Created by ashvayka on 15.03.18.
23 23 */
24   -//TODO: add all "See" references
25 24 public enum MsgType {
26 25
27 26 /**
... ... @@ -97,6 +96,7 @@ public enum MsgType {
97 96
98 97 STATS_PERSIST_TICK_MSG,
99 98
  99 + STATS_PERSIST_MSG,
100 100
101 101 /**
102 102 * Message that is sent by TransportRuleEngineService to Device Actor. Represents messages from the device itself.
... ...
... ... @@ -16,8 +16,9 @@
16 16 package org.thingsboard.server.common.msg.aware;
17 17
18 18 import org.thingsboard.server.common.data.id.DeviceId;
  19 +import org.thingsboard.server.common.msg.TbActorMsg;
19 20
20   -public interface DeviceAwareMsg {
  21 +public interface DeviceAwareMsg extends TbActorMsg {
21 22
22 23 DeviceId getDeviceId();
23 24 }
... ...
... ... @@ -16,8 +16,9 @@
16 16 package org.thingsboard.server.common.msg.aware;
17 17
18 18 import org.thingsboard.server.common.data.id.RuleChainId;
  19 +import org.thingsboard.server.common.msg.TbActorMsg;
19 20
20   -public interface RuleChainAwareMsg {
  21 +public interface RuleChainAwareMsg extends TbActorMsg {
21 22
22 23 RuleChainId getRuleChainId();
23 24
... ...
... ... @@ -16,8 +16,9 @@
16 16 package org.thingsboard.server.common.msg.aware;
17 17
18 18 import org.thingsboard.server.common.data.id.TenantId;
  19 +import org.thingsboard.server.common.msg.TbActorMsg;
19 20
20   -public interface TenantAwareMsg {
  21 +public interface TenantAwareMsg extends TbActorMsg {
21 22
22 23 TenantId getTenantId();
23 24
... ...
... ... @@ -62,8 +62,6 @@
62 62 <jackson-annotations.version>2.10.2</jackson-annotations.version>
63 63 <jackson-core.version>2.10.2</jackson-core.version>
64 64 <json-schema-validator.version>2.2.6</json-schema-validator.version>
65   - <scala.version>2.13</scala.version>
66   - <akka.version>2.6.3</akka.version>
67 65 <californium.version>1.0.2</californium.version>
68 66 <gson.version>2.6.2</gson.version>
69 67 <velocity.version>1.7</velocity.version>
... ... @@ -782,6 +780,11 @@
782 780 </dependency>
783 781 <dependency>
784 782 <groupId>org.thingsboard.common</groupId>
  783 + <artifactId>actor</artifactId>
  784 + <version>${project.version}</version>
  785 + </dependency>
  786 + <dependency>
  787 + <groupId>org.thingsboard.common</groupId>
785 788 <artifactId>dao-api</artifactId>
786 789 <version>${project.version}</version>
787 790 </dependency>
... ... @@ -1114,16 +1117,6 @@
1114 1117 </exclusions>
1115 1118 </dependency>
1116 1119 <dependency>
1117   - <groupId>com.typesafe.akka</groupId>
1118   - <artifactId>akka-actor_${scala.version}</artifactId>
1119   - <version>${akka.version}</version>
1120   - </dependency>
1121   - <dependency>
1122   - <groupId>com.typesafe.akka</groupId>
1123   - <artifactId>akka-slf4j_${scala.version}</artifactId>
1124   - <version>${akka.version}</version>
1125   - </dependency>
1126   - <dependency>
1127 1120 <groupId>org.eclipse.californium</groupId>
1128 1121 <artifactId>californium-core</artifactId>
1129 1122 <version>${californium.version}</version>
... ...