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

Kafka Rule Node

Browse Source
This commit is contained in:
Igor Kulikov 2018-05-07 16:53:42 +03:00
parent dfd3252313
commit 3f706984b6
8 changed files with 209 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
View File

@ -68,8 +68,13 @@ class DefaultTbContext implements TbContext {
@Override
public void tellNext(TbMsg msg, String relationType) {
tellNext(msg, relationType, null);
}
@Override
public void tellNext(TbMsg msg, String relationType, Throwable th) {
if (nodeCtx.getSelf().isDebugMode()) {
mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg);
mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg, th);
}
nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getId(), relationType, msg), nodeCtx.getSelfActor());
}

2
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java
View File

@ -44,6 +44,8 @@ public interface TbContext {
void tellNext(TbMsg msg, String relationType);
void tellNext(TbMsg msg, String relationType, Throwable th);
void tellNext(TbMsg msg, Set<String> relationTypes);
void tellSelf(TbMsg msg, long delayMs);

4
rule-engine/rule-engine-components/pom.xml
View File

@ -86,6 +86,10 @@
<artifactId>spring-web</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>

19
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/TbNodeUtils.java
View File

@ -19,6 +19,9 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.server.common.msg.TbMsgMetaData;
import java.util.Map;
/**
* Created by ashvayka on 19.01.18.
@ -27,6 +30,9 @@ public class TbNodeUtils {
private static final ObjectMapper mapper = new ObjectMapper();
private static final String VARIABLE_TEMPLATE = "${%s}";
public static <T> T convert(TbNodeConfiguration configuration, Class<T> clazz) throws TbNodeException {
try {
return mapper.treeToValue(configuration.getData(), clazz);
@ -35,4 +41,17 @@ public class TbNodeUtils {
}
}
public static String processPattern(String pattern, TbMsgMetaData metaData) {
String result = new String(pattern);
for (Map.Entry<String,String> keyVal : metaData.values().entrySet()) {
result = processVar(result, keyVal.getKey(), keyVal.getValue());
}
return result;
}
private static String processVar(String pattern, String key, String val) {
String varPattern = String.format(VARIABLE_TEMPLATE, key);
return pattern.replace(varPattern, val);
}
}

117
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java Normal file
View File

@ -0,0 +1,117 @@
/**
* 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.rule.engine.kafka;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.thingsboard.rule.engine.TbNodeUtils;
import org.thingsboard.rule.engine.api.*;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
@Slf4j
@RuleNode(
type = ComponentType.ACTION,
name = "kafka",
configClazz = TbKafkaNodeConfiguration.class,
nodeDescription = "Publish messages to Kafka server",
nodeDetails = "Expects messages with any message type. Will send record via Kafka producer to Kafka server.",
uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = "tbActionNodeKafkaConfig"
)
public class TbKafkaNode implements TbNode {
private static final String OFFSET = "offset";
private static final String PARTITION = "partition";
private static final String TOPIC = "topic";
private static final String ERROR = "error";
private TbKafkaNodeConfiguration config;
private Producer<?, String> producer;
@Override
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
this.config = TbNodeUtils.convert(configuration, TbKafkaNodeConfiguration.class);
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, config.getValueSerializer());
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer());
properties.put(ProducerConfig.ACKS_CONFIG, config.getAcks());
properties.put(ProducerConfig.RETRIES_CONFIG, config.getRetries());
properties.put(ProducerConfig.BATCH_SIZE_CONFIG, config.getBatchSize());
properties.put(ProducerConfig.LINGER_MS_CONFIG, config.getLinger());
properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, config.getBufferMemory());
if (config.getOtherProperties() != null) {
config.getOtherProperties()
.forEach((k,v) -> properties.put(k, v));
}
try {
this.producer = new KafkaProducer<>(properties);
} catch (Exception e) {
throw new TbNodeException(e);
}
}
@Override
public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
String topic = TbNodeUtils.processPattern(config.getTopicPattern(), msg.getMetaData());
try {
producer.send(new ProducerRecord<>(topic, msg.getData()),
(metadata, e) -> {
if (metadata != null) {
TbMsg next = processResponse(ctx, msg, metadata);
ctx.tellNext(next, TbRelationTypes.SUCCESS);
} else {
TbMsg next = processException(ctx, msg, e);
ctx.tellNext(next, TbRelationTypes.FAILURE, e);
}
});
} catch (Exception e) {
ctx.tellError(msg, e);
}
}
@Override
public void destroy() {
if (this.producer != null) {
try {
this.producer.close();
} catch (Exception e) {
log.error("Failed to close producer during destroy()", e);
}
}
}
private TbMsg processResponse(TbContext ctx, TbMsg origMsg, RecordMetadata recordMetadata) {
TbMsgMetaData metaData = origMsg.getMetaData().copy();
metaData.putValue(OFFSET, String.valueOf(recordMetadata.offset()));
metaData.putValue(PARTITION, String.valueOf(recordMetadata.partition()));
metaData.putValue(TOPIC, recordMetadata.topic());
return ctx.transformMsg(origMsg, origMsg.getType(), origMsg.getOriginator(), metaData, origMsg.getData());
}
private TbMsg processException(TbContext ctx, TbMsg origMsg, Exception e) {
TbMsgMetaData metaData = origMsg.getMetaData().copy();
metaData.putValue(ERROR, e.getClass() + ": " + e.getMessage());
return ctx.transformMsg(origMsg, origMsg.getType(), origMsg.getOriginator(), metaData, origMsg.getData());
}
}

55
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeConfiguration.java Normal file
View File

@ -0,0 +1,55 @@
/**
* 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.rule.engine.kafka;
import lombok.Data;
import org.apache.kafka.common.serialization.StringSerializer;
import org.thingsboard.rule.engine.api.NodeConfiguration;
import java.util.Collections;
import java.util.Map;
@Data
public class TbKafkaNodeConfiguration implements NodeConfiguration<TbKafkaNodeConfiguration> {
private String topicPattern;
private String bootstrapServers;
private int retries;
private int batchSize;
private int linger;
private int bufferMemory;
private String acks;
private String keySerializer;
private String valueSerializer;
private Map<String, String> otherProperties;
@Override
public TbKafkaNodeConfiguration defaultConfiguration() {
TbKafkaNodeConfiguration configuration = new TbKafkaNodeConfiguration();
configuration.setTopicPattern("my-topic");
configuration.setBootstrapServers("localhost:9092");
configuration.setRetries(0);
configuration.setBatchSize(16384);
configuration.setLinger(0);
configuration.setBufferMemory(33554432);
configuration.setAcks("-1");
configuration.setKeySerializer(StringSerializer.class.getName());
configuration.setValueSerializer(StringSerializer.class.getName());
configuration.setOtherProperties(Collections.emptyMap());
return configuration;
}
}

21
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNode.java
View File

@ -50,7 +50,6 @@ import java.util.concurrent.TimeUnit;
)
public class TbRestApiCallNode implements TbNode {
private static final String VARIABLE_TEMPLATE = "${%s}";
private static final String STATUS = "status";
private static final String STATUS_CODE = "statusCode";
private static final String STATUS_REASON = "statusReason";
@ -77,19 +76,19 @@ public class TbRestApiCallNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
String endpointUrl = processPattern(config.getRestEndpointUrlPattern(), msg.getMetaData());
String endpointUrl = TbNodeUtils.processPattern(config.getRestEndpointUrlPattern(), msg.getMetaData());
HttpHeaders headers = prepareHeaders(msg.getMetaData());
HttpMethod method = HttpMethod.valueOf(config.getRequestMethod());
HttpEntity<String> entity = new HttpEntity<>(msg.getData(), headers);
ListenableFuture<ResponseEntity<String>> future =httpClient.exchange(
ListenableFuture<ResponseEntity<String>> future = httpClient.exchange(
endpointUrl, method, entity, String.class);
future.addCallback(new ListenableFutureCallback<ResponseEntity<String>>() {
@Override
public void onFailure(Throwable throwable) {
TbMsg next = processException(ctx, msg, throwable);
ctx.tellNext(next, TbRelationTypes.FAILURE);
ctx.tellNext(next, TbRelationTypes.FAILURE, throwable);
}
@Override
@ -145,21 +144,9 @@ public class TbRestApiCallNode implements TbNode {
private HttpHeaders prepareHeaders(TbMsgMetaData metaData) {
HttpHeaders headers = new HttpHeaders();
config.getHeaders().forEach((k,v) -> {
headers.add(processPattern(k, metaData), processPattern(v, metaData));
headers.add(TbNodeUtils.processPattern(k, metaData), TbNodeUtils.processPattern(v, metaData));
});
return headers;
}
private String processPattern(String pattern, TbMsgMetaData metaData) {
String result = new String(pattern);
for (Map.Entry<String,String> keyVal : metaData.values().entrySet()) {
result = processVar(result, keyVal.getKey(), keyVal.getValue());
}
return result;
}
private String processVar(String pattern, String key, String val) {
String varPattern = String.format(VARIABLE_TEMPLATE, key);
return pattern.replace(varPattern, val);
}
}

4
rule-engine/rule-engine-components/src/main/resources/public/static/rulenode/rulenode-core-config.js
View File

File diff suppressed because one or more lines are too long