From 24f5872adf3f50f36629ef0d5fb3b791c92be838 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Mon, 19 Oct 2020 18:00:20 +0300 Subject: [PATCH] Queue interfaces with no implementation --- .../queue/provider/TbCoreQueueFactory.java | 13 ++- .../provider/TbQueueProducerProvider.java | 11 ++ .../provider/TbRuleEngineQueueFactory.java | 2 +- .../provider/TbTransportQueueFactory.java | 10 +- .../TbTransportQueueProducerProvider.java | 7 ++ .../TbUsageStatsClientQueueFactory.java | 29 +++++ .../scheduler/DefaultSchedulerComponent.java | 61 ++++++++++ .../queue/scheduler/SchedulerComponent.java | 32 ++++++ .../usagestats/DefaultTbUsageStatsClient.java | 106 ++++++++++++++++++ .../queue/usagestats/TbUsageStatsClient.java | 26 +++++ common/queue/src/main/proto/queue.proto | 13 +++ .../transport/mqtt/MqttTransportHandler.java | 5 +- .../common/transport/TransportContext.java | 3 + .../common/transport/TransportService.java | 8 +- .../service/DefaultTransportService.java | 24 ++-- 15 files changed, 318 insertions(+), 32 deletions(-) create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/provider/TbUsageStatsClientQueueFactory.java create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/scheduler/DefaultSchedulerComponent.java create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/scheduler/SchedulerComponent.java create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbUsageStatsClient.java create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/usagestats/TbUsageStatsClient.java 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 59a8f449b0..d009757997 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 @@ -16,6 +16,7 @@ package org.thingsboard.server.queue.provider; import org.thingsboard.server.gen.js.JsInvokeProtos; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -33,7 +34,7 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg; * Responsible for initialization of various Producers and Consumers used by TB Core Node. * Implementation Depends on the queue queue.type from yml or TB_QUEUE_TYPE environment variable */ -public interface TbCoreQueueFactory { +public interface TbCoreQueueFactory extends TbUsageStatsClientQueueFactory { /** * Used to push messages to instances of TB Transport Service @@ -77,6 +78,16 @@ public interface TbCoreQueueFactory { */ TbQueueConsumer> createToCoreMsgConsumer(); + /** + * Used to consume messages about usage statistics by TB Core Service + * + * @return + */ + default TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { + //TODO: implement + return null; + } + /** * Used to consume high priority messages by TB Core Service * diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java index 34c6ee577f..fa20f11960 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbQueueProducerProvider.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.queue.provider; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -63,4 +64,14 @@ public interface TbQueueProducerProvider { */ TbQueueProducer> getTbCoreNotificationsMsgProducer(); + /** + * Used to push messages to other instances of TB Core Service + * + * @return + */ + default TbQueueProducer> getTbUsageStatsMsgProducer() { + //TODO: implement + 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 561b3e84ea..2f80bbce45 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 @@ -32,7 +32,7 @@ import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; * Responsible for initialization of various Producers and Consumers used by TB Core Node. * Implementation Depends on the queue queue.type from yml or TB_QUEUE_TYPE environment variable */ -public interface TbRuleEngineQueueFactory { +public interface TbRuleEngineQueueFactory extends TbUsageStatsClientQueueFactory { /** * Used to push messages to instances of TB Transport Service diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueFactory.java index dc1d2c449c..bd6c6e0584 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueFactory.java @@ -15,17 +15,17 @@ */ package org.thingsboard.server.queue.provider; -import org.thingsboard.server.queue.TbQueueConsumer; -import org.thingsboard.server.queue.TbQueueProducer; -import org.thingsboard.server.queue.TbQueueRequestTemplate; -import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; -public interface TbTransportQueueFactory { +public interface TbTransportQueueFactory extends TbUsageStatsClientQueueFactory { TbQueueRequestTemplate, TbProtoQueueMsg> createTransportApiRequestTemplate(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java index bced3f83ac..9d11aa727b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbTransportQueueProducerProvider.java @@ -30,6 +30,7 @@ public class TbTransportQueueProducerProvider implements TbQueueProducerProvider private final TbTransportQueueFactory tbQueueProvider; private TbQueueProducer> toRuleEngine; private TbQueueProducer> toTbCore; + private TbQueueProducer> toUsageStats; public TbTransportQueueProducerProvider(TbTransportQueueFactory tbQueueProvider) { this.tbQueueProvider = tbQueueProvider; @@ -39,6 +40,7 @@ public class TbTransportQueueProducerProvider implements TbQueueProducerProvider public void init() { this.toTbCore = tbQueueProvider.createTbCoreMsgProducer(); this.toRuleEngine = tbQueueProvider.createRuleEngineMsgProducer(); + this.toUsageStats = tbQueueProvider.createToUsageStatsServiceMsgProducer(); } @Override @@ -65,4 +67,9 @@ public class TbTransportQueueProducerProvider implements TbQueueProducerProvider public TbQueueProducer> getTbCoreNotificationsMsgProducer() { throw new RuntimeException("Not Implemented! Should not be used by Transport!"); } + + @Override + public TbQueueProducer> getTbUsageStatsMsgProducer() { + return toUsageStats; + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbUsageStatsClientQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbUsageStatsClientQueueFactory.java new file mode 100644 index 0000000000..1811877338 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/TbUsageStatsClientQueueFactory.java @@ -0,0 +1,29 @@ +/** + * 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.queue.provider; + +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; +import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; + +public interface TbUsageStatsClientQueueFactory { + + default TbQueueProducer> createToUsageStatsServiceMsgProducer() { + //TODO: implement + return null; + } + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/scheduler/DefaultSchedulerComponent.java b/common/queue/src/main/java/org/thingsboard/server/queue/scheduler/DefaultSchedulerComponent.java new file mode 100644 index 0000000000..02ce60496e --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/scheduler/DefaultSchedulerComponent.java @@ -0,0 +1,61 @@ +/** + * 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.queue.scheduler; + +import org.springframework.stereotype.Component; +import org.thingsboard.common.util.ThingsBoardThreadFactory; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import java.util.concurrent.Callable; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +@Component +public class DefaultSchedulerComponent implements SchedulerComponent{ + + protected ScheduledExecutorService schedulerExecutor; + + @PostConstruct + public void init(){ + this.schedulerExecutor = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("queue-scheduler")); + } + + @PreDestroy + public void destroy() { + if (schedulerExecutor != null) { + schedulerExecutor.shutdownNow(); + } + } + + public ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit) { + return schedulerExecutor.schedule(command, delay, unit); + } + + public ScheduledFuture schedule(Callable callable, long delay, TimeUnit unit) { + return schedulerExecutor.schedule(callable, delay, unit); + } + + public ScheduledFuture scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) { + return schedulerExecutor.scheduleAtFixedRate(command, initialDelay, period, unit); + } + + public ScheduledFuture scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) { + return schedulerExecutor.scheduleWithFixedDelay(command, initialDelay, delay, unit); + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/scheduler/SchedulerComponent.java b/common/queue/src/main/java/org/thingsboard/server/queue/scheduler/SchedulerComponent.java new file mode 100644 index 0000000000..3c12be43a5 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/scheduler/SchedulerComponent.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.queue.scheduler; + +import java.util.concurrent.Callable; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +public interface SchedulerComponent { + + ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit); + + ScheduledFuture schedule(Callable callable, long delay, TimeUnit unit); + + ScheduledFuture scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit); + + ScheduledFuture scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit); + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbUsageStatsClient.java b/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbUsageStatsClient.java new file mode 100644 index 0000000000..87eb5e767a --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/DefaultTbUsageStatsClient.java @@ -0,0 +1,106 @@ +/** + * 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.queue.usagestats; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.UsageRecordKey; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.gen.transport.TransportProtos.ToUsageStatsServiceMsg; +import org.thingsboard.server.gen.transport.TransportProtos.UsageStatsKVProto; +import org.thingsboard.server.queue.TbQueueCallback; +import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.provider.TbQueueProducerProvider; +import org.thingsboard.server.queue.scheduler.SchedulerComponent; + +import javax.annotation.PostConstruct; +import java.util.Random; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +@Component +public class DefaultTbUsageStatsClient implements TbUsageStatsClient { + + @Value("${usage.stats.report.enabled:true}") + private boolean enabled; + @Value("${usage.stats.report.interval:600}") + private int interval; + + private final ConcurrentMap[] values = new ConcurrentMap[UsageRecordKey.values().length]; + private final PartitionService partitionService; + private final SchedulerComponent scheduler; + private final TbQueueProducerProvider producerProvider; + private TbQueueProducer> msgProducer; + + public DefaultTbUsageStatsClient(PartitionService partitionService, SchedulerComponent scheduler, TbQueueProducerProvider producerProvider) { + this.partitionService = partitionService; + this.scheduler = scheduler; + this.producerProvider = producerProvider; + } + + @PostConstruct + private void init() { + if (enabled) { + msgProducer = this.producerProvider.getTbUsageStatsMsgProducer(); + for (UsageRecordKey key : UsageRecordKey.values()) { + values[key.ordinal()] = new ConcurrentHashMap<>(); + } + scheduler.scheduleWithFixedDelay(this::reportStats, new Random().nextInt(interval), interval, TimeUnit.SECONDS); + } + } + + private void reportStats() { + ConcurrentMap report = new ConcurrentHashMap<>(); + + for (UsageRecordKey key : UsageRecordKey.values()) { + values[key.ordinal()].forEach(((tenantId, atomicLong) -> { + long value = atomicLong.getAndSet(0); + if (value > 0) { + ToUsageStatsServiceMsg.Builder msgBuilder = report.computeIfAbsent(tenantId, id -> { + ToUsageStatsServiceMsg.Builder msg = ToUsageStatsServiceMsg.newBuilder(); + msg.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); + msg.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); + return msg; + }); + msgBuilder.addValues(UsageStatsKVProto.newBuilder().setKey(key.name()).setValue(value).build()); + } + })); + } + + report.forEach(((tenantId, builder) -> { + //TODO: figure out how to minimize messages into the queue. Maybe group by 100s of messages? + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, tenantId); + msgProducer.send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), builder.build()), null); + })); + } + + @Override + public void report(TenantId tenantId, UsageRecordKey key, long value) { + if (enabled) { + ConcurrentMap map = values[key.ordinal()]; + AtomicLong atomicValue = map.computeIfAbsent(tenantId, id -> new AtomicLong()); + atomicValue.addAndGet(value); + } + } + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/TbUsageStatsClient.java b/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/TbUsageStatsClient.java new file mode 100644 index 0000000000..e4e659b5c5 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/usagestats/TbUsageStatsClient.java @@ -0,0 +1,26 @@ +/** + * 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.queue.usagestats; + +import org.thingsboard.server.common.data.UsageRecordKey; +import org.thingsboard.server.common.data.id.TenantId; + +public interface TbUsageStatsClient { + + void report(TenantId tenantId, UsageRecordKey key, long value); + + +} diff --git a/common/queue/src/main/proto/queue.proto b/common/queue/src/main/proto/queue.proto index 01d8ca4524..f716bd90ae 100644 --- a/common/queue/src/main/proto/queue.proto +++ b/common/queue/src/main/proto/queue.proto @@ -532,3 +532,16 @@ message ToTransportMsg { EntityDeleteMsg entityDeleteMsg = 9; ProvisionDeviceResponseMsg provisionResponse = 10; } + +message UsageStatsKVProto{ + string key = 1; + int64 value = 2; +} + +message ToUsageStatsServiceMsg { + int64 tenantIdMSB = 1; + int64 tenantIdLSB = 2; + int64 entityIdMSB = 3; + int64 entityIdLSB = 4; + repeated UsageStatsKVProto values = 5; +} diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index 3723b19f6d..2c4b2db7df 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -57,6 +57,7 @@ import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceResponseMsg; import org.thingsboard.server.gen.transport.TransportProtos.SessionEvent; import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg; +import org.thingsboard.server.queue.scheduler.SchedulerComponent; import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor; import org.thingsboard.server.transport.mqtt.session.DeviceSessionCtx; import org.thingsboard.server.transport.mqtt.session.GatewaySessionHandler; @@ -97,6 +98,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement private final UUID sessionId; private final MqttTransportContext context; private final TransportService transportService; + private final SchedulerComponent scheduler; private final SslHandler sslHandler; private final ConcurrentMap mqttQoSMap; @@ -108,6 +110,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement this.sessionId = UUID.randomUUID(); this.context = context; this.transportService = context.getTransportService(); + this.scheduler = context.getScheduler(); this.sslHandler = sslHandler; this.mqttQoSMap = new ConcurrentHashMap<>(); this.deviceSessionCtx = new DeviceSessionCtx(sessionId, mqttQoSMap, context); @@ -333,7 +336,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement } else { deviceSessionCtx.getContext().getProtoMqttAdaptor().convertToPublish(deviceSessionCtx, provisionResponseMsg).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush); } - transportService.getSchedulerExecutor().schedule(() -> processDisconnect(ctx), 60, TimeUnit.SECONDS); + scheduler.schedule(() -> processDisconnect(ctx), 60, TimeUnit.SECONDS); } catch (Exception e) { log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e); } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportContext.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportContext.java index 66809f9e61..4b7677dd69 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportContext.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportContext.java @@ -23,6 +23,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Service; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.queue.scheduler.SchedulerComponent; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; @@ -44,6 +45,8 @@ public abstract class TransportContext { private TransportService transportService; @Autowired private TbServiceInfoProvider serviceInfoProvider; + @Autowired + private SchedulerComponent scheduler; @Getter private ExecutorService executor; diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java index 8f57db3897..4872d8e477 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java @@ -17,16 +17,14 @@ package org.thingsboard.server.common.transport; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.DeviceTransportType; -import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; import org.thingsboard.server.common.transport.limits.TransportRateLimitType; -import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg; -import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileResponseMsg; +import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.PostAttributeMsg; import org.thingsboard.server.gen.transport.TransportProtos.PostTelemetryMsg; import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceRequestMsg; @@ -42,8 +40,6 @@ import org.thingsboard.server.gen.transport.TransportProtos.ValidateBasicMqttCre import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg; -import java.util.concurrent.ScheduledExecutorService; - /** * Created by ashvayka on 04.10.18. */ @@ -92,8 +88,6 @@ public interface TransportService { void process(SessionInfoProto sessionInfo, ClaimDeviceMsg msg, TransportServiceCallback callback); - ScheduledExecutorService getSchedulerExecutor(); - void registerAsyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener); void registerSyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener, long timeout); 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 3f37685383..774a437e39 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 @@ -23,7 +23,6 @@ import com.google.gson.JsonObject; import com.google.protobuf.ByteString; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.data.DeviceProfile; @@ -53,7 +52,6 @@ import org.thingsboard.server.common.transport.TransportTenantProfileCache; import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse; import org.thingsboard.server.common.transport.auth.TransportDeviceInfo; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; -import org.thingsboard.server.common.transport.limits.TransportRateLimit; import org.thingsboard.server.common.transport.limits.TransportRateLimitService; import org.thingsboard.server.common.transport.limits.TransportRateLimitType; import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult; @@ -79,6 +77,7 @@ import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.provider.TbTransportQueueFactory; +import org.thingsboard.server.queue.scheduler.SchedulerComponent; import org.thingsboard.server.queue.util.TbTransportComponent; import javax.annotation.PostConstruct; @@ -94,7 +93,6 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -126,6 +124,7 @@ public class DefaultTransportService implements TransportService { private final TransportTenantProfileCache tenantProfileCache; private final TransportRateLimitService rateLimitService; private final DataDecodingEncodingService dataDecodingEncodingService; + private final SchedulerComponent scheduler; protected TbQueueRequestTemplate, TbProtoQueueMsg> transportApiRequestTemplate; protected TbQueueProducer> ruleEngineMsgProducer; @@ -136,7 +135,6 @@ public class DefaultTransportService implements TransportService { protected MessagesStats tbCoreProducerStats; protected MessagesStats transportApiStats; - protected ScheduledExecutorService schedulerExecutor; protected ExecutorService transportCallbackExecutor; private ExecutorService mainConsumerExecutor; @@ -152,7 +150,7 @@ public class DefaultTransportService implements TransportService { StatsFactory statsFactory, TransportDeviceProfileCache deviceProfileCache, TransportTenantProfileCache tenantProfileCache, - TransportRateLimitService rateLimitService, DataDecodingEncodingService dataDecodingEncodingService) { + TransportRateLimitService rateLimitService, DataDecodingEncodingService dataDecodingEncodingService, SchedulerComponent scheduler) { this.serviceInfoProvider = serviceInfoProvider; this.queueProvider = queueProvider; this.producerProvider = producerProvider; @@ -162,6 +160,7 @@ public class DefaultTransportService implements TransportService { this.tenantProfileCache = tenantProfileCache; this.rateLimitService = rateLimitService; this.dataDecodingEncodingService = dataDecodingEncodingService; + this.scheduler = scheduler; } @PostConstruct @@ -169,9 +168,8 @@ public class DefaultTransportService implements TransportService { this.ruleEngineProducerStats = statsFactory.createMessagesStats(StatsType.RULE_ENGINE.getName() + ".producer"); this.tbCoreProducerStats = statsFactory.createMessagesStats(StatsType.CORE.getName() + ".producer"); this.transportApiStats = statsFactory.createMessagesStats(StatsType.TRANSPORT.getName() + ".producer"); - this.schedulerExecutor = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("transport-scheduler")); this.transportCallbackExecutor = Executors.newWorkStealingPool(20); - this.schedulerExecutor.scheduleAtFixedRate(this::checkInactivityAndReportActivity, new Random().nextInt((int) sessionReportTimeout), sessionReportTimeout, TimeUnit.MILLISECONDS); + this.scheduler.scheduleAtFixedRate(this::checkInactivityAndReportActivity, new Random().nextInt((int) sessionReportTimeout), sessionReportTimeout, TimeUnit.MILLISECONDS); transportApiRequestTemplate = queueProvider.createTransportApiRequestTemplate(); transportApiRequestTemplate.setMessagesStats(transportApiStats); ruleEngineMsgProducer = producerProvider.getRuleEngineMsgProducer(); @@ -217,9 +215,6 @@ public class DefaultTransportService implements TransportService { if (transportNotificationsConsumer != null) { transportNotificationsConsumer.unsubscribe(); } - if (schedulerExecutor != null) { - schedulerExecutor.shutdownNow(); - } if (transportCallbackExecutor != null) { transportCallbackExecutor.shutdownNow(); } @@ -231,11 +226,6 @@ public class DefaultTransportService implements TransportService { } } - @Override - public ScheduledExecutorService getSchedulerExecutor() { - return this.schedulerExecutor; - } - @Override public void registerAsyncSession(TransportProtos.SessionInfoProto sessionInfo, SessionMsgListener listener) { sessions.putIfAbsent(toSessionId(sessionInfo), new SessionMetaData(sessionInfo, TransportProtos.SessionType.ASYNC, listener)); @@ -488,7 +478,7 @@ public class DefaultTransportService implements TransportService { sendToRuleEngine(tenantId, tbMsg, new TransportTbQueueCallback(callback)); String requestId = sessionId + "-" + msg.getRequestId(); toServerRpcPendingMap.put(requestId, new RpcRequestMetadata(sessionId, msg.getRequestId())); - schedulerExecutor.schedule(() -> processTimeout(requestId), clientSideRpcTimeout, TimeUnit.MILLISECONDS); + scheduler.schedule(() -> processTimeout(requestId), clientSideRpcTimeout, TimeUnit.MILLISECONDS); } } @@ -561,7 +551,7 @@ public class DefaultTransportService implements TransportService { SessionMetaData currentSession = new SessionMetaData(sessionInfo, TransportProtos.SessionType.SYNC, listener); sessions.putIfAbsent(toSessionId(sessionInfo), currentSession); - ScheduledFuture executorFuture = schedulerExecutor.schedule(() -> { + ScheduledFuture executorFuture = scheduler.schedule(() -> { listener.onRemoteSessionCloseCommand(TransportProtos.SessionCloseNotificationProto.getDefaultInstance()); deregisterSession(sessionInfo); }, timeout, TimeUnit.MILLISECONDS);