|
|
|
@ -18,7 +18,9 @@ package org.thingsboard.server.service.queue; |
|
|
|
import com.google.protobuf.ByteString; |
|
|
|
import lombok.RequiredArgsConstructor; |
|
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
|
import org.springframework.beans.factory.annotation.Value; |
|
|
|
import org.springframework.context.annotation.Lazy; |
|
|
|
import org.springframework.scheduling.annotation.Scheduled; |
|
|
|
import org.springframework.stereotype.Service; |
|
|
|
import org.thingsboard.rule.engine.api.msg.DeviceEdgeUpdateMsg; |
|
|
|
@ -27,6 +29,7 @@ import org.thingsboard.server.cluster.TbClusterService; |
|
|
|
import org.thingsboard.server.common.data.EdgeUtils; |
|
|
|
import org.thingsboard.server.common.data.edge.EdgeEventActionType; |
|
|
|
import org.thingsboard.server.common.data.edge.EdgeEventType; |
|
|
|
import org.thingsboard.server.common.data.queue.Queue; |
|
|
|
import org.thingsboard.server.common.msg.ToDeviceActorNotificationMsg; |
|
|
|
import org.thingsboard.server.common.data.ApiUsageState; |
|
|
|
import org.thingsboard.server.common.data.Device; |
|
|
|
@ -60,6 +63,7 @@ import org.thingsboard.server.queue.TbQueueCallback; |
|
|
|
import org.thingsboard.server.queue.TbQueueProducer; |
|
|
|
import org.thingsboard.server.queue.common.MultipleTbQueueCallbackWrapper; |
|
|
|
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|
|
|
import org.thingsboard.server.queue.discovery.NotificationsTopicService; |
|
|
|
import org.thingsboard.server.queue.discovery.PartitionService; |
|
|
|
import org.thingsboard.server.queue.provider.TbQueueProducerProvider; |
|
|
|
import org.thingsboard.server.service.gateway_device.GatewayNotificationsService; |
|
|
|
@ -88,8 +92,12 @@ public class DefaultTbClusterService implements TbClusterService { |
|
|
|
private final AtomicInteger toRuleEngineNfs = new AtomicInteger(0); |
|
|
|
private final AtomicInteger toTransportNfs = new AtomicInteger(0); |
|
|
|
|
|
|
|
@Autowired |
|
|
|
@Lazy |
|
|
|
private PartitionService partitionService; |
|
|
|
|
|
|
|
private final TbQueueProducerProvider producerProvider; |
|
|
|
private final PartitionService partitionService; |
|
|
|
private final NotificationsTopicService notificationsTopicService; |
|
|
|
private final DataDecodingEncodingService encodingService; |
|
|
|
private final TbDeviceProfileCache deviceProfileCache; |
|
|
|
private final OtaPackageStateService otaPackageStateService; |
|
|
|
@ -120,7 +128,7 @@ public class DefaultTbClusterService implements TbClusterService { |
|
|
|
|
|
|
|
@Override |
|
|
|
public void pushNotificationToCore(String serviceId, FromDeviceRpcResponse response, TbQueueCallback callback) { |
|
|
|
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); |
|
|
|
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); |
|
|
|
log.trace("PUSHING msg: {} to:{}", response, tpi); |
|
|
|
FromDeviceRPCResponseProto.Builder builder = FromDeviceRPCResponseProto.newBuilder() |
|
|
|
.setRequestIdMSB(response.getId().getMostSignificantBits()) |
|
|
|
@ -185,7 +193,7 @@ public class DefaultTbClusterService implements TbClusterService { |
|
|
|
|
|
|
|
@Override |
|
|
|
public void pushNotificationToRuleEngine(String serviceId, FromDeviceRpcResponse response, TbQueueCallback callback) { |
|
|
|
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceId); |
|
|
|
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceId); |
|
|
|
log.trace("PUSHING msg: {} to:{}", response, tpi); |
|
|
|
FromDeviceRPCResponseProto.Builder builder = FromDeviceRPCResponseProto.newBuilder() |
|
|
|
.setRequestIdMSB(response.getId().getMostSignificantBits()) |
|
|
|
@ -199,14 +207,14 @@ public class DefaultTbClusterService implements TbClusterService { |
|
|
|
|
|
|
|
@Override |
|
|
|
public void pushNotificationToTransport(String serviceId, ToTransportMsg response, TbQueueCallback callback) { |
|
|
|
if (serviceId == null || serviceId.isEmpty()){ |
|
|
|
if (serviceId == null || serviceId.isEmpty()) { |
|
|
|
log.trace("pushNotificationToTransport: skipping message without serviceId [{}], (ToTransportMsg) response [{}]", serviceId, response); |
|
|
|
if (callback != null) { |
|
|
|
callback.onSuccess(null); //callback that message already sent, no useful payload expected
|
|
|
|
} |
|
|
|
return; |
|
|
|
} |
|
|
|
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_TRANSPORT, serviceId); |
|
|
|
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, serviceId); |
|
|
|
log.trace("PUSHING msg: {} to:{}", response, tpi); |
|
|
|
producerProvider.getTransportNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), response), callback); |
|
|
|
toTransportNfs.incrementAndGet(); |
|
|
|
@ -314,7 +322,7 @@ public class DefaultTbClusterService implements TbClusterService { |
|
|
|
Set<String> tbTransportServices = partitionService.getAllServiceIds(ServiceType.TB_TRANSPORT); |
|
|
|
TbQueueCallback proxyCallback = callback != null ? new MultipleTbQueueCallbackWrapper(tbTransportServices.size(), callback) : null; |
|
|
|
for (String transportServiceId : tbTransportServices) { |
|
|
|
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_TRANSPORT, transportServiceId); |
|
|
|
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, transportServiceId); |
|
|
|
toTransportNfProducer.send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), transportMsg), proxyCallback); |
|
|
|
toTransportNfs.incrementAndGet(); |
|
|
|
} |
|
|
|
@ -328,7 +336,7 @@ public class DefaultTbClusterService implements TbClusterService { |
|
|
|
TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> toCoreNfProducer = producerProvider.getTbCoreNotificationsMsgProducer(); |
|
|
|
Set<String> tbCoreServices = partitionService.getAllServiceIds(ServiceType.TB_CORE); |
|
|
|
for (String serviceId : tbCoreServices) { |
|
|
|
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); |
|
|
|
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); |
|
|
|
ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setEdgeEventUpdateMsg(ByteString.copyFrom(msgBytes)).build(); |
|
|
|
toCoreNfProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEdgeId().getId(), toCoreMsg), null); |
|
|
|
toCoreNfs.incrementAndGet(); |
|
|
|
@ -349,7 +357,7 @@ public class DefaultTbClusterService implements TbClusterService { |
|
|
|
TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> toCoreNfProducer = producerProvider.getTbCoreNotificationsMsgProducer(); |
|
|
|
Set<String> tbCoreServices = partitionService.getAllServiceIds(ServiceType.TB_CORE); |
|
|
|
for (String serviceId : tbCoreServices) { |
|
|
|
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); |
|
|
|
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); |
|
|
|
ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setComponentLifecycleMsg(ByteString.copyFrom(msgBytes)).build(); |
|
|
|
toCoreNfProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEntityId().getId(), toCoreMsg), null); |
|
|
|
toCoreNfs.incrementAndGet(); |
|
|
|
@ -358,7 +366,7 @@ public class DefaultTbClusterService implements TbClusterService { |
|
|
|
tbRuleEngineServices.removeAll(tbCoreServices); |
|
|
|
} |
|
|
|
for (String serviceId : tbRuleEngineServices) { |
|
|
|
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceId); |
|
|
|
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceId); |
|
|
|
ToRuleEngineNotificationMsg toRuleEngineMsg = ToRuleEngineNotificationMsg.newBuilder().setComponentLifecycleMsg(ByteString.copyFrom(msgBytes)).build(); |
|
|
|
toRuleEngineProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEntityId().getId(), toRuleEngineMsg), null); |
|
|
|
toRuleEngineNfs.incrementAndGet(); |
|
|
|
@ -473,4 +481,93 @@ public class DefaultTbClusterService implements TbClusterService { |
|
|
|
break; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void onQueueChange(Queue queue, TbQueueCallback callback) { |
|
|
|
log.trace("[{}][{}] Processing queue change [{}] event", queue.getTenantId(), queue.getId(), queue.getName()); |
|
|
|
|
|
|
|
// TransportProtos.QueueUpdateMsg queueUpdateMsg = TransportProtos.QueueUpdateMsg.newBuilder()
|
|
|
|
// .setTenantIdMSB(queue.getTenantId().getId().getMostSignificantBits())
|
|
|
|
// .setTenantIdLSB(queue.getTenantId().getId().getLeastSignificantBits())
|
|
|
|
// .setQueueName(queue.getName())
|
|
|
|
// .setQueueTopic(queue.getTopic())
|
|
|
|
// .setPartitions(queue.getPartitions())
|
|
|
|
// .build();
|
|
|
|
//
|
|
|
|
// if ("Main".equals(queue.getName())) {
|
|
|
|
// Set<TransportProtos.ServiceInfo> tbTransportServices = partitionService.getAllServices(ServiceType.TB_TRANSPORT);
|
|
|
|
// ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setQueueUpdateMsg(queueUpdateMsg).build();
|
|
|
|
// for (TransportProtos.ServiceInfo transportService : tbTransportServices) {
|
|
|
|
// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, transportService.getServiceId());
|
|
|
|
// producerProvider.getTransportNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), transportMsg), callback);
|
|
|
|
// toTransportNfs.incrementAndGet();
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// Set<TransportProtos.ServiceInfo> tbCoreServices = partitionService.getAllServices(ServiceType.TB_CORE);
|
|
|
|
// ToCoreNotificationMsg coreMsg = ToCoreNotificationMsg.newBuilder().setQueueUpdateMsg(queueUpdateMsg).build();
|
|
|
|
// for (TransportProtos.ServiceInfo coreService : tbCoreServices) {
|
|
|
|
// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, coreService.getServiceId());
|
|
|
|
// producerProvider.getTbCoreNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), coreMsg), callback);
|
|
|
|
// toCoreNfs.incrementAndGet();
|
|
|
|
// }
|
|
|
|
// } else {
|
|
|
|
// Set<TransportProtos.ServiceInfo> tbRuleEngineServices = partitionService.getAllServices(ServiceType.TB_RULE_ENGINE);
|
|
|
|
// ToRuleEngineNotificationMsg ruleEngineMsg = ToRuleEngineNotificationMsg.newBuilder().setQueueUpdateMsg(queueUpdateMsg).build();
|
|
|
|
// for (TransportProtos.ServiceInfo ruleEngineService : tbRuleEngineServices) {
|
|
|
|
// TenantId tenantId = new TenantId(new UUID(ruleEngineService.getTenantIdMSB(), ruleEngineService.getTenantIdLSB()));
|
|
|
|
// if (tenantId.equals(queue.getTenantId())) {
|
|
|
|
// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, ruleEngineService.getServiceId());
|
|
|
|
// producerProvider.getRuleEngineNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), ruleEngineMsg), callback);
|
|
|
|
// toRuleEngineNfs.incrementAndGet();
|
|
|
|
// }
|
|
|
|
// }
|
|
|
|
// }
|
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void onQueueDelete(Queue queue, TbQueueCallback callback) { |
|
|
|
log.trace("[{}][{}] Processing queue delete [{}] event", queue.getTenantId(), queue.getId(), queue.getName()); |
|
|
|
|
|
|
|
// TransportProtos.QueueDeleteMsg queueDeleteMsg = TransportProtos.QueueDeleteMsg.newBuilder()
|
|
|
|
// .setTenantIdMSB(queue.getTenantId().getId().getMostSignificantBits())
|
|
|
|
// .setTenantIdLSB(queue.getTenantId().getId().getLeastSignificantBits())
|
|
|
|
// .setQueueName(queue.getName())
|
|
|
|
// .build();
|
|
|
|
//
|
|
|
|
// if ("Main".equals(queue.getName())) {
|
|
|
|
// Set<TransportProtos.ServiceInfo> tbTransportServices = partitionService.getAllServices(ServiceType.TB_TRANSPORT);
|
|
|
|
// ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setQueueDeleteMsg(queueDeleteMsg).build();
|
|
|
|
// for (TransportProtos.ServiceInfo transportService : tbTransportServices) {
|
|
|
|
// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, transportService.getServiceId());
|
|
|
|
// producerProvider.getTransportNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), transportMsg), callback);
|
|
|
|
// toTransportNfs.incrementAndGet();
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// Set<TransportProtos.ServiceInfo> tbCoreServices = partitionService.getAllServices(ServiceType.TB_CORE);
|
|
|
|
// ToCoreNotificationMsg coreMsg = ToCoreNotificationMsg.newBuilder().setQueueDeleteMsg(queueDeleteMsg).build();
|
|
|
|
// for (TransportProtos.ServiceInfo coreService : tbCoreServices) {
|
|
|
|
// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, coreService.getServiceId());
|
|
|
|
// producerProvider.getTbCoreNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), coreMsg), callback);
|
|
|
|
// toCoreNfs.incrementAndGet();
|
|
|
|
// }
|
|
|
|
// } else {
|
|
|
|
// //TODO: 3.2 should be changed with smth like TransportApiRequestTemplate and get responses from all RE's
|
|
|
|
// Set<TransportProtos.ServiceInfo> tbRuleEngineServices = partitionService.getAllServices(ServiceType.TB_RULE_ENGINE);
|
|
|
|
// ToRuleEngineNotificationMsg ruleEngineMsg = ToRuleEngineNotificationMsg.newBuilder().setQueueDeleteMsg(queueDeleteMsg).build();
|
|
|
|
// for (TransportProtos.ServiceInfo ruleEngineService : tbRuleEngineServices) {
|
|
|
|
// TenantId tenantId = new TenantId(new UUID(ruleEngineService.getTenantIdMSB(), ruleEngineService.getTenantIdLSB()));
|
|
|
|
// if (tenantId.equals(queue.getTenantId())) {
|
|
|
|
// TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, ruleEngineService.getServiceId());
|
|
|
|
// log.info("Send notification about deleting queue [{}] to Rule Engine [{}]", queue.getName(), tpi.getFullTopicName());
|
|
|
|
// producerProvider.getRuleEngineNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), ruleEngineMsg), callback);
|
|
|
|
// toRuleEngineNfs.incrementAndGet();
|
|
|
|
// }
|
|
|
|
// }
|
|
|
|
// try {
|
|
|
|
// Thread.sleep(3000);
|
|
|
|
// } catch (InterruptedException e) {
|
|
|
|
// e.printStackTrace();
|
|
|
|
// }
|
|
|
|
// }
|
|
|
|
} |
|
|
|
} |
|
|
|
|
YevhenBondarenko
4 years ago