Browse Source

Merge branch 'master' of github.com:thingsboard/thingsboard into improvement/ruleNodes

pull/1483/head
ShvaykaD 8 years ago
parent
commit
8d53bf4c65
  1. 1
      application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java
  2. 1
      application/src/main/java/org/thingsboard/server/service/transport/RemoteRuleEngineTransportService.java
  3. 1
      application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java
  4. 5
      common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java
  5. 2
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RemoteTransportService.java
  6. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java
  7. 6
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNode.java
  8. 18
      ui/src/app/components/widget/widget.controller.js
  9. 23
      ui/src/app/widget/lib/timeseries-table-widget.js

1
application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java

@ -77,6 +77,7 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService {
public void init() { public void init() {
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<JsInvokeProtos.RemoteJsRequest> requestBuilder = TBKafkaProducerTemplate.builder(); TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<JsInvokeProtos.RemoteJsRequest> requestBuilder = TBKafkaProducerTemplate.builder();
requestBuilder.settings(kafkaSettings); requestBuilder.settings(kafkaSettings);
requestBuilder.clientId("producer-js-invoke-" + nodeIdProvider.getNodeId());
requestBuilder.defaultTopic(requestTopic); requestBuilder.defaultTopic(requestTopic);
requestBuilder.encoder(new RemoteJsRequestEncoder()); requestBuilder.encoder(new RemoteJsRequestEncoder());
requestBuilder.enricher((request, responseTopic, requestId) -> { requestBuilder.enricher((request, responseTopic, requestId) -> {

1
application/src/main/java/org/thingsboard/server/service/transport/RemoteRuleEngineTransportService.java

@ -112,6 +112,7 @@ public class RemoteRuleEngineTransportService implements RuleEngineTransportServ
public void init() { public void init() {
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<ToTransportMsg> notificationsProducerBuilder = TBKafkaProducerTemplate.builder(); TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<ToTransportMsg> notificationsProducerBuilder = TBKafkaProducerTemplate.builder();
notificationsProducerBuilder.settings(kafkaSettings); notificationsProducerBuilder.settings(kafkaSettings);
notificationsProducerBuilder.clientId("producer-transport-notification-" + nodeIdProvider.getNodeId());
notificationsProducerBuilder.encoder(new ToTransportMsgEncoder()); notificationsProducerBuilder.encoder(new ToTransportMsgEncoder());
notificationsProducer = notificationsProducerBuilder.build(); notificationsProducer = notificationsProducerBuilder.build();

1
application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java

@ -68,6 +68,7 @@ public class RemoteTransportApiService {
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TransportApiResponseMsg> responseBuilder = TBKafkaProducerTemplate.builder(); TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TransportApiResponseMsg> responseBuilder = TBKafkaProducerTemplate.builder();
responseBuilder.settings(kafkaSettings); responseBuilder.settings(kafkaSettings);
responseBuilder.clientId("producer-transport-api-response-" + nodeIdProvider.getNodeId());
responseBuilder.encoder(new TransportApiResponseEncoder()); responseBuilder.encoder(new TransportApiResponseEncoder());
TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<TransportApiRequestMsg> requestBuilder = TBKafkaConsumerTemplate.builder(); TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder<TransportApiRequestMsg> requestBuilder = TBKafkaConsumerTemplate.builder();

5
common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java

@ -62,10 +62,13 @@ public class TBKafkaProducerTemplate<T> {
@Builder @Builder
private TBKafkaProducerTemplate(TbKafkaSettings settings, TbKafkaEncoder<T> encoder, TbKafkaEnricher<T> enricher, private TBKafkaProducerTemplate(TbKafkaSettings settings, TbKafkaEncoder<T> encoder, TbKafkaEnricher<T> enricher,
TbKafkaPartitioner<T> partitioner, String defaultTopic) { TbKafkaPartitioner<T> partitioner, String defaultTopic, String clientId) {
Properties props = settings.toProps(); Properties props = settings.toProps();
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
if (!StringUtils.isEmpty(clientId)) {
props.put(ProducerConfig.CLIENT_ID_CONFIG, clientId);
}
this.settings = settings; this.settings = settings;
this.producer = new KafkaProducer<>(props); this.producer = new KafkaProducer<>(props);
this.encoder = encoder; this.encoder = encoder;

2
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RemoteTransportService.java

@ -104,6 +104,7 @@ public class RemoteTransportService extends AbstractTransportService {
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TransportApiRequestMsg> requestBuilder = TBKafkaProducerTemplate.builder(); TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TransportApiRequestMsg> requestBuilder = TBKafkaProducerTemplate.builder();
requestBuilder.settings(kafkaSettings); requestBuilder.settings(kafkaSettings);
requestBuilder.clientId("producer-transport-api-request-" + nodeIdProvider.getNodeId());
requestBuilder.defaultTopic(transportApiRequestsTopic); requestBuilder.defaultTopic(transportApiRequestsTopic);
requestBuilder.encoder(new TransportApiRequestEncoder()); requestBuilder.encoder(new TransportApiRequestEncoder());
@ -128,6 +129,7 @@ public class RemoteTransportService extends AbstractTransportService {
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<ToRuleEngineMsg> ruleEngineProducerBuilder = TBKafkaProducerTemplate.builder(); TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<ToRuleEngineMsg> ruleEngineProducerBuilder = TBKafkaProducerTemplate.builder();
ruleEngineProducerBuilder.settings(kafkaSettings); ruleEngineProducerBuilder.settings(kafkaSettings);
ruleEngineProducerBuilder.clientId("producer-rule-engine-request-" + nodeIdProvider.getNodeId());
ruleEngineProducerBuilder.defaultTopic(ruleEngineTopic); ruleEngineProducerBuilder.defaultTopic(ruleEngineTopic);
ruleEngineProducerBuilder.encoder(new ToRuleEngineMsgEncoder()); ruleEngineProducerBuilder.encoder(new ToRuleEngineMsgEncoder());
ruleEngineProducer = ruleEngineProducerBuilder.build(); ruleEngineProducer = ruleEngineProducerBuilder.build();

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

@ -25,6 +25,7 @@ import org.thingsboard.server.common.msg.TbMsgMetaData;
import java.util.Properties; import java.util.Properties;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
@Slf4j @Slf4j
@RuleNode( @RuleNode(
@ -54,6 +55,7 @@ public class TbKafkaNode implements TbNode {
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
this.config = TbNodeUtils.convert(configuration, TbKafkaNodeConfiguration.class); this.config = TbNodeUtils.convert(configuration, TbKafkaNodeConfiguration.class);
Properties properties = new Properties(); Properties properties = new Properties();
properties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-tb-kafka-node-" + ctx.getSelfId().getId().toString() + "-" + ctx.getNodeId());
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, config.getValueSerializer()); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, config.getValueSerializer());
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer()); properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer());

6
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNode.java

@ -24,6 +24,7 @@ import com.google.common.util.concurrent.ListenableFuture;
import lombok.Data; import lombok.Data;
import lombok.NoArgsConstructor; import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils; import org.apache.commons.lang3.math.NumberUtils;
import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbContext;
@ -66,6 +67,8 @@ import static org.thingsboard.server.common.data.kv.Aggregation.NONE;
configDirective = "tbEnrichmentNodeGetTelemetryFromDatabase") configDirective = "tbEnrichmentNodeGetTelemetryFromDatabase")
public class TbGetTelemetryNode implements TbNode { public class TbGetTelemetryNode implements TbNode {
private static final String DESC_ORDER = "DESC";
private TbGetTelemetryNodeConfiguration config; private TbGetTelemetryNodeConfiguration config;
private List<String> tsKeyNames; private List<String> tsKeyNames;
private int limit; private int limit;
@ -80,6 +83,9 @@ public class TbGetTelemetryNode implements TbNode {
limit = config.getFetchMode().equals(FETCH_MODE_ALL) ? MAX_FETCH_SIZE : 1; limit = config.getFetchMode().equals(FETCH_MODE_ALL) ? MAX_FETCH_SIZE : 1;
fetchMode = config.getFetchMode(); fetchMode = config.getFetchMode();
orderBy = config.getOrderBy(); orderBy = config.getOrderBy();
if (StringUtils.isEmpty(orderBy)) {
orderBy = DESC_ORDER;
}
mapper = new ObjectMapper(); mapper = new ObjectMapper();
mapper.configure(JsonGenerator.Feature.QUOTE_FIELD_NAMES, false); mapper.configure(JsonGenerator.Feature.QUOTE_FIELD_NAMES, false);
mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true); mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true);

18
ui/src/app/components/widget/widget.controller.js

@ -138,7 +138,7 @@ export default function WidgetController($scope, $state, $timeout, $window, $ele
headerAction.icon = descriptor.icon; headerAction.icon = descriptor.icon;
headerAction.descriptor = descriptor; headerAction.descriptor = descriptor;
headerAction.onAction = function($event) { headerAction.onAction = function($event) {
var entityInfo = getFirstEntityInfo(); var entityInfo = getActiveEntityInfo();
var entityId = entityInfo ? entityInfo.entityId : null; var entityId = entityInfo ? entityInfo.entityId : null;
var entityName = entityInfo ? entityInfo.entityName : null; var entityName = entityInfo ? entityInfo.entityName : null;
handleWidgetAction($event, this.descriptor, entityId, entityName); handleWidgetAction($event, this.descriptor, entityId, entityName);
@ -502,13 +502,15 @@ export default function WidgetController($scope, $state, $timeout, $window, $ele
} }
} }
function getFirstEntityInfo() { function getActiveEntityInfo() {
var entityInfo; var entityInfo = widgetContext.activeEntityInfo;
for (var id in widgetContext.subscriptions) { if (!entityInfo) {
var subscription = widgetContext.subscriptions[id]; for (var id in widgetContext.subscriptions) {
entityInfo = subscription.getFirstEntityInfo(); var subscription = widgetContext.subscriptions[id];
if (entityInfo) { entityInfo = subscription.getFirstEntityInfo();
break; if (entityInfo) {
break;
}
} }
} }
return entityInfo; return entityInfo;

23
ui/src/app/widget/lib/timeseries-table-widget.js

@ -44,7 +44,7 @@ function TimeseriesTableWidget() {
} }
/*@ngInject*/ /*@ngInject*/
function TimeseriesTableWidgetController($element, $scope, $filter, $timeout) { function TimeseriesTableWidgetController($element, $scope, $filter, $timeout, types) {
var vm = this; var vm = this;
let dateFormatFilter = 'yyyy-MM-dd HH:mm:ss'; let dateFormatFilter = 'yyyy-MM-dd HH:mm:ss';
@ -228,9 +228,29 @@ function TimeseriesTableWidgetController($element, $scope, $filter, $timeout) {
$scope.$watch('vm.sourceIndex', function(newIndex, oldIndex) { $scope.$watch('vm.sourceIndex', function(newIndex, oldIndex) {
if (newIndex != oldIndex) { if (newIndex != oldIndex) {
updateSourceData(vm.sources[vm.sourceIndex]); updateSourceData(vm.sources[vm.sourceIndex]);
updateActiveEntityInfo();
} }
}); });
function updateActiveEntityInfo() {
var source = vm.sources[vm.sourceIndex];
var activeEntityInfo = null;
if (source) {
var datasource = source.datasource;
if (datasource.type === types.datasourceType.entity &&
datasource.entityType && datasource.entityId) {
activeEntityInfo = {
entityId: {
entityType: datasource.entityType,
id: datasource.entityId
},
entityName: datasource.entityName
};
}
}
vm.ctx.activeEntityInfo = activeEntityInfo;
}
function updateDatasources() { function updateDatasources() {
vm.sources = []; vm.sources = [];
vm.sourceIndex = 0; vm.sourceIndex = 0;
@ -314,6 +334,7 @@ function TimeseriesTableWidgetController($element, $scope, $filter, $timeout) {
vm.sources.push(source); vm.sources.push(source);
} }
} }
updateActiveEntityInfo();
} }
function updatePage(source) { function updatePage(source) {

Loading…
Cancel
Save