diff --git a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java index 0a12e93171..f1248a3aad 100644 --- a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java @@ -134,12 +134,12 @@ public class AppActor extends ContextAwareActor { private void onQueueToRuleEngineMsg(QueueToRuleEngineMsg msg) { if (TenantId.SYS_TENANT_ID.equals(msg.getTenantId())) { - msg.getTbMsg().getCallback().onFailure(new RuleEngineException("Message has system tenant id!")); + msg.getMsg().getCallback().onFailure(new RuleEngineException("Message has system tenant id!")); } else { if (!deletedTenants.contains(msg.getTenantId())) { getOrCreateTenantActor(msg.getTenantId()).tell(msg); } else { - msg.getTbMsg().getCallback().onSuccess(); + msg.getMsg().getCallback().onSuccess(); } } } diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 51100a8377..6d27516811 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -19,7 +19,6 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import io.netty.channel.EventLoopGroup; import lombok.extern.slf4j.Slf4j; -import org.springframework.data.redis.core.RedisTemplate; import org.springframework.util.StringUtils; import org.thingsboard.common.util.ListeningExecutor; import org.thingsboard.rule.engine.api.MailService; @@ -34,7 +33,6 @@ import org.thingsboard.rule.engine.api.TbRelationTypes; import org.thingsboard.rule.engine.api.sms.SmsSenderFactory; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.TbActorRef; -import org.thingsboard.server.common.data.ApiUsageRecordKey; import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; @@ -90,10 +88,12 @@ class DefaultTbContext implements TbContext { public final static ObjectMapper mapper = new ObjectMapper(); private final ActorSystemContext mainCtx; + private final String ruleChainName; private final RuleNodeCtx nodeCtx; - public DefaultTbContext(ActorSystemContext mainCtx, RuleNodeCtx nodeCtx) { + public DefaultTbContext(ActorSystemContext mainCtx, String ruleChainName, RuleNodeCtx nodeCtx) { this.mainCtx = mainCtx; + this.ruleChainName = ruleChainName; this.nodeCtx = nodeCtx; } @@ -117,13 +117,13 @@ class DefaultTbContext implements TbContext { relationTypes.forEach(relationType -> mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg, relationType, th)); } msg.getCallback().onProcessingEnd(nodeCtx.getSelf().getId()); - nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getId(), relationTypes, msg, th != null ? th.getMessage() : null)); + nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId(), relationTypes, msg, th != null ? th.getMessage() : null)); } @Override public void tellSelf(TbMsg msg, long delayMs) { //TODO: add persistence layer - scheduleMsgWithDelay(new RuleNodeToSelfMsg(msg), delayMs, nodeCtx.getSelfActor()); + scheduleMsgWithDelay(new RuleNodeToSelfMsg(this, msg), delayMs, nodeCtx.getSelfActor()); } @Override @@ -254,7 +254,8 @@ class DefaultTbContext implements TbContext { } else { failureMessage = null; } - nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getId(), Collections.singleton(TbRelationTypes.FAILURE), + nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getRuleChainId(), + nodeCtx.getSelf().getId(), Collections.singleton(TbRelationTypes.FAILURE), msg, failureMessage)); } @@ -301,6 +302,16 @@ class DefaultTbContext implements TbContext { return nodeCtx.getSelf().getId(); } + @Override + public RuleNode getSelf() { + return nodeCtx.getSelf(); + } + + @Override + public String getRuleChainName() { + return ruleChainName; + } + @Override public TenantId getTenantId() { return nodeCtx.getTenantId(); diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java index 579fae829e..cc0c9da8f2 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java @@ -23,7 +23,6 @@ import org.thingsboard.server.actors.TbActorRef; import org.thingsboard.server.actors.TbEntityActorId; import org.thingsboard.server.actors.service.DefaultActorService; import org.thingsboard.server.actors.shared.ComponentMsgProcessor; -import org.thingsboard.server.common.data.ApiUsageRecordKey; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.RuleChainId; @@ -197,11 +196,11 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor relationTypes; - private final TbMsg msg; + @Getter private final String failureMessage; + public RuleNodeToRuleChainTellNextMsg(RuleChainId ruleChainId, RuleNodeId originator, Set relationTypes, TbMsg tbMsg, String failureMessage) { + super(tbMsg); + this.ruleChainId = ruleChainId; + this.originator = originator; + this.relationTypes = relationTypes; + this.failureMessage = failureMessage; + } + + @Override + public void onTbActorStopped(TbActorStopReason reason) { + String message = reason == TbActorStopReason.STOPPED ? String.format("Rule chain [%s] stopped", ruleChainId.getId()) : String.format("Failed to initialize rule chain [%s]!", ruleChainId.getId()); + msg.getCallback().onFailure(new RuleEngineException(message)); + } + @Override public MsgType getMsgType() { return MsgType.RULE_TO_RULE_CHAIN_TELL_NEXT_MSG; diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeToSelfMsg.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeToSelfMsg.java index bdd33f2381..3ebaa7ffa8 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeToSelfMsg.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeToSelfMsg.java @@ -15,18 +15,25 @@ */ package org.thingsboard.server.actors.ruleChain; -import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.ToString; +import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.server.common.msg.MsgType; -import org.thingsboard.server.common.msg.TbActorMsg; +import org.thingsboard.server.common.msg.TbActorStopReason; import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbRuleEngineActorMsg; +import org.thingsboard.server.common.msg.queue.RuleNodeException; /** * Created by ashvayka on 19.03.18. */ -@Data -final class RuleNodeToSelfMsg implements TbActorMsg { +@EqualsAndHashCode(callSuper = true) +@ToString +final class RuleNodeToSelfMsg extends TbToRuleNodeActorMsg { - private final TbMsg msg; + public RuleNodeToSelfMsg(TbContext ctx, TbMsg tbMsg) { + super(ctx, tbMsg); + } @Override public MsgType getMsgType() { diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/TbToRuleNodeActorMsg.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/TbToRuleNodeActorMsg.java new file mode 100644 index 0000000000..6d492f630a --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/TbToRuleNodeActorMsg.java @@ -0,0 +1,42 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.actors.ruleChain; + +import lombok.EqualsAndHashCode; +import lombok.Getter; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.server.common.msg.TbActorStopReason; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbRuleEngineActorMsg; +import org.thingsboard.server.common.msg.queue.RuleNodeException; + +@EqualsAndHashCode(callSuper = true) +public abstract class TbToRuleNodeActorMsg extends TbRuleEngineActorMsg { + + @Getter + private final TbContext ctx; + + public TbToRuleNodeActorMsg(TbContext ctx, TbMsg tbMsg) { + super(tbMsg); + this.ctx = ctx; + } + + @Override + public void onTbActorStopped(TbActorStopReason reason) { + String message = reason == TbActorStopReason.STOPPED ? "Rule node stopped" : "Failed to initialize rule node!"; + msg.getCallback().onFailure(new RuleNodeException(message, ctx.getRuleChainName(), ctx.getSelf())); + } +} diff --git a/application/src/main/java/org/thingsboard/server/actors/stats/StatsPersistMsg.java b/application/src/main/java/org/thingsboard/server/actors/stats/StatsPersistMsg.java index 5e6f69278d..1f64c71629 100644 --- a/application/src/main/java/org/thingsboard/server/actors/stats/StatsPersistMsg.java +++ b/application/src/main/java/org/thingsboard/server/actors/stats/StatsPersistMsg.java @@ -28,10 +28,10 @@ import org.thingsboard.server.common.msg.TbActorMsg; @ToString public final class StatsPersistMsg implements TbActorMsg { - private long messagesProcessed; - private long errorsOccurred; - private TenantId tenantId; - private EntityId entityId; + private final long messagesProcessed; + private final long errorsOccurred; + private final TenantId tenantId; + private final EntityId entityId; @Override public MsgType getMsgType() { diff --git a/application/src/main/java/org/thingsboard/server/actors/stats/StatsPersistTick.java b/application/src/main/java/org/thingsboard/server/actors/stats/StatsPersistTick.java index f8edccf692..012e26e216 100644 --- a/application/src/main/java/org/thingsboard/server/actors/stats/StatsPersistTick.java +++ b/application/src/main/java/org/thingsboard/server/actors/stats/StatsPersistTick.java @@ -18,7 +18,7 @@ package org.thingsboard.server.actors.stats; import org.thingsboard.server.common.msg.MsgType; import org.thingsboard.server.common.msg.TbActorMsg; -public final class StatsPersistTick implements TbActorMsg{ +public final class StatsPersistTick implements TbActorMsg { @Override public MsgType getMsgType() { return MsgType.STATS_PERSIST_TICK_MSG; diff --git a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java index acd583e795..d6d1d6aa9a 100644 --- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java @@ -119,7 +119,7 @@ public class TenantActor extends RuleChainManagerActor { log.info("[{}] Processing missing Tenant msg: {}", tenantId, msg); if (msg.getMsgType().equals(MsgType.QUEUE_TO_RULE_ENGINE_MSG)) { QueueToRuleEngineMsg queueMsg = (QueueToRuleEngineMsg) msg; - queueMsg.getTbMsg().getCallback().onSuccess(); + queueMsg.getMsg().getCallback().onSuccess(); } else if (msg.getMsgType().equals(MsgType.TRANSPORT_TO_DEVICE_ACTOR_MSG)) { TransportToDeviceActorMsgWrapper transportMsg = (TransportToDeviceActorMsgWrapper) msg; transportMsg.getCallback().onSuccess(); @@ -177,7 +177,7 @@ public class TenantActor extends RuleChainManagerActor { log.warn("RECEIVED INVALID MESSAGE: {}", msg); return; } - TbMsg tbMsg = msg.getTbMsg(); + TbMsg tbMsg = msg.getMsg(); if (apiUsageState.isReExecEnabled()) { if (tbMsg.getRuleChainId() == null) { if (getRootChainActor() != null) { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index 98e46d5a50..106ca40c94 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -181,7 +181,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< new TbMsgPackCallback(id, tenantId, ctx, stats.getTimer(tenantId, SUCCESSFUL_STATUS), stats.getTimer(tenantId, FAILED_STATUS)) : new TbMsgPackCallback(id, tenantId, ctx); try { - if (toRuleEngineMsg.getTbMsg() != null && !toRuleEngineMsg.getTbMsg().isEmpty()) { + if (!toRuleEngineMsg.getTbMsg().isEmpty()) { forwardToRuleEngineActor(configuration.getName(), tenantId, toRuleEngineMsg, callback); } else { callback.onSuccess(); @@ -209,6 +209,9 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< if (statsEnabled) { stats.log(result, decision.isCommit()); } + + ctx.cleanup(); + if (decision.isCommit()) { submitStrategy.stop(); break; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContext.java b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContext.java index 4ccdce8ff8..d7a064c4ca 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContext.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContext.java @@ -147,4 +147,10 @@ public class TbMsgPackProcessingContext { .forEach(info -> log.info("[{}][{}] execution count: {}. {}", queueName, info.getRuleNodeId(), info.getExecutionCount(), info.getLabel())); } } + + public void cleanup() { + pendingMap.clear(); + successMap.clear(); + failedMap.clear(); + } } diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/ToDeviceRpcRequestActorMsg.java b/application/src/main/java/org/thingsboard/server/service/rpc/ToDeviceRpcRequestActorMsg.java index b75cf4d7ca..64a08b2ef4 100644 --- a/application/src/main/java/org/thingsboard/server/service/rpc/ToDeviceRpcRequestActorMsg.java +++ b/application/src/main/java/org/thingsboard/server/service/rpc/ToDeviceRpcRequestActorMsg.java @@ -31,6 +31,8 @@ import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; @RequiredArgsConstructor public class ToDeviceRpcRequestActorMsg implements ToDeviceActorNotificationMsg { + private static final long serialVersionUID = -8592877558138716589L; + @Getter private final String serviceId; @Getter diff --git a/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java b/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java index d542bdbe61..74d81c0307 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java @@ -34,6 +34,8 @@ import java.util.UUID; @Data public class TransportToDeviceActorMsgWrapper implements TbActorMsg, DeviceAwareMsg, TenantAwareMsg, Serializable { + private static final long serialVersionUID = 7191333353202935941L; + private final TenantId tenantId; private final DeviceId deviceId; private final TransportToDeviceActorMsg msg; diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbActor.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbActor.java index 4acadaff2c..0d2ec046a0 100644 --- a/common/actor/src/main/java/org/thingsboard/server/actors/TbActor.java +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbActor.java @@ -30,7 +30,7 @@ public interface TbActor { } default InitFailureStrategy onInitFailure(int attempt, Throwable t) { - return InitFailureStrategy.retryWithDelay(5000 * attempt); + return InitFailureStrategy.retryWithDelay(5000L * attempt); } default ProcessFailureStrategy onProcessFailure(Throwable t) { diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorException.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorException.java index 66ccb3f81c..86329cc641 100644 --- a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorException.java +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorException.java @@ -17,6 +17,8 @@ package org.thingsboard.server.actors; public class TbActorException extends Exception { + private static final long serialVersionUID = 8209771144711980882L; + public TbActorException(String message, Throwable cause) { super(message, cause); } diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java index 225b9351ad..689a657d6d 100644 --- a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java @@ -18,6 +18,7 @@ package org.thingsboard.server.actors; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.msg.TbActorMsg; +import org.thingsboard.server.common.msg.TbActorStopReason; import java.util.List; import java.util.concurrent.ConcurrentLinkedQueue; @@ -49,6 +50,7 @@ public final class TbActorMailbox implements TbActorCtx { private final AtomicBoolean busy = new AtomicBoolean(FREE); private final AtomicBoolean ready = new AtomicBoolean(NOT_READY); private final AtomicBoolean destroyInProgress = new AtomicBoolean(); + private volatile TbActorStopReason stopReason; public void initActor() { dispatcher.getExecutor().execute(() -> tryInit(1)); @@ -70,6 +72,7 @@ public final class TbActorMailbox implements TbActorCtx { InitFailureStrategy strategy = actor.onInitFailure(attempt, t); if (strategy.isStop() || (settings.getMaxActorInitAttempts() > 0 && attemptIdx > settings.getMaxActorInitAttempts())) { log.info("[{}] Failed to init actor, attempt {}, going to stop attempts.", selfId, attempt, t); + stopReason = TbActorStopReason.INIT_FAILED; system.stop(selfId); } else if (strategy.getRetryDelay() > 0) { log.info("[{}] Failed to init actor, attempt {}, going to retry in attempts in {}ms", selfId, attempt, strategy.getRetryDelay()); @@ -84,12 +87,16 @@ public final class TbActorMailbox implements TbActorCtx { } private void enqueue(TbActorMsg msg, boolean highPriority) { - if (highPriority) { - highPriorityMsgs.add(msg); + if (!destroyInProgress.get()) { + if (highPriority) { + highPriorityMsgs.add(msg); + } else { + normalPriorityMsgs.add(msg); + } + tryProcessQueue(true); } else { - normalPriorityMsgs.add(msg); + msg.onTbActorStopped(stopReason); } - tryProcessQueue(true); } private void tryProcessQueue(boolean newMsg) { @@ -180,11 +187,16 @@ public final class TbActorMailbox implements TbActorCtx { } public void destroy() { + if (stopReason == null) { + stopReason = TbActorStopReason.STOPPED; + } destroyInProgress.set(true); dispatcher.getExecutor().execute(() -> { try { ready.set(NOT_READY); actor.destroy(); + highPriorityMsgs.forEach(msg -> msg.onTbActorStopped(stopReason)); + normalPriorityMsgs.forEach(msg -> msg.onTbActorStopped(stopReason)); } catch (Throwable t) { log.warn("[{}] Failed to destroy actor: {}", selfId, t); } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java index 495d5bf5ea..ffc574d5c2 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java @@ -66,11 +66,6 @@ public enum MsgType { */ REMOTE_TO_RULE_CHAIN_TELL_NEXT_MSG, - /** - * Message that is sent by RuleActor implementation to RuleActor itself to log the error. - */ - RULE_TO_SELF_ERROR_MSG, - /** * Message that is sent by RuleActor implementation to RuleActor itself to process the message. */ diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbActorMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbActorMsg.java index fe35d47d88..b7ba251704 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbActorMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbActorMsg.java @@ -22,4 +22,12 @@ public interface TbActorMsg { MsgType getMsgType(); + /** + * Executed when the target TbActor is stopped or destroyed. + * For example, rule node failed to initialize or removed from rule chain. + * Implementation should cleanup the resources. + */ + default void onTbActorStopped(TbActorStopReason reason) { + } + } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbActorStopReason.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbActorStopReason.java new file mode 100644 index 0000000000..32035a7352 --- /dev/null +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbActorStopReason.java @@ -0,0 +1,22 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.msg; + +public enum TbActorStopReason { + + INIT_FAILED, STOPPED + +} diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeToSelfErrorMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbRuleEngineActorMsg.java similarity index 56% rename from application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeToSelfErrorMsg.java rename to common/message/src/main/java/org/thingsboard/server/common/msg/TbRuleEngineActorMsg.java index b62c513780..64bb772b4f 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeToSelfErrorMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbRuleEngineActorMsg.java @@ -13,25 +13,18 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.actors.ruleChain; +package org.thingsboard.server.common.msg; -import lombok.Data; -import org.thingsboard.server.common.msg.MsgType; -import org.thingsboard.server.common.msg.TbActorMsg; -import org.thingsboard.server.common.msg.TbMsg; +import lombok.EqualsAndHashCode; +import lombok.Getter; -/** - * Created by ashvayka on 19.03.18. - */ -@Data -final class RuleNodeToSelfErrorMsg implements TbActorMsg { +@EqualsAndHashCode +public abstract class TbRuleEngineActorMsg implements TbActorMsg { - private final TbMsg msg; - private final Throwable error; + @Getter + protected final TbMsg msg; - @Override - public MsgType getMsgType() { - return MsgType.RULE_TO_SELF_ERROR_MSG; + public TbRuleEngineActorMsg(TbMsg msg) { + this.msg = msg; } - } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/QueueToRuleEngineMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/QueueToRuleEngineMsg.java index ded2fa722c..7648fdab6d 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/QueueToRuleEngineMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/QueueToRuleEngineMsg.java @@ -15,32 +15,58 @@ */ package org.thingsboard.server.common.msg.queue; -import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.ToString; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.MsgType; -import org.thingsboard.server.common.msg.TbActorMsg; +import org.thingsboard.server.common.msg.TbActorStopReason; import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbRuleEngineActorMsg; -import java.io.Serializable; import java.util.Set; /** * Created by ashvayka on 15.03.18. */ -@Data -public final class QueueToRuleEngineMsg implements TbActorMsg { +@ToString +@EqualsAndHashCode(callSuper = true) +public final class QueueToRuleEngineMsg extends TbRuleEngineActorMsg { + @Getter private final TenantId tenantId; - private final TbMsg tbMsg; + @Getter private final Set relationTypes; + @Getter private final String failureMessage; + public QueueToRuleEngineMsg(TenantId tenantId, TbMsg tbMsg, Set relationTypes, String failureMessage) { + super(tbMsg); + this.tenantId = tenantId; + this.relationTypes = relationTypes; + this.failureMessage = failureMessage; + } + @Override public MsgType getMsgType() { return MsgType.QUEUE_TO_RULE_ENGINE_MSG; } + @Override + public void onTbActorStopped(TbActorStopReason reason) { + String message; + if (msg.getRuleChainId() != null) { + message = reason == TbActorStopReason.STOPPED ? + String.format("Rule chain [%s] stopped", msg.getRuleChainId().getId()) : + String.format("Failed to initialize rule chain [%s]!", msg.getRuleChainId().getId()); + } else { + message = reason == TbActorStopReason.STOPPED ? "Rule chain stopped" : "Failed to initialize rule chain!"; + } + msg.getCallback().onFailure(new RuleEngineException(message)); + } + public boolean isTellNext() { return relationTypes != null && !relationTypes.isEmpty(); } + } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/RuleNodeException.java b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/RuleNodeException.java index 8e68d1f94c..cd073f3a6e 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/RuleNodeException.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/RuleNodeException.java @@ -24,6 +24,9 @@ import org.thingsboard.server.common.data.rule.RuleNode; @Slf4j public class RuleNodeException extends RuleEngineException { + + private static final long serialVersionUID = -1776681087370749776L; + @Getter private final String ruleChainName; @Getter @@ -33,6 +36,7 @@ public class RuleNodeException extends RuleEngineException { @Getter private final RuleNodeId ruleNodeId; + public RuleNodeException(String message, String ruleChainName, RuleNode ruleNode) { super(message); this.ruleChainName = ruleChainName; diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 7a7319f742..5f182c2c34 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; +import org.thingsboard.server.common.data.rule.RuleNode; import org.thingsboard.server.common.data.rule.RuleNodeState; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -161,6 +162,10 @@ public interface TbContext { RuleNodeId getSelfId(); + RuleNode getSelf(); + + String getRuleChainName(); + TenantId getTenantId(); AttributesService getAttributesService(); diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/msg/DeviceCredentialsUpdateNotificationMsg.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/msg/DeviceCredentialsUpdateNotificationMsg.java index ff1f1e40b1..ea776f8918 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/msg/DeviceCredentialsUpdateNotificationMsg.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/msg/DeviceCredentialsUpdateNotificationMsg.java @@ -31,6 +31,8 @@ import java.util.Set; @Data public class DeviceCredentialsUpdateNotificationMsg implements ToDeviceActorNotificationMsg { + private static final long serialVersionUID = -3956907402411126990L; + private final TenantId tenantId; private final DeviceId deviceId; diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/msg/DeviceNameOrTypeUpdateMsg.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/msg/DeviceNameOrTypeUpdateMsg.java index e92f0f2842..0418986998 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/msg/DeviceNameOrTypeUpdateMsg.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/msg/DeviceNameOrTypeUpdateMsg.java @@ -24,6 +24,9 @@ import org.thingsboard.server.common.msg.MsgType; @Data @AllArgsConstructor public class DeviceNameOrTypeUpdateMsg implements ToDeviceActorNotificationMsg { + + private static final long serialVersionUID = -5738949227650536685L; + private final TenantId tenantId; private final DeviceId deviceId; private final String deviceName; diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java index ffa9da7e8b..9282d2c357 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java @@ -18,6 +18,7 @@ package org.thingsboard.rule.engine.profile; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.AdditionalAnswers; @@ -29,15 +30,22 @@ import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.alarm.AlarmSeverity; import org.thingsboard.server.common.data.device.profile.AlarmCondition; import org.thingsboard.server.common.data.device.profile.AlarmRule; import org.thingsboard.server.common.data.device.profile.DeviceProfileAlarm; import org.thingsboard.server.common.data.device.profile.DeviceProfileData; +import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.query.DynamicValue; +import org.thingsboard.server.common.data.query.DynamicValueSourceType; import org.thingsboard.server.common.data.query.EntityKey; import org.thingsboard.server.common.data.query.EntityKeyType; import org.thingsboard.server.common.data.query.EntityKeyValueType; @@ -48,13 +56,19 @@ import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.common.msg.session.SessionMsgType; +import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.device.DeviceService; +import org.thingsboard.server.dao.model.sql.AttributeKvCompositeKey; +import org.thingsboard.server.dao.model.sql.AttributeKvEntity; import org.thingsboard.server.dao.timeseries.TimeseriesService; import java.util.Collections; +import java.util.List; +import java.util.Optional; import java.util.TreeMap; import java.util.UUID; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.verify; @RunWith(MockitoJUnitRunner.class) @@ -74,9 +88,12 @@ public class TbDeviceProfileNodeTest { private RuleEngineAlarmService alarmService; @Mock private DeviceService deviceService; + @Mock + private AttributesService attributesService; private TenantId tenantId = new TenantId(UUID.randomUUID()); private DeviceId deviceId = new DeviceId(UUID.randomUUID()); + private CustomerId customerId = new CustomerId(UUID.randomUUID()); private DeviceProfileId deviceProfileId = new DeviceProfileId(UUID.randomUUID()); @Test @@ -186,12 +203,248 @@ public class TbDeviceProfileNodeTest { } + @Test + public void testCurrentDeviceAttributeForDynamicValue() throws Exception { + init(); + + DeviceProfile deviceProfile = new DeviceProfile(); + deviceProfile.setId(deviceProfileId); + DeviceProfileData deviceProfileData = new DeviceProfileData(); + + Device device = new Device(); + device.setId(deviceId); + device.setCustomerId(customerId); + + AttributeKvCompositeKey compositeKey = new AttributeKvCompositeKey( + EntityType.TENANT, deviceId.getId(), "SERVER_SCOPE", "greaterAttribute" + ); + + AttributeKvEntity attributeKvEntity = new AttributeKvEntity(); + attributeKvEntity.setId(compositeKey); + attributeKvEntity.setLongValue(30L); + attributeKvEntity.setLastUpdateTs(0L); + + AttributeKvEntry entry = attributeKvEntity.toData(); + ListenableFuture> listListenableFutureWithLess = + Futures.immediateFuture(Collections.singletonList(entry)); + + KeyFilter highTempFilter = new KeyFilter(); + highTempFilter.setKey(new EntityKey(EntityKeyType.TIME_SERIES, "temperature")); + highTempFilter.setValueType(EntityKeyValueType.NUMERIC); + NumericFilterPredicate highTemperaturePredicate = new NumericFilterPredicate(); + highTemperaturePredicate.setOperation(NumericFilterPredicate.NumericOperation.GREATER); + highTemperaturePredicate.setValue(new FilterPredicateValue<>( + 0.0, + null, + new DynamicValue<>(DynamicValueSourceType.CURRENT_DEVICE, "greaterAttribute") + )); + highTempFilter.setPredicate(highTemperaturePredicate); + AlarmCondition alarmCondition = new AlarmCondition(); + alarmCondition.setCondition(Collections.singletonList(highTempFilter)); + AlarmRule alarmRule = new AlarmRule(); + alarmRule.setCondition(alarmCondition); + DeviceProfileAlarm dpa = new DeviceProfileAlarm(); + dpa.setId("highTemperatureAlarmID"); + dpa.setAlarmType("highTemperatureAlarm"); + dpa.setCreateRules(new TreeMap<>(Collections.singletonMap(AlarmSeverity.CRITICAL, alarmRule))); + + deviceProfileData.setAlarms(Collections.singletonList(dpa)); + deviceProfile.setProfileData(deviceProfileData); + + Mockito.when(cache.get(tenantId, deviceId)).thenReturn(deviceProfile); + Mockito.when(timeseriesService.findLatest(tenantId, deviceId, Collections.singleton("temperature"))) + .thenReturn(Futures.immediateFuture(Collections.emptyList())); + Mockito.when(alarmService.findLatestByOriginatorAndType(tenantId, deviceId, "highTemperatureAlarm")) + .thenReturn(Futures.immediateFuture(null)); + Mockito.when(alarmService.createOrUpdateAlarm(Mockito.any())).thenAnswer(AdditionalAnswers.returnsFirstArg()); + Mockito.when(ctx.getAttributesService()).thenReturn(attributesService); + Mockito.when(attributesService.find(eq(tenantId), eq(deviceId), Mockito.anyString(), Mockito.anySet())) + .thenReturn(listListenableFutureWithLess); + + TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), ""); + Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyString())) + .thenReturn(theMsg); + + ObjectNode data = mapper.createObjectNode(); + data.put("temperature", 35); + TbMsg msg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, new TbMsgMetaData(), + TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null); + + node.onMsg(ctx, msg); + verify(ctx).tellSuccess(msg); + verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); + } + + @Test + public void testCurrentCustomersAttributeForDynamicValue() throws Exception { + init(); + + DeviceProfile deviceProfile = new DeviceProfile(); + deviceProfile.setId(deviceProfileId); + DeviceProfileData deviceProfileData = new DeviceProfileData(); + + Device device = new Device(); + device.setId(deviceId); + device.setCustomerId(customerId); + + AttributeKvCompositeKey compositeKey = new AttributeKvCompositeKey( + EntityType.TENANT, deviceId.getId(), "SERVER_SCOPE", "lessAttribute" + ); + + AttributeKvEntity attributeKvEntity = new AttributeKvEntity(); + attributeKvEntity.setId(compositeKey); + attributeKvEntity.setLongValue(30L); + attributeKvEntity.setLastUpdateTs(0L); + + AttributeKvEntry entry = attributeKvEntity.toData(); + ListenableFuture> listListenableFutureWithLess = + Futures.immediateFuture(Collections.singletonList(entry)); + ListenableFuture> optionalListenableFutureWithLess = + Futures.immediateFuture(Optional.of(entry)); + + KeyFilter lowTempFilter = new KeyFilter(); + lowTempFilter.setKey(new EntityKey(EntityKeyType.TIME_SERIES, "temperature")); + lowTempFilter.setValueType(EntityKeyValueType.NUMERIC); + NumericFilterPredicate lowTempPredicate = new NumericFilterPredicate(); + lowTempPredicate.setOperation(NumericFilterPredicate.NumericOperation.LESS); + lowTempPredicate.setValue( + new FilterPredicateValue<>( + 20.0, + null, + new DynamicValue<>(DynamicValueSourceType.CURRENT_CUSTOMER, "lessAttribute")) + ); + lowTempFilter.setPredicate(lowTempPredicate); + AlarmCondition alarmCondition = new AlarmCondition(); + alarmCondition.setCondition(Collections.singletonList(lowTempFilter)); + AlarmRule alarmRule = new AlarmRule(); + alarmRule.setCondition(alarmCondition); + DeviceProfileAlarm dpa = new DeviceProfileAlarm(); + dpa.setId("lesstempID"); + dpa.setAlarmType("lessTemperatureAlarm"); + dpa.setCreateRules(new TreeMap<>(Collections.singletonMap(AlarmSeverity.CRITICAL, alarmRule))); + + deviceProfileData.setAlarms(Collections.singletonList(dpa)); + deviceProfile.setProfileData(deviceProfileData); + + Mockito.when(cache.get(tenantId, deviceId)).thenReturn(deviceProfile); + Mockito.when(timeseriesService.findLatest(tenantId, deviceId, Collections.singleton("temperature"))) + .thenReturn(Futures.immediateFuture(Collections.emptyList())); + Mockito.when(alarmService.findLatestByOriginatorAndType(tenantId, deviceId, "lessTemperatureAlarm")) + .thenReturn(Futures.immediateFuture(null)); + Mockito.when(alarmService.createOrUpdateAlarm(Mockito.any())).thenAnswer(AdditionalAnswers.returnsFirstArg()); + Mockito.when(ctx.getAttributesService()).thenReturn(attributesService); + Mockito.when(attributesService.find(eq(tenantId), eq(deviceId), Mockito.anyString(), Mockito.anySet())) + .thenReturn(listListenableFutureWithLess); + Mockito.when(ctx.getDeviceService().findDeviceById(tenantId, deviceId)) + .thenReturn(device); + Mockito.when(attributesService.find(eq(tenantId), eq(customerId), eq(DataConstants.SERVER_SCOPE), Mockito.anyString())) + .thenReturn(optionalListenableFutureWithLess); + + TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), ""); + Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyString())) + .thenReturn(theMsg); + + ObjectNode data = mapper.createObjectNode(); + data.put("temperature", 25); + TbMsg msg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, new TbMsgMetaData(), + TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null); + + node.onMsg(ctx, msg); + verify(ctx).tellSuccess(msg); + verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); + } + + @Test + public void testCurrentTenantAttributeForDynamicValue() throws Exception { + init(); + + DeviceProfile deviceProfile = new DeviceProfile(); + DeviceProfileData deviceProfileData = new DeviceProfileData(); + + Device device = new Device(); + device.setId(deviceId); + device.setCustomerId(customerId); + + AttributeKvCompositeKey compositeKey = new AttributeKvCompositeKey( + EntityType.TENANT, deviceId.getId(), "SERVER_SCOPE", "lessAttribute" + ); + + AttributeKvEntity attributeKvEntity = new AttributeKvEntity(); + attributeKvEntity.setId(compositeKey); + attributeKvEntity.setLongValue(50L); + attributeKvEntity.setLastUpdateTs(0L); + + AttributeKvEntry entry = attributeKvEntity.toData(); + ListenableFuture> listListenableFutureWithLess = + Futures.immediateFuture(Collections.singletonList(entry)); + ListenableFuture> optionalListenableFutureWithLess = + Futures.immediateFuture(Optional.of(entry)); + + KeyFilter lowTempFilter = new KeyFilter(); + lowTempFilter.setKey(new EntityKey(EntityKeyType.TIME_SERIES, "temperature")); + lowTempFilter.setValueType(EntityKeyValueType.NUMERIC); + NumericFilterPredicate lowTempPredicate = new NumericFilterPredicate(); + lowTempPredicate.setOperation(NumericFilterPredicate.NumericOperation.LESS); + lowTempPredicate.setValue( + new FilterPredicateValue<>( + 32.0, + null, + new DynamicValue<>(DynamicValueSourceType.CURRENT_TENANT, "lessAttribute")) + ); + lowTempFilter.setPredicate(lowTempPredicate); + AlarmCondition alarmCondition = new AlarmCondition(); + alarmCondition.setCondition(Collections.singletonList(lowTempFilter)); + AlarmRule alarmRule = new AlarmRule(); + alarmRule.setCondition(alarmCondition); + DeviceProfileAlarm dpa = new DeviceProfileAlarm(); + dpa.setId("lesstempID"); + dpa.setAlarmType("lessTemperatureAlarm"); + dpa.setCreateRules(new TreeMap<>(Collections.singletonMap(AlarmSeverity.CRITICAL, alarmRule))); + + deviceProfileData.setAlarms(Collections.singletonList(dpa)); + deviceProfile.setProfileData(deviceProfileData); + + Mockito.when(cache.get(tenantId, deviceId)).thenReturn(deviceProfile); + Mockito.when(timeseriesService.findLatest(tenantId, deviceId, Collections.singleton("temperature"))) + .thenReturn(Futures.immediateFuture(Collections.emptyList())); + Mockito.when(alarmService.findLatestByOriginatorAndType(tenantId, deviceId, "lessTemperatureAlarm")) + .thenReturn(Futures.immediateFuture(null)); + Mockito.when(alarmService.createOrUpdateAlarm(Mockito.any())) + .thenAnswer(AdditionalAnswers.returnsFirstArg()); + Mockito.when(ctx.getAttributesService()).thenReturn(attributesService); + Mockito.when(attributesService.find(eq(tenantId), eq(deviceId), Mockito.anyString(), Mockito.anySet())) + .thenReturn(listListenableFutureWithLess); + Mockito.when(attributesService.find(eq(tenantId), eq(tenantId), eq(DataConstants.SERVER_SCOPE), Mockito.anyString())) + .thenReturn(optionalListenableFutureWithLess); + + TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), ""); + Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.anyString())) + .thenReturn(theMsg); + + ObjectNode data = mapper.createObjectNode(); + data.put("temperature", 40); + TbMsg msg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, new TbMsgMetaData(), + TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null); + + node.onMsg(ctx, msg); + verify(ctx).tellSuccess(msg); + verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); + } + private void init() throws TbNodeException { + + UUID uuid = new UUID(6041557255264276971L, -9019477126543226049L); + System.out.println(uuid); + Mockito.when(ctx.getTenantId()).thenReturn(tenantId); Mockito.when(ctx.getDeviceProfileCache()).thenReturn(cache); Mockito.when(ctx.getTimeseriesService()).thenReturn(timeseriesService); Mockito.when(ctx.getAlarmService()).thenReturn(alarmService); Mockito.when(ctx.getDeviceService()).thenReturn(deviceService); + Mockito.when(ctx.getAttributesService()).thenReturn(attributesService); TbNodeConfiguration nodeConfiguration = new TbNodeConfiguration(mapper.createObjectNode()); node = new TbDeviceProfileNode(); node.init(ctx, nodeConfiguration);