dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feature/data-pattern' of https://github.com/YevhenBondarenko/thingsboard into YevhenBondarenko-feature/data-pattern

Browse Source
This commit is contained in:
Andrii Shvaika 2021-02-08 18:20:11 +02:00
commit 674ab2adb2
17 changed files with 89 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/util/TbNodeUtils.java
View File

@ -16,15 +16,20 @@
package org.thingsboard.rule.engine.api.util; package org.thingsboard.rule.engine.api.util;
import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.common.msg.TbMsgMetaData;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors; import java.util.stream.Collectors;
/** /**
@ -34,8 +39,11 @@ public class TbNodeUtils {
private static final ObjectMapper mapper = new ObjectMapper(); private static final ObjectMapper mapper = new ObjectMapper();
private static final String VARIABLE_TEMPLATE = "${%s}"; private static final String METADATA_VARIABLE_TEMPLATE = "${%s}";
private static final Pattern DATA_PATTERN = Pattern.compile("(\\$\\[)(.*?)(\\])");
private static final String DATA_VARIABLE_TEMPLATE = "$[%s]";
public static <T> T convert(TbNodeConfiguration configuration, Class<T> clazz) throws TbNodeException { public static <T> T convert(TbNodeConfiguration configuration, Class<T> clazz) throws TbNodeException {
try { try {
@ -45,6 +53,44 @@ public class TbNodeUtils {
} }
} }
public static List<String> processPatterns(List<String> patterns, TbMsg tbMsg) {
if (!CollectionUtils.isEmpty(patterns)) {
return patterns.stream().map(p -> processPattern(p, tbMsg)).collect(Collectors.toList());
}
return Collections.emptyList();
}
public static String processPattern(String pattern, TbMsg tbMsg) {
try {
String result = processPattern(pattern, tbMsg.getMetaData());
JsonNode json = mapper.readTree(tbMsg.getData());
if (json.isObject()) {
Matcher matcher = DATA_PATTERN.matcher(result);
while (matcher.find()) {
String group = matcher.group(2);
String[] keys = group.split("\\.");
JsonNode jsonNode = json;
for (String key : keys) {
if (StringUtils.isNotEmpty(key) && jsonNode != null) {
jsonNode = jsonNode.get(key);
} else {
jsonNode = null;
break;
}
}
if (jsonNode != null && !jsonNode.isObject() && !jsonNode.isArray()) {
result = result.replace(String.format(DATA_VARIABLE_TEMPLATE, group), jsonNode.asText());
}
}
}
return result;
} catch (Exception e) {
throw new RuntimeException("Failed to process pattern!", e);
}
}
public static List<String> processPatterns(List<String> patterns, TbMsgMetaData metaData) { public static List<String> processPatterns(List<String> patterns, TbMsgMetaData metaData) {
if (!CollectionUtils.isEmpty(patterns)) { if (!CollectionUtils.isEmpty(patterns)) {
return patterns.stream().map(p -> processPattern(p, metaData)).collect(Collectors.toList()); return patterns.stream().map(p -> processPattern(p, metaData)).collect(Collectors.toList());
@ -53,15 +99,15 @@ public class TbNodeUtils {
} }
public static String processPattern(String pattern, TbMsgMetaData metaData) { public static String processPattern(String pattern, TbMsgMetaData metaData) {
String result = new String(pattern); String result = pattern;
for (Map.Entry<String,String> keyVal : metaData.values().entrySet()) { for (Map.Entry<String, String> keyVal : metaData.values().entrySet()) {
result = processVar(result, keyVal.getKey(), keyVal.getValue()); result = processVar(result, keyVal.getKey(), keyVal.getValue());
} }
return result; return result;
} }
private static String processVar(String pattern, String key, String val) { private static String processVar(String pattern, String key, String val) {
String varPattern = String.format(VARIABLE_TEMPLATE, key); String varPattern = String.format(METADATA_VARIABLE_TEMPLATE, key);
return pattern.replace(varPattern, val); return pattern.replace(varPattern, val);
} }

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractCustomerActionNode.java
View File

@ -79,7 +79,7 @@ public abstract class TbAbstractCustomerActionNode<C extends TbAbstractCustomerA
protected abstract void doProcessCustomerAction(TbContext ctx, TbMsg msg, CustomerId customerId); protected abstract void doProcessCustomerAction(TbContext ctx, TbMsg msg, CustomerId customerId);
protected ListenableFuture<CustomerId> getCustomer(TbContext ctx, TbMsg msg) { protected ListenableFuture<CustomerId> getCustomer(TbContext ctx, TbMsg msg) {
String customerTitle = TbNodeUtils.processPattern(this.config.getCustomerNamePattern(), msg.getMetaData()); String customerTitle = TbNodeUtils.processPattern(this.config.getCustomerNamePattern(), msg);
CustomerKey key = new CustomerKey(customerTitle); CustomerKey key = new CustomerKey(customerTitle);
return ctx.getDbCallbackExecutor().executeAsync(() -> { return ctx.getDbCallbackExecutor().executeAsync(() -> {
Optional<CustomerId> customerId = customerIdCache.get(key); Optional<CustomerId> customerId = customerIdCache.get(key);

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractRelationActionNode.java
View File

@ -140,7 +140,7 @@ public abstract class TbAbstractRelationActionNode<C extends TbAbstractRelationA
} }
protected String processPattern(TbMsg msg, String pattern) { protected String processPattern(TbMsg msg, String pattern) {
return TbNodeUtils.processPattern(pattern, msg.getMetaData()); return TbNodeUtils.processPattern(pattern, msg);
} }
@Data @Data

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbClearAlarmNode.java
View File

@ -56,7 +56,7 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode<TbClearAlarmNodeConfig
@Override @Override
protected ListenableFuture<TbAlarmResult> processAlarm(TbContext ctx, TbMsg msg) { protected ListenableFuture<TbAlarmResult> processAlarm(TbContext ctx, TbMsg msg) {
String alarmType = TbNodeUtils.processPattern(this.config.getAlarmType(), msg.getMetaData()); String alarmType = TbNodeUtils.processPattern(this.config.getAlarmType(), msg);
ListenableFuture<Alarm> alarmFuture; ListenableFuture<Alarm> alarmFuture;
if (msg.getOriginator().getEntityType().equals(EntityType.ALARM)) { if (msg.getOriginator().getEntityType().equals(EntityType.ALARM)) {
alarmFuture = ctx.getAlarmService().findAlarmByIdAsync(ctx.getTenantId(), new AlarmId(msg.getOriginator().getId())); alarmFuture = ctx.getAlarmService().findAlarmByIdAsync(ctx.getTenantId(), new AlarmId(msg.getOriginator().getId()));

4
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateAlarmNode.java
View File

@ -70,7 +70,7 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf
final Alarm msgAlarm; final Alarm msgAlarm;
if (!config.isUseMessageAlarmData()) { if (!config.isUseMessageAlarmData()) {
alarmType = TbNodeUtils.processPattern(this.config.getAlarmType(), msg.getMetaData()); alarmType = TbNodeUtils.processPattern(this.config.getAlarmType(), msg);
msgAlarm = null; msgAlarm = null;
} else { } else {
try { try {
@ -151,7 +151,7 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConf
.status(AlarmStatus.ACTIVE_UNACK) .status(AlarmStatus.ACTIVE_UNACK)
.severity(config.getSeverity()) .severity(config.getSeverity())
.propagate(config.isPropagate()) .propagate(config.isPropagate())
.type(TbNodeUtils.processPattern(this.config.getAlarmType(), msg.getMetaData())) .type(TbNodeUtils.processPattern(this.config.getAlarmType(), msg))
.propagateRelationTypes(relationTypes) .propagateRelationTypes(relationTypes)
//todo-vp: alarm date should be taken from Message or current Time should be used? //todo-vp: alarm date should be taken from Message or current Time should be used?
// .startTs(System.currentTimeMillis()) // .startTs(System.currentTimeMillis())

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sns/TbSnsNode.java
View File

@ -83,7 +83,7 @@ public class TbSnsNode implements TbNode {
} }
private TbMsg publishMessage(TbContext ctx, TbMsg msg) { private TbMsg publishMessage(TbContext ctx, TbMsg msg) {
String topicArn = TbNodeUtils.processPattern(this.config.getTopicArnPattern(), msg.getMetaData()); String topicArn = TbNodeUtils.processPattern(this.config.getTopicArnPattern(), msg);
PublishRequest publishRequest = new PublishRequest() PublishRequest publishRequest = new PublishRequest()
.withTopicArn(topicArn) .withTopicArn(topicArn)
.withMessage(msg.getData()); .withMessage(msg.getData());

6
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNode.java
View File

@ -91,14 +91,14 @@ public class TbSqsNode implements TbNode {
} }
private TbMsg publishMessage(TbContext ctx, TbMsg msg) { private TbMsg publishMessage(TbContext ctx, TbMsg msg) {
String queueUrl = TbNodeUtils.processPattern(this.config.getQueueUrlPattern(), msg.getMetaData()); String queueUrl = TbNodeUtils.processPattern(this.config.getQueueUrlPattern(), msg);
SendMessageRequest sendMsgRequest = new SendMessageRequest(); SendMessageRequest sendMsgRequest = new SendMessageRequest();
sendMsgRequest.withQueueUrl(queueUrl); sendMsgRequest.withQueueUrl(queueUrl);
sendMsgRequest.withMessageBody(msg.getData()); sendMsgRequest.withMessageBody(msg.getData());
Map<String, MessageAttributeValue> messageAttributes = new HashMap<>(); Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
this.config.getMessageAttributes().forEach((k,v) -> { this.config.getMessageAttributes().forEach((k,v) -> {
String name = TbNodeUtils.processPattern(k, msg.getMetaData()); String name = TbNodeUtils.processPattern(k, msg);
String val = TbNodeUtils.processPattern(v, msg.getMetaData()); String val = TbNodeUtils.processPattern(v, msg);
messageAttributes.put(name, new MessageAttributeValue().withDataType("String").withStringValue(val)); messageAttributes.put(name, new MessageAttributeValue().withDataType("String").withStringValue(val));
}); });
sendMsgRequest.setMessageAttributes(messageAttributes); sendMsgRequest.setMessageAttributes(messageAttributes);

4
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
View File

@ -84,7 +84,7 @@ public class TbMsgDelayNode implements TbNode {
int periodInSeconds; int periodInSeconds;
if (config.isUseMetadataPeriodInSecondsPatterns()) { if (config.isUseMetadataPeriodInSecondsPatterns()) {
if (isParsable(msg, config.getPeriodInSecondsPattern())) { if (isParsable(msg, config.getPeriodInSecondsPattern())) {
periodInSeconds = Integer.parseInt(TbNodeUtils.processPattern(config.getPeriodInSecondsPattern(), msg.getMetaData())); periodInSeconds = Integer.parseInt(TbNodeUtils.processPattern(config.getPeriodInSecondsPattern(), msg));
} else { } else {
throw new RuntimeException("Can't parse period in seconds from metadata using pattern: " + config.getPeriodInSecondsPattern()); throw new RuntimeException("Can't parse period in seconds from metadata using pattern: " + config.getPeriodInSecondsPattern());
} }
@ -95,7 +95,7 @@ public class TbMsgDelayNode implements TbNode {
} }
private boolean isParsable(TbMsg msg, String pattern) { private boolean isParsable(TbMsg msg, String pattern) {
return NumberUtils.isParsable(TbNodeUtils.processPattern(pattern, msg.getMetaData())); return NumberUtils.isParsable(TbNodeUtils.processPattern(pattern, msg));
} }
@Override @Override

4
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/gcp/pubsub/TbPubSubNode.java
View File

@ -91,8 +91,8 @@ public class TbPubSubNode implements TbNode {
PubsubMessage.Builder pubsubMessageBuilder = PubsubMessage.newBuilder(); PubsubMessage.Builder pubsubMessageBuilder = PubsubMessage.newBuilder();
pubsubMessageBuilder.setData(data); pubsubMessageBuilder.setData(data);
this.config.getMessageAttributes().forEach((k, v) -> { this.config.getMessageAttributes().forEach((k, v) -> {
String name = TbNodeUtils.processPattern(k, msg.getMetaData()); String name = TbNodeUtils.processPattern(k, msg);
String val = TbNodeUtils.processPattern(v, msg.getMetaData()); String val = TbNodeUtils.processPattern(v, msg);
pubsubMessageBuilder.putAttributes(name, val); pubsubMessageBuilder.putAttributes(name, val);
}); });
ApiFuture<String> messageIdFuture = this.pubSubClient.publish(pubsubMessageBuilder.build()); ApiFuture<String> messageIdFuture = this.pubSubClient.publish(pubsubMessageBuilder.build());

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java
View File

@ -94,7 +94,7 @@ public class TbKafkaNode implements TbNode {
@Override @Override
public void onMsg(TbContext ctx, TbMsg msg) { public void onMsg(TbContext ctx, TbMsg msg) {
String topic = TbNodeUtils.processPattern(config.getTopicPattern(), msg.getMetaData()); String topic = TbNodeUtils.processPattern(config.getTopicPattern(), msg);
try { try {
ctx.getExternalCallExecutor().executeAsync(() -> { ctx.getExternalCallExecutor().executeAsync(() -> {
publish(ctx, msg, topic); publish(ctx, msg, topic);

16
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mail/TbMsgToEmailNode.java
View File

@ -72,18 +72,18 @@ public class TbMsgToEmailNode implements TbNode {
private EmailPojo convert(TbMsg msg) throws IOException { private EmailPojo convert(TbMsg msg) throws IOException {
EmailPojo.EmailPojoBuilder builder = EmailPojo.builder(); EmailPojo.EmailPojoBuilder builder = EmailPojo.builder();
builder.from(fromTemplate(this.config.getFromTemplate(), msg.getMetaData())); builder.from(fromTemplate(this.config.getFromTemplate(), msg));
builder.to(fromTemplate(this.config.getToTemplate(), msg.getMetaData())); builder.to(fromTemplate(this.config.getToTemplate(), msg));
builder.cc(fromTemplate(this.config.getCcTemplate(), msg.getMetaData())); builder.cc(fromTemplate(this.config.getCcTemplate(), msg));
builder.bcc(fromTemplate(this.config.getBccTemplate(), msg.getMetaData())); builder.bcc(fromTemplate(this.config.getBccTemplate(), msg));
builder.subject(fromTemplate(this.config.getSubjectTemplate(), msg.getMetaData())); builder.subject(fromTemplate(this.config.getSubjectTemplate(), msg));
builder.body(fromTemplate(this.config.getBodyTemplate(), msg.getMetaData())); builder.body(fromTemplate(this.config.getBodyTemplate(), msg));
return builder.build(); return builder.build();
} }
private String fromTemplate(String template, TbMsgMetaData metaData) { private String fromTemplate(String template, TbMsg msg) {
if (!StringUtils.isEmpty(template)) { if (!StringUtils.isEmpty(template)) {
return TbNodeUtils.processPattern(template, metaData); return TbNodeUtils.processPattern(template, msg);
} else { } else {
return null; return null;
} }

8
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java
View File

@ -92,10 +92,10 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
} }
ConcurrentHashMap<String, List<String>> failuresMap = new ConcurrentHashMap<>(); ConcurrentHashMap<String, List<String>> failuresMap = new ConcurrentHashMap<>();
ListenableFuture<List<Void>> allFutures = Futures.allAsList( ListenableFuture<List<Void>> allFutures = Futures.allAsList(
putLatestTelemetry(ctx, entityId, msg, LATEST_TS, TbNodeUtils.processPatterns(config.getLatestTsKeyNames(), msg.getMetaData()), failuresMap), putLatestTelemetry(ctx, entityId, msg, LATEST_TS, TbNodeUtils.processPatterns(config.getLatestTsKeyNames(), msg), failuresMap),
putAttrAsync(ctx, entityId, msg, CLIENT_SCOPE, TbNodeUtils.processPatterns(config.getClientAttributeNames(), msg.getMetaData()), failuresMap, "cs_"), putAttrAsync(ctx, entityId, msg, CLIENT_SCOPE, TbNodeUtils.processPatterns(config.getClientAttributeNames(), msg), failuresMap, "cs_"),
putAttrAsync(ctx, entityId, msg, SHARED_SCOPE, TbNodeUtils.processPatterns(config.getSharedAttributeNames(), msg.getMetaData()), failuresMap, "shared_"), putAttrAsync(ctx, entityId, msg, SHARED_SCOPE, TbNodeUtils.processPatterns(config.getSharedAttributeNames(), msg), failuresMap, "shared_"),
putAttrAsync(ctx, entityId, msg, SERVER_SCOPE, TbNodeUtils.processPatterns(config.getServerAttributeNames(), msg.getMetaData()), failuresMap, "ss_") putAttrAsync(ctx, entityId, msg, SERVER_SCOPE, TbNodeUtils.processPatterns(config.getServerAttributeNames(), msg), failuresMap, "ss_")
); );
withCallback(allFutures, i -> { withCallback(allFutures, i -> {
if (!failuresMap.isEmpty()) { if (!failuresMap.isEmpty()) {

8
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNode.java
View File

@ -103,7 +103,7 @@ public class TbGetTelemetryNode implements TbNode {
if (config.isUseMetadataIntervalPatterns()) { if (config.isUseMetadataIntervalPatterns()) {
checkMetadataKeyPatterns(msg); checkMetadataKeyPatterns(msg);
} }
List<String> keys = TbNodeUtils.processPatterns(tsKeyNames, msg.getMetaData()); List<String> keys = TbNodeUtils.processPatterns(tsKeyNames, msg);
ListenableFuture<List<TsKvEntry>> list = ctx.getTimeseriesService().findAll(ctx.getTenantId(), msg.getOriginator(), buildQueries(msg, keys)); ListenableFuture<List<TsKvEntry>> list = ctx.getTimeseriesService().findAll(ctx.getTenantId(), msg.getOriginator(), buildQueries(msg, keys));
DonAsynchron.withCallback(list, data -> { DonAsynchron.withCallback(list, data -> {
process(data, msg, keys); process(data, msg, keys);
@ -197,10 +197,10 @@ public class TbGetTelemetryNode implements TbNode {
Interval interval = new Interval(); Interval interval = new Interval();
if (config.isUseMetadataIntervalPatterns()) { if (config.isUseMetadataIntervalPatterns()) {
if (isParsable(msg, config.getStartIntervalPattern())) { if (isParsable(msg, config.getStartIntervalPattern())) {
interval.setStartTs(Long.parseLong(TbNodeUtils.processPattern(config.getStartIntervalPattern(), msg.getMetaData()))); interval.setStartTs(Long.parseLong(TbNodeUtils.processPattern(config.getStartIntervalPattern(), msg)));
} }
if (isParsable(msg, config.getEndIntervalPattern())) { if (isParsable(msg, config.getEndIntervalPattern())) {
interval.setEndTs(Long.parseLong(TbNodeUtils.processPattern(config.getEndIntervalPattern(), msg.getMetaData()))); interval.setEndTs(Long.parseLong(TbNodeUtils.processPattern(config.getEndIntervalPattern(), msg)));
} }
} else { } else {
long ts = System.currentTimeMillis(); long ts = System.currentTimeMillis();
@ -211,7 +211,7 @@ public class TbGetTelemetryNode implements TbNode {
} }
private boolean isParsable(TbMsg msg, String pattern) { private boolean isParsable(TbMsg msg, String pattern) {
return NumberUtils.isParsable(TbNodeUtils.processPattern(pattern, msg.getMetaData())); return NumberUtils.isParsable(TbNodeUtils.processPattern(pattern, msg));
} }
private void checkMetadataKeyPatterns(TbMsg msg) { private void checkMetadataKeyPatterns(TbMsg msg) {

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java
View File

@ -76,7 +76,7 @@ public class TbMqttNode implements TbNode {
@Override @Override
public void onMsg(TbContext ctx, TbMsg msg) { public void onMsg(TbContext ctx, TbMsg msg) {
String topic = TbNodeUtils.processPattern(this.mqttNodeConfiguration.getTopicPattern(), msg.getMetaData()); String topic = TbNodeUtils.processPattern(this.mqttNodeConfiguration.getTopicPattern(), msg);
this.mqttClient.publish(topic, Unpooled.wrappedBuffer(msg.getData().getBytes(UTF8)), MqttQoS.AT_LEAST_ONCE) this.mqttClient.publish(topic, Unpooled.wrappedBuffer(msg.getData().getBytes(UTF8)), MqttQoS.AT_LEAST_ONCE)
.addListener(future -> { .addListener(future -> {
if (future.isSuccess()) { if (future.isSuccess()) {

4
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rabbitmq/TbRabbitMqNode.java
View File

@ -90,11 +90,11 @@ public class TbRabbitMqNode implements TbNode {
private TbMsg publishMessage(TbContext ctx, TbMsg msg) throws Exception { private TbMsg publishMessage(TbContext ctx, TbMsg msg) throws Exception {
String exchangeName = ""; String exchangeName = "";
if (!StringUtils.isEmpty(this.config.getExchangeNamePattern())) { if (!StringUtils.isEmpty(this.config.getExchangeNamePattern())) {
exchangeName = TbNodeUtils.processPattern(this.config.getExchangeNamePattern(), msg.getMetaData()); exchangeName = TbNodeUtils.processPattern(this.config.getExchangeNamePattern(), msg);
} }
String routingKey = ""; String routingKey = "";
if (!StringUtils.isEmpty(this.config.getRoutingKeyPattern())) { if (!StringUtils.isEmpty(this.config.getRoutingKeyPattern())) {
routingKey = TbNodeUtils.processPattern(this.config.getRoutingKeyPattern(), msg.getMetaData()); routingKey = TbNodeUtils.processPattern(this.config.getRoutingKeyPattern(), msg);
} }
AMQP.BasicProperties properties = null; AMQP.BasicProperties properties = null;
if (!StringUtils.isEmpty(this.config.getMessageProperties())) { if (!StringUtils.isEmpty(this.config.getMessageProperties())) {

8
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbHttpClient.java
View File

@ -172,8 +172,8 @@ public class TbHttpClient {
} }
public void processMessage(TbContext ctx, TbMsg msg) { public void processMessage(TbContext ctx, TbMsg msg) {
String endpointUrl = TbNodeUtils.processPattern(config.getRestEndpointUrlPattern(), msg.getMetaData()); String endpointUrl = TbNodeUtils.processPattern(config.getRestEndpointUrlPattern(), msg);
HttpHeaders headers = prepareHeaders(msg.getMetaData()); HttpHeaders headers = prepareHeaders(msg);
HttpMethod method = HttpMethod.valueOf(config.getRequestMethod()); HttpMethod method = HttpMethod.valueOf(config.getRequestMethod());
HttpEntity<String> entity = new HttpEntity<>(msg.getData(), headers); HttpEntity<String> entity = new HttpEntity<>(msg.getData(), headers);
@ -232,9 +232,9 @@ public class TbHttpClient {
return ctx.transformMsg(origMsg, origMsg.getType(), origMsg.getOriginator(), metaData, origMsg.getData()); return ctx.transformMsg(origMsg, origMsg.getType(), origMsg.getOriginator(), metaData, origMsg.getData());
} }
private HttpHeaders prepareHeaders(TbMsgMetaData metaData) { private HttpHeaders prepareHeaders(TbMsg msg) {
HttpHeaders headers = new HttpHeaders(); HttpHeaders headers = new HttpHeaders();
config.getHeaders().forEach((k, v) -> headers.add(TbNodeUtils.processPattern(k, metaData), TbNodeUtils.processPattern(v, metaData))); config.getHeaders().forEach((k, v) -> headers.add(TbNodeUtils.processPattern(k, msg), TbNodeUtils.processPattern(v, msg)));
ClientCredentials credentials = config.getCredentials(); ClientCredentials credentials = config.getCredentials();
if (CredentialsType.BASIC == credentials.getType()) { if (CredentialsType.BASIC == credentials.getType()) {
BasicCredentials basicCredentials = (BasicCredentials) credentials; BasicCredentials basicCredentials = (BasicCredentials) credentials;

4
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/sms/TbSendSmsNode.java
View File

@ -72,8 +72,8 @@ public class TbSendSmsNode implements TbNode {
} }
private void sendSms(TbContext ctx, TbMsg msg) throws Exception { private void sendSms(TbContext ctx, TbMsg msg) throws Exception {
String numbersTo = TbNodeUtils.processPattern(this.config.getNumbersToTemplate(), msg.getMetaData()); String numbersTo = TbNodeUtils.processPattern(this.config.getNumbersToTemplate(), msg);
String message = TbNodeUtils.processPattern(this.config.getSmsMessageTemplate(), msg.getMetaData()); String message = TbNodeUtils.processPattern(this.config.getSmsMessageTemplate(), msg);
String[] numbersToList = numbersTo.split(","); String[] numbersToList = numbersTo.split(",");
if (this.config.isUseSystemSmsSettings()) { if (this.config.isUseSystemSmsSettings()) {
ctx.getSmsService().sendSms(ctx.getTenantId(), numbersToList, message); ctx.getSmsService().sendSms(ctx.getTenantId(), numbersToList, message);