From 2a815058e7279ee4aefbcaa2c5ba3adff00b7532 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Tue, 17 Mar 2020 12:38:28 +0200 Subject: [PATCH] Improvements and bugfix to QueueProviders --- .../server/actors/ActorSystemContext.java | 23 ++++---- .../actors/ruleChain/DefaultTbContext.java | 4 +- .../RuleChainActorMessageProcessor.java | 2 +- .../queue/DefaultTbCoreConsumerService.java | 11 +++- .../DefaultTbRuleEngineConsumerService.java | 16 ++++-- .../service/queue/TbCoreConsumerService.java | 4 +- .../queue/TbRuleEngineConsumerService.java | 5 +- .../queue/TbRuleEngineConsumerStats.java | 2 +- .../DefaultTbCoreToTransportService.java | 4 +- .../src/main/resources/thingsboard.yml | 21 +++----- ...=> ConsistentHashParitionServiceTest.java} | 13 ++--- .../thingsboard/server/TbQueueConsumer.java | 2 + .../thingsboard/server/TbQueueProducer.java | 2 +- .../server/TbQueueRequestTemplate.java | 4 ++ .../server/TbQueueRuleEngineSettings.java | 31 +++++++++++ .../server/TbQueueTransportApiSettings.java | 7 +++ .../TbQueueTransportNotificationSettings.java | 32 +++++++++++ .../common/DefaultTbQueueRequestTemplate.java | 4 +- .../discovery/ConsistentHashCircle.java | 2 +- .../ConsistentHashPartitionService.java | 54 ++++++++----------- .../DefaultTbServiceInfoProvider.java | 2 +- .../discovery/DummyDiscoveryService.java | 15 ++++++ .../discovery/PartitionChangeEvent.java | 6 ++- .../server/discovery/PartitionService.java | 15 ++++++ .../server/discovery/ServiceKey.java | 47 ++++++++++++++++ .../ZkPartitionDiscoveryService.java | 2 +- .../server/kafka/TBKafkaConsumerTemplate.java | 40 +++++++++----- .../server/kafka/TBKafkaProducerTemplate.java | 2 +- .../server/memory/InMemoryStorage.java | 39 +++++++++++--- .../memory/InMemoryTbQueueConsumer.java | 8 ++- .../memory/InMemoryTbQueueProducer.java | 11 ++-- .../InMemoryMonolithQueueProvider.java | 36 +++++++------ .../InMemoryTransportQueueProvider.java | 28 +++++++--- .../provider/KafkaMonolithQueueProvider.java | 4 +- .../provider/KafkaTbCoreQueueProvider.java | 4 +- .../KafkaTbRuleEngineQueueProvider.java | 2 +- .../provider/KafkaTransportQueueProvider.java | 2 +- .../server/provider/TbCoreQueueProvider.java | 2 + .../provider/TbRuleEngineQueueProvider.java | 2 +- .../service/DefaultTransportService.java | 12 +++-- 40 files changed, 384 insertions(+), 138 deletions(-) rename application/src/test/java/org/thingsboard/server/service/cluster/routing/{ConsistentClusterRoutingServiceTest.java => ConsistentHashParitionServiceTest.java} (92%) create mode 100644 common/queue/src/main/java/org/thingsboard/server/TbQueueRuleEngineSettings.java create mode 100644 common/queue/src/main/java/org/thingsboard/server/TbQueueTransportNotificationSettings.java create mode 100644 common/queue/src/main/java/org/thingsboard/server/discovery/ServiceKey.java diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index e5a0e6a25a..3db5c82ee2 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -35,6 +35,7 @@ import org.springframework.context.annotation.Lazy; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; +import org.springframework.stereotype.Service; import org.thingsboard.rule.engine.api.MailService; import org.thingsboard.rule.engine.api.RuleChainTransactionService; import org.thingsboard.server.actors.service.ActorService; @@ -64,6 +65,7 @@ import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.user.UserService; +import org.thingsboard.server.discovery.TbServiceInfoProvider; import org.thingsboard.server.kafka.TbNodeIdProvider; import org.thingsboard.server.service.component.ComponentDiscoveryService; import org.thingsboard.server.service.encoding.DataDecodingEncodingService; @@ -102,6 +104,11 @@ public class ActorSystemContext { return debugPerTenantLimits; } + @Autowired + @Getter + @Setter + private TbServiceInfoProvider serviceInfoProvider; + @Getter @Setter private ActorService actorService; @@ -233,10 +240,6 @@ public class ActorSystemContext { @Getter private RuleChainTransactionService ruleChainTransactionService; - @Value("${cluster.partition_id}") - @Getter - private long queuePartitionId; - @Value("${actors.session.max_concurrent_sessions_per_device:1}") @Getter private long maxConcurrentSessionsPerDevice; @@ -359,7 +362,7 @@ public class ActorSystemContext { event.setEntityId(entityId); event.setType(DataConstants.ERROR); //TODO 2.5 -// event.setBody(toBodyJson(discoveryService.getCurrentServer().getServerAddress(), method, toString(e))); + event.setBody(toBodyJson(serviceInfoProvider.getServiceInfo().getServiceId(), method, toString(e))); persistEvent(event); } @@ -369,7 +372,7 @@ public class ActorSystemContext { event.setEntityId(entityId); event.setType(DataConstants.LC_EVENT); //TODO 2.5 -// event.setBody(toBodyJson(discoveryService.getCurrentServer().getServerAddress(), lcEvent, Optional.ofNullable(e))); + event.setBody(toBodyJson(serviceInfoProvider.getServiceInfo().getServiceId(), lcEvent, Optional.ofNullable(e))); persistEvent(event); } @@ -383,8 +386,8 @@ public class ActorSystemContext { return sw.toString(); } - private JsonNode toBodyJson(ServerAddress server, ComponentLifecycleEvent event, Optional e) { - ObjectNode node = mapper.createObjectNode().put("server", server.toString()).put("event", event.name()); + private JsonNode toBodyJson(String serviceId, ComponentLifecycleEvent event, Optional e) { + ObjectNode node = mapper.createObjectNode().put("server", serviceId).put("event", event.name()); if (e.isPresent()) { node = node.put("success", false); node = node.put("error", toString(e.get())); @@ -394,8 +397,8 @@ public class ActorSystemContext { return node; } - private JsonNode toBodyJson(ServerAddress server, String method, String body) { - return mapper.createObjectNode().put("server", server.toString()).put("method", method).put("error", body); + private JsonNode toBodyJson(String serviceId, String method, String body) { + return mapper.createObjectNode().put("server", serviceId).put("method", method).put("error", body); } diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index f5ea5c399f..3e982b520e 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -146,12 +146,12 @@ class DefaultTbContext implements TbContext { @Override public TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data) { - return new TbMsg(UUIDs.timeBased(), type, originator, metaData.copy(), data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId(), mainCtx.getQueuePartitionId()); + return new TbMsg(UUIDs.timeBased(), type, originator, metaData.copy(), data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId(), 0); } @Override public TbMsg transformMsg(TbMsg origMsg, String type, EntityId originator, TbMsgMetaData metaData, String data) { - return new TbMsg(origMsg.getId(), type, originator, metaData.copy(), origMsg.getDataType(), data, origMsg.getTransactionData(), origMsg.getRuleChainId(), origMsg.getRuleNodeId(), mainCtx.getQueuePartitionId()); + return new TbMsg(origMsg.getId(), type, originator, metaData.copy(), origMsg.getDataType(), data, origMsg.getTransactionData(), origMsg.getRuleChainId(), origMsg.getRuleNodeId(), 0); } @Override diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java index 095af7f62d..fe95c448a0 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java @@ -331,6 +331,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index ef2b44bb6c..6f8d9dc9ac 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -15,7 +15,6 @@ */ package org.thingsboard.server.service.queue; -import akka.actor.ActorRef; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; @@ -27,10 +26,10 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.TbQueueConsumer; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.common.TbProtoQueueMsg; +import org.thingsboard.server.discovery.PartitionChangeEvent; +import org.thingsboard.server.discovery.ServiceType; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.provider.TbCoreQueueProvider; import org.thingsboard.server.provider.TbRuleEngineQueueProvider; -import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; @@ -70,10 +69,17 @@ public class DefaultTbRuleEngineConsumerService implements TbRuleEngineConsumerS @PostConstruct public void init() { - this.consumer.subscribe(); this.mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-core-consumer")); } + @Override + public void onApplicationEvent(PartitionChangeEvent partitionChangeEvent) { + if (partitionChangeEvent.getServiceKey().getServiceType() == ServiceType.TB_RULE_ENGINE) { + log.info("Subscribing to partitions: {}", partitionChangeEvent.getPartitions()); + this.consumer.subscribe(partitionChangeEvent.getPartitions()); + } + } + @EventListener(ApplicationReadyEvent.class) public void onApplicationEvent(ApplicationReadyEvent event) { mainConsumerExecutor.execute(() -> { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerService.java index 1e1b352645..8a39b5df4a 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerService.java @@ -15,10 +15,12 @@ */ package org.thingsboard.server.service.queue; +import org.springframework.context.ApplicationListener; +import org.thingsboard.server.discovery.PartitionChangeEvent; import org.thingsboard.server.gen.transport.TransportProtos; import java.util.function.Consumer; -public interface TbCoreConsumerService { +public interface TbCoreConsumerService extends ApplicationListener { } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerService.java index 671cd72262..968fbb2274 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerService.java @@ -15,6 +15,9 @@ */ package org.thingsboard.server.service.queue; -public interface TbRuleEngineConsumerService { +import org.springframework.context.ApplicationListener; +import org.thingsboard.server.discovery.PartitionChangeEvent; + +public interface TbRuleEngineConsumerService extends ApplicationListener { } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java index a5e8ba94af..27792c5a61 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java @@ -5,7 +5,7 @@ * 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 + * 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, 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 f7afd105dc..a6e1a37eca 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 @@ -5,7 +5,7 @@ * 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 + * 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, @@ -40,7 +40,7 @@ public class DefaultTbCoreToTransportService implements TbCoreToTransportService private final TbQueueProducer> tbTransportProducer; - @Value("${queue.notifications.topic}") + @Value("${queue.transport.notifications_topic}") private String notificationsTopic; public DefaultTbCoreToTransportService(TbCoreQueueProvider tbCoreQueueProvider) { diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 57d837de31..fc49716d25 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -70,16 +70,7 @@ zk: # Name of the directory in zookeeper 'filesystem' zk_dir: "${ZOOKEEPER_NODES_DIR:/thingsboard}" -# Clustering properties related to consistent-hashing. See architecture docs for more details. cluster: - # Unique id for this node (autogenerated if empty) - node_id: "${CLUSTER_NODE_ID:}" - # Name of hash function used for consistent hash ring. - hash_function_name: "${CLUSTER_HASH_FUNCTION_NAME:murmur3_128}" - # Amount of virtual nodes in consistent hash ring. - vitrual_nodes_size: "${CLUSTER_VIRTUAL_NODES_SIZE:16}" - # Queue partition id for current node - partition_id: "${QUEUE_PARTITION_ID:0}" stats: enabled: "${TB_CLUSTER_STATS_ENABLED:false}" print_interval_ms: "${TB_CLUSTER_STATS_PRINT_INTERVAL_MS:10000}" @@ -536,6 +527,9 @@ queue: batch.size: "${TB_KAFKA_BATCH_SIZE:16384}" linger.ms: "${TB_KAFKA_LINGER_MS:1}" buffer.memory: "${TB_BUFFER_MEMORY:33554432}" + partitions: + hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" + virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}" transport_api: requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb.transport.api.requests}" responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb.transport.api.responses}" @@ -547,7 +541,7 @@ queue: core: topic: "${TB_QUEUE_CORE_TOPIC:tb.core}" poll_interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" - partitions: "${TB_QUEUE_CORE_PARTITIONS:100}" + partitions: "${TB_QUEUE_CORE_PARTITIONS:10}" pack_processing_timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:60000}" stats: enabled: "${TB_QUEUE_CORE_STATS_ENABLED:false}" @@ -555,13 +549,14 @@ queue: rule_engine: topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb.rule-engine}" poll_interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" - partitions: "${TB_QUEUE_RULE_ENGINE_PARTITIONS:100}" + partitions: "${TB_QUEUE_RULE_ENGINE_PARTITIONS:10}" pack_processing_timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}" stats: enabled: "${TB_QUEUE_RULE_ENGINE_STATS_ENABLED:false}" print_interval_ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}" - notifications: - topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}" + transport: + notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}" + poll_interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" service: type: "${TB_SERVICE_TYPE:monolith}" # monolith or tb-core or tb-rule-engine or tb-transport diff --git a/application/src/test/java/org/thingsboard/server/service/cluster/routing/ConsistentClusterRoutingServiceTest.java b/application/src/test/java/org/thingsboard/server/service/cluster/routing/ConsistentHashParitionServiceTest.java similarity index 92% rename from application/src/test/java/org/thingsboard/server/service/cluster/routing/ConsistentClusterRoutingServiceTest.java rename to application/src/test/java/org/thingsboard/server/service/cluster/routing/ConsistentHashParitionServiceTest.java index 0c34bbf9d6..9442202688 100644 --- a/application/src/test/java/org/thingsboard/server/service/cluster/routing/ConsistentClusterRoutingServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/cluster/routing/ConsistentHashParitionServiceTest.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -21,11 +21,10 @@ import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.runners.MockitoJUnitRunner; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.test.util.ReflectionTestUtils; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.msg.cluster.ServerAddress; -import org.thingsboard.server.common.msg.cluster.ServerType; import org.thingsboard.server.discovery.ConsistentHashPartitionService; import org.thingsboard.server.discovery.ServiceType; import org.thingsboard.server.discovery.TbServiceInfoProvider; @@ -44,12 +43,13 @@ import static org.mockito.Mockito.mock; @Slf4j @RunWith(MockitoJUnitRunner.class) -public class ConsistentClusterRoutingServiceTest { +public class ConsistentHashParitionServiceTest { public static final int ITERATIONS = 1000000; private ConsistentHashPartitionService clusterRoutingService; private TbServiceInfoProvider discoveryService; + private ApplicationEventPublisher applicationEventPublisher; private String hashFunctionName = "murmur3_128"; private Integer virtualNodesSize = 16; @@ -58,7 +58,8 @@ public class ConsistentClusterRoutingServiceTest { @Before public void setup() throws Exception { discoveryService = mock(TbServiceInfoProvider.class); - clusterRoutingService = new ConsistentHashPartitionService(discoveryService); + applicationEventPublisher = mock(ApplicationEventPublisher.class); + clusterRoutingService = new ConsistentHashPartitionService(discoveryService, applicationEventPublisher); ReflectionTestUtils.setField(clusterRoutingService, "coreTopic", "tb.core"); ReflectionTestUtils.setField(clusterRoutingService, "corePartitions", 3); ReflectionTestUtils.setField(clusterRoutingService, "ruleEngineTopic", "tb.rule-engine"); @@ -103,7 +104,7 @@ public class ConsistentClusterRoutingServiceTest { List> data = map.entrySet().stream().sorted(Comparator.comparingInt(Map.Entry::getValue)).collect(Collectors.toList()); long end = System.currentTimeMillis(); double diff = (data.get(data.size() - 1).getValue() - data.get(0).getValue()); - System.out.println("Size: " + virtualNodesSize + " Time: " + (end - start) + " Diff: " + diff + "(" + String.format("%f", (diff/ITERATIONS) * 100.0) + "%)"); + System.out.println("Size: " + virtualNodesSize + " Time: " + (end - start) + " Diff: " + diff + "(" + String.format("%f", (diff / ITERATIONS) * 100.0) + "%)"); for (Map.Entry entry : data) { System.out.println(entry.getKey() + ": " + entry.getValue()); diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueConsumer.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueConsumer.java index ddf9d7d9b3..ac4d7611f6 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueConsumer.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueConsumer.java @@ -23,6 +23,8 @@ public interface TbQueueConsumer { void subscribe(); + void subscribe(List partitions); + void unsubscribe(); List poll(long durationInMillis); diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueProducer.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueProducer.java index b3778c2f8e..2498117c46 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueProducer.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueProducer.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueRequestTemplate.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueRequestTemplate.java index 5182bb7260..2eb429b082 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueRequestTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueRequestTemplate.java @@ -19,6 +19,10 @@ import com.google.common.util.concurrent.ListenableFuture; public interface TbQueueRequestTemplate { + void init(); + ListenableFuture send(Request request); + void stop(); + } diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueRuleEngineSettings.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueRuleEngineSettings.java new file mode 100644 index 0000000000..4acfec1091 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueRuleEngineSettings.java @@ -0,0 +1,31 @@ +/** + * Copyright © 2016-2020 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; + +import lombok.Data; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +@Data +@Component +public class TbQueueRuleEngineSettings { + + @Value("${queue.rule_engine.topic}") + private String topic; + + @Value("${queue.rule_engine.partitions}") + private int partitions; +} diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueTransportApiSettings.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueTransportApiSettings.java index 603d787fa8..b98bd3b76d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueTransportApiSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueTransportApiSettings.java @@ -22,6 +22,7 @@ import org.springframework.stereotype.Component; @Data @Component public class TbQueueTransportApiSettings { + @Value("${queue.transport_api.requests_topic}") private String requestsTopic; @@ -34,6 +35,12 @@ public class TbQueueTransportApiSettings { @Value("${queue.transport_api.max_requests_timeout}") private int maxRequestsTimeout; + @Value("${queue.transport_api.max_callback_threads}") + private int maxCallbackThreads; + + @Value("${queue.transport_api.request_poll_interval}") + private long requestPollInterval; + @Value("${queue.transport_api.response_poll_interval}") private long responsePollInterval; diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueTransportNotificationSettings.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueTransportNotificationSettings.java new file mode 100644 index 0000000000..2666045e7e --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueTransportNotificationSettings.java @@ -0,0 +1,32 @@ +/** + * Copyright © 2016-2020 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; + +import lombok.Data; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +@Data +@Component +public class TbQueueTransportNotificationSettings { + + @Value("${queue.transport.notifications_topic}") + private String notificationsTopic; + + @Value("${queue.transport.poll_interval}") + private long transportPollInterval; + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueRequestTemplate.java b/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueRequestTemplate.java index 53914953f3..0388daebb6 100644 --- a/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueRequestTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueRequestTemplate.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -79,6 +79,7 @@ public class DefaultTbQueueRequestTemplate partitionTopics = new ConcurrentHashMap<>(); private final ConcurrentMap partitionSizes = new ConcurrentHashMap<>(); @@ -53,8 +68,9 @@ public class ConsistentHashPartitionService implements PartitionService { private HashFunction hashFunction; - public ConsistentHashPartitionService(TbServiceInfoProvider serviceInfoProvider) { + public ConsistentHashPartitionService(TbServiceInfoProvider serviceInfoProvider, ApplicationEventPublisher applicationEventPublisher) { this.serviceInfoProvider = serviceInfoProvider; + this.applicationEventPublisher = applicationEventPublisher; } @PostConstruct @@ -128,13 +144,15 @@ public class ConsistentHashPartitionService implements PartitionService { for (int i = 0; i < size; i++) { ServiceInfo serviceInfo = resolveByPartitionIdx(newCircles.get(type), i); if (currentService.equals(serviceInfo)) { - myPartitions.putIfAbsent(new ServiceKey(type, getTenantId(serviceInfo)), new ArrayList<>()); + ServiceKey serviceKey = new ServiceKey(type, getTenantId(serviceInfo)); + myPartitions.computeIfAbsent(serviceKey, key -> new ArrayList<>()).add(i); } } }); myPartitions.forEach((serviceKey, partitions) -> { if (!partitions.equals(oldPartitions.get(serviceKey))) { log.info("[{}] NEW PARTITIONS: {}", serviceKey, partitions); + applicationEventPublisher.publishEvent(new PartitionChangeEvent(this, serviceKey, partitions)); } }); } @@ -178,32 +196,6 @@ public class ConsistentHashPartitionService implements PartitionService { return hashFunction.newHasher().putString(instance.getServiceId(), StandardCharsets.UTF_8).putInt(i).hash(); } - private static class ServiceKey { - @Getter - private final ServiceType serviceType; - @Getter - private final TenantId tenantId; - - public ServiceKey(ServiceType serviceType, TenantId tenantId) { - this.serviceType = serviceType; - this.tenantId = tenantId; - } - - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - ServiceKey that = (ServiceKey) o; - return serviceType == that.serviceType && - Objects.equals(tenantId, that.tenantId); - } - - @Override - public int hashCode() { - return Objects.hash(serviceType, tenantId); - } - } - public static HashFunction forName(String name) { switch (name) { case "murmur3_32": diff --git a/common/queue/src/main/java/org/thingsboard/server/discovery/DefaultTbServiceInfoProvider.java b/common/queue/src/main/java/org/thingsboard/server/discovery/DefaultTbServiceInfoProvider.java index 1fd87a40b6..80040a017b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/discovery/DefaultTbServiceInfoProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/discovery/DefaultTbServiceInfoProvider.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/common/queue/src/main/java/org/thingsboard/server/discovery/DummyDiscoveryService.java b/common/queue/src/main/java/org/thingsboard/server/discovery/DummyDiscoveryService.java index ef9aca72a0..8983cf1e97 100644 --- a/common/queue/src/main/java/org/thingsboard/server/discovery/DummyDiscoveryService.java +++ b/common/queue/src/main/java/org/thingsboard/server/discovery/DummyDiscoveryService.java @@ -1,3 +1,18 @@ +/** + * Copyright © 2016-2020 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.discovery; import lombok.extern.slf4j.Slf4j; diff --git a/common/queue/src/main/java/org/thingsboard/server/discovery/PartitionChangeEvent.java b/common/queue/src/main/java/org/thingsboard/server/discovery/PartitionChangeEvent.java index 7d083ba873..570a774c9c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/discovery/PartitionChangeEvent.java +++ b/common/queue/src/main/java/org/thingsboard/server/discovery/PartitionChangeEvent.java @@ -16,6 +16,7 @@ package org.thingsboard.server.discovery; import lombok.Getter; +import lombok.Setter; import org.springframework.context.ApplicationEvent; import java.util.List; @@ -23,11 +24,14 @@ import java.util.List; public class PartitionChangeEvent extends ApplicationEvent { + @Getter + private final ServiceKey serviceKey; @Getter private final List partitions; - public PartitionChangeEvent(Object source, List partitions) { + public PartitionChangeEvent(Object source, ServiceKey serviceKey, List partitions) { super(source); + this.serviceKey = serviceKey; this.partitions = partitions; } } diff --git a/common/queue/src/main/java/org/thingsboard/server/discovery/PartitionService.java b/common/queue/src/main/java/org/thingsboard/server/discovery/PartitionService.java index 90d43f4af3..ad61a63807 100644 --- a/common/queue/src/main/java/org/thingsboard/server/discovery/PartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/discovery/PartitionService.java @@ -1,3 +1,18 @@ +/** + * Copyright © 2016-2020 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.discovery; import org.thingsboard.server.common.data.id.EntityId; diff --git a/common/queue/src/main/java/org/thingsboard/server/discovery/ServiceKey.java b/common/queue/src/main/java/org/thingsboard/server/discovery/ServiceKey.java new file mode 100644 index 0000000000..0179be968c --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/discovery/ServiceKey.java @@ -0,0 +1,47 @@ +/** + * Copyright © 2016-2020 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.discovery; + +import lombok.Getter; +import org.thingsboard.server.common.data.id.TenantId; + +import java.util.Objects; + +public class ServiceKey { + @Getter + private final ServiceType serviceType; + @Getter + private final TenantId tenantId; + + public ServiceKey(ServiceType serviceType, TenantId tenantId) { + this.serviceType = serviceType; + this.tenantId = tenantId; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + ServiceKey that = (ServiceKey) o; + return serviceType == that.serviceType && + Objects.equals(tenantId, that.tenantId); + } + + @Override + public int hashCode() { + return Objects.hash(serviceType, tenantId); + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/discovery/ZkPartitionDiscoveryService.java b/common/queue/src/main/java/org/thingsboard/server/discovery/ZkPartitionDiscoveryService.java index d9af5bb894..07f11617ac 100644 --- a/common/queue/src/main/java/org/thingsboard/server/discovery/ZkPartitionDiscoveryService.java +++ b/common/queue/src/main/java/org/thingsboard/server/discovery/ZkPartitionDiscoveryService.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaConsumerTemplate.java index 558d556d09..51b8725118 100644 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaConsumerTemplate.java @@ -31,6 +31,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Properties; +import java.util.stream.Collectors; /** * Created by ashvayka on 24.09.18. @@ -40,6 +41,7 @@ public class TBKafkaConsumerTemplate implements TbQueueCon private final KafkaConsumer consumer; private final TbKafkaDecoder decoder; + private volatile boolean subscribed; @Getter private final String topic; @@ -69,23 +71,37 @@ public class TBKafkaConsumerTemplate implements TbQueueCon @Override public void subscribe() { consumer.subscribe(Collections.singletonList(topic)); + subscribed = true; + } + + @Override + public void subscribe(List partitions) { + consumer.subscribe(partitions.stream().map(partition -> topic + "." + partition).collect(Collectors.toList())); + subscribed = true; } @Override public List poll(long durationInMillis) { - ConsumerRecords records = consumer.poll(Duration.ofMillis(durationInMillis)); - if (records.count() > 0) { - List result = new ArrayList<>(); - records.forEach(record -> { - try { - result.add(decode(record)); - } catch (IOException e) { - log.error("Failed decode record: [{}]", record); - } - }); - return result; + if (!subscribed) { + try { + Thread.sleep(durationInMillis); + } catch (InterruptedException e) { + log.debug("Failed to await subscription", e); + } + } else { + ConsumerRecords records = consumer.poll(Duration.ofMillis(durationInMillis)); + if (records.count() > 0) { + List result = new ArrayList<>(); + records.forEach(record -> { + try { + result.add(decode(record)); + } catch (IOException e) { + log.error("Failed decode record: [{}]", record); + } + }); + return result; + } } - return Collections.emptyList(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java index 8072a2da66..1b5d57f1d2 100644 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryStorage.java b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryStorage.java index ded4cd9810..86d103ea70 100644 --- a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryStorage.java +++ b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryStorage.java @@ -15,16 +15,24 @@ */ package org.thingsboard.server.memory; +import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.TbQueueMsg; +import java.util.ArrayList; +import java.util.Collections; import java.util.LinkedList; +import java.util.List; import java.util.Map; import java.util.Queue; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +@Slf4j public final class InMemoryStorage { private static InMemoryStorage instance; - private final Map> storage; + private final Map> storage; private InMemoryStorage() { storage = new ConcurrentHashMap<>(); @@ -42,19 +50,38 @@ public final class InMemoryStorage { } public boolean put(String topic, TbQueueMsg msg) { - return storage.computeIfAbsent(topic, (t) -> new LinkedList<>()).add(msg); + return storage.computeIfAbsent(topic, (t) -> new LinkedBlockingQueue<>()).add(msg); } - public TbQueueMsg get(String topic) { + public List get(String topic, long durationInMillis) { if (storage.containsKey(topic)) { - return storage.get(topic).peek(); + try { + List entities; + T first = (T) storage.get(topic).poll(durationInMillis, TimeUnit.MILLISECONDS); + if (first != null) { + entities = new ArrayList<>(); + entities.add(first); + } else { + entities = Collections.emptyList(); + List otherList = new ArrayList<>(); + storage.get(topic).drainTo(otherList, 100); + for (TbQueueMsg other : otherList) { + entities.add((T) other); + } + } + return entities; + } catch (InterruptedException e) { + log.warn("Queue was interrupted", e); + return Collections.emptyList(); + } } - return null; + return Collections.emptyList(); } public void commit(String topic) { + //TODO: 2.5 Until someone calls commit you should not allow to poll new elements. if (storage.containsKey(topic)) { - storage.get(topic).remove(); +// storage.get(topic).remove(); } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueConsumer.java b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueConsumer.java index b8ddcff89c..3f1e844efc 100644 --- a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueConsumer.java +++ b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueConsumer.java @@ -18,7 +18,6 @@ package org.thingsboard.server.memory; import org.thingsboard.server.TbQueueConsumer; import org.thingsboard.server.TbQueueMsg; -import java.util.Collections; import java.util.List; public class InMemoryTbQueueConsumer implements TbQueueConsumer { @@ -40,6 +39,11 @@ public class InMemoryTbQueueConsumer implements TbQueueCon } + @Override + public void subscribe(List partitions) { + + } + @Override public void unsubscribe() { @@ -47,7 +51,7 @@ public class InMemoryTbQueueConsumer implements TbQueueCon @Override public List poll(long durationInMillis) { - return Collections.singletonList((T)storage.get(topic)); + return storage.get(topic, durationInMillis); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueProducer.java b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueProducer.java index 8abb5e2d43..79fa1d6b39 100644 --- a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueProducer.java +++ b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueProducer.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -41,10 +41,13 @@ public class InMemoryTbQueueProducer implements TbQueuePro public void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback) { boolean result = storage.put(tpi.getTopic(), msg); if (result) { - callback.onSuccess(null); + if (callback != null) { + callback.onSuccess(null); + } } else { - Exception e = new RuntimeException("Failure add msg to InMemoryQueue"); - callback.onFailure(e); + if (callback != null) { + callback.onFailure(new RuntimeException("Failure add msg to InMemoryQueue")); + } } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryMonolithQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryMonolithQueueProvider.java index c7839e70e9..4fa10b7c44 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryMonolithQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryMonolithQueueProvider.java @@ -21,7 +21,11 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.TbQueueConsumer; import org.thingsboard.server.TbQueueCoreSettings; import org.thingsboard.server.TbQueueProducer; +import org.thingsboard.server.TbQueueRuleEngineSettings; +import org.thingsboard.server.TbQueueTransportApiSettings; +import org.thingsboard.server.TbQueueTransportNotificationSettings; import org.thingsboard.server.common.TbProtoQueueMsg; +import org.thingsboard.server.discovery.PartitionChangeEvent; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; @@ -36,50 +40,52 @@ import org.thingsboard.server.memory.InMemoryTbQueueProducer; public class InMemoryMonolithQueueProvider implements TbCoreQueueProvider, TbRuleEngineQueueProvider { private final TbQueueCoreSettings coreSettings; + private final TbQueueRuleEngineSettings ruleEngineSettings; + private final TbQueueTransportApiSettings transportApiSettings; + private final TbQueueTransportNotificationSettings notificationSettings; - public InMemoryMonolithQueueProvider(TbQueueCoreSettings coreSettings) { + public InMemoryMonolithQueueProvider(TbQueueCoreSettings coreSettings, + TbQueueRuleEngineSettings ruleEngineSettings, + TbQueueTransportApiSettings transportApiSettings, + TbQueueTransportNotificationSettings notificationSettings) { this.coreSettings = coreSettings; + this.ruleEngineSettings = ruleEngineSettings; + this.transportApiSettings = transportApiSettings; + this.notificationSettings = notificationSettings; } @Override public TbQueueProducer> getTransportMsgProducer() { - InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); - return producer; + return new InMemoryTbQueueProducer<>(notificationSettings.getNotificationsTopic()); } @Override public TbQueueProducer> getRuleEngineMsgProducer() { - InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); - return producer; + return new InMemoryTbQueueProducer<>(ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> getTbCoreMsgProducer() { - InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); - return producer; + return new InMemoryTbQueueProducer<>(coreSettings.getTopic()); } @Override public TbQueueConsumer> getToRuleEngineMsgConsumer() { - InMemoryTbQueueConsumer> consumer = new InMemoryTbQueueConsumer<>(coreSettings.getTopic()); - return consumer; + return new InMemoryTbQueueConsumer<>(ruleEngineSettings.getTopic()); } @Override public TbQueueConsumer> getToCoreMsgConsumer() { - InMemoryTbQueueConsumer> consumer = new InMemoryTbQueueConsumer<>(coreSettings.getTopic()); - return consumer; + return new InMemoryTbQueueConsumer<>(coreSettings.getTopic()); } @Override public TbQueueConsumer> getTransportApiRequestConsumer() { - InMemoryTbQueueConsumer> consumer = new InMemoryTbQueueConsumer<>(coreSettings.getTopic()); - return consumer; + return new InMemoryTbQueueConsumer<>(transportApiSettings.getRequestsTopic()); } @Override public TbQueueProducer> getTransportApiResponseProducer() { - InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); - return producer; + return new InMemoryTbQueueProducer<>(transportApiSettings.getResponsesTopic()); } } diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryTransportQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryTransportQueueProvider.java index 789ea1cf5a..2d7205e516 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryTransportQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryTransportQueueProvider.java @@ -15,13 +15,19 @@ */ package org.thingsboard.server.provider; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Component; +import org.thingsboard.server.TbQueueAdmin; import org.thingsboard.server.TbQueueConsumer; +import org.thingsboard.server.TbQueueCoreSettings; import org.thingsboard.server.TbQueueProducer; import org.thingsboard.server.TbQueueRequestTemplate; +import org.thingsboard.server.TbQueueRuleEngineSettings; import org.thingsboard.server.TbQueueTransportApiSettings; +import org.thingsboard.server.TbQueueTransportNotificationSettings; import org.thingsboard.server.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.common.TbProtoQueueMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; @@ -37,20 +43,29 @@ import org.thingsboard.server.memory.InMemoryTbQueueProducer; @Slf4j public class InMemoryTransportQueueProvider implements TransportQueueProvider { + private final TbQueueCoreSettings coreSettings; + private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueTransportApiSettings transportApiSettings; + private final TbQueueTransportNotificationSettings notificationSettings; - public InMemoryTransportQueueProvider(TbQueueTransportApiSettings transportApiSettings) { + public InMemoryTransportQueueProvider(TbQueueCoreSettings coreSettings, + TbQueueRuleEngineSettings ruleEngineSettings, + TbQueueTransportApiSettings transportApiSettings, + TbQueueTransportNotificationSettings notificationSettings) { + this.coreSettings = coreSettings; + this.ruleEngineSettings = ruleEngineSettings; this.transportApiSettings = transportApiSettings; + this.notificationSettings = notificationSettings; } @Override public TbQueueRequestTemplate, TbProtoQueueMsg> getTransportApiRequestTemplate() { InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic()); - InMemoryTbQueueConsumer> consumer = new InMemoryTbQueueConsumer<>(transportApiSettings.getResponsesTopic()); DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder , TbProtoQueueMsg> templateBuilder = DefaultTbQueueRequestTemplate.builder(); + templateBuilder.queueAdmin(topic -> Futures.immediateFuture(null)); templateBuilder.requestTemplate(producer); templateBuilder.responseTemplate(consumer); templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests()); @@ -61,19 +76,16 @@ public class InMemoryTransportQueueProvider implements TransportQueueProvider { @Override public TbQueueProducer> getRuleEngineMsgProducer() { - InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic()); - return producer; + return new InMemoryTbQueueProducer<>(ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> getTbCoreMsgProducer() { - InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic()); - return producer; + return new InMemoryTbQueueProducer<>(coreSettings.getTopic()); } @Override public TbQueueConsumer> getTransportNotificationsConsumer() { - InMemoryTbQueueConsumer> consumer = new InMemoryTbQueueConsumer<>(transportApiSettings.getResponsesTopic()); - return consumer; + return new InMemoryTbQueueConsumer<>(notificationSettings.getNotificationsTopic()); } } diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaMonolithQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaMonolithQueueProvider.java index 6aaf76420b..f06897ebe6 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaMonolithQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaMonolithQueueProvider.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -21,6 +21,8 @@ import org.thingsboard.server.TbQueueConsumer; import org.thingsboard.server.TbQueueCoreSettings; import org.thingsboard.server.TbQueueProducer; import org.thingsboard.server.common.TbProtoQueueMsg; +import org.thingsboard.server.discovery.PartitionChangeEvent; +import org.thingsboard.server.discovery.ServiceType; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbCoreQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbCoreQueueProvider.java index 426a30eea1..5cbe93196a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbCoreQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbCoreQueueProvider.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -21,6 +21,8 @@ import org.thingsboard.server.TbQueueConsumer; import org.thingsboard.server.TbQueueCoreSettings; import org.thingsboard.server.TbQueueProducer; import org.thingsboard.server.common.TbProtoQueueMsg; +import org.thingsboard.server.discovery.PartitionChangeEvent; +import org.thingsboard.server.discovery.ServiceType; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbRuleEngineQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbRuleEngineQueueProvider.java index 6057ebba6f..eb0bcb5d6e 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbRuleEngineQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbRuleEngineQueueProvider.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTransportQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTransportQueueProvider.java index c7b5b6f839..dcf6474e1e 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTransportQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTransportQueueProvider.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/TbCoreQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/TbCoreQueueProvider.java index 6123fc0859..ef9cd23ac3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/TbCoreQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/TbCoreQueueProvider.java @@ -15,9 +15,11 @@ */ package org.thingsboard.server.provider; +import org.springframework.context.ApplicationListener; import org.thingsboard.server.TbQueueConsumer; import org.thingsboard.server.TbQueueProducer; import org.thingsboard.server.common.TbProtoQueueMsg; +import org.thingsboard.server.discovery.PartitionChangeEvent; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/TbRuleEngineQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/TbRuleEngineQueueProvider.java index 46f1a3351d..2efa9ffbc6 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/TbRuleEngineQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/TbRuleEngineQueueProvider.java @@ -5,7 +5,7 @@ * 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 + * 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, 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 661411cf89..19bf43f7db 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 @@ -5,7 +5,7 @@ * 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 + * 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, @@ -78,13 +78,12 @@ public class DefaultTransportService implements TransportService { private long sessionInactivityTimeout; @Value("${transport.sessions.report_timeout}") private long sessionReportTimeout; + @Value("${queue.transport.poll_interval}") + private int notificationsPollDuration; private final TransportQueueProvider queueProvider; private final PartitionService partitionService; - @Value("${kafka.notifications.poll_interval}") - private int notificationsPollDuration; - protected TbQueueRequestTemplate, TbProtoQueueMsg> transportApiRequestTemplate; protected TbQueueProducer> ruleEngineMsgProducer; protected TbQueueProducer> tbCoreMsgProducer; @@ -120,7 +119,7 @@ public class DefaultTransportService implements TransportService { ruleEngineMsgProducer = queueProvider.getRuleEngineMsgProducer(); tbCoreMsgProducer = queueProvider.getTbCoreMsgProducer(); transportNotificationsConsumer = queueProvider.getTransportNotificationsConsumer(); - + transportApiRequestTemplate.init(); mainConsumerExecutor.execute(() -> { while (!stopped) { try { @@ -163,6 +162,9 @@ public class DefaultTransportService implements TransportService { if (mainConsumerExecutor != null) { mainConsumerExecutor.shutdownNow(); } + if (transportApiRequestTemplate != null) { + transportApiRequestTemplate.stop(); + } } @Override