diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java index 4344507dd9..ebe99d99e0 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java @@ -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 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> toCoreNfProducer = producerProvider.getTbCoreNotificationsMsgProducer(); Set 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> toCoreNfProducer = producerProvider.getTbCoreNotificationsMsgProducer(); Set 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 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 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 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 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 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 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(); +// } +// } + } } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java index 83398e29ed..b158a51b77 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java @@ -50,6 +50,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionUpdate import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.queue.TbQueueProducer; 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.discovery.TbApplicationEventListener; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; @@ -88,6 +89,9 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene @Autowired private TimeseriesService tsService; + @Autowired + private NotificationsTopicService notificationsTopicService; + @Autowired private PartitionService partitionService; @@ -264,7 +268,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene updateDeviceInactivityTimeout(tenantId, entityId, attributes); } else if (TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(scope) && notifyDevice) { clusterService.pushMsgToCore(DeviceAttributesEventNotificationMsg.onUpdate(tenantId, - new DeviceId(entityId.getId()), DataConstants.SHARED_SCOPE, new ArrayList<>(attributes)) + new DeviceId(entityId.getId()), DataConstants.SHARED_SCOPE, new ArrayList<>(attributes)) , null); } } @@ -377,7 +381,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(s.getSubscriptionId(), subscriptionUpdate); localSubscriptionService.onSubscriptionUpdate(s.getSessionId(), update, TbCallback.EMPTY); } else { - TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId()); + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId()); toCoreNotificationsProducer.send(tpi, toProto(s, subscriptionUpdate, ignoreEmptyUpdates), null); } } @@ -400,7 +404,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene AlarmSubscriptionUpdate update = new AlarmSubscriptionUpdate(s.getSubscriptionId(), alarm, deleted); localSubscriptionService.onSubscriptionUpdate(s.getSessionId(), update, TbCallback.EMPTY); } else { - TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId()); + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId()); toCoreNotificationsProducer.send(tpi, toProto(s, alarm, deleted), null); } } @@ -441,7 +445,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene } }); if (!missedUpdates.isEmpty()) { - TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId()); + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId()); toCoreNotificationsProducer.send(tpi, toProto(subscription, missedUpdates), null); } }, @@ -464,7 +468,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene DonAsynchron.withCallback(tsService.findLatest(subscription.getTenantId(), subscription.getEntityId(), subscription.getKeyStates().keySet()), missedUpdates -> { if (missedUpdates != null && !missedUpdates.isEmpty()) { - TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId()); + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId()); toCoreNotificationsProducer.send(tpi, toProto(subscription, missedUpdates), null); } }, @@ -483,7 +487,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene DonAsynchron.withCallback(tsService.findAll(subscription.getTenantId(), subscription.getEntityId(), queries), missedUpdates -> { if (missedUpdates != null && !missedUpdates.isEmpty()) { - TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId()); + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId()); toCoreNotificationsProducer.send(tpi, toProto(subscription, missedUpdates), null); } }, @@ -533,7 +537,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene }); ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setToLocalSubscriptionServiceMsg( - LocalSubscriptionServiceMsgProto.newBuilder().setSubUpdate(builder.build()).build()) + LocalSubscriptionServiceMsgProto.newBuilder().setSubUpdate(builder.build()).build()) .build(); return new TbProtoQueueMsg<>(subscription.getEntityId().getId(), toCoreMsg); } @@ -547,8 +551,8 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene builder.setDeleted(deleted); ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setToLocalSubscriptionServiceMsg( - LocalSubscriptionServiceMsgProto.newBuilder() - .setAlarmSubUpdate(builder.build()).build()) + LocalSubscriptionServiceMsgProto.newBuilder() + .setAlarmSubUpdate(builder.build()).build()) .build(); return new TbProtoQueueMsg<>(subscription.getEntityId().getId(), toCoreMsg); } diff --git a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTbCoreToTransportService.java b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTbCoreToTransportService.java index 883cfd645c..0228743076 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTbCoreToTransportService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTbCoreToTransportService.java @@ -16,7 +16,6 @@ package org.thingsboard.server.service.transport; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; @@ -25,7 +24,7 @@ import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueMsgMetadata; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.util.TbCoreComponent; @@ -39,11 +38,11 @@ import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID; @TbCoreComponent public class DefaultTbCoreToTransportService implements TbCoreToTransportService { - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbQueueProducer> tbTransportProducer; - public DefaultTbCoreToTransportService(PartitionService partitionService, TbQueueProducerProvider tbQueueProducerProvider) { - this.partitionService = partitionService; + public DefaultTbCoreToTransportService(NotificationsTopicService notificationsTopicService, TbQueueProducerProvider tbQueueProducerProvider) { + this.notificationsTopicService = notificationsTopicService; this.tbTransportProducer = tbQueueProducerProvider.getTransportNotificationsMsgProducer(); } @@ -54,14 +53,14 @@ public class DefaultTbCoreToTransportService implements TbCoreToTransportService @Override public void process(String nodeId, ToTransportMsg msg, Runnable onSuccess, Consumer onFailure) { - if (nodeId == null || nodeId.isEmpty()){ + if (nodeId == null || nodeId.isEmpty()) { log.trace("process: skipping message without nodeId [{}], (ToTransportMsg) msg [{}]", nodeId, msg); if (onSuccess != null) { onSuccess.run(); } return; } - TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_TRANSPORT, nodeId); + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, nodeId); UUID sessionId = new UUID(msg.getSessionIdMSB(), msg.getSessionIdLSB()); log.trace("[{}][{}] Pushing session data to topic: {}", tpi.getFullTopicName(), sessionId, msg); TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(NULL_UUID, msg); diff --git a/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java b/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java index 7fb3a83425..22953f259a 100644 --- a/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java +++ b/common/cluster-api/src/main/java/org/thingsboard/server/cluster/TbClusterService.java @@ -15,31 +15,31 @@ */ package org.thingsboard.server.cluster; -import org.thingsboard.server.common.data.edge.EdgeEventActionType; -import org.thingsboard.server.common.data.edge.EdgeEventType; -import org.thingsboard.server.common.msg.ToDeviceActorNotificationMsg; import org.thingsboard.server.common.data.ApiUsageState; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.TbResource; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.TenantProfile; -import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.edge.EdgeEventActionType; +import org.thingsboard.server.common.data.edge.EdgeEventType; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.ToDeviceActorNotificationMsg; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.queue.TbQueueCallback; +import org.thingsboard.server.queue.TbQueueClusterService; import java.util.UUID; -public interface TbClusterService { +public interface TbClusterService extends TbQueueClusterService { void pushMsgToCore(TopicPartitionInfo tpi, UUID msgKey, ToCoreMsg msg, TbQueueCallback callback); diff --git a/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueClusterService.java b/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueClusterService.java new file mode 100644 index 0000000000..d37760211f --- /dev/null +++ b/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueClusterService.java @@ -0,0 +1,24 @@ +/** + * Copyright © 2016-2022 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.server.queue; + +import org.thingsboard.server.common.data.queue.Queue; + +public interface TbQueueClusterService { + void onQueueChange(Queue queue, TbQueueCallback callback); + + void onQueueDelete(Queue queue, TbQueueCallback callback); +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java index 4882a96f72..96df34186c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java @@ -213,20 +213,6 @@ public class HashPartitionService implements PartitionService { return result; } - @Override - public TopicPartitionInfo getNotificationsTopic(ServiceType serviceType, String serviceId) { - switch (serviceType) { - case TB_CORE: - return tbCoreNotificationTopics.computeIfAbsent(serviceId, - id -> buildNotificationsTopicPartitionInfo(serviceType, serviceId)); - case TB_RULE_ENGINE: - return tbRuleEngineNotificationTopics.computeIfAbsent(serviceId, - id -> buildNotificationsTopicPartitionInfo(serviceType, serviceId)); - default: - return buildNotificationsTopicPartitionInfo(serviceType, serviceId); - } - } - @Override public int resolvePartitionIndex(UUID entityId, int partitions) { int hash = hashFunction.newHasher() diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/NotificationsTopicService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/NotificationsTopicService.java new file mode 100644 index 0000000000..2c259a3ce0 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/NotificationsTopicService.java @@ -0,0 +1,54 @@ +/** + * Copyright © 2016-2022 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.server.queue.discovery; + +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; + +import java.util.HashMap; +import java.util.Map; + +@Service +public class NotificationsTopicService { + + private Map tbCoreNotificationTopics = new HashMap<>(); + private Map tbRuleEngineNotificationTopics = new HashMap<>(); + + /** + * Each Service should start a consumer for messages that target individual service instance based on serviceId. + * This topic is likely to have single partition, and is always assigned to the service. + * @param serviceType + * @param serviceId + * @return + */ + public TopicPartitionInfo getNotificationsTopic(ServiceType serviceType, String serviceId) { + switch (serviceType) { + case TB_CORE: + return tbCoreNotificationTopics.computeIfAbsent(serviceId, + id -> buildNotificationsTopicPartitionInfo(serviceType, serviceId)); + case TB_RULE_ENGINE: + return tbRuleEngineNotificationTopics.computeIfAbsent(serviceId, + id -> buildNotificationsTopicPartitionInfo(serviceType, serviceId)); + default: + return buildNotificationsTopicPartitionInfo(serviceType, serviceId); + } + } + + private TopicPartitionInfo buildNotificationsTopicPartitionInfo(ServiceType serviceType, String serviceId) { + return new TopicPartitionInfo(serviceType.name().toLowerCase() + ".notifications." + serviceId, null, null, false); + } +} \ No newline at end of file diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java index c011e6545a..c222b252a5 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java @@ -49,15 +49,6 @@ public interface PartitionService { */ Set getAllServiceIds(ServiceType serviceType); - /** - * Each Service should start a consumer for messages that target individual service instance based on serviceId. - * This topic is likely to have single partition, and is always assigned to the service. - * @param serviceType - * @param serviceId - * @return - */ - TopicPartitionInfo getNotificationsTopic(ServiceType serviceType, String serviceId); - int resolvePartitionIndex(UUID entityId, int partitions); int countTransportsByType(String type); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java index 9979e03608..873719c832 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java @@ -22,7 +22,15 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; -import org.thingsboard.server.gen.transport.TransportProtos.*; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToOtaPackageStateServiceMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.TbQueueProducer; @@ -30,7 +38,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; @@ -51,7 +59,7 @@ import java.nio.charset.StandardCharsets; @ConditionalOnExpression("'${queue.type:null}'=='aws-sqs' && '${service.type:null}'=='monolith'") public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngineQueueFactory { - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbQueueCoreSettings coreSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRuleEngineSettings ruleEngineSettings; @@ -66,7 +74,7 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng private final TbQueueAdmin transportApiAdmin; private final TbQueueAdmin notificationAdmin; - public AwsSqsMonolithQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, + public AwsSqsMonolithQueueFactory(NotificationsTopicService notificationsTopicService, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueTransportApiSettings transportApiSettings, @@ -74,7 +82,7 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng TbAwsSqsSettings sqsSettings, TbAwsSqsQueueAttributes sqsQueueAttributes, TbQueueRemoteJsInvokeSettings jsInvokeSettings) { - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.coreSettings = coreSettings; this.serviceInfoProvider = serviceInfoProvider; this.ruleEngineSettings = ruleEngineSettings; @@ -124,7 +132,7 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbAwsSqsConsumerTemplate<>(notificationAdmin, sqsSettings, - partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @@ -137,7 +145,7 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbAwsSqsConsumerTemplate<>(notificationAdmin, sqsSettings, - partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java index a09e832892..81736c2b55 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java @@ -37,7 +37,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; @@ -61,7 +61,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueCoreSettings coreSettings; private final TbQueueTransportApiSettings transportApiSettings; - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; @@ -76,7 +76,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { TbQueueCoreSettings coreSettings, TbQueueTransportApiSettings transportApiSettings, TbQueueRuleEngineSettings ruleEngineSettings, - PartitionService partitionService, + NotificationsTopicService notificationsTopicService, TbServiceInfoProvider serviceInfoProvider, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbAwsSqsQueueAttributes sqsQueueAttributes, @@ -85,7 +85,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { this.coreSettings = coreSettings; this.transportApiSettings = transportApiSettings; this.ruleEngineSettings = ruleEngineSettings; - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.serviceInfoProvider = serviceInfoProvider; this.jsInvokeSettings = jsInvokeSettings; this.transportNotificationSettings = transportNotificationSettings; @@ -131,7 +131,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbAwsSqsConsumerTemplate<>(notificationAdmin, sqsSettings, - partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java index ac4898e48d..9c223f08f5 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java @@ -32,7 +32,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; @@ -52,7 +52,7 @@ import java.nio.charset.StandardCharsets; @ConditionalOnExpression("'${queue.type:null}'=='aws-sqs' && '${service.type:null}'=='tb-rule-engine'") public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbQueueCoreSettings coreSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRuleEngineSettings ruleEngineSettings; @@ -65,14 +65,14 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory private final TbQueueAdmin jsExecutorAdmin; private final TbQueueAdmin notificationAdmin; - public AwsSqsTbRuleEngineQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, + public AwsSqsTbRuleEngineQueueFactory(NotificationsTopicService notificationsTopicService, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbServiceInfoProvider serviceInfoProvider, TbAwsSqsSettings sqsSettings, TbAwsSqsQueueAttributes sqsQueueAttributes, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueTransportNotificationSettings transportNotificationSettings) { - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.coreSettings = coreSettings; this.serviceInfoProvider = serviceInfoProvider; this.ruleEngineSettings = ruleEngineSettings; @@ -120,7 +120,7 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbAwsSqsConsumerTemplate<>(notificationAdmin, sqsSettings, - partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportProtos.ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java index 0548603cb7..e3c81df1dd 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryMonolithQueueFactory.java @@ -27,7 +27,7 @@ import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.memory.InMemoryStorage; import org.thingsboard.server.queue.memory.InMemoryTbQueueConsumer; @@ -43,7 +43,7 @@ import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; @ConditionalOnExpression("'${queue.type:null}'=='in-memory' && '${service.type:null}'=='monolith'") public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngineQueueFactory { - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbQueueCoreSettings coreSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRuleEngineSettings ruleEngineSettings; @@ -51,12 +51,12 @@ public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE private final TbQueueTransportNotificationSettings transportNotificationSettings; private final InMemoryStorage storage; - public InMemoryMonolithQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, + public InMemoryMonolithQueueFactory(NotificationsTopicService notificationsTopicService, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings) { - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.coreSettings = coreSettings; this.serviceInfoProvider = serviceInfoProvider; this.ruleEngineSettings = ruleEngineSettings; @@ -97,7 +97,7 @@ public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { - return new InMemoryTbQueueConsumer<>(partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); + return new InMemoryTbQueueConsumer<>(notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); } @Override @@ -107,7 +107,7 @@ public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { - return new InMemoryTbQueueConsumer<>(partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); + return new InMemoryTbQueueConsumer<>(notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java index cf75a77135..ce171497de 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java @@ -37,7 +37,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.kafka.TbKafkaAdmin; import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService; @@ -60,7 +60,7 @@ import java.util.concurrent.atomic.AtomicLong; @ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='monolith'") public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngineQueueFactory { - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbKafkaSettings kafkaSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueCoreSettings coreSettings; @@ -78,7 +78,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi private final TbQueueAdmin fwUpdatesAdmin; private final AtomicLong consumerCount = new AtomicLong(); - public KafkaMonolithQueueFactory(PartitionService partitionService, TbKafkaSettings kafkaSettings, + public KafkaMonolithQueueFactory(NotificationsTopicService notificationsTopicService, TbKafkaSettings kafkaSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, @@ -87,7 +87,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbKafkaConsumerStatsService consumerStatsService, TbKafkaTopicConfigs kafkaTopicConfigs) { - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.kafkaSettings = kafkaSettings; this.serviceInfoProvider = serviceInfoProvider; this.coreSettings = coreSettings; @@ -173,7 +173,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); - consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); + consumerBuilder.topic(notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("monolith-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); @@ -199,7 +199,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); - consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); + consumerBuilder.topic(notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("monolith-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java index 4f12f797f0..58dec0792a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java @@ -37,7 +37,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.kafka.TbKafkaAdmin; import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService; @@ -58,7 +58,7 @@ import java.nio.charset.StandardCharsets; @ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='tb-core'") public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbKafkaSettings kafkaSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueCoreSettings coreSettings; @@ -75,7 +75,8 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueAdmin notificationAdmin; private final TbQueueAdmin fwUpdatesAdmin; - public KafkaTbCoreQueueFactory(PartitionService partitionService, TbKafkaSettings kafkaSettings, + public KafkaTbCoreQueueFactory(NotificationsTopicService notificationsTopicService, + TbKafkaSettings kafkaSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, @@ -84,7 +85,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { TbKafkaConsumerStatsService consumerStatsService, TbQueueTransportNotificationSettings transportNotificationSettings, TbKafkaTopicConfigs kafkaTopicConfigs) { - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.kafkaSettings = kafkaSettings; this.serviceInfoProvider = serviceInfoProvider; this.coreSettings = coreSettings; @@ -169,7 +170,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); - consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); + consumerBuilder.topic(notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("tb-core-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("tb-core-notifications-node-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java index 2ed638ac35..645cbee906 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java @@ -35,7 +35,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.kafka.TbKafkaAdmin; import org.thingsboard.server.queue.kafka.TbKafkaConsumerStatsService; @@ -57,7 +57,7 @@ import java.util.concurrent.atomic.AtomicLong; @ConditionalOnExpression("'${queue.type:null}'=='kafka' && '${service.type:null}'=='tb-rule-engine'") public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbKafkaSettings kafkaSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueCoreSettings coreSettings; @@ -73,7 +73,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { private final TbQueueAdmin fwUpdatesAdmin; private final AtomicLong consumerCount = new AtomicLong(); - public KafkaTbRuleEngineQueueFactory(PartitionService partitionService, TbKafkaSettings kafkaSettings, + public KafkaTbRuleEngineQueueFactory(NotificationsTopicService notificationsTopicService, TbKafkaSettings kafkaSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, @@ -81,7 +81,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { TbKafkaConsumerStatsService consumerStatsService, TbQueueTransportNotificationSettings transportNotificationSettings, TbKafkaTopicConfigs kafkaTopicConfigs) { - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.kafkaSettings = kafkaSettings; this.serviceInfoProvider = serviceInfoProvider; this.coreSettings = coreSettings; @@ -175,7 +175,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder> consumerBuilder = TbKafkaConsumerTemplate.builder(); consumerBuilder.settings(kafkaSettings); - consumerBuilder.topic(partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); + consumerBuilder.topic(notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName()); consumerBuilder.clientId("tb-rule-engine-notifications-consumer-" + serviceInfoProvider.getServiceId()); consumerBuilder.groupId("tb-rule-engine-notifications-node-" + serviceInfoProvider.getServiceId()); consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java index 3b31531583..3388f0d0f3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java @@ -38,7 +38,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.pubsub.TbPubSubAdmin; import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate; @@ -64,7 +64,7 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; @@ -79,7 +79,7 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng TbQueueRuleEngineSettings ruleEngineSettings, TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, - PartitionService partitionService, + NotificationsTopicService notificationsTopicService, TbServiceInfoProvider serviceInfoProvider, TbPubSubSubscriptionSettings pubSubSubscriptionSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings) { @@ -88,7 +88,7 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng this.ruleEngineSettings = ruleEngineSettings; this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.serviceInfoProvider = serviceInfoProvider; this.coreAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getCoreSettings()); @@ -134,7 +134,7 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbPubSubConsumerTemplate<>(notificationAdmin, pubSubSettings, - partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @@ -147,7 +147,7 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbPubSubConsumerTemplate<>(notificationAdmin, pubSubSettings, - partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java index 3b1c3c81c9..18a6668581 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java @@ -37,7 +37,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.pubsub.TbPubSubAdmin; import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate; @@ -60,7 +60,7 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { private final TbPubSubSettings pubSubSettings; private final TbQueueCoreSettings coreSettings; private final TbQueueTransportApiSettings transportApiSettings; - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; @@ -75,7 +75,7 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { public PubSubTbCoreQueueFactory(TbPubSubSettings pubSubSettings, TbQueueCoreSettings coreSettings, TbQueueTransportApiSettings transportApiSettings, - PartitionService partitionService, + NotificationsTopicService notificationsTopicService, TbServiceInfoProvider serviceInfoProvider, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueTransportNotificationSettings transportNotificationSettings, @@ -84,7 +84,7 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { this.pubSubSettings = pubSubSettings; this.coreSettings = coreSettings; this.transportApiSettings = transportApiSettings; - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.serviceInfoProvider = serviceInfoProvider; this.jsInvokeSettings = jsInvokeSettings; this.transportNotificationSettings = transportNotificationSettings; @@ -131,7 +131,7 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbPubSubConsumerTemplate<>(notificationAdmin, pubSubSettings, - partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java index d5acbeb4c8..8a3b88846f 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java @@ -35,7 +35,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.pubsub.TbPubSubAdmin; import org.thingsboard.server.queue.pubsub.TbPubSubConsumerTemplate; @@ -58,7 +58,7 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory private final TbPubSubSettings pubSubSettings; private final TbQueueCoreSettings coreSettings; private final TbQueueRuleEngineSettings ruleEngineSettings; - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; @@ -71,7 +71,7 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory public PubSubTbRuleEngineQueueFactory(TbPubSubSettings pubSubSettings, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, - PartitionService partitionService, + NotificationsTopicService notificationsTopicService, TbServiceInfoProvider serviceInfoProvider, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueTransportNotificationSettings transportNotificationSettings, @@ -79,7 +79,7 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory this.pubSubSettings = pubSubSettings; this.coreSettings = coreSettings; this.ruleEngineSettings = ruleEngineSettings; - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.serviceInfoProvider = serviceInfoProvider; this.jsInvokeSettings = jsInvokeSettings; this.transportNotificationSettings = transportNotificationSettings; @@ -124,7 +124,7 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbPubSubConsumerTemplate<>(notificationAdmin, pubSubSettings, - partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java index 80ae1d24a9..a22f6811e3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java @@ -38,7 +38,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.rabbitmq.TbRabbitMqAdmin; import org.thingsboard.server.queue.rabbitmq.TbRabbitMqConsumerTemplate; @@ -59,7 +59,7 @@ import java.nio.charset.StandardCharsets; @ConditionalOnExpression("'${queue.type:null}'=='rabbitmq' && '${service.type:null}'=='monolith'") public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngineQueueFactory { - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbQueueCoreSettings coreSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRuleEngineSettings ruleEngineSettings; @@ -74,7 +74,7 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE private final TbQueueAdmin transportApiAdmin; private final TbQueueAdmin notificationAdmin; - public RabbitMqMonolithQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, + public RabbitMqMonolithQueueFactory(NotificationsTopicService notificationsTopicService, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueTransportApiSettings transportApiSettings, @@ -82,7 +82,7 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE TbRabbitMqSettings rabbitMqSettings, TbRabbitMqQueueArguments queueArguments, TbQueueRemoteJsInvokeSettings jsInvokeSettings) { - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.coreSettings = coreSettings; this.serviceInfoProvider = serviceInfoProvider; this.ruleEngineSettings = ruleEngineSettings; @@ -132,7 +132,7 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbRabbitMqConsumerTemplate<>(notificationAdmin, rabbitMqSettings, - partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @@ -145,7 +145,7 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbRabbitMqConsumerTemplate<>(notificationAdmin, rabbitMqSettings, - partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java index e2abcde93c..e728be6085 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java @@ -37,7 +37,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.rabbitmq.TbRabbitMqAdmin; import org.thingsboard.server.queue.rabbitmq.TbRabbitMqConsumerTemplate; @@ -61,7 +61,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueCoreSettings coreSettings; private final TbQueueTransportApiSettings transportApiSettings; - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; @@ -76,7 +76,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { TbQueueCoreSettings coreSettings, TbQueueTransportApiSettings transportApiSettings, TbQueueRuleEngineSettings ruleEngineSettings, - PartitionService partitionService, + NotificationsTopicService notificationsTopicService, TbServiceInfoProvider serviceInfoProvider, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueTransportNotificationSettings transportNotificationSettings, @@ -85,7 +85,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { this.coreSettings = coreSettings; this.transportApiSettings = transportApiSettings; this.ruleEngineSettings = ruleEngineSettings; - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.serviceInfoProvider = serviceInfoProvider; this.jsInvokeSettings = jsInvokeSettings; this.transportNotificationSettings = transportNotificationSettings; @@ -131,7 +131,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbRabbitMqConsumerTemplate<>(notificationAdmin, rabbitMqSettings, - partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java index c17ce6a058..21e152787d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java @@ -35,7 +35,7 @@ import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.rabbitmq.TbRabbitMqAdmin; import org.thingsboard.server.queue.rabbitmq.TbRabbitMqConsumerTemplate; @@ -55,7 +55,7 @@ import java.nio.charset.StandardCharsets; @ConditionalOnExpression("'${queue.type:null}'=='rabbitmq' && '${service.type:null}'=='tb-rule-engine'") public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbQueueCoreSettings coreSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRuleEngineSettings ruleEngineSettings; @@ -68,14 +68,14 @@ public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactor private final TbQueueAdmin jsExecutorAdmin; private final TbQueueAdmin notificationAdmin; - public RabbitMqTbRuleEngineQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, + public RabbitMqTbRuleEngineQueueFactory(NotificationsTopicService notificationsTopicService, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbServiceInfoProvider serviceInfoProvider, TbRabbitMqSettings rabbitMqSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueTransportNotificationSettings transportNotificationSettings, TbRabbitMqQueueArguments queueArguments) { - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.coreSettings = coreSettings; this.serviceInfoProvider = serviceInfoProvider; this.ruleEngineSettings = ruleEngineSettings; @@ -123,7 +123,7 @@ public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactor @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbRabbitMqConsumerTemplate<>(notificationAdmin, rabbitMqSettings, - partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java index 86dafd18d3..da219d9084 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java @@ -42,7 +42,7 @@ import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; @@ -58,7 +58,7 @@ import java.nio.charset.StandardCharsets; @ConditionalOnExpression("'${queue.type:null}'=='service-bus' && '${service.type:null}'=='monolith'") public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngineQueueFactory { - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbQueueCoreSettings coreSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRuleEngineSettings ruleEngineSettings; @@ -73,7 +73,7 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul private final TbQueueAdmin transportApiAdmin; private final TbQueueAdmin notificationAdmin; - public ServiceBusMonolithQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, + public ServiceBusMonolithQueueFactory(NotificationsTopicService notificationsTopicService, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueTransportApiSettings transportApiSettings, @@ -81,7 +81,7 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul TbServiceBusSettings serviceBusSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbServiceBusQueueConfigs serviceBusQueueConfigs) { - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.coreSettings = coreSettings; this.serviceInfoProvider = serviceInfoProvider; this.ruleEngineSettings = ruleEngineSettings; @@ -131,7 +131,7 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbServiceBusConsumerTemplate<>(notificationAdmin, serviceBusSettings, - partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } @@ -144,7 +144,7 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbServiceBusConsumerTemplate<>(notificationAdmin, serviceBusSettings, - partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java index 815f4c8b38..e1eb41b2b2 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java @@ -42,7 +42,7 @@ import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; @@ -61,7 +61,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueCoreSettings coreSettings; private final TbQueueTransportApiSettings transportApiSettings; - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; @@ -76,7 +76,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { TbQueueCoreSettings coreSettings, TbQueueTransportApiSettings transportApiSettings, TbQueueRuleEngineSettings ruleEngineSettings, - PartitionService partitionService, + NotificationsTopicService notificationsTopicService, TbServiceInfoProvider serviceInfoProvider, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueTransportNotificationSettings transportNotificationSettings, @@ -85,7 +85,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { this.coreSettings = coreSettings; this.transportApiSettings = transportApiSettings; this.ruleEngineSettings = ruleEngineSettings; - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.serviceInfoProvider = serviceInfoProvider; this.jsInvokeSettings = jsInvokeSettings; this.transportNotificationSettings = transportNotificationSettings; @@ -131,7 +131,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueConsumer> createToCoreNotificationsMsgConsumer() { return new TbServiceBusConsumerTemplate<>(notificationAdmin, serviceBusSettings, - partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToCoreNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java index daa6db1a6e..a6ef991f57 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java @@ -40,7 +40,7 @@ import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings; import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; @@ -55,7 +55,7 @@ import java.nio.charset.StandardCharsets; @ConditionalOnExpression("'${queue.type:null}'=='service-bus' && '${service.type:null}'=='tb-rule-engine'") public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { - private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbQueueCoreSettings coreSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRuleEngineSettings ruleEngineSettings; @@ -68,14 +68,14 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact private final TbQueueAdmin jsExecutorAdmin; private final TbQueueAdmin notificationAdmin; - public ServiceBusTbRuleEngineQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, + public ServiceBusTbRuleEngineQueueFactory(NotificationsTopicService notificationsTopicService, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, TbServiceInfoProvider serviceInfoProvider, TbServiceBusSettings serviceBusSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbQueueTransportNotificationSettings transportNotificationSettings, TbServiceBusQueueConfigs serviceBusQueueConfigs) { - this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.coreSettings = coreSettings; this.serviceInfoProvider = serviceInfoProvider; this.ruleEngineSettings = ruleEngineSettings; @@ -123,7 +123,7 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact @Override public TbQueueConsumer> createToRuleEngineNotificationsMsgConsumer() { return new TbServiceBusConsumerTemplate<>(notificationAdmin, serviceBusSettings, - partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), + notificationsTopicService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceInfoProvider.getServiceId()).getFullTopicName(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToRuleEngineNotificationMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 1ca7d80394..77738a1e9d 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -85,6 +85,7 @@ import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.AsyncCallbackTemplate; 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.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; @@ -157,6 +158,7 @@ public class DefaultTransportService implements TransportService { private final TbTransportQueueFactory queueProvider; private final TbQueueProducerProvider producerProvider; private final PartitionService partitionService; + private final NotificationsTopicService notificationsTopicService; private final TbServiceInfoProvider serviceInfoProvider; private final StatsFactory statsFactory; private final TransportDeviceProfileCache deviceProfileCache; @@ -190,6 +192,7 @@ public class DefaultTransportService implements TransportService { TbTransportQueueFactory queueProvider, TbQueueProducerProvider producerProvider, PartitionService partitionService, + NotificationsTopicService notificationsTopicService, StatsFactory statsFactory, TransportDeviceProfileCache deviceProfileCache, TransportTenantProfileCache tenantProfileCache, @@ -200,6 +203,7 @@ public class DefaultTransportService implements TransportService { this.queueProvider = queueProvider; this.producerProvider = producerProvider; this.partitionService = partitionService; + this.notificationsTopicService = notificationsTopicService; this.statsFactory = statsFactory; this.deviceProfileCache = deviceProfileCache; this.tenantProfileCache = tenantProfileCache; @@ -224,7 +228,7 @@ public class DefaultTransportService implements TransportService { ruleEngineMsgProducer = producerProvider.getRuleEngineMsgProducer(); tbCoreMsgProducer = producerProvider.getTbCoreMsgProducer(); transportNotificationsConsumer = queueProvider.createTransportNotificationsConsumer(); - TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_TRANSPORT, serviceInfoProvider.getServiceId()); + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_TRANSPORT, serviceInfoProvider.getServiceId()); transportNotificationsConsumer.subscribe(Collections.singleton(tpi)); transportApiRequestTemplate.init(); mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("transport-consumer")); diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/BaseQueueService.java b/dao/src/main/java/org/thingsboard/server/dao/queue/BaseQueueService.java index 75978ea371..7049d8ae1c 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/queue/BaseQueueService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/queue/BaseQueueService.java @@ -39,6 +39,7 @@ import org.thingsboard.server.dao.service.PaginatedRemover; import org.thingsboard.server.dao.service.Validator; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; import org.thingsboard.server.queue.TbQueueAdmin; +import org.thingsboard.server.queue.TbQueueClusterService; import java.util.List; @@ -56,8 +57,8 @@ public class BaseQueueService extends AbstractEntityService implements QueueServ @Autowired(required = false) private TbQueueAdmin tbQueueAdmin; -// @Autowired(required = false) -// private TbQueueClusterService queueClusterService; + @Autowired(required = false) + private TbQueueClusterService queueClusterService; // @Autowired // private QueueStatsService queueStatsService; @@ -99,7 +100,7 @@ public class BaseQueueService extends AbstractEntityService implements QueueServ //TODO: remove if partitions can't be deleted. if (currentPartitions != oldPartitions && tbQueueAdmin != null) { -// queueClusterService.onQueueDelete(queue, null); + queueClusterService.onQueueDelete(queue, null); if (currentPartitions > oldPartitions) { log.info("Added [{}] new partitions to [{}] queue", currentPartitions - oldPartitions, queue.getName()); for (int i = oldPartitions; i < currentPartitions; i++) { @@ -121,9 +122,9 @@ public class BaseQueueService extends AbstractEntityService implements QueueServ public void deleteQueue(TenantId tenantId, QueueId queueId) { log.trace("Executing deleteQueue, queueId: [{}]", queueId); Queue queue = findQueueById(tenantId, queueId); -// if (queueClusterService != null) { -// queueClusterService.onQueueDelete(queue, null); -// } + if (queueClusterService != null) { + queueClusterService.onQueueDelete(queue, null); + } // queueStatsService.deleteQueueStatsByQueueId(tenantId, queueId); boolean result = queueDao.removeById(tenantId, queueId.getId()); if (result && tbQueueAdmin != null) {