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

Fix for Transport API subscriptions

Browse Source
This commit is contained in:
Andrii Shvaika 2020-05-20 15:37:06 +03:00
parent acf900e1de
commit 676868b4d3
2 changed files with 2 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -200,7 +200,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
consumerBuilder.settings(kafkaSettings); consumerBuilder.settings(kafkaSettings);
consumerBuilder.topic(transportApiSettings.getRequestsTopic()); consumerBuilder.topic(transportApiSettings.getRequestsTopic());
consumerBuilder.clientId("monolith-transport-api-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.clientId("monolith-transport-api-consumer-" + serviceInfoProvider.getServiceId());
consumerBuilder.groupId("monolith-transport-api-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("monolith-transport-api-consumer");
consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders()));
consumerBuilder.admin(transportApiAdmin); consumerBuilder.admin(transportApiAdmin);
return consumerBuilder.build(); return consumerBuilder.build();

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

@ -170,7 +170,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
consumerBuilder.settings(kafkaSettings); consumerBuilder.settings(kafkaSettings);
consumerBuilder.topic(transportApiSettings.getRequestsTopic()); consumerBuilder.topic(transportApiSettings.getRequestsTopic());
consumerBuilder.clientId("tb-core-transport-api-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.clientId("tb-core-transport-api-consumer-" + serviceInfoProvider.getServiceId());
consumerBuilder.groupId("tb-core-transport-api-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("tb-core-transport-api-consumer");
consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders())); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiRequestMsg.parseFrom(msg.getData()), msg.getHeaders()));
consumerBuilder.admin(transportApiAdmin); consumerBuilder.admin(transportApiAdmin);
return consumerBuilder.build(); return consumerBuilder.build();