diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java index 9c0aa101c4..9038f004d3 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java @@ -64,7 +64,6 @@ import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.edge.EdgeContextComponent; -import org.thingsboard.server.service.state.DefaultDeviceStateService; import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import java.io.IOException; @@ -84,6 +83,10 @@ import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; +import static org.thingsboard.server.service.state.DefaultDeviceStateService.ACTIVITY_STATE; +import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_CONNECT_TIME; +import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_DISCONNECT_TIME; + @Service @Slf4j @ConditionalOnProperty(prefix = "edges", value = "enabled", havingValue = "true") @@ -95,7 +98,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i private final Map sessionNewEvents = new HashMap<>(); private final ConcurrentMap> sessionEdgeEventChecks = new ConcurrentHashMap<>(); private final ConcurrentMap> localSyncEdgeRequests = new ConcurrentHashMap<>(); - private final ConcurrentMap edgeEventsProcessed = new ConcurrentHashMap<>(); + private final ConcurrentMap edgeEventsMigrationProcessed = new ConcurrentHashMap<>(); @Value("${edges.rpc.port}") private int rpcPort; @@ -275,7 +278,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i if (session != null && session.isConnected()) { log.info("[{}] Closing and removing session for edge [{}]", tenantId, edgeId); session.destroy(); - session.deleteTopic(edgeId); + session.cleanUp(); session.close(); sessions.remove(edgeId); final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); @@ -334,13 +337,13 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } finally { newEventLock.unlock(); } - save(tenantId, edgeId, DefaultDeviceStateService.ACTIVITY_STATE, true); + save(tenantId, edgeId, ACTIVITY_STATE, true); long lastConnectTs = System.currentTimeMillis(); - save(tenantId, edgeId, DefaultDeviceStateService.LAST_CONNECT_TIME, lastConnectTs); + save(tenantId, edgeId, LAST_CONNECT_TIME, lastConnectTs); edgeIdServiceIdCache.put(edgeId, serviceInfoProvider.getServiceId()); pushRuleEngineMessage(tenantId, edge, lastConnectTs, TbMsgType.CONNECT_EVENT); cancelScheduleEdgeEventsCheck(edgeId); - edgeEventsProcessed.putIfAbsent(edgeId, Boolean.FALSE); + edgeEventsMigrationProcessed.putIfAbsent(edgeId, Boolean.FALSE); scheduleEdgeEventsCheck(edgeGrpcSession); } @@ -447,14 +450,14 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } private void processEdgeEventMigrationIfNeeded(EdgeGrpcSession session, EdgeId edgeId) throws Exception { - boolean isMigrationProcessed = edgeEventsProcessed.getOrDefault(edgeId, Boolean.FALSE); + boolean isMigrationProcessed = edgeEventsMigrationProcessed.getOrDefault(edgeId, Boolean.FALSE); if (!isMigrationProcessed) { Boolean eventsExist = session.migrateEdgeEvents().get(); if (Boolean.TRUE.equals(eventsExist)) { sessionNewEvents.put(edgeId, true); scheduleEdgeEventsCheck(session); } else if (Boolean.FALSE.equals(eventsExist)) { - edgeEventsProcessed.put(edgeId, true); + edgeEventsMigrationProcessed.put(edgeId, true); } else { scheduleEdgeEventsCheck(session); } @@ -487,9 +490,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i } toRemove.destroy(); TenantId tenantId = toRemove.getEdge().getTenantId(); - save(tenantId, edgeId, DefaultDeviceStateService.ACTIVITY_STATE, false); + save(tenantId, edgeId, ACTIVITY_STATE, false); long lastDisconnectTs = System.currentTimeMillis(); - save(tenantId, edgeId, DefaultDeviceStateService.LAST_DISCONNECT_TIME, lastDisconnectTs); + save(tenantId, edgeId, LAST_DISCONNECT_TIME, lastDisconnectTs); pushRuleEngineMessage(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); cancelScheduleEdgeEventsCheck(edgeId); } else { @@ -554,11 +557,11 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i ObjectNode edgeState = JacksonUtil.newObjectNode(); boolean isConnected = TbMsgType.CONNECT_EVENT.equals(msgType); if (isConnected) { - edgeState.put(DefaultDeviceStateService.ACTIVITY_STATE, true); - edgeState.put(DefaultDeviceStateService.LAST_CONNECT_TIME, ts); + edgeState.put(ACTIVITY_STATE, true); + edgeState.put(LAST_CONNECT_TIME, ts); } else { - edgeState.put(DefaultDeviceStateService.ACTIVITY_STATE, false); - edgeState.put(DefaultDeviceStateService.LAST_DISCONNECT_TIME, ts); + edgeState.put(ACTIVITY_STATE, false); + edgeState.put(LAST_DISCONNECT_TIME, ts); } ctx.getNotificationRuleProcessor().process(EdgeConnectionTrigger.builder() .tenantId(tenantId) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 4ad9c04353..f0bcbf2768 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -923,13 +923,9 @@ public abstract class EdgeGrpcSession implements Closeable { return Futures.allAsList(result); } - protected void destroy() { - // used for KafkaEdgeGrpcSession only - } + protected void destroy() {} - protected void deleteTopic(EdgeId edgeId) { - // used for KafkaEdgeGrpcSession only - } + protected void cleanUp() {} @Override public void close() { diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeEventService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeEventService.java index 9b93238400..454645ca65 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeEventService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeEventService.java @@ -20,7 +20,6 @@ import com.google.common.util.concurrent.ListenableFuture; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; -import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; @@ -39,9 +38,7 @@ import java.util.UUID; @ConditionalOnExpression("'${queue.type:null}'=='kafka'") public class KafkaEdgeEventService extends BaseEdgeEventService { - @Lazy private final TopicService topicService; - @Lazy private final TbQueueProducerProvider producerProvider; @Override diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java index 1b2faebbee..5754f7c8f3 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java @@ -137,8 +137,8 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession { } @Override - public void deleteTopic(EdgeId edgeId) { - String topic = topicService.getEdgeEventNotificationsTopic(tenantId, edgeId).getTopic(); + public void cleanUp() { + String topic = topicService.getEdgeEventNotificationsTopic(tenantId, edge.getId()).getTopic(); TbKafkaAdmin kafkaAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeEventConfigs()); kafkaAdmin.deleteTopic(topic); } diff --git a/application/src/main/java/org/thingsboard/server/service/ttl/KafkaEdgeTopicsCleanUpService.java b/application/src/main/java/org/thingsboard/server/service/ttl/KafkaEdgeTopicsCleanUpService.java index 29e3b971d4..03b46921eb 100644 --- a/application/src/main/java/org/thingsboard/server/service/ttl/KafkaEdgeTopicsCleanUpService.java +++ b/application/src/main/java/org/thingsboard/server/service/ttl/KafkaEdgeTopicsCleanUpService.java @@ -36,17 +36,18 @@ import org.thingsboard.server.queue.kafka.TbKafkaAdmin; import org.thingsboard.server.queue.kafka.TbKafkaSettings; import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; import org.thingsboard.server.queue.util.TbCoreComponent; -import org.thingsboard.server.service.state.DefaultDeviceStateService; import java.time.Instant; import java.util.Date; import java.util.concurrent.TimeUnit; +import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_CONNECT_TIME; + @Slf4j @Service @TbCoreComponent @RequiredArgsConstructor -@ConditionalOnExpression("'${queue.type:null}'=='kafka' && ${edges.enabled:true}") +@ConditionalOnExpression("'${queue.type:null}'=='kafka' && ${edges.enabled:true} && ${sql.ttl.edge_events.edge_events_ttl:0} > 0") public class KafkaEdgeTopicsCleanUpService { private final EdgeService edgeService; @@ -81,11 +82,11 @@ public class KafkaEdgeTopicsCleanUpService { PageDataIterable edgeIds = new PageDataIterable<>(link -> edgeService.findEdgeIdsByTenantId(tenantId, link), 1024); long currentTimeMillis = System.currentTimeMillis(); - long ttlMillis = TimeUnit.SECONDS.toChronoUnit().getDuration().multipliedBy(ttlSeconds).toMillis(); + long ttlMillis = TimeUnit.SECONDS.toMillis(ttlSeconds); for (EdgeId edgeId : edgeIds) { TbKafkaAdmin kafkaAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeEventConfigs()); - attributesService.find(tenantId, edgeId, AttributeScope.SERVER_SCOPE, DefaultDeviceStateService.LAST_CONNECT_TIME).get() + attributesService.find(tenantId, edgeId, AttributeScope.SERVER_SCOPE, LAST_CONNECT_TIME).get() .flatMap(AttributeKvEntry::getLongValue) .filter(lastConnectTime -> isTopicExpired(lastConnectTime, ttlMillis, currentTimeMillis)) .ifPresent(lastConnectTime -> { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java index c209c574e2..2d750ae38b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java @@ -28,12 +28,14 @@ import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.util.PropertyUtils; import java.util.Collections; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.stream.Collectors; /** * Created by ashvayka on 24.09.18. @@ -178,17 +180,24 @@ public class TbKafkaAdmin implements TbQueueAdmin { public boolean isTopicEmpty(String topic) { try { TopicDescription topicDescription = settings.getAdminClient().describeTopics(Collections.singletonList(topic)).topicNameValues().get(topic).get(); - TopicPartition topicPartition = new TopicPartition(topic, topicDescription.partitions().get(0).partition()); + List partitions = topicDescription.partitions().stream().map(partitionInfo -> new TopicPartition(topic, partitionInfo.partition())).toList(); - Map beginningOffsets = - settings.getAdminClient().listOffsets(Collections.singletonMap(topicPartition, OffsetSpec.earliest())).all().get(); - Map endOffsets = - settings.getAdminClient().listOffsets(Collections.singletonMap(topicPartition, OffsetSpec.latest())).all().get(); + Map beginningOffsets = settings.getAdminClient().listOffsets(partitions.stream() + .collect(Collectors.toMap(partition -> partition, partition -> OffsetSpec.earliest()))).all().get(); - long beginningOffset = beginningOffsets.get(topicPartition).offset(); - long endOffset = endOffsets.get(topicPartition).offset(); + Map endOffsets = settings.getAdminClient().listOffsets(partitions.stream() + .collect(Collectors.toMap(partition -> partition, partition -> OffsetSpec.latest()))).all().get(); - return beginningOffset == endOffset; + for (TopicPartition partition : partitions) { + long beginningOffset = beginningOffsets.get(partition).offset(); + long endOffset = endOffsets.get(partition).offset(); + + if (beginningOffset != endOffset) { + log.debug("Partition [{}] of topic [{}] is not empty. Returning false.", partition.partition(), topic); + return false; + } + } + return true; } catch (InterruptedException | ExecutionException e) { log.error("Failed to check if topic [{}] is empty.", topic, e); return false; 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 d563307390..310cc1d1c5 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 @@ -20,8 +20,6 @@ import jakarta.annotation.PreDestroy; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.id.EdgeId; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest; @@ -284,11 +282,6 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueConsumer> createEdgeEventMsgConsumer(TenantId tenantId, EdgeId edgeId) { - return null; - } - @Override public TbQueueProducer> createEdgeEventMsgProducer() { return null; 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 056fbb2508..515c1bba44 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 @@ -20,13 +20,10 @@ import jakarta.annotation.PreDestroy; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.id.EdgeId; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; -import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; @@ -264,16 +261,6 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueConsumer> createEdgeEventMsgConsumer(TenantId tenantId, EdgeId edgeId) { - return null; - } - - @Override - public TbQueueProducer> createEdgeEventMsgProducer() { - return null; - } - @PreDestroy private void destroy() { if (coreAdmin != null) { 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 04ca87eabb..a93aba2764 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 @@ -25,7 +25,6 @@ import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; -import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; @@ -136,11 +135,6 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory return new TbAwsSqsProducerTemplate<>(notificationAdmin, sqsSettings, topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueProducer> createEdgeEventMsgProducer() { - return null; - } - @Override public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbAwsSqsConsumerTemplate<>(ruleEngineAdmin, sqsSettings, topicService.buildTopicName(configuration.getTopic()), 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 44b2469da8..d70cad159b 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 @@ -19,8 +19,6 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.id.EdgeId; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; @@ -206,11 +204,6 @@ public class InMemoryMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE return new InMemoryTbQueueProducer<>(storage, topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueConsumer> createEdgeEventMsgConsumer(TenantId tenantId, EdgeId edgeId) { - return null; - } - @Override public TbQueueProducer> createEdgeEventMsgProducer() { return null; 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 74f95743b6..87a1a69c2e 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 @@ -198,7 +198,6 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-edge-event-" + serviceInfoProvider.getServiceId()); - requestBuilder.defaultTopic(topicService.buildTopicName("edge-events")); requestBuilder.admin(edgeEventAdmin); return requestBuilder.build(); } 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 f0bb294e20..05bef819b7 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 @@ -20,8 +20,6 @@ import jakarta.annotation.PreDestroy; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.id.EdgeId; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest; @@ -285,11 +283,6 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueConsumer> createEdgeEventMsgConsumer(TenantId tenantId, EdgeId edgeId) { - return null; - } - @Override public TbQueueProducer> createEdgeEventMsgProducer() { return null; 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 f885f1070e..919d895a97 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 @@ -20,13 +20,10 @@ import jakarta.annotation.PreDestroy; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.id.EdgeId; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; -import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; @@ -257,16 +254,6 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueConsumer> createEdgeEventMsgConsumer(TenantId tenantId, EdgeId edgeId) { - return null; - } - - @Override - public TbQueueProducer> createEdgeEventMsgProducer() { - return null; - } - @PreDestroy private void destroy() { if (coreAdmin != null) { 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 a444276d19..3c59923def 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 @@ -20,8 +20,6 @@ import jakarta.annotation.PreDestroy; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.id.EdgeId; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest; @@ -283,11 +281,6 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueConsumer> createEdgeEventMsgConsumer(TenantId tenantId, EdgeId edgeId) { - return null; - } - @Override public TbQueueProducer> createEdgeEventMsgProducer() { return null; 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 053d682580..16f9838384 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 @@ -20,13 +20,10 @@ import jakarta.annotation.PreDestroy; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.id.EdgeId; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; -import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; @@ -173,16 +170,6 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueConsumer> createEdgeEventMsgConsumer(TenantId tenantId, EdgeId edgeId) { - return null; - } - - @Override - public TbQueueProducer> createEdgeEventMsgProducer() { - return null; - } - @Override public TbQueueConsumer> createTransportApiRequestConsumer() { return new TbRabbitMqConsumerTemplate<>(transportApiAdmin, rabbitMqSettings, topicService.buildTopicName(transportApiSettings.getRequestsTopic()), 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 2f6a5c9eb6..a7af729e1f 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 @@ -25,7 +25,6 @@ import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; -import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; @@ -134,11 +133,6 @@ public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactor return new TbRabbitMqProducerTemplate<>(notificationAdmin, rabbitMqSettings, topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueProducer> createEdgeEventMsgProducer() { - return null; - } - @Override public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbRabbitMqConsumerTemplate<>(ruleEngineAdmin, rabbitMqSettings, topicService.buildTopicName(configuration.getTopic()), 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 2772dd4ff9..932bbd21e5 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 @@ -20,8 +20,6 @@ import jakarta.annotation.PreDestroy; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.id.EdgeId; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; @@ -282,11 +280,6 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueConsumer> createEdgeEventMsgConsumer(TenantId tenantId, EdgeId edgeId) { - return null; - } - @Override public TbQueueProducer> createEdgeEventMsgProducer() { return null; 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 640019a8a7..71a7efe50b 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 @@ -20,13 +20,10 @@ import jakarta.annotation.PreDestroy; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.id.EdgeId; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; -import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; @@ -257,16 +254,6 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueConsumer> createEdgeEventMsgConsumer(TenantId tenantId, EdgeId edgeId) { - return null; - } - - @Override - public TbQueueProducer> createEdgeEventMsgProducer() { - return null; - } - @PreDestroy private void destroy() { if (coreAdmin != null) { 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 916c778722..54661f695c 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 @@ -25,7 +25,6 @@ import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; -import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; @@ -134,11 +133,6 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact return new TbServiceBusProducerTemplate<>(notificationAdmin, serviceBusSettings, topicService.getEdgeNotificationsTopic(serviceInfoProvider.getServiceId()).getFullTopicName()); } - @Override - public TbQueueProducer> createEdgeEventMsgProducer() { - return null; - } - @Override public TbQueueConsumer> createToRuleEngineMsgConsumer(Queue configuration) { return new TbServiceBusConsumerTemplate<>(ruleEngineAdmin, serviceBusSettings, topicService.buildTopicName(configuration.getTopic()), diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java index 2fa95443ac..c4002f4d3e 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbCoreQueueFactory.java @@ -151,8 +151,12 @@ public interface TbCoreQueueFactory extends TbUsageStatsClientQueueFactory, Hous TbQueueProducer> createEdgeNotificationsMsgProducer(); - TbQueueConsumer> createEdgeEventMsgConsumer(TenantId tenantId, EdgeId edgeId); + default TbQueueConsumer> createEdgeEventMsgConsumer(TenantId tenantId, EdgeId edgeId) { + return null; + } - TbQueueProducer> createEdgeEventMsgProducer(); + default TbQueueProducer> createEdgeEventMsgProducer() { + return null; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java index a25b590b9a..c406aeb311 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbRuleEngineQueueFactory.java @@ -77,7 +77,9 @@ public interface TbRuleEngineQueueFactory extends TbUsageStatsClientQueueFactory TbQueueProducer> createEdgeNotificationsMsgProducer(); - TbQueueProducer> createEdgeEventMsgProducer(); + default TbQueueProducer> createEdgeEventMsgProducer() { + return null; + } /** * Used to consume messages about firmware update notifications to TB Core Service