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

refactored

Browse Source
This commit is contained in:
YevhenBondarenko 2020-12-30 15:13:23 +02:00 committed by Andrew Shvayka
parent 4de258f2ae
commit 1b905ef727
4 changed files with 9 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java
View File

@ -165,7 +165,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
consumerBuilder.settings(kafkaSettings); consumerBuilder.settings(kafkaSettings);
consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName());
consumerBuilder.clientId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.clientId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId());
consumerBuilder.groupId("monolith-rule-engine-notifications-consumer"); consumerBuilder.groupId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId());
consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
consumerBuilder.admin(notificationAdmin); consumerBuilder.admin(notificationAdmin);
return consumerBuilder.build(); return consumerBuilder.build();
@ -189,7 +189,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
consumerBuilder.settings(kafkaSettings); consumerBuilder.settings(kafkaSettings);
consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName());
consumerBuilder.clientId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.clientId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId());
consumerBuilder.groupId("monolith-core-notifications-consumer"); consumerBuilder.groupId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId());
consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
consumerBuilder.admin(notificationAdmin); consumerBuilder.admin(notificationAdmin);
return consumerBuilder.build(); return consumerBuilder.build();
@ -230,7 +230,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
responseBuilder.settings(kafkaSettings); responseBuilder.settings(kafkaSettings);
responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId());
responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId());
responseBuilder.groupId("rule-engine-node"); responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId());
responseBuilder.decoder(msg -> { responseBuilder.decoder(msg -> {
JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder();
JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder);

4
common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java
View File

@ -159,7 +159,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
consumerBuilder.settings(kafkaSettings); consumerBuilder.settings(kafkaSettings);
consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName());
consumerBuilder.clientId("tb-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.clientId("tb-core-notifications-consumer-" + serviceInfoProvider.getServiceId());
consumerBuilder.groupId("tb-core-notifications-node"); consumerBuilder.groupId("tb-core-notifications-node-" + serviceInfoProvider.getServiceId());
consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
consumerBuilder.admin(notificationAdmin); consumerBuilder.admin(notificationAdmin);
return consumerBuilder.build(); return consumerBuilder.build();
@ -200,7 +200,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
responseBuilder.settings(kafkaSettings); responseBuilder.settings(kafkaSettings);
responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId());
responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId());
responseBuilder.groupId("rule-engine-node"); responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId());
responseBuilder.decoder(msg -> { responseBuilder.decoder(msg -> {
JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder();
JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder);

4
common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java
View File

@ -154,7 +154,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
consumerBuilder.settings(kafkaSettings); consumerBuilder.settings(kafkaSettings);
consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName());
consumerBuilder.clientId("tb-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.clientId("tb-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId());
consumerBuilder.groupId("tb-rule-engine-notifications-node"); consumerBuilder.groupId("tb-rule-engine-notifications-node-" + serviceInfoProvider.getServiceId());
consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
consumerBuilder.admin(notificationAdmin); consumerBuilder.admin(notificationAdmin);
return consumerBuilder.build(); return consumerBuilder.build();
@ -173,7 +173,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory {
responseBuilder.settings(kafkaSettings); responseBuilder.settings(kafkaSettings);
responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.topic(jsInvokeSettings.getResponseTopic() + "." + serviceInfoProvider.getServiceId());
responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId()); responseBuilder.clientId("js-" + serviceInfoProvider.getServiceId());
responseBuilder.groupId("rule-engine-node"); responseBuilder.groupId("rule-engine-node-" + serviceInfoProvider.getServiceId());
responseBuilder.decoder(msg -> { responseBuilder.decoder(msg -> {
JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder(); JsInvokeProtos.RemoteJsResponse.Builder builder = JsInvokeProtos.RemoteJsResponse.newBuilder();
JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder); JsonFormat.parser().ignoringUnknownFields().merge(new String(msg.getData(), StandardCharsets.UTF_8), builder);

4
common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbTransportQueueFactory.java
View File

@ -92,7 +92,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory {
responseBuilder.settings(kafkaSettings); responseBuilder.settings(kafkaSettings);
responseBuilder.topic(transportApiSettings.getResponsesTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.topic(transportApiSettings.getResponsesTopic() + "." + serviceInfoProvider.getServiceId());
responseBuilder.clientId("transport-api-response-" + serviceInfoProvider.getServiceId()); responseBuilder.clientId("transport-api-response-" + serviceInfoProvider.getServiceId());
responseBuilder.groupId("transport-node"); responseBuilder.groupId("transport-node-" + serviceInfoProvider.getServiceId());
responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders()));
responseBuilder.admin(transportApiAdmin); responseBuilder.admin(transportApiAdmin);
@ -133,7 +133,7 @@ public class KafkaTbTransportQueueFactory implements TbTransportQueueFactory {
responseBuilder.settings(kafkaSettings); responseBuilder.settings(kafkaSettings);
responseBuilder.topic(transportNotificationSettings.getNotificationsTopic() + "." + serviceInfoProvider.getServiceId()); responseBuilder.topic(transportNotificationSettings.getNotificationsTopic() + "." + serviceInfoProvider.getServiceId());
responseBuilder.clientId("transport-api-notifications-" + serviceInfoProvider.getServiceId()); responseBuilder.clientId("transport-api-notifications-" + serviceInfoProvider.getServiceId());
responseBuilder.groupId("transport-node"); responseBuilder.groupId("transport-node-" + serviceInfoProvider.getServiceId());
responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders())); responseBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToTransportMsg.parseFrom(msg.getData()), msg.getHeaders()));
responseBuilder.admin(notificationAdmin); responseBuilder.admin(notificationAdmin);
return responseBuilder.build(); return responseBuilder.build();