From fb722c5915da9673535528a9b7904aacc497fb15 Mon Sep 17 00:00:00 2001 From: Andrew Shvayka Date: Thu, 26 Apr 2018 12:25:38 +0300 Subject: [PATCH 1/3] Queue usage implementation --- .../server/actors/ActorSystemContext.java | 5 ++ .../device/DeviceActorMessageProcessor.java | 53 ++++++------ .../actors/device/PendingSessionMsgData.java | 3 + .../actors/ruleChain/DefaultTbContext.java | 1 - .../RuleChainActorMessageProcessor.java | 84 ++++++++++++------- .../ruleChain/RuleChainToRuleChainMsg.java | 40 +++++++++ .../actors/shared/ComponentMsgProcessor.java | 26 +++++- .../controller/RuleChainController.java | 2 +- .../service/script/NashornJsEngine.java | 2 +- .../server/common/msg/MsgType.java | 5 ++ .../thingsboard/server/common/msg/TbMsg.java | 42 +++++++--- common/message/src/main/proto/tbmsg.proto | 17 +++- .../queue/{MsqQueue.java => MsgQueue.java} | 2 +- .../server/dao/queue/QueueBenchmark.java | 15 ++-- ...raMsqQueue.java => CassandraMsgQueue.java} | 4 +- .../cassandra/UnprocessedMsgFilterTest.java | 4 +- .../impl/CassandraMsgRepositoryTest.java | 11 ++- .../rule/engine/action/TbAlarmNode.java | 10 +-- .../rule/engine/debug/TbMsgGeneratorNode.java | 8 +- .../rule/engine/mail/TbMsgToEmailNode.java | 6 +- .../rule/engine/mail/TbSendEmailNode.java | 3 - .../transform/TbChangeOriginatorNode.java | 2 +- .../rule/engine/action/TbAlarmNodeTest.java | 17 ++-- .../engine/filter/TbJsFilterNodeTest.java | 11 ++- .../engine/filter/TbJsSwitchNodeTest.java | 7 +- .../engine/mail/TbMsgToEmailNodeTest.java | 7 +- .../TbGetCustomerAttributeNodeTest.java | 22 +++-- .../transform/TbChangeOriginatorNodeTest.java | 17 +++- .../transform/TbTransformMsgNodeTest.java | 18 ++-- 29 files changed, 312 insertions(+), 132 deletions(-) create mode 100644 application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainToRuleChainMsg.java rename dao/src/main/java/org/thingsboard/server/dao/queue/{MsqQueue.java => MsgQueue.java} (97%) rename dao/src/main/java/org/thingsboard/server/dao/service/queue/cassandra/{CassandraMsqQueue.java => CassandraMsgQueue.java} (96%) diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index c1fd8782e5..fad7a1bee2 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -49,6 +49,7 @@ import org.thingsboard.server.dao.customer.CustomerService; import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.dao.event.EventService; import org.thingsboard.server.dao.plugin.PluginService; +import org.thingsboard.server.dao.queue.MsgQueue; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.dao.rule.RuleService; @@ -187,6 +188,10 @@ public class ActorSystemContext { @Getter private MailService mailService; + @Autowired + @Getter + private MsgQueue msgQueue; + @Value("${actors.session.sync.timeout}") @Getter private long syncSessionTimeout; diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 728fef1377..cd043bf51f 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -204,9 +204,13 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso void processQueueAck(ActorContext context, RuleEngineQueuePutAckMsg msg) { PendingSessionMsgData data = pendingMsgs.remove(msg.getId()); if (data != null && data.isReplyOnQueueAck()) { - logger.debug("[{}] Queue put [{}] ack detected!", deviceId, msg.getId()); - ToDeviceMsg toDeviceMsg = BasicStatusCodeResponse.onSuccess(data.getSessionMsgType(), data.getRequestId()); - sendMsgToSessionActor(new BasicToDeviceSessionActorMsg(toDeviceMsg, data.getSessionId()), data.getServerAddress()); + int remainingAcks = data.getAckMsgCount() - 1; + data.setAckMsgCount(remainingAcks); + logger.debug("[{}] Queue put [{}] ack detected. Remaining acks: {}!", deviceId, msg.getId(), remainingAcks); + if (remainingAcks == 0) { + ToDeviceMsg toDeviceMsg = BasicStatusCodeResponse.onSuccess(data.getSessionMsgType(), data.getRequestId()); + sendMsgToSessionActor(new BasicToDeviceSessionActorMsg(toDeviceMsg, data.getSessionId()), data.getServerAddress()); + } } } @@ -320,8 +324,10 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso kv.getStrValue().ifPresent(v -> json.addProperty(kv.getKey(), v)); } - TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), deviceId, defaultMetaData, TbMsgDataType.JSON, gson.toJson(json)); - pushToRuleEngineWithTimeout(context, tbMsg, src, request); + TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), deviceId, defaultMetaData, TbMsgDataType.JSON, gson.toJson(json), null, null, 0L); + PendingSessionMsgData msgData = new PendingSessionMsgData(src.getSessionId(), src.getServerAddress(), + SessionMsgType.POST_ATTRIBUTES_REQUEST, request.getRequestId(), true, 1); + pushToRuleEngineWithTimeout(context, tbMsg, msgData); } private void handlePostTelemetryRequest(ActorContext context, DeviceToDeviceActorMsg src) { @@ -329,10 +335,12 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso Map> tsData = request.getData(); - JsonArray json = new JsonArray(); + PendingSessionMsgData msgData = new PendingSessionMsgData(src.getSessionId(), src.getServerAddress(), + SessionMsgType.POST_TELEMETRY_REQUEST, request.getRequestId(), true, tsData.size()); + for (Map.Entry> entry : tsData.entrySet()) { - JsonObject ts = new JsonObject(); - ts.addProperty("ts", entry.getKey()); + JsonObject json = new JsonObject(); + json.addProperty("ts", entry.getKey()); JsonObject values = new JsonObject(); for (KvEntry kv : entry.getValue()) { kv.getBooleanValue().ifPresent(v -> values.addProperty(kv.getKey(), v)); @@ -340,12 +348,10 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso kv.getDoubleValue().ifPresent(v -> values.addProperty(kv.getKey(), v)); kv.getStrValue().ifPresent(v -> values.addProperty(kv.getKey(), v)); } - ts.add("values", values); - json.add(ts); + json.add("values", values); + TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, defaultMetaData, TbMsgDataType.JSON, gson.toJson(json), null, null, 0L); + pushToRuleEngineWithTimeout(context, tbMsg, msgData); } - - TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, defaultMetaData, TbMsgDataType.JSON, gson.toJson(json)); - pushToRuleEngineWithTimeout(context, tbMsg, src, request); } private void handleClientSideRPCRequest(ActorContext context, DeviceToDeviceActorMsg src) { @@ -357,8 +363,9 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso TbMsgMetaData requestMetaData = defaultMetaData.copy(); requestMetaData.putValue("requestId", Integer.toString(request.getRequestId())); - TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.TO_SERVER_RPC_REQUEST.name(), deviceId, requestMetaData, TbMsgDataType.JSON, gson.toJson(json)); - pushToRuleEngineWithTimeout(context, tbMsg, src, request); + TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.TO_SERVER_RPC_REQUEST.name(), deviceId, requestMetaData, TbMsgDataType.JSON, gson.toJson(json), null, null, 0L); + PendingSessionMsgData msgData = new PendingSessionMsgData(src.getSessionId(), src.getServerAddress(), SessionMsgType.TO_SERVER_RPC_REQUEST, request.getRequestId(), false, 1); + pushToRuleEngineWithTimeout(context, tbMsg, msgData); scheduleMsgWithDelay(context, new DeviceActorClientSideRpcTimeoutMsg(request.getRequestId(), systemContext.getClientSideRpcTimeout()), systemContext.getClientSideRpcTimeout()); toServerRpcPendingMap.put(request.getRequestId(), new ToServerRpcRequestMetadata(src.getSessionId(), src.getSessionType(), src.getServerAddress())); @@ -380,19 +387,15 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } } - private void pushToRuleEngineWithTimeout(ActorContext context, TbMsg tbMsg, DeviceToDeviceActorMsg src, FromDeviceRequestMsg fromDeviceRequestMsg) { - pushToRuleEngineWithTimeout(context, tbMsg, src, fromDeviceRequestMsg, true); - } - - private void pushToRuleEngineWithTimeout(ActorContext context, TbMsg tbMsg, DeviceToDeviceActorMsg src, FromDeviceRequestMsg fromDeviceRequestMsg, boolean replyOnAck) { - SessionMsgType sessionMsgType = fromDeviceRequestMsg.getMsgType(); - int requestId = fromDeviceRequestMsg.getRequestId(); + private void pushToRuleEngineWithTimeout(ActorContext context, TbMsg tbMsg, PendingSessionMsgData pendingMsgData) { + SessionMsgType sessionMsgType = pendingMsgData.getSessionMsgType(); + int requestId = pendingMsgData.getRequestId(); if (systemContext.isQueuePersistenceEnabled()) { - pendingMsgs.put(tbMsg.getId(), new PendingSessionMsgData(src.getSessionId(), src.getServerAddress(), sessionMsgType, requestId, replyOnAck)); + pendingMsgs.put(tbMsg.getId(), pendingMsgData); scheduleMsgWithDelay(context, new DeviceActorQueueTimeoutMsg(tbMsg.getId(), systemContext.getQueuePersistenceTimeout()), systemContext.getQueuePersistenceTimeout()); } else { - ToDeviceSessionActorMsg response = new BasicToDeviceSessionActorMsg(BasicStatusCodeResponse.onSuccess(sessionMsgType, requestId), src.getSessionId()); - sendMsgToSessionActor(response, src.getServerAddress()); + ToDeviceSessionActorMsg response = new BasicToDeviceSessionActorMsg(BasicStatusCodeResponse.onSuccess(sessionMsgType, requestId), pendingMsgData.getSessionId()); + sendMsgToSessionActor(response, pendingMsgData.getServerAddress()); } context.parent().tell(new DeviceActorToRuleEngineMsg(context.self(), tbMsg), context.self()); } diff --git a/application/src/main/java/org/thingsboard/server/actors/device/PendingSessionMsgData.java b/application/src/main/java/org/thingsboard/server/actors/device/PendingSessionMsgData.java index 2ce05b2a4e..23ad966049 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/PendingSessionMsgData.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/PendingSessionMsgData.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.actors.device; +import lombok.AllArgsConstructor; import lombok.Data; import org.thingsboard.server.common.data.id.SessionId; import org.thingsboard.server.common.msg.cluster.ServerAddress; @@ -26,6 +27,7 @@ import java.util.Optional; * Created by ashvayka on 17.04.18. */ @Data +@AllArgsConstructor public final class PendingSessionMsgData { private final SessionId sessionId; @@ -33,5 +35,6 @@ public final class PendingSessionMsgData { private final SessionMsgType sessionMsgType; private final int requestId; private final boolean replyOnQueueAck; + private int ackMsgCount; } 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 bf6132a2d3..16e79b6eeb 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 @@ -239,7 +239,6 @@ class DefaultTbContext implements TbContext { .build()); }); } - }; } } 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 f05b9a8607..847c49430e 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 @@ -19,6 +19,7 @@ import akka.actor.ActorContext; import akka.actor.ActorRef; import akka.actor.Props; import akka.event.LoggingAdapter; +import com.datastax.driver.core.utils.UUIDs; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.device.DeviceActorToRuleEngineMsg; import org.thingsboard.server.actors.device.RuleEngineQueuePutAckMsg; @@ -41,6 +42,7 @@ import org.thingsboard.server.common.msg.system.ServiceToRuleEngineMsg; import org.thingsboard.server.dao.rule.RuleChainService; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -52,6 +54,7 @@ import java.util.stream.Collectors; */ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor { + private static final long DEFAULT_CLUSTER_PARTITION = 0L; private final ActorRef parent; private final ActorRef self; private final Map nodeActors; @@ -83,6 +86,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor relations = service.getRuleNodeRelations(ruleNode.getId()); - for (EntityRelation relation : relations) { - if (relation.getTo().getEntityType() == EntityType.RULE_NODE) { - RuleNodeCtx ruleNodeCtx = nodeActors.get(new RuleNodeId(relation.getTo().getId())); - if (ruleNodeCtx == null) { - throw new IllegalArgumentException("Rule Node [" + relation.getFrom() + "] has invalid relation to Rule node [" + relation.getTo() + "]"); + if (relations.size() == 0) { + nodeRoutes.put(ruleNode.getId(), Collections.emptyList()); + } else { + for (EntityRelation relation : relations) { + if (relation.getTo().getEntityType() == EntityType.RULE_NODE) { + RuleNodeCtx ruleNodeCtx = nodeActors.get(new RuleNodeId(relation.getTo().getId())); + if (ruleNodeCtx == null) { + throw new IllegalArgumentException("Rule Node [" + relation.getFrom() + "] has invalid relation to Rule node [" + relation.getTo() + "]"); + } } + nodeRoutes.computeIfAbsent(ruleNode.getId(), k -> new ArrayList<>()) + .add(new RuleNodeRelation(ruleNode.getId(), relation.getTo(), relation.getType())); } - nodeRoutes.computeIfAbsent(ruleNode.getId(), k -> new ArrayList<>()) - .add(new RuleNodeRelation(ruleNode.getId(), relation.getTo(), relation.getType())); } } @@ -161,45 +169,61 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor pushMsgToNode(firstNode, msg)); } - public void onDeviceActorToRuleEngineMsg(DeviceActorToRuleEngineMsg envelope) { + void onDeviceActorToRuleEngineMsg(DeviceActorToRuleEngineMsg envelope) { checkActive(); - TbMsg tbMsg = envelope.getTbMsg(); - //TODO: push to queue and act on ack in async way - pushMsgToNode(firstNode, tbMsg); - envelope.getCallbackRef().tell(new RuleEngineQueuePutAckMsg(tbMsg.getId()), self); + putToQueue(envelope.getTbMsg(), msg -> { + pushMsgToNode(firstNode, msg); + envelope.getCallbackRef().tell(new RuleEngineQueuePutAckMsg(msg.getId()), self); + }); } void onTellNext(RuleNodeToRuleChainTellNextMsg envelope) { checkActive(); RuleNodeId originator = envelope.getOriginator(); String targetRelationType = envelope.getRelationType(); - List relations = nodeRoutes.get(originator); - if (relations == null) { - return; - } - boolean copy = relations.size() > 1; - for (RuleNodeRelation relation : relations) { - TbMsg msg = envelope.getMsg(); - if (copy) { - msg = msg.copy(); + List relations = nodeRoutes.get(originator).stream() + .filter(r -> targetRelationType == null || targetRelationType.equals(r.getType())) + .collect(Collectors.toList()); + + TbMsg msg = envelope.getMsg(); + int relationsCount = relations.size(); + if (relationsCount == 0) { + queue.ack(msg, msg.getRuleNodeId().getId(), msg.getClusterPartition()); + } else if (relationsCount == 1) { + for (RuleNodeRelation relation : relations) { + pushToTarget(msg, relation.getOut()); } - if (targetRelationType == null || targetRelationType.equalsIgnoreCase(relation.getType())) { - switch (relation.getOut().getEntityType()) { + } else { + for (RuleNodeRelation relation : relations) { + EntityId target = relation.getOut(); + switch (target.getEntityType()) { case RULE_NODE: - RuleNodeId targetRuleNodeId = new RuleNodeId(relation.getOut().getId()); - RuleNodeCtx targetRuleNode = nodeActors.get(targetRuleNodeId); - pushMsgToNode(targetRuleNode, msg); + RuleNodeId targetId = new RuleNodeId(target.getId()); + RuleNodeCtx targetNodeCtx = nodeActors.get(targetId); + TbMsg copy = msg.copy(UUIDs.timeBased(), entityId, targetId, DEFAULT_CLUSTER_PARTITION); + putToQueue(copy, queuedMsg -> pushMsgToNode(targetNodeCtx, queuedMsg)); break; case RULE_CHAIN: -// TODO: implement + parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, msg, true), self); break; } } + //TODO: Ideally this should happen in async way when all targets confirm that the copied messages are successfully written to corresponding target queues. + queue.ack(msg, msg.getRuleNodeId().getId(), msg.getClusterPartition()); + } + } + + private void pushToTarget(TbMsg msg, EntityId target) { + switch (target.getEntityType()) { + case RULE_NODE: + pushMsgToNode(nodeActors.get(new RuleNodeId(target.getId())), msg); + break; + case RULE_CHAIN: + parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, msg, false), self); + break; } } diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainToRuleChainMsg.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainToRuleChainMsg.java new file mode 100644 index 0000000000..f0fca215a8 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainToRuleChainMsg.java @@ -0,0 +1,40 @@ +/** + * Copyright © 2016-2018 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.actors.ruleChain; + +import lombok.Data; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.msg.MsgType; +import org.thingsboard.server.common.msg.TbActorMsg; +import org.thingsboard.server.common.msg.TbMsg; + +/** + * Created by ashvayka on 19.03.18. + */ +@Data +final class RuleChainToRuleChainMsg implements TbActorMsg { + + private final RuleChainId target; + private final RuleChainId source; + private final TbMsg msg; + private final boolean enqueue; + + @Override + public MsgType getMsgType() { + return MsgType.RULE_CHAIN_TO_RULE_CHAIN_MSG; + } +} diff --git a/application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java b/application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java index e25d3a7af9..a9cf7955fe 100644 --- a/application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java @@ -17,22 +17,32 @@ package org.thingsboard.server.actors.shared; import akka.actor.ActorContext; import akka.event.LoggingAdapter; +import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Futures; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.stats.StatsPersistTick; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.plugin.ComponentLifecycleState; +import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.cluster.ClusterEventMsg; +import org.thingsboard.server.dao.queue.MsgQueue; -public abstract class ComponentMsgProcessor extends AbstractContextAwareMsgProcessor { +import javax.annotation.Nullable; +import java.util.function.Consumer; + +public abstract class ComponentMsgProcessor extends AbstractContextAwareMsgProcessor { protected final TenantId tenantId; protected final T entityId; + protected final MsgQueue queue; protected ComponentLifecycleState state; protected ComponentMsgProcessor(ActorSystemContext systemContext, LoggingAdapter logger, TenantId tenantId, T id) { super(systemContext, logger); this.tenantId = tenantId; this.entityId = id; + this.queue = systemContext.getMsgQueue(); } public abstract void start(ActorContext context) throws Exception; @@ -75,4 +85,18 @@ public abstract class ComponentMsgProcessor extends AbstractContextAwareMsgPr throw new IllegalStateException("Rule chain is not active!"); } } + + protected void putToQueue(final TbMsg tbMsg, final Consumer onSuccess) { + Futures.addCallback(queue.put(tbMsg, entityId.getId(), 0), new FutureCallback() { + @Override + public void onSuccess(@Nullable Void result) { + onSuccess.accept(tbMsg); + } + + @Override + public void onFailure(Throwable t) { + logger.debug("Failed to push message [{}] to queue due to [{}]", tbMsg, t); + } + }); + } } diff --git a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java index 48c9cd5342..05e7836786 100644 --- a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java +++ b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java @@ -237,7 +237,7 @@ public class RuleChainController extends BaseController { ScriptEngine engine = null; try { engine = new NashornJsEngine(script, functionName, argNames); - TbMsg inMsg = new TbMsg(UUIDs.timeBased(), msgType, null, new TbMsgMetaData(metadata), data); + TbMsg inMsg = new TbMsg(UUIDs.timeBased(), msgType, null, new TbMsgMetaData(metadata), data, null, null, 0L); switch (scriptType) { case "update": output = msgToOutput(engine.executeUpdate(inMsg)); diff --git a/application/src/main/java/org/thingsboard/server/service/script/NashornJsEngine.java b/application/src/main/java/org/thingsboard/server/service/script/NashornJsEngine.java index d68f6fed29..8e689ec32d 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/NashornJsEngine.java +++ b/application/src/main/java/org/thingsboard/server/service/script/NashornJsEngine.java @@ -118,7 +118,7 @@ public class NashornJsEngine implements org.thingsboard.rule.engine.api.ScriptEn String newData = data != null ? data : msg.getData(); TbMsgMetaData newMetadata = metadata != null ? new TbMsgMetaData(metadata) : msg.getMetaData(); String newMessageType = !StringUtils.isEmpty(messageType) ? messageType : msg.getType(); - return new TbMsg(msg.getId(), newMessageType, msg.getOriginator(), newMetadata, newData); + return new TbMsg(msg.getId(), newMessageType, msg.getOriginator(), newMetadata, newData, msg.getRuleChainId(), msg.getRuleNodeId(), msg.getClusterPartition()); } catch (Throwable th) { th.printStackTrace(); throw new RuntimeException("Failed to unbind message data from javascript result", th); 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 74c08da043..d63456ea67 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 @@ -47,6 +47,11 @@ public enum MsgType { */ RULE_CHAIN_TO_RULE_MSG, + /** + * Message that is sent by RuleChainActor to other RuleChainActor with command to process TbMsg. + */ + RULE_CHAIN_TO_RULE_CHAIN_MSG, + /** * Message that is sent by RuleActor to RuleChainActor with command to process TbMsg by next nodes in chain. */ diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java index 1c7de3bf10..dd80bf785c 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java @@ -15,12 +15,13 @@ */ package org.thingsboard.server.common.msg; -import com.google.protobuf.ByteString; import com.google.protobuf.InvalidProtocolBufferException; import lombok.AllArgsConstructor; import lombok.Data; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityIdFactory; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.msg.gen.MsgProtos; import java.io.Serializable; @@ -41,22 +42,40 @@ public final class TbMsg implements Serializable { private final TbMsgDataType dataType; private final String data; - public TbMsg(UUID id, String type, EntityId originator, TbMsgMetaData metaData, String data) { + //The following fields are not persisted to DB, because they can always be recovered from the context; + private final RuleChainId ruleChainId; + private final RuleNodeId ruleNodeId; + private final long clusterPartition; + + public TbMsg(UUID id, String type, EntityId originator, TbMsgMetaData metaData, String data, + RuleChainId ruleChainId, RuleNodeId ruleNodeId, long clusterPartition) { this.id = id; this.type = type; this.originator = originator; this.metaData = metaData; - this.dataType = TbMsgDataType.JSON; this.data = data; + this.dataType = TbMsgDataType.JSON; + this.ruleChainId = ruleChainId; + this.ruleNodeId = ruleNodeId; + this.clusterPartition = clusterPartition; } public static ByteBuffer toBytes(TbMsg msg) { MsgProtos.TbMsgProto.Builder builder = MsgProtos.TbMsgProto.newBuilder(); builder.setId(msg.getId().toString()); builder.setType(msg.getType()); - if (msg.getOriginator() != null) { - builder.setEntityType(msg.getOriginator().getEntityType().name()); - builder.setEntityId(msg.getOriginator().getId().toString()); + builder.setEntityType(msg.getOriginator().getEntityType().name()); + builder.setEntityIdMSB(msg.getOriginator().getId().getMostSignificantBits()); + builder.setEntityIdLSB(msg.getOriginator().getId().getLeastSignificantBits()); + + if (msg.getRuleChainId() != null) { + builder.setRuleChainIdMSB(msg.getRuleChainId().getId().getMostSignificantBits()); + builder.setRuleChainIdLSB(msg.getRuleChainId().getId().getLeastSignificantBits()); + } + + if (msg.getRuleNodeId() != null) { + builder.setRuleNodeIdMSB(msg.getRuleNodeId().getId().getMostSignificantBits()); + builder.setRuleNodeIdLSB(msg.getRuleNodeId().getId().getLeastSignificantBits()); } if (msg.getMetaData() != null) { @@ -73,15 +92,18 @@ public final class TbMsg implements Serializable { try { MsgProtos.TbMsgProto proto = MsgProtos.TbMsgProto.parseFrom(buffer.array()); TbMsgMetaData metaData = new TbMsgMetaData(proto.getMetaData().getDataMap()); - EntityId entityId = EntityIdFactory.getByTypeAndId(proto.getEntityType(), proto.getEntityId()); + EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB())); + RuleChainId ruleChainId = new RuleChainId(new UUID(proto.getRuleChainIdMSB(), proto.getRuleChainIdLSB())); + RuleNodeId ruleNodeId = new RuleNodeId(new UUID(proto.getRuleNodeIdMSB(), proto.getRuleChainIdLSB())); TbMsgDataType dataType = TbMsgDataType.values()[proto.getDataType()]; - return new TbMsg(UUID.fromString(proto.getId()), proto.getType(), entityId, metaData, dataType, proto.getData()); + return new TbMsg(UUID.fromString(proto.getId()), proto.getType(), entityId, metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, proto.getClusterPartition()); } catch (InvalidProtocolBufferException e) { throw new IllegalStateException("Could not parse protobuf for TbMsg", e); } } - public TbMsg copy() { - return new TbMsg(id, type, originator, metaData.copy(), dataType, data); + public TbMsg copy(UUID newId, RuleChainId ruleChainId, RuleNodeId ruleNodeId, long clusterPartition) { + return new TbMsg(newId, type, originator, metaData, dataType, data, ruleChainId, ruleNodeId, clusterPartition); } + } diff --git a/common/message/src/main/proto/tbmsg.proto b/common/message/src/main/proto/tbmsg.proto index 4ce1fb69bc..60003dc420 100644 --- a/common/message/src/main/proto/tbmsg.proto +++ b/common/message/src/main/proto/tbmsg.proto @@ -27,10 +27,19 @@ message TbMsgProto { string id = 1; string type = 2; string entityType = 3; - string entityId = 4; + int64 entityIdMSB = 4; + int64 entityIdLSB = 5; - TbMsgMetaDataProto metaData = 5; + int64 ruleChainIdMSB = 6; + int64 ruleChainIdLSB = 7; + + int64 ruleNodeIdMSB = 8; + int64 ruleNodeIdLSB = 9; + int64 clusterPartition = 10; + + TbMsgMetaDataProto metaData = 11; + + int32 dataType = 12; + string data = 13; - int32 dataType = 6; - string data = 7; } \ No newline at end of file diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/MsqQueue.java b/dao/src/main/java/org/thingsboard/server/dao/queue/MsgQueue.java similarity index 97% rename from dao/src/main/java/org/thingsboard/server/dao/queue/MsqQueue.java rename to dao/src/main/java/org/thingsboard/server/dao/queue/MsgQueue.java index 345012892c..e49b89eb81 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/queue/MsqQueue.java +++ b/dao/src/main/java/org/thingsboard/server/dao/queue/MsgQueue.java @@ -20,7 +20,7 @@ import org.thingsboard.server.common.msg.TbMsg; import java.util.UUID; -public interface MsqQueue { +public interface MsgQueue { ListenableFuture put(TbMsg msg, UUID nodeId, long clusterPartition); diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/QueueBenchmark.java b/dao/src/main/java/org/thingsboard/server/dao/queue/QueueBenchmark.java index 85f42ae169..ef55bcb1de 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/queue/QueueBenchmark.java +++ b/dao/src/main/java/org/thingsboard/server/dao/queue/QueueBenchmark.java @@ -27,10 +27,9 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.CommandLineRunner; import org.springframework.boot.SpringApplication; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.ComponentScan; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -61,11 +60,11 @@ public class QueueBenchmark implements CommandLineRunner { } @Autowired - private MsqQueue msqQueue; + private MsgQueue msgQueue; @Override public void run(String... strings) throws Exception { - System.out.println("It works + " + msqQueue); + System.out.println("It works + " + msgQueue); long start = System.currentTimeMillis(); @@ -81,8 +80,8 @@ public class QueueBenchmark implements CommandLineRunner { try { TbMsg msg = randomMsg(); UUID nodeId = UUIDs.timeBased(); - ListenableFuture put = msqQueue.put(msg, nodeId, 100L); -// ListenableFuture put = msqQueue.ack(msg, nodeId, 100L); + ListenableFuture put = msgQueue.put(msg, nodeId, 100L); +// ListenableFuture put = msgQueue.ack(msg, nodeId, 100L); Futures.addCallback(put, new FutureCallback() { @Override public void onSuccess(@Nullable Void result) { @@ -126,7 +125,7 @@ public class QueueBenchmark implements CommandLineRunner { TbMsgMetaData metaData = new TbMsgMetaData(); metaData.putValue("key", "value"); String dataStr = "someContent"; - return new TbMsg(UUIDs.timeBased(), "type", null, metaData, TbMsgDataType.JSON, dataStr); + return new TbMsg(UUIDs.timeBased(), "type", null, metaData, TbMsgDataType.JSON, dataStr, new RuleChainId(UUIDs.timeBased()), new RuleNodeId(UUIDs.timeBased()), 0L); } @Bean diff --git a/dao/src/main/java/org/thingsboard/server/dao/service/queue/cassandra/CassandraMsqQueue.java b/dao/src/main/java/org/thingsboard/server/dao/service/queue/cassandra/CassandraMsgQueue.java similarity index 96% rename from dao/src/main/java/org/thingsboard/server/dao/service/queue/cassandra/CassandraMsqQueue.java rename to dao/src/main/java/org/thingsboard/server/dao/service/queue/cassandra/CassandraMsgQueue.java index 59e10d8f4c..361b18ce4b 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/service/queue/cassandra/CassandraMsqQueue.java +++ b/dao/src/main/java/org/thingsboard/server/dao/service/queue/cassandra/CassandraMsgQueue.java @@ -22,7 +22,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.thingsboard.server.common.msg.TbMsg; -import org.thingsboard.server.dao.queue.MsqQueue; +import org.thingsboard.server.dao.queue.MsgQueue; import org.thingsboard.server.dao.service.queue.cassandra.repository.AckRepository; import org.thingsboard.server.dao.service.queue.cassandra.repository.MsgRepository; import org.thingsboard.server.dao.util.NoSqlDao; @@ -33,7 +33,7 @@ import java.util.UUID; @Component @Slf4j @NoSqlDao -public class CassandraMsqQueue implements MsqQueue { +public class CassandraMsgQueue implements MsgQueue { @Autowired private MsgRepository msgRepository; diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/queue/cassandra/UnprocessedMsgFilterTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/queue/cassandra/UnprocessedMsgFilterTest.java index 3935c9b92e..caccc85932 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/queue/cassandra/UnprocessedMsgFilterTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/queue/cassandra/UnprocessedMsgFilterTest.java @@ -33,8 +33,8 @@ public class UnprocessedMsgFilterTest { public void acknowledgedMsgsAreFilteredOut() { UUID id1 = UUID.randomUUID(); UUID id2 = UUID.randomUUID(); - TbMsg msg1 = new TbMsg(id1, "T", null, null, null, null); - TbMsg msg2 = new TbMsg(id2, "T", null, null, null, null); + TbMsg msg1 = new TbMsg(id1, "T", null, null, null, null, null, null, 0L); + TbMsg msg2 = new TbMsg(id2, "T", null, null, null, null, null, null, 0L); List msgs = Lists.newArrayList(msg1, msg2); List acks = Lists.newArrayList(new MsgAck(id2, UUID.randomUUID(), 1L, 1L)); Collection actual = msgFilter.filter(msgs, acks); diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/queue/cassandra/repository/impl/CassandraMsgRepositoryTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/queue/cassandra/repository/impl/CassandraMsgRepositoryTest.java index a766aa48ce..ae252ae44a 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/queue/cassandra/repository/impl/CassandraMsgRepositoryTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/queue/cassandra/repository/impl/CassandraMsgRepositoryTest.java @@ -23,6 +23,8 @@ import org.junit.Before; import org.junit.Test; import org.springframework.beans.factory.annotation.Autowired; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -45,7 +47,8 @@ public class CassandraMsgRepositoryTest extends AbstractServiceTest { @Test public void msgCanBeSavedAndRead() throws ExecutionException, InterruptedException { - TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), null, TbMsgDataType.JSON, "0000"); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), null, TbMsgDataType.JSON, "0000", + new RuleChainId(UUIDs.timeBased()), new RuleNodeId(UUIDs.timeBased()), 0L); UUID nodeId = UUIDs.timeBased(); ListenableFuture future = msgRepository.save(msg, nodeId, 1L, 1L, 1L); future.get(); @@ -55,7 +58,8 @@ public class CassandraMsgRepositoryTest extends AbstractServiceTest { @Test public void expiredMsgsAreNotReturned() throws ExecutionException, InterruptedException { - TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), null, TbMsgDataType.JSON, "0000"); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), null, TbMsgDataType.JSON, "0000", + new RuleChainId(UUIDs.timeBased()), new RuleNodeId(UUIDs.timeBased()), 0L); UUID nodeId = UUIDs.timeBased(); ListenableFuture future = msgRepository.save(msg, nodeId, 2L, 2L, 2L); future.get(); @@ -68,7 +72,8 @@ public class CassandraMsgRepositoryTest extends AbstractServiceTest { TbMsgMetaData metaData = new TbMsgMetaData(); metaData.putValue("key", "value"); String dataStr = "someContent"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), metaData, TbMsgDataType.JSON, dataStr); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "type", new DeviceId(UUIDs.timeBased()), metaData, TbMsgDataType.JSON, dataStr, + new RuleChainId(UUIDs.timeBased()), new RuleNodeId(UUIDs.timeBased()), 0L); UUID nodeId = UUIDs.timeBased(); ListenableFuture future = msgRepository.save(msg, nodeId, 1L, 1L, 1L); future.get(); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAlarmNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAlarmNode.java index a242093efc..42ad237e48 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAlarmNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAlarmNode.java @@ -91,11 +91,11 @@ public class TbAlarmNode implements TbNode { if (alarmResult.alarm == null) { ctx.tellNext(msg, "False"); } else if (alarmResult.isCreated) { - ctx.tellNext(toAlarmMsg(alarmResult, msg), "Created"); + ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), "Created"); } else if (alarmResult.isUpdated) { - ctx.tellNext(toAlarmMsg(alarmResult, msg), "Updated"); + ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), "Updated"); } else if (alarmResult.isCleared) { - ctx.tellNext(toAlarmMsg(alarmResult, msg), "Cleared"); + ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), "Cleared"); } }, t -> ctx.tellError(msg, t)); @@ -176,7 +176,7 @@ public class TbAlarmNode implements TbNode { return ctx.getJsExecutor().executeAsync(() -> buildDetailsJsEngine.executeJson(msg)); } - private TbMsg toAlarmMsg(AlarmResult alarmResult, TbMsg originalMsg) { + private TbMsg toAlarmMsg(TbContext ctx, AlarmResult alarmResult, TbMsg originalMsg) { JsonNode jsonNodes = mapper.valueToTree(alarmResult.alarm); String data = jsonNodes.toString(); TbMsgMetaData metaData = originalMsg.getMetaData().copy(); @@ -187,7 +187,7 @@ public class TbAlarmNode implements TbNode { } else if (alarmResult.isCleared) { metaData.putValue(IS_CLEARED_ALARM, Boolean.TRUE.toString()); } - return new TbMsg(UUIDs.timeBased(), "ALARM", originalMsg.getOriginator(), metaData, data); + return ctx.newMsg("ALARM", originalMsg.getOriginator(), metaData, data); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java index 65b13710e3..2e95f4a835 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java @@ -78,18 +78,18 @@ public class TbMsgGeneratorNode implements TbNode { } private void sentTickMsg(TbContext ctx) { - TbMsg tickMsg = new TbMsg(UUIDs.timeBased(), TB_MSG_GENERATOR_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), ""); + TbMsg tickMsg = ctx.newMsg(TB_MSG_GENERATOR_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), ""); nextTickId = tickMsg.getId(); ctx.tellSelf(tickMsg, delay); } - protected ListenableFuture generate(TbContext ctx) { + private ListenableFuture generate(TbContext ctx) { return ctx.getJsExecutor().executeAsync(() -> { if (prevMsg == null) { - prevMsg = new TbMsg(UUIDs.timeBased(), "", originatorId, new TbMsgMetaData(), "{}"); + prevMsg = ctx.newMsg( "", originatorId, new TbMsgMetaData(), "{}"); } TbMsg generated = jsEngine.executeGenerate(prevMsg); - prevMsg = new TbMsg(UUIDs.timeBased(), generated.getType(), originatorId, generated.getMetaData(), generated.getData()); + prevMsg = ctx.newMsg(generated.getType(), originatorId, generated.getMetaData(), generated.getData()); return prevMsg; }); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNode.java index cae2058417..7c7e685678 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNode.java @@ -76,7 +76,7 @@ public class TbMsgToEmailNode implements TbNode { public void onMsg(TbContext ctx, TbMsg msg) { try { EmailPojo email = convert(msg); - TbMsg emailMsg = buildEmailMsg(msg, email); + TbMsg emailMsg = buildEmailMsg(ctx, msg, email); ctx.tellNext(emailMsg); } catch (Exception ex) { log.warn("Can not convert message to email " + ex.getMessage()); @@ -84,9 +84,9 @@ public class TbMsgToEmailNode implements TbNode { } } - private TbMsg buildEmailMsg(TbMsg msg, EmailPojo email) throws JsonProcessingException { + private TbMsg buildEmailMsg(TbContext ctx, TbMsg msg, EmailPojo email) throws JsonProcessingException { String emailJson = MAPPER.writeValueAsString(email); - return new TbMsg(UUIDs.timeBased(), SEND_EMAIL_TYPE, msg.getOriginator(), msg.getMetaData().copy(), emailJson); + return ctx.newMsg(SEND_EMAIL_TYPE, msg.getOriginator(), msg.getMetaData().copy(), emailJson); } private EmailPojo convert(TbMsg msg) throws IOException { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mail/TbSendEmailNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mail/TbSendEmailNode.java index 407476b5ae..0b74d6efa7 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mail/TbSendEmailNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mail/TbSendEmailNode.java @@ -63,8 +63,6 @@ public class TbSendEmailNode implements TbNode { } catch (Exception ex) { ctx.tellError(msg, ex); } - - } private EmailPojo getEmail(TbMsg msg) throws IOException { @@ -84,6 +82,5 @@ public class TbSendEmailNode implements TbNode { @Override public void destroy() { - } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNode.java index 05ad49c144..36287fea2c 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNode.java @@ -60,7 +60,7 @@ public class TbChangeOriginatorNode extends TbAbstractTransformNode { @Override protected ListenableFuture transform(TbContext ctx, TbMsg msg) { ListenableFuture newOriginator = getNewOriginator(ctx, msg.getOriginator()); - return Futures.transform(newOriginator, (Function) n -> new TbMsg(msg.getId(), msg.getType(), n, msg.getMetaData(), msg.getData())); + return Futures.transform(newOriginator, (Function) n -> ctx.newMsg(msg.getType(), n, msg.getMetaData(), msg.getData())); } private ListenableFuture getNewOriginator(TbContext ctx, EntityId original) { diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java index c0aec73693..28713038a3 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java @@ -35,6 +35,8 @@ import org.thingsboard.rule.engine.api.*; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -73,6 +75,9 @@ public class TbAlarmNodeTest { @Mock private ScriptEngine detailsJs; + private RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased()); + private RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); + private ListeningExecutor dbExecutor; private EntityId originator = new DeviceId(UUIDs.timeBased()); @@ -103,7 +108,7 @@ public class TbAlarmNodeTest { public void newAlarmCanBeCreated() throws ScriptException, IOException { initWithScript(); metaData.putValue("key", "value"); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson, ruleChainId, ruleNodeId, 0L); when(createJs.executeFilter(msg)).thenReturn(true); when(detailsJs.executeJson(msg)).thenReturn(null); @@ -143,7 +148,7 @@ public class TbAlarmNodeTest { public void shouldCreateScriptThrowsException() throws ScriptException { initWithScript(); metaData.putValue("key", "value"); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson, ruleChainId, ruleNodeId, 0L); when(createJs.executeFilter(msg)).thenThrow(new NotImplementedException("message")); @@ -165,7 +170,7 @@ public class TbAlarmNodeTest { public void buildDetailsThrowsException() throws ScriptException, IOException { initWithScript(); metaData.putValue("key", "value"); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson, ruleChainId, ruleNodeId, 0L); when(createJs.executeFilter(msg)).thenReturn(true); when(detailsJs.executeJson(msg)).thenThrow(new NotImplementedException("message")); @@ -191,7 +196,7 @@ public class TbAlarmNodeTest { public void ifAlarmClearedCreateNew() throws ScriptException, IOException { initWithScript(); metaData.putValue("key", "value"); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson, ruleChainId, ruleNodeId, 0L); Alarm clearedAlarm = Alarm.builder().status(CLEARED_ACK).build(); @@ -233,7 +238,7 @@ public class TbAlarmNodeTest { public void alarmCanBeUpdated() throws ScriptException, IOException { initWithScript(); metaData.putValue("key", "value"); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson, ruleChainId, ruleNodeId, 0L); long oldEndDate = System.currentTimeMillis(); Alarm activeAlarm = Alarm.builder().type("SomeType").tenantId(tenantId).originator(originator).status(ACTIVE_UNACK).severity(WARNING).endTs(oldEndDate).build(); @@ -279,7 +284,7 @@ public class TbAlarmNodeTest { public void alarmCanBeCleared() throws ScriptException, IOException { initWithScript(); metaData.putValue("key", "value"); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson, ruleChainId, ruleNodeId, 0L); long oldEndDate = System.currentTimeMillis(); Alarm activeAlarm = Alarm.builder().type("SomeType").tenantId(tenantId).originator(originator).status(ACTIVE_UNACK).severity(WARNING).endTs(oldEndDate).build(); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsFilterNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsFilterNodeTest.java index 08a22f085d..d0b13ebc2e 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsFilterNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsFilterNodeTest.java @@ -27,6 +27,8 @@ import org.mockito.Mock; import org.mockito.runners.MockitoJUnitRunner; import org.mockito.stubbing.Answer; import org.thingsboard.rule.engine.api.*; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -48,10 +50,13 @@ public class TbJsFilterNodeTest { @Mock private ScriptEngine scriptEngine; + private RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased()); + private RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); + @Test public void falseEvaluationDoNotSendMsg() throws TbNodeException, ScriptException { initWithScript(); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, new TbMsgMetaData(), "{}"); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); mockJsExecutor(); when(scriptEngine.executeFilter(msg)).thenReturn(false); @@ -64,7 +69,7 @@ public class TbJsFilterNodeTest { public void exceptionInJsThrowsException() throws TbNodeException, ScriptException { initWithScript(); TbMsgMetaData metaData = new TbMsgMetaData(); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, "{}"); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, "{}", ruleChainId, ruleNodeId, 0L); mockJsExecutor(); when(scriptEngine.executeFilter(msg)).thenThrow(new ScriptException("error")); @@ -77,7 +82,7 @@ public class TbJsFilterNodeTest { public void metadataConditionCanBeTrue() throws TbNodeException, ScriptException { initWithScript(); TbMsgMetaData metaData = new TbMsgMetaData(); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, "{}"); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, "{}", ruleChainId, ruleNodeId, 0L); mockJsExecutor(); when(scriptEngine.executeFilter(msg)).thenReturn(true); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsSwitchNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsSwitchNodeTest.java index a495124afc..cc4c2972ce 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsSwitchNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/filter/TbJsSwitchNodeTest.java @@ -28,6 +28,8 @@ import org.mockito.Mock; import org.mockito.runners.MockitoJUnitRunner; import org.mockito.stubbing.Answer; import org.thingsboard.rule.engine.api.*; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -51,6 +53,9 @@ public class TbJsSwitchNodeTest { @Mock private ScriptEngine scriptEngine; + private RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased()); + private RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); + @Test public void multipleRoutesAreAllowed() throws TbNodeException, ScriptException { initWithScript(); @@ -59,7 +64,7 @@ public class TbJsSwitchNodeTest { metaData.putValue("humidity", "99"); String rawJson = "{\"name\": \"Vit\", \"passed\": 5}"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson, ruleChainId, ruleNodeId, 0L); mockJsExecutor(); when(scriptEngine.executeSwitch(msg)).thenReturn(Sets.newHashSet("one", "three")); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNodeTest.java index 877047ce3a..2e5d947d18 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNodeTest.java @@ -27,6 +27,8 @@ import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -48,12 +50,15 @@ public class TbMsgToEmailNodeTest { private TbMsgMetaData metaData = new TbMsgMetaData(); private String rawJson = "{\"name\": \"temp\", \"passed\": 5 , \"complex\": {\"val\":12, \"count\":100}}"; + private RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased()); + private RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); + @Test public void msgCanBeConverted() throws IOException { initWithScript(); metaData.putValue("username", "oreo"); metaData.putValue("userEmail", "user@email.io"); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", originator, metaData, rawJson, ruleChainId, ruleNodeId, 0L); emailNode.onMsg(ctx, msg); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/TbGetCustomerAttributeNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/TbGetCustomerAttributeNodeTest.java index e26312b7ca..158b7cd2b9 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/TbGetCustomerAttributeNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/TbGetCustomerAttributeNodeTest.java @@ -34,6 +34,8 @@ import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.kv.*; import org.thingsboard.server.common.msg.TbMsg; @@ -77,6 +79,9 @@ public class TbGetCustomerAttributeNodeTest { private TbMsg msg; + private RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased()); + private RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); + @Before public void init() throws TbNodeException { TbGetEntityAttrNodeConfiguration config = new TbGetEntityAttrNodeConfiguration(); @@ -98,7 +103,8 @@ public class TbGetCustomerAttributeNodeTest { User user = new User(); user.setCustomerId(customerId); - msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), "{}"); + + msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); when(ctx.getUserService()).thenReturn(userService); when(userService.findUserByIdAsync(userId)).thenReturn(Futures.immediateFuture(user)); @@ -123,7 +129,7 @@ public class TbGetCustomerAttributeNodeTest { User user = new User(); user.setCustomerId(customerId); - msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), "{}"); + msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); when(ctx.getUserService()).thenReturn(userService); when(userService.findUserByIdAsync(userId)).thenReturn(Futures.immediateFuture(user)); @@ -148,7 +154,7 @@ public class TbGetCustomerAttributeNodeTest { User user = new User(); user.setCustomerId(customerId); - msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), "{}"); + msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); when(ctx.getUserService()).thenReturn(userService); when(userService.findUserByIdAsync(userId)).thenReturn(Futures.immediateFuture(null)); @@ -166,7 +172,7 @@ public class TbGetCustomerAttributeNodeTest { @Test public void customerAttributeAddedInMetadata() { CustomerId customerId = new CustomerId(UUIDs.timeBased()); - msg = new TbMsg(UUIDs.timeBased(), "CUSTOMER", customerId, new TbMsgMetaData(), "{}"); + msg = new TbMsg(UUIDs.timeBased(), "CUSTOMER", customerId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); entityAttributeFetched(customerId); } @@ -177,7 +183,7 @@ public class TbGetCustomerAttributeNodeTest { User user = new User(); user.setCustomerId(customerId); - msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), "{}"); + msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); when(ctx.getUserService()).thenReturn(userService); when(userService.findUserByIdAsync(userId)).thenReturn(Futures.immediateFuture(user)); @@ -192,7 +198,7 @@ public class TbGetCustomerAttributeNodeTest { Asset asset = new Asset(); asset.setCustomerId(customerId); - msg = new TbMsg(UUIDs.timeBased(), "USER", assetId, new TbMsgMetaData(), "{}"); + msg = new TbMsg(UUIDs.timeBased(), "USER", assetId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); when(ctx.getAssetService()).thenReturn(assetService); when(assetService.findAssetByIdAsync(assetId)).thenReturn(Futures.immediateFuture(asset)); @@ -207,7 +213,7 @@ public class TbGetCustomerAttributeNodeTest { Device device = new Device(); device.setCustomerId(customerId); - msg = new TbMsg(UUIDs.timeBased(), "USER", deviceId, new TbMsgMetaData(), "{}"); + msg = new TbMsg(UUIDs.timeBased(), "USER", deviceId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); when(ctx.getDeviceService()).thenReturn(deviceService); when(deviceService.findDeviceByIdAsync(deviceId)).thenReturn(Futures.immediateFuture(device)); @@ -234,7 +240,7 @@ public class TbGetCustomerAttributeNodeTest { Device device = new Device(); device.setCustomerId(customerId); - msg = new TbMsg(UUIDs.timeBased(), "USER", deviceId, new TbMsgMetaData(), "{}"); + msg = new TbMsg(UUIDs.timeBased(), "USER", deviceId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); when(ctx.getDeviceService()).thenReturn(deviceService); when(deviceService.findDeviceByIdAsync(deviceId)).thenReturn(Futures.immediateFuture(device)); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeTest.java index 1b66433085..7f864ba109 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeTest.java @@ -29,6 +29,8 @@ import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.dao.asset.AssetService; @@ -57,7 +59,10 @@ public class TbChangeOriginatorNodeTest { Asset asset = new Asset(); asset.setCustomerId(customerId); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "ASSET", assetId, new TbMsgMetaData(), "{}"); + RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased()); + RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); + + TbMsg msg = new TbMsg(UUIDs.timeBased(), "ASSET", assetId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); when(ctx.getAssetService()).thenReturn(assetService); when(assetService.findAssetByIdAsync(assetId)).thenReturn(Futures.immediateFuture(asset)); @@ -78,7 +83,10 @@ public class TbChangeOriginatorNodeTest { Asset asset = new Asset(); asset.setCustomerId(customerId); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "ASSET", assetId, new TbMsgMetaData(), "{}"); + RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased()); + RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); + + TbMsg msg = new TbMsg(UUIDs.timeBased(), "ASSET", assetId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); when(ctx.getAssetService()).thenReturn(assetService); when(assetService.findAssetByIdAsync(assetId)).thenReturn(Futures.immediateFuture(asset)); @@ -99,7 +107,10 @@ public class TbChangeOriginatorNodeTest { Asset asset = new Asset(); asset.setCustomerId(customerId); - TbMsg msg = new TbMsg(UUIDs.timeBased(), "ASSET", assetId, new TbMsgMetaData(), "{}"); + RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased()); + RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); + + TbMsg msg = new TbMsg(UUIDs.timeBased(), "ASSET", assetId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); when(ctx.getAssetService()).thenReturn(assetService); when(assetService.findAssetByIdAsync(assetId)).thenReturn(Futures.immediateFailedFuture(new IllegalStateException("wrong"))); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbTransformMsgNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbTransformMsgNodeTest.java index b904d7e95f..a29bc9055b 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbTransformMsgNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbTransformMsgNodeTest.java @@ -27,6 +27,8 @@ import org.mockito.Mock; import org.mockito.runners.MockitoJUnitRunner; import org.mockito.stubbing.Answer; import org.thingsboard.rule.engine.api.*; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -56,8 +58,10 @@ public class TbTransformMsgNodeTest { metaData.putValue("temp", "7"); String rawJson = "{\"passed\": 5}"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson); - TbMsg transformedMsg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, "{new}"); + RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased()); + RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson, ruleChainId, ruleNodeId, 0L); + TbMsg transformedMsg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, "{new}", ruleChainId, ruleNodeId, 0L); mockJsExecutor(); when(scriptEngine.executeUpdate(msg)).thenReturn(transformedMsg); @@ -77,8 +81,10 @@ public class TbTransformMsgNodeTest { metaData.putValue("temp", "7"); String rawJson = "{\"passed\": 5"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson); - TbMsg transformedMsg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, "{new}"); + RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased()); + RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson, ruleChainId, ruleNodeId, 0L); + TbMsg transformedMsg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, "{new}", ruleChainId, ruleNodeId, 0L); mockJsExecutor(); when(scriptEngine.executeUpdate(msg)).thenReturn(transformedMsg); @@ -97,7 +103,9 @@ public class TbTransformMsgNodeTest { metaData.putValue("temp", "7"); String rawJson = "{\"passed\": 5"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson); + RuleChainId ruleChainId = new RuleChainId(UUIDs.timeBased()); + RuleNodeId ruleNodeId = new RuleNodeId(UUIDs.timeBased()); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson, ruleChainId, ruleNodeId, 0L); mockJsExecutor(); when(scriptEngine.executeUpdate(msg)).thenThrow(new IllegalStateException("error")); From a2d2b559ac1bd2a94a0359fc92f210b5e79625d5 Mon Sep 17 00:00:00 2001 From: Andrew Shvayka Date: Fri, 27 Apr 2018 18:49:22 +0300 Subject: [PATCH 2/3] Queue put and ack --- .../actors/ruleChain/DefaultTbContext.java | 3 +- .../RuleChainActorMessageProcessor.java | 14 ++- .../actors/shared/ComponentMsgProcessor.java | 1 + ...tractMqttServerSideRpcIntegrationTest.java | 2 - .../MqttServerSideRpcSqlIntegrationTest.java | 2 +- ...AbstractRuleEngineFlowIntegrationTest.java | 2 +- ...actRuleEngineLifecycleIntegrationTest.java | 3 +- .../service/script/NashornJsEngineTest.java | 14 +-- .../dao/sql/queue/DummySqlMsgQueue.java | 50 +++++++++++ .../rule/engine/filter/TbJsSwitchNode.java | 2 +- .../rule/engine/action/TbAlarmNodeTest.java | 89 +++++++++++-------- .../engine/mail/TbMsgToEmailNodeTest.java | 20 +++-- .../transform/TbChangeOriginatorNodeTest.java | 26 +++--- 13 files changed, 155 insertions(+), 73 deletions(-) create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/queue/DummySqlMsgQueue.java 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 16e79b6eeb..40b32d3072 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 @@ -53,7 +53,6 @@ import java.util.function.Consumer; */ class DefaultTbContext implements TbContext { - private static final Function, ? extends Void> LIST_VOID_FUNCTION = v -> null; private final ActorSystemContext mainCtx; private final RuleNodeCtx nodeCtx; @@ -120,7 +119,7 @@ class DefaultTbContext implements TbContext { @Override public TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data) { - return new TbMsg(UUIDs.timeBased(), type, originator, metaData, data); + return new TbMsg(UUIDs.timeBased(), type, originator, metaData, data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId(), 0L); } @Override 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 847c49430e..fb7bc4902d 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 @@ -169,12 +169,12 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor pushMsgToNode(firstNode, msg)); + putToQueue(enrichWithRuleChainId(envelope.getTbMsg()), msg -> pushMsgToNode(firstNode, msg)); } void onDeviceActorToRuleEngineMsg(DeviceActorToRuleEngineMsg envelope) { checkActive(); - putToQueue(envelope.getTbMsg(), msg -> { + putToQueue(enrichWithRuleChainId(envelope.getTbMsg()), msg -> { pushMsgToNode(firstNode, msg); envelope.getCallbackRef().tell(new RuleEngineQueuePutAckMsg(msg.getId()), self); }); @@ -185,7 +185,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor relations = nodeRoutes.get(originator).stream() - .filter(r -> targetRelationType == null || targetRelationType.equals(r.getType())) + .filter(r -> targetRelationType == null || targetRelationType.equalsIgnoreCase(r.getType())) .collect(Collectors.toList()); TbMsg msg = envelope.getMsg(); @@ -212,7 +212,8 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor extends Abstract } protected void putToQueue(final TbMsg tbMsg, final Consumer onSuccess) { + EntityId entityId = tbMsg.getRuleNodeId() != null ? tbMsg.getRuleNodeId() : tbMsg.getRuleChainId(); Futures.addCallback(queue.put(tbMsg, entityId.getId(), 0), new FutureCallback() { @Override public void onSuccess(@Nullable Void result) { diff --git a/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java b/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java index af477640e6..b732e66f48 100644 --- a/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java @@ -80,7 +80,6 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractC } } - @Ignore @Test public void testServerMqttOneWayRpc() throws Exception { Device device = new Device(); @@ -107,7 +106,6 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractC Assert.assertTrue(StringUtils.isEmpty(result)); } - @Ignore @Test public void testServerMqttOneWayRpcDeviceOffline() throws Exception { Device device = new Device(); diff --git a/application/src/test/java/org/thingsboard/server/mqtt/rpc/sql/MqttServerSideRpcSqlIntegrationTest.java b/application/src/test/java/org/thingsboard/server/mqtt/rpc/sql/MqttServerSideRpcSqlIntegrationTest.java index 7c9c0586f0..d2287f04a9 100644 --- a/application/src/test/java/org/thingsboard/server/mqtt/rpc/sql/MqttServerSideRpcSqlIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/mqtt/rpc/sql/MqttServerSideRpcSqlIntegrationTest.java @@ -15,7 +15,7 @@ */ package org.thingsboard.server.mqtt.rpc.sql; -import org.thingsboard.server.dao.service.DaoNoSqlTest; +import org.junit.Ignore; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.mqtt.rpc.AbstractMqttServerSideRpcIntegrationTest; diff --git a/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java index f45e30335f..2d56772df3 100644 --- a/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java @@ -150,7 +150,7 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule "CUSTOM", device.getId(), new TbMsgMetaData(), - "{}"); + "{}", null, null, 0L); actorService.onMsg(new ServiceToRuleEngineMsg(savedTenant.getId(), tbMsg)); Thread.sleep(3000); diff --git a/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java index 29e8b73561..18583fdc7e 100644 --- a/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java @@ -138,7 +138,8 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac "CUSTOM", device.getId(), new TbMsgMetaData(), - "{}"); + "{}", + null, null, 0L); actorService.onMsg(new ServiceToRuleEngineMsg(savedTenant.getId(), tbMsg)); Thread.sleep(3000); diff --git a/application/src/test/java/org/thingsboard/server/service/script/NashornJsEngineTest.java b/application/src/test/java/org/thingsboard/server/service/script/NashornJsEngineTest.java index e6a48e2838..1981287f6d 100644 --- a/application/src/test/java/org/thingsboard/server/service/script/NashornJsEngineTest.java +++ b/application/src/test/java/org/thingsboard/server/service/script/NashornJsEngineTest.java @@ -42,7 +42,7 @@ public class NashornJsEngineTest { metaData.putValue("humidity", "99"); String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson, null, null, 0L); TbMsg actual = scriptEngine.executeUpdate(msg); assertEquals("70", actual.getMetaData().getValue("temp")); @@ -57,7 +57,7 @@ public class NashornJsEngineTest { metaData.putValue("humidity", "99"); String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson, null, null, 0L); TbMsg actual = scriptEngine.executeUpdate(msg); assertEquals("94", actual.getMetaData().getValue("newAttr")); @@ -72,7 +72,7 @@ public class NashornJsEngineTest { metaData.putValue("humidity", "99"); String rawJson = "{\"name\":\"Vit\",\"passed\": 5,\"bigObj\":{\"prop\":42}}"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson, null, null, 0L); TbMsg actual = scriptEngine.executeUpdate(msg); @@ -89,7 +89,7 @@ public class NashornJsEngineTest { metaData.putValue("humidity", "99"); String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson, null, null, 0L); assertFalse(scriptEngine.executeFilter(msg)); } @@ -102,7 +102,7 @@ public class NashornJsEngineTest { metaData.putValue("humidity", "99"); String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson, null, null, 0L); assertTrue(scriptEngine.executeFilter(msg)); } @@ -122,7 +122,7 @@ public class NashornJsEngineTest { metaData.putValue("humidity", "99"); String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson, null, null, 0L); Set actual = scriptEngine.executeSwitch(msg); assertEquals(Sets.newHashSet("one"), actual); } @@ -143,7 +143,7 @@ public class NashornJsEngineTest { metaData.putValue("humidity", "99"); String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson); + TbMsg msg = new TbMsg(UUIDs.timeBased(), "USER", null, metaData, rawJson, null, null, 0L); Set actual = scriptEngine.executeSwitch(msg); assertEquals(Sets.newHashSet("one", "three"), actual); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/queue/DummySqlMsgQueue.java b/dao/src/main/java/org/thingsboard/server/dao/sql/queue/DummySqlMsgQueue.java new file mode 100644 index 0000000000..0281d409d2 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/queue/DummySqlMsgQueue.java @@ -0,0 +1,50 @@ +/** + * Copyright © 2016-2018 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.dao.sql.queue; + +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.dao.queue.MsgQueue; +import org.thingsboard.server.dao.util.SqlDao; + +import java.util.Collections; +import java.util.UUID; + +/** + * Created by ashvayka on 27.04.18. + */ +@Component +@Slf4j +@SqlDao +public class DummySqlMsgQueue implements MsgQueue { + @Override + public ListenableFuture put(TbMsg msg, UUID nodeId, long clusterPartition) { + return Futures.immediateFuture(null); + } + + @Override + public ListenableFuture ack(TbMsg msg, UUID nodeId, long clusterPartition) { + return Futures.immediateFuture(null); + } + + @Override + public Iterable findUnprocessed(UUID nodeId, long clusterPartition) { + return Collections.emptyList(); + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsSwitchNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsSwitchNode.java index 3c6704b013..4ee8bd5461 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsSwitchNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsSwitchNode.java @@ -61,7 +61,7 @@ public class TbJsSwitchNode implements TbNode { ctx.tellNext(msg, nextRelations); } - @Override + @Override public void destroy() { if (jsEngine != null) { jsEngine.destroy(); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java index 28713038a3..8213195711 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java @@ -118,17 +118,21 @@ public class TbAlarmNodeTest { node.onMsg(ctx, msg); - ArgumentCaptor captor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctx).tellNext(captor.capture(), eq("Created")); - TbMsg actualMsg = captor.getValue(); + verify(ctx).tellNext(any(), eq("Created")); - assertEquals("ALARM", actualMsg.getType()); - assertEquals(originator, actualMsg.getOriginator()); - assertEquals("value", actualMsg.getMetaData().getValue("key")); - assertEquals(Boolean.TRUE.toString(), actualMsg.getMetaData().getValue(IS_NEW_ALARM)); - assertNotSame(metaData, actualMsg.getMetaData()); + ArgumentCaptor typeCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor originatorCaptor = ArgumentCaptor.forClass(EntityId.class); + ArgumentCaptor metadataCaptor = ArgumentCaptor.forClass(TbMsgMetaData.class); + ArgumentCaptor dataCaptor = ArgumentCaptor.forClass(String.class); + verify(ctx).newMsg(typeCaptor.capture(), originatorCaptor.capture(), metadataCaptor.capture(), dataCaptor.capture()); - Alarm actualAlarm = new ObjectMapper().readValue(actualMsg.getData().getBytes(), Alarm.class); + assertEquals("ALARM", typeCaptor.getValue()); + assertEquals(originator, originatorCaptor.getValue()); + assertEquals("value", metadataCaptor.getValue().getValue("key")); + assertEquals(Boolean.TRUE.toString(), metadataCaptor.getValue().getValue(IS_NEW_ALARM)); + assertNotSame(metaData, metadataCaptor.getValue()); + + Alarm actualAlarm = new ObjectMapper().readValue(dataCaptor.getValue().getBytes(), Alarm.class); Alarm expectedAlarm = Alarm.builder() .tenantId(tenantId) .originator(originator) @@ -208,17 +212,22 @@ public class TbAlarmNodeTest { node.onMsg(ctx, msg); - ArgumentCaptor captor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctx).tellNext(captor.capture(), eq("Created")); - TbMsg actualMsg = captor.getValue(); + verify(ctx).tellNext(any(), eq("Created")); + + ArgumentCaptor typeCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor originatorCaptor = ArgumentCaptor.forClass(EntityId.class); + ArgumentCaptor metadataCaptor = ArgumentCaptor.forClass(TbMsgMetaData.class); + ArgumentCaptor dataCaptor = ArgumentCaptor.forClass(String.class); + verify(ctx).newMsg(typeCaptor.capture(), originatorCaptor.capture(), metadataCaptor.capture(), dataCaptor.capture()); + + assertEquals("ALARM", typeCaptor.getValue()); + assertEquals(originator, originatorCaptor.getValue()); + assertEquals("value", metadataCaptor.getValue().getValue("key")); + assertEquals(Boolean.TRUE.toString(), metadataCaptor.getValue().getValue(IS_NEW_ALARM)); + assertNotSame(metaData, metadataCaptor.getValue()); - assertEquals("ALARM", actualMsg.getType()); - assertEquals(originator, actualMsg.getOriginator()); - assertEquals("value", actualMsg.getMetaData().getValue("key")); - assertEquals(Boolean.TRUE.toString(), actualMsg.getMetaData().getValue(IS_NEW_ALARM)); - assertNotSame(metaData, actualMsg.getMetaData()); - Alarm actualAlarm = new ObjectMapper().readValue(actualMsg.getData().getBytes(), Alarm.class); + Alarm actualAlarm = new ObjectMapper().readValue(dataCaptor.getValue().getBytes(), Alarm.class); Alarm expectedAlarm = Alarm.builder() .tenantId(tenantId) .originator(originator) @@ -252,17 +261,21 @@ public class TbAlarmNodeTest { node.onMsg(ctx, msg); - ArgumentCaptor captor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctx).tellNext(captor.capture(), eq("Updated")); - TbMsg actualMsg = captor.getValue(); + verify(ctx).tellNext(any(), eq("Updated")); - assertEquals("ALARM", actualMsg.getType()); - assertEquals(originator, actualMsg.getOriginator()); - assertEquals("value", actualMsg.getMetaData().getValue("key")); - assertEquals(Boolean.TRUE.toString(), actualMsg.getMetaData().getValue(IS_EXISTING_ALARM)); - assertNotSame(metaData, actualMsg.getMetaData()); + ArgumentCaptor typeCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor originatorCaptor = ArgumentCaptor.forClass(EntityId.class); + ArgumentCaptor metadataCaptor = ArgumentCaptor.forClass(TbMsgMetaData.class); + ArgumentCaptor dataCaptor = ArgumentCaptor.forClass(String.class); + verify(ctx).newMsg(typeCaptor.capture(), originatorCaptor.capture(), metadataCaptor.capture(), dataCaptor.capture()); - Alarm actualAlarm = new ObjectMapper().readValue(actualMsg.getData().getBytes(), Alarm.class); + assertEquals("ALARM", typeCaptor.getValue()); + assertEquals(originator, originatorCaptor.getValue()); + assertEquals("value", metadataCaptor.getValue().getValue("key")); + assertEquals(Boolean.TRUE.toString(), metadataCaptor.getValue().getValue(IS_EXISTING_ALARM)); + assertNotSame(metaData, metadataCaptor.getValue()); + + Alarm actualAlarm = new ObjectMapper().readValue(dataCaptor.getValue().getBytes(), Alarm.class); assertTrue(activeAlarm.getEndTs() > oldEndDate); Alarm expectedAlarm = Alarm.builder() .tenantId(tenantId) @@ -298,17 +311,21 @@ public class TbAlarmNodeTest { node.onMsg(ctx, msg); - ArgumentCaptor captor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctx).tellNext(captor.capture(), eq("Cleared")); - TbMsg actualMsg = captor.getValue(); + verify(ctx).tellNext(any(), eq("Cleared")); + + ArgumentCaptor typeCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor originatorCaptor = ArgumentCaptor.forClass(EntityId.class); + ArgumentCaptor metadataCaptor = ArgumentCaptor.forClass(TbMsgMetaData.class); + ArgumentCaptor dataCaptor = ArgumentCaptor.forClass(String.class); + verify(ctx).newMsg(typeCaptor.capture(), originatorCaptor.capture(), metadataCaptor.capture(), dataCaptor.capture()); - assertEquals("ALARM", actualMsg.getType()); - assertEquals(originator, actualMsg.getOriginator()); - assertEquals("value", actualMsg.getMetaData().getValue("key")); - assertEquals(Boolean.TRUE.toString(), actualMsg.getMetaData().getValue(IS_CLEARED_ALARM)); - assertNotSame(metaData, actualMsg.getMetaData()); + assertEquals("ALARM", typeCaptor.getValue()); + assertEquals(originator, originatorCaptor.getValue()); + assertEquals("value", metadataCaptor.getValue().getValue("key")); + assertEquals(Boolean.TRUE.toString(), metadataCaptor.getValue().getValue(IS_CLEARED_ALARM)); + assertNotSame(metaData, metadataCaptor.getValue()); - Alarm actualAlarm = new ObjectMapper().readValue(actualMsg.getData().getBytes(), Alarm.class); + Alarm actualAlarm = new ObjectMapper().readValue(dataCaptor.getValue().getBytes(), Alarm.class); Alarm expectedAlarm = Alarm.builder() .tenantId(tenantId) .originator(originator) diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNodeTest.java index 2e5d947d18..fe19a6d245 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNodeTest.java @@ -36,7 +36,9 @@ import java.io.IOException; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotSame; +import static org.mockito.Matchers.any; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; @RunWith(MockitoJUnitRunner.class) public class TbMsgToEmailNodeTest { @@ -62,17 +64,19 @@ public class TbMsgToEmailNodeTest { emailNode.onMsg(ctx, msg); - ArgumentCaptor captor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctx).tellNext(captor.capture()); - TbMsg actualMsg = captor.getValue(); + ArgumentCaptor typeCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor originatorCaptor = ArgumentCaptor.forClass(EntityId.class); + ArgumentCaptor metadataCaptor = ArgumentCaptor.forClass(TbMsgMetaData.class); + ArgumentCaptor dataCaptor = ArgumentCaptor.forClass(String.class); + verify(ctx).newMsg(typeCaptor.capture(), originatorCaptor.capture(), metadataCaptor.capture(), dataCaptor.capture()); - assertEquals("SEND_EMAIL", actualMsg.getType()); - assertEquals(originator, actualMsg.getOriginator()); - assertEquals("oreo", actualMsg.getMetaData().getValue("username")); - assertNotSame(metaData, actualMsg.getMetaData()); + assertEquals("SEND_EMAIL", typeCaptor.getValue()); + assertEquals(originator, originatorCaptor.getValue()); + assertEquals("oreo", metadataCaptor.getValue().getValue("username")); + assertNotSame(metaData, metadataCaptor.getValue()); - EmailPojo actual = new ObjectMapper().readValue(actualMsg.getData().getBytes(), EmailPojo.class); + EmailPojo actual = new ObjectMapper().readValue(dataCaptor.getValue().getBytes(), EmailPojo.class); EmailPojo expected = new EmailPojo.EmailPojoBuilder() .from("test@mail.org") diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeTest.java index 7f864ba109..1f75ed364c 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNodeTest.java @@ -29,6 +29,7 @@ import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.msg.TbMsg; @@ -68,11 +69,14 @@ public class TbChangeOriginatorNodeTest { when(assetService.findAssetByIdAsync(assetId)).thenReturn(Futures.immediateFuture(asset)); node.onMsg(ctx, msg); - ArgumentCaptor captor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctx).tellNext(captor.capture()); - TbMsg actualMsg = captor.getValue(); - assertEquals(customerId, actualMsg.getOriginator()); - assertEquals(msg.getId(), actualMsg.getId()); + + ArgumentCaptor typeCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor originatorCaptor = ArgumentCaptor.forClass(EntityId.class); + ArgumentCaptor metadataCaptor = ArgumentCaptor.forClass(TbMsgMetaData.class); + ArgumentCaptor dataCaptor = ArgumentCaptor.forClass(String.class); + verify(ctx).newMsg(typeCaptor.capture(), originatorCaptor.capture(), metadataCaptor.capture(), dataCaptor.capture()); + + assertEquals(customerId, originatorCaptor.getValue()); } @Test @@ -92,11 +96,13 @@ public class TbChangeOriginatorNodeTest { when(assetService.findAssetByIdAsync(assetId)).thenReturn(Futures.immediateFuture(asset)); node.onMsg(ctx, msg); - ArgumentCaptor captor = ArgumentCaptor.forClass(TbMsg.class); - verify(ctx).spawn(captor.capture()); - TbMsg actualMsg = captor.getValue(); - assertEquals(customerId, actualMsg.getOriginator()); - assertEquals(msg.getId(), actualMsg.getId()); + ArgumentCaptor typeCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor originatorCaptor = ArgumentCaptor.forClass(EntityId.class); + ArgumentCaptor metadataCaptor = ArgumentCaptor.forClass(TbMsgMetaData.class); + ArgumentCaptor dataCaptor = ArgumentCaptor.forClass(String.class); + verify(ctx).newMsg(typeCaptor.capture(), originatorCaptor.capture(), metadataCaptor.capture(), dataCaptor.capture()); + + assertEquals(customerId, originatorCaptor.getValue()); } @Test From 647ccafc5c4878804306dcd2020ed9338bae81f1 Mon Sep 17 00:00:00 2001 From: Andrew Shvayka Date: Fri, 27 Apr 2018 19:45:57 +0300 Subject: [PATCH 3/3] Build fix after merge --- .../controller/BaseComponentDescriptorControllerTest.java | 4 ++-- .../mqtt/rpc/sql/MqttServerSideRpcSqlIntegrationTest.java | 1 - 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseComponentDescriptorControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseComponentDescriptorControllerTest.java index d04fb640f4..3a07d2989d 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseComponentDescriptorControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/BaseComponentDescriptorControllerTest.java @@ -36,7 +36,7 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers. public abstract class BaseComponentDescriptorControllerTest extends AbstractControllerTest { - private static final int AMOUNT_OF_DEFAULT_FILTER_NODES = 3; + private static final int AMOUNT_OF_DEFAULT_FILTER_NODES = 4; private Tenant savedTenant; private User tenantAdmin; @@ -87,7 +87,7 @@ public abstract class BaseComponentDescriptorControllerTest extends AbstractCont }); Assert.assertNotNull(descriptors); - Assert.assertEquals(AMOUNT_OF_DEFAULT_FILTER_NODES, descriptors.size()); + Assert.assertTrue(descriptors.size() >= AMOUNT_OF_DEFAULT_FILTER_NODES); for (ComponentType type : ComponentType.values()) { doGet("/api/components/" + type).andExpect(status().isOk()); diff --git a/application/src/test/java/org/thingsboard/server/mqtt/rpc/sql/MqttServerSideRpcSqlIntegrationTest.java b/application/src/test/java/org/thingsboard/server/mqtt/rpc/sql/MqttServerSideRpcSqlIntegrationTest.java index d2287f04a9..f3d69bacd5 100644 --- a/application/src/test/java/org/thingsboard/server/mqtt/rpc/sql/MqttServerSideRpcSqlIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/mqtt/rpc/sql/MqttServerSideRpcSqlIntegrationTest.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.mqtt.rpc.sql; -import org.junit.Ignore; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.mqtt.rpc.AbstractMqttServerSideRpcIntegrationTest;