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 c1a2c17f9c..b888bc33e4 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 @@ -25,7 +25,9 @@ import org.thingsboard.rule.engine.api.RuleEngineRpcService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.ScriptEngine; import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.api.TbRelationTypes; import org.thingsboard.server.actors.ActorSystemContext; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.RuleNodeId; @@ -98,11 +100,11 @@ class DefaultTbContext implements TbContext { } @Override - public void tellError(TbMsg msg, Throwable th) { + public void tellFailure(TbMsg msg, Throwable th) { if (nodeCtx.getSelf().isDebugMode()) { - mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg, "", th); + mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg, TbRelationTypes.FAILURE, th); } - nodeCtx.getSelfActor().tell(new RuleNodeToSelfErrorMsg(msg, th), nodeCtx.getSelfActor()); + nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getId(), Collections.singleton(TbRelationTypes.FAILURE), msg), nodeCtx.getSelfActor()); } @Override diff --git a/application/src/main/java/org/thingsboard/server/service/component/AnnotationComponentDiscoveryService.java b/application/src/main/java/org/thingsboard/server/service/component/AnnotationComponentDiscoveryService.java index d49a63d683..096b2bc4e6 100644 --- a/application/src/main/java/org/thingsboard/server/service/component/AnnotationComponentDiscoveryService.java +++ b/application/src/main/java/org/thingsboard/server/service/component/AnnotationComponentDiscoveryService.java @@ -29,6 +29,8 @@ import org.springframework.stereotype.Service; import org.thingsboard.rule.engine.api.NodeConfiguration; import org.thingsboard.rule.engine.api.NodeDefinition; import org.thingsboard.rule.engine.api.RuleNode; +import org.thingsboard.rule.engine.api.TbRelationTypes; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.plugin.ComponentDescriptor; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.dao.component.ComponentDescriptorService; @@ -36,6 +38,7 @@ import org.thingsboard.server.dao.component.ComponentDescriptorService; import javax.annotation.PostConstruct; import java.lang.annotation.Annotation; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -163,7 +166,7 @@ public class AnnotationComponentDiscoveryService implements ComponentDiscoverySe nodeDefinition.setDescription(nodeAnnotation.nodeDescription()); nodeDefinition.setInEnabled(nodeAnnotation.inEnabled()); nodeDefinition.setOutEnabled(nodeAnnotation.outEnabled()); - nodeDefinition.setRelationTypes(nodeAnnotation.relationTypes()); + nodeDefinition.setRelationTypes(getRelationTypesWithFailureRelation(nodeAnnotation)); nodeDefinition.setCustomRelations(nodeAnnotation.customRelations()); Class configClazz = nodeAnnotation.configClazz(); NodeConfiguration config = configClazz.newInstance(); @@ -176,6 +179,14 @@ public class AnnotationComponentDiscoveryService implements ComponentDiscoverySe return nodeDefinition; } + private String[] getRelationTypesWithFailureRelation(RuleNode nodeAnnotation) { + List relationTypes = new ArrayList<>(Arrays.asList(nodeAnnotation.relationTypes())); + if (!relationTypes.contains(TbRelationTypes.FAILURE)) { + relationTypes.add(TbRelationTypes.FAILURE); + } + return relationTypes.toArray(new String[relationTypes.size()]); + } + private Set getBeanDefinitions(Class componentType) { ClassPathScanningCandidateComponentProvider scanner = new ClassPathScanningCandidateComponentProvider(false); scanner.addIncludeFilter(new AnnotationTypeFilter(componentType)); diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 734917651b..a3d6db4ffe 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -46,7 +46,7 @@ public interface TbContext { void tellSelf(TbMsg msg, long delayMs); - void tellError(TbMsg msg, Throwable th); + void tellFailure(TbMsg msg, Throwable th); void updateSelf(RuleNode self); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java index 8cf00ccc8d..6b4296e0a0 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java @@ -62,7 +62,7 @@ public abstract class TbAbstractAlarmNode ctx.tellError(msg, t)); + t -> ctx.tellFailure(msg, t)); } protected abstract ListenableFuture processAlarm(TbContext ctx, TbMsg msg); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbLogNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbLogNode.java index 892b4d76a6..895862e0b1 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbLogNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbLogNode.java @@ -57,7 +57,7 @@ public class TbLogNode implements TbNode { log.info(toString); ctx.tellNext(msg, SUCCESS); }, - t -> ctx.tellError(msg, t)); + t -> ctx.tellFailure(msg, t)); } @Override diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sns/TbSnsNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sns/TbSnsNode.java index 22db39474c..a509d9abf3 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sns/TbSnsNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sns/TbSnsNode.java @@ -77,15 +77,15 @@ public class TbSnsNode implements TbNode { m -> ctx.tellNext(m, TbRelationTypes.SUCCESS), t -> { TbMsg next = processException(ctx, msg, t); - ctx.tellNext(next, TbRelationTypes.FAILURE, t); + ctx.tellFailure(next, t); }); } - ListenableFuture publishMessageAsync(TbContext ctx, TbMsg msg) { + private ListenableFuture publishMessageAsync(TbContext ctx, TbMsg msg) { return ctx.getExternalCallExecutor().executeAsync(() -> publishMessage(ctx, msg)); } - TbMsg publishMessage(TbContext ctx, TbMsg msg) { + private TbMsg publishMessage(TbContext ctx, TbMsg msg) { String topicArn = TbNodeUtils.processPattern(this.config.getTopicArnPattern(), msg.getMetaData()); PublishRequest publishRequest = new PublishRequest() .withTopicArn(topicArn) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNode.java index c4ac13e834..c46c56f7c8 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNode.java @@ -83,15 +83,15 @@ public class TbSqsNode implements TbNode { m -> ctx.tellNext(m, TbRelationTypes.SUCCESS), t -> { TbMsg next = processException(ctx, msg, t); - ctx.tellNext(next, TbRelationTypes.FAILURE, t); + ctx.tellFailure(next, t); }); } - ListenableFuture publishMessageAsync(TbContext ctx, TbMsg msg) { + private ListenableFuture publishMessageAsync(TbContext ctx, TbMsg msg) { return ctx.getExternalCallExecutor().executeAsync(() -> publishMessage(ctx, msg)); } - TbMsg publishMessage(TbContext ctx, TbMsg msg) { + private TbMsg publishMessage(TbContext ctx, TbMsg msg) { String queueUrl = TbNodeUtils.processPattern(this.config.getQueueUrlPattern(), msg.getMetaData()); SendMessageRequest sendMsgRequest = new SendMessageRequest(); sendMsgRequest.withQueueUrl(queueUrl); 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 5de3dff990..faea63a778 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 @@ -74,7 +74,7 @@ public class TbMsgGeneratorNode implements TbNode { if (msg.getType().equals(TB_MSG_GENERATOR_NODE_MSG) && msg.getId().equals(nextTickId)) { withCallback(generate(ctx), m -> {ctx.tellNext(m, SUCCESS); sentTickMsg(ctx);}, - t -> {ctx.tellError(msg, t); sentTickMsg(ctx);}); + t -> {ctx.tellFailure(msg, t); sentTickMsg(ctx);}); } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsFilterNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsFilterNode.java index 50755783e3..f12cfd0c92 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsFilterNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbJsFilterNode.java @@ -52,8 +52,8 @@ public class TbJsFilterNode implements TbNode { public void onMsg(TbContext ctx, TbMsg msg) { ListeningExecutor jsExecutor = ctx.getJsExecutor(); withCallback(jsExecutor.executeAsync(() -> jsEngine.executeFilter(msg)), - filterResult -> ctx.tellNext(msg, filterResult.booleanValue() ? "True" : "False"), - t -> ctx.tellError(msg, t)); + filterResult -> ctx.tellNext(msg, filterResult ? "True" : "False"), + t -> ctx.tellFailure(msg, t)); } @Override 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 ac39eb5668..d48148e3d1 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 @@ -55,7 +55,7 @@ public class TbJsSwitchNode implements TbNode { ListeningExecutor jsExecutor = ctx.getJsExecutor(); withCallback(jsExecutor.executeAsync(() -> jsEngine.executeSwitch(msg)), result -> processSwitch(ctx, msg, result), - t -> ctx.tellError(msg, t)); + t -> ctx.tellFailure(msg, t)); } private void processSwitch(TbContext ctx, TbMsg msg, Set nextRelations) { diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java index 083f77d799..2d1e34dc5b 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java @@ -82,11 +82,11 @@ public class TbKafkaNode implements TbNode { ctx.tellNext(next, TbRelationTypes.SUCCESS); } else { TbMsg next = processException(ctx, msg, e); - ctx.tellNext(next, TbRelationTypes.FAILURE, e); + ctx.tellFailure(next, e); } }); } catch (Exception e) { - ctx.tellError(msg, e); + ctx.tellFailure(msg, e); } } 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 90a4c5b6ae..62bbbf93d4 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 @@ -61,7 +61,7 @@ public class TbMsgToEmailNode implements TbNode { ctx.tellNext(emailMsg, SUCCESS); } catch (Exception ex) { log.warn("Can not convert message to email " + ex.getMessage()); - ctx.tellError(msg, ex); + ctx.tellFailure(msg, ex); } } 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 3c23c750b2..c31f912490 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 @@ -15,9 +15,7 @@ */ package org.thingsboard.rule.engine.mail; -import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.mail.javamail.JavaMailSenderImpl; @@ -78,9 +76,9 @@ public class TbSendEmailNode implements TbNode { return null; }), ok -> ctx.tellNext(msg, SUCCESS), - fail -> ctx.tellError(msg, fail)); + fail -> ctx.tellFailure(msg, fail)); } catch (Exception ex) { - ctx.tellError(msg, ex); + ctx.tellFailure(msg, ex); } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java index a85a91dc80..c6212cc045 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java @@ -52,9 +52,9 @@ public abstract class TbAbstractGetAttributesNode safePutAttributes(ctx, msg, entityId), - t -> ctx.tellError(msg, t), ctx.getDbCallbackExecutor()); + t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); } catch (Throwable th) { - ctx.tellError(msg, th); + ctx.tellFailure(msg, th); } } @@ -69,7 +69,7 @@ public abstract class TbAbstractGetAttributesNode ctx.tellNext(msg, SUCCESS), t -> ctx.tellError(msg, t)); + withCallback(allFutures, i -> ctx.tellNext(msg, SUCCESS), t -> ctx.tellFailure(msg, t)); } private ListenableFuture putAttrAsync(TbContext ctx, EntityId entityId, TbMsg msg, String scope, List keys, String prefix) { @@ -77,7 +77,7 @@ public abstract class TbAbstractGetAttributesNode> latest = ctx.getAttributesService().find(entityId, scope, keys); - return Futures.transform(latest, (Function, Void>) l -> { + return Futures.transform(latest, l -> { l.forEach(r -> msg.getMetaData().putValue(prefix + r.getKey(), r.getValueAsString())); return null; }); @@ -88,7 +88,7 @@ public abstract class TbAbstractGetAttributesNode> latest = ctx.getTimeseriesService().findLatest(entityId, keys); - return Futures.transform(latest, (Function, Void>) l -> { + return Futures.transform(latest, l -> { l.forEach(r -> msg.getMetaData().putValue(r.getKey(), r.getValueAsString())); return null; }); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbEntityGetAttrNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbEntityGetAttrNode.java index 82805eb339..749c528e97 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbEntityGetAttrNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbEntityGetAttrNode.java @@ -54,9 +54,9 @@ public abstract class TbEntityGetAttrNode implements TbNode withCallback( findEntityAsync(ctx, msg.getOriginator()), entityId -> safeGetAttributes(ctx, msg, entityId), - t -> ctx.tellError(msg, t)); + t -> ctx.tellFailure(msg, t)); } catch (Throwable th) { - ctx.tellError(msg, th); + ctx.tellFailure(msg, th); } } @@ -68,18 +68,18 @@ public abstract class TbEntityGetAttrNode implements TbNode withCallback(config.isTelemetry() ? getLatestTelemetry(ctx, entityId) : getAttributesAsync(ctx, entityId), attributes -> putAttributesAndTell(ctx, msg, attributes), - t -> ctx.tellError(msg, t)); + t -> ctx.tellFailure(msg, t)); } private ListenableFuture> getAttributesAsync(TbContext ctx, EntityId entityId) { ListenableFuture> latest = ctx.getAttributesService().find(entityId, SERVER_SCOPE, config.getAttrMapping().keySet()); - return Futures.transform(latest, (Function, ? extends List>) l -> + return Futures.transform(latest, l -> l.stream().map(i -> (KvEntry) i).collect(Collectors.toList())); } private ListenableFuture> getLatestTelemetry(TbContext ctx, EntityId entityId) { ListenableFuture> latest = ctx.getTimeseriesService().findLatest(entityId, config.getAttrMapping().keySet()); - return Futures.transform(latest, (Function, ? extends List>) l -> + return Futures.transform(latest, l -> l.stream().map(i -> (KvEntry) i).collect(Collectors.toList())); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java index ce54a73f4b..455e6c49fe 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java @@ -83,7 +83,7 @@ public class TbMqttNode implements TbNode { ctx.tellNext(msg, TbRelationTypes.SUCCESS); } else { TbMsg next = processException(ctx, msg, future.cause()); - ctx.tellNext(next, TbRelationTypes.FAILURE, future.cause()); + ctx.tellFailure(next, future.cause()); } } ); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rabbitmq/TbRabbitMqNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rabbitmq/TbRabbitMqNode.java index 993b172ddf..c14590b93f 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rabbitmq/TbRabbitMqNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rabbitmq/TbRabbitMqNode.java @@ -80,15 +80,15 @@ public class TbRabbitMqNode implements TbNode { m -> ctx.tellNext(m, TbRelationTypes.SUCCESS), t -> { TbMsg next = processException(ctx, msg, t); - ctx.tellNext(next, TbRelationTypes.FAILURE, t); + ctx.tellFailure(next, t); }); } - ListenableFuture publishMessageAsync(TbContext ctx, TbMsg msg) { + private ListenableFuture publishMessageAsync(TbContext ctx, TbMsg msg) { return ctx.getExternalCallExecutor().executeAsync(() -> publishMessage(ctx, msg)); } - TbMsg publishMessage(TbContext ctx, TbMsg msg) throws Exception { + private TbMsg publishMessage(TbContext ctx, TbMsg msg) throws Exception { String exchangeName = ""; if (!StringUtils.isEmpty(this.config.getExchangeNamePattern())) { exchangeName = TbNodeUtils.processPattern(this.config.getExchangeNamePattern(), msg.getMetaData()); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNode.java index 6b902c6ae5..8f634cabb3 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNode.java @@ -89,7 +89,7 @@ public class TbRestApiCallNode implements TbNode { @Override public void onFailure(Throwable throwable) { TbMsg next = processException(ctx, msg, throwable); - ctx.tellNext(next, TbRelationTypes.FAILURE, throwable); + ctx.tellFailure(next, throwable); } @Override diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNode.java index 52cf1aad9e..1cbea2da2e 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNode.java @@ -52,11 +52,11 @@ public class TbSendRPCReplyNode implements TbNode { public void onMsg(TbContext ctx, TbMsg msg) { String requestIdStr = msg.getMetaData().getValue(config.getRequestIdMetaDataAttribute()); if (msg.getOriginator().getEntityType() != EntityType.DEVICE) { - ctx.tellError(msg, new RuntimeException("Message originator is not a device entity!")); + ctx.tellFailure(msg, new RuntimeException("Message originator is not a device entity!")); } else if (StringUtils.isEmpty(requestIdStr)) { - ctx.tellError(msg, new RuntimeException("Request id is not present in the metadata!")); + ctx.tellFailure(msg, new RuntimeException("Request id is not present in the metadata!")); } else if (StringUtils.isEmpty(msg.getData())) { - ctx.tellError(msg, new RuntimeException("Request body is empty!")); + ctx.tellFailure(msg, new RuntimeException("Request body is empty!")); } else { ctx.getRpcService().sendRpcReply(new DeviceId(msg.getOriginator().getId()), Integer.parseInt(requestIdStr), msg.getData()); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java index 904c6945b8..3ebbf6d260 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java @@ -63,15 +63,15 @@ public class TbSendRPCRequestNode implements TbNode { JsonObject json = jsonParser.parse(msg.getData()).getAsJsonObject(); if (msg.getOriginator().getEntityType() != EntityType.DEVICE) { - ctx.tellError(msg, new RuntimeException("Message originator is not a device entity!")); + ctx.tellFailure(msg, new RuntimeException("Message originator is not a device entity!")); } else if (!json.has("method")) { - ctx.tellError(msg, new RuntimeException("Method is not present in the message!")); + ctx.tellFailure(msg, new RuntimeException("Method is not present in the message!")); } else if (!json.has("params")) { - ctx.tellError(msg, new RuntimeException("Params are not present in the message!")); + ctx.tellFailure(msg, new RuntimeException("Params are not present in the message!")); } else { int requestId = json.has("requestId") ? json.get("requestId").getAsInt() : random.nextInt(); RuleEngineDeviceRpcRequest request = RuleEngineDeviceRpcRequest.builder() - .method(gson.toJson(json.get("method"))) + .method(json.get("method").getAsString()) .body(gson.toJson(json.get("params"))) .deviceId(new DeviceId(msg.getOriginator().getId())) .requestId(requestId) @@ -84,8 +84,7 @@ public class TbSendRPCRequestNode implements TbNode { ctx.tellNext(next, TbRelationTypes.SUCCESS); } else { TbMsg next = ctx.transformMsg(msg, msg.getType(), msg.getOriginator(), msg.getMetaData(), wrap("error", ruleEngineDeviceRpcResponse.getError().get().name())); - ctx.tellNext(next, TbRelationTypes.FAILURE); - ctx.tellError(msg, new RuntimeException(ruleEngineDeviceRpcResponse.getError().get().name())); + ctx.tellFailure(next, new RuntimeException(ruleEngineDeviceRpcResponse.getError().get().name())); } }); } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java index ad415044ba..db96364e1a 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java @@ -17,7 +17,6 @@ package org.thingsboard.rule.engine.telemetry; import com.google.gson.JsonParser; import lombok.extern.slf4j.Slf4j; -import org.springframework.util.StringUtils; import org.thingsboard.rule.engine.TbNodeUtils; import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; @@ -25,19 +24,12 @@ import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.server.common.data.kv.AttributeKvEntry; -import org.thingsboard.server.common.data.kv.BasicTsKvEntry; -import org.thingsboard.server.common.data.kv.KvEntry; -import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; -import org.thingsboard.server.common.msg.core.AttributesUpdateRequest; -import org.thingsboard.server.common.msg.core.TelemetryUploadRequest; import org.thingsboard.server.common.msg.session.SessionMsgType; import org.thingsboard.server.common.transport.adaptor.JsonConverter; import java.util.ArrayList; -import java.util.List; -import java.util.Map; import java.util.Set; @Slf4j @@ -63,7 +55,7 @@ public class TbMsgAttributesNode implements TbNode { @Override public void onMsg(TbContext ctx, TbMsg msg) { if (!msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name())) { - ctx.tellError(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType())); + ctx.tellFailure(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType())); return; } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java index d91d5d41ad..ea4446dbe5 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java @@ -60,7 +60,7 @@ public class TbMsgTimeseriesNode implements TbNode { @Override public void onMsg(TbContext ctx, TbMsg msg) { if (!msg.getType().equals(SessionMsgType.POST_TELEMETRY_REQUEST.name())) { - ctx.tellError(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType())); + ctx.tellFailure(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType())); return; } long ts = -1; @@ -71,14 +71,14 @@ public class TbMsgTimeseriesNode implements TbNode { } catch (NumberFormatException e) {} } if (ts == -1) { - ctx.tellError(msg, new IllegalArgumentException("Msg metadata doesn't contain valid ts value: " + msg.getMetaData())); + ctx.tellFailure(msg, new IllegalArgumentException("Msg metadata doesn't contain valid ts value: " + msg.getMetaData())); return; } String src = msg.getData(); TelemetryUploadRequest telemetryUploadRequest = JsonConverter.convertToTelemetry(new JsonParser().parse(src), ts); Map> tsKvMap = telemetryUploadRequest.getData(); if (tsKvMap == null) { - ctx.tellError(msg, new IllegalArgumentException("Msg body is empty: " + src)); + ctx.tellFailure(msg, new IllegalArgumentException("Msg body is empty: " + src)); return; } List tsKvEntryList = new ArrayList<>(); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TelemetryNodeCallback.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TelemetryNodeCallback.java index d66b7685d3..8a9faab798 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TelemetryNodeCallback.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TelemetryNodeCallback.java @@ -39,6 +39,6 @@ class TelemetryNodeCallback implements FutureCallback { @Override public void onFailure(Throwable t) { - ctx.tellError(msg, t); + ctx.tellFailure(msg, t); } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbAbstractTransformNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbAbstractTransformNode.java index 74579a8e01..2616407a7b 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbAbstractTransformNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbAbstractTransformNode.java @@ -51,7 +51,7 @@ public abstract class TbAbstractTransformNode implements TbNode { ctx.tellNext(msg, FAILURE); } }, - t -> ctx.tellError(msg, t)); + t -> ctx.tellFailure(msg, t)); } protected abstract ListenableFuture transform(TbContext ctx, TbMsg msg); 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 bdc61e9880..5d6bdee8c3 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 @@ -368,7 +368,7 @@ public class TbAlarmNodeTest { private void verifyError(TbMsg msg, String message, Class expectedClass) { ArgumentCaptor captor = ArgumentCaptor.forClass(Throwable.class); - verify(ctx).tellError(same(msg), captor.capture()); + verify(ctx).tellFailure(same(msg), captor.capture()); Throwable value = captor.getValue(); assertEquals(expectedClass, value.getClass()); 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 aeb20bbf19..b3e097bafd 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 @@ -117,7 +117,7 @@ public class TbJsFilterNodeTest { private void verifyError(TbMsg msg, String message, Class expectedClass) { ArgumentCaptor captor = ArgumentCaptor.forClass(Throwable.class); - verify(ctx).tellError(same(msg), captor.capture()); + verify(ctx).tellFailure(same(msg), captor.capture()); Throwable value = captor.getValue(); assertEquals(expectedClass, value.getClass()); 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 6d62009990..f547807a80 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 @@ -99,7 +99,7 @@ public class TbJsSwitchNodeTest { private void verifyError(TbMsg msg, String message, Class expectedClass) { ArgumentCaptor captor = ArgumentCaptor.forClass(Throwable.class); - verify(ctx).tellError(same(msg), captor.capture()); + verify(ctx).tellFailure(same(msg), captor.capture()); Throwable value = captor.getValue(); assertEquals(expectedClass, value.getClass()); 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 997b68413a..4a855aed3f 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 @@ -100,7 +100,6 @@ public class TbGetCustomerAttributeNodeTest { User user = new User(); user.setCustomerId(customerId); - msg = new TbMsg(UUIDs.timeBased(), "USER", userId, new TbMsgMetaData(), "{}", ruleChainId, ruleNodeId, 0L); when(ctx.getUserService()).thenReturn(userService); @@ -112,7 +111,7 @@ public class TbGetCustomerAttributeNodeTest { node.onMsg(ctx, msg); final ArgumentCaptor captor = ArgumentCaptor.forClass(Throwable.class); - verify(ctx).tellError(same(msg), captor.capture()); + verify(ctx).tellFailure(same(msg), captor.capture()); Throwable value = captor.getValue(); assertEquals("something wrong", value.getMessage()); @@ -137,7 +136,7 @@ public class TbGetCustomerAttributeNodeTest { node.onMsg(ctx, msg); final ArgumentCaptor captor = ArgumentCaptor.forClass(Throwable.class); - verify(ctx).tellError(same(msg), captor.capture()); + verify(ctx).tellFailure(same(msg), captor.capture()); Throwable value = captor.getValue(); assertEquals("something wrong", value.getMessage()); 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 48e6353984..1daeb19593 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 @@ -117,7 +117,7 @@ public class TbTransformMsgNodeTest { private void verifyError(TbMsg msg, String message, Class expectedClass) { ArgumentCaptor captor = ArgumentCaptor.forClass(Throwable.class); - verify(ctx).tellError(same(msg), captor.capture()); + verify(ctx).tellFailure(same(msg), captor.capture()); Throwable value = captor.getValue(); assertEquals(expectedClass, value.getClass());