Queue interfaces with no implementation

@@ -16,6 +16,7 @@
 package org.thingsboard.server.queue.provider;
 package org.thingsboard.server.queue.provider;
 
 
 import org.thingsboard.server.gen.js.JsInvokeProtos;
 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.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
 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.
  * 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
  * 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
      * Used to push messages to instances of TB Transport Service
@@ -78,6 +79,16 @@ public interface TbCoreQueueFactory {
     TbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> createToCoreMsgConsumer();
     TbQueueConsumer<TbProtoQueueMsg<ToCoreMsg>> createToCoreMsgConsumer();
 
 
     /**
     /**
+     * Used to consume messages about usage statistics by TB Core Service
+     *
+     * @return
+     */
+    default TbQueueConsumer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> createToUsageStatsServiceMsgConsumer() {
+        //TODO: implement
+        return null;
+    }
+
+    /**
      * Used to consume high priority messages by TB Core Service
      * Used to consume high priority messages by TB Core Service
      *
      *
      * @return
      * @return

@@ -15,6 +15,7 @@
  */
  */
 package org.thingsboard.server.queue.provider;
 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.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
@@ -63,4 +64,14 @@ public interface TbQueueProducerProvider {
      */
      */
     TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer();
     TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer();
 
 
+    /**
+     * Used to push messages to other instances of TB Core Service
+     *
+     * @return
+     */
+    default TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer() {
+        //TODO: implement
+        return null;
+    }
+
 }
 }

@@ -32,7 +32,7 @@ import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
  * Responsible for initialization of various Producers and Consumers used by TB Core Node.
  * 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
  * 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
      * Used to push messages to instances of TB Transport Service

@@ -15,17 +15,17 @@
  */
  */
 package org.thingsboard.server.queue.provider;
 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.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
 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<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> createTransportApiRequestTemplate();
     TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> createTransportApiRequestTemplate();
 
 

@@ -30,6 +30,7 @@ public class TbTransportQueueProducerProvider implements TbQueueProducerProvider
     private final TbTransportQueueFactory tbQueueProvider;
     private final TbTransportQueueFactory tbQueueProvider;
     private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> toRuleEngine;
     private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> toRuleEngine;
     private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> toTbCore;
     private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> toTbCore;
+    private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToUsageStatsServiceMsg>> toUsageStats;
 
 
     public TbTransportQueueProducerProvider(TbTransportQueueFactory tbQueueProvider) {
     public TbTransportQueueProducerProvider(TbTransportQueueFactory tbQueueProvider) {
         this.tbQueueProvider = tbQueueProvider;
         this.tbQueueProvider = tbQueueProvider;
@@ -39,6 +40,7 @@ public class TbTransportQueueProducerProvider implements TbQueueProducerProvider
     public void init() {
     public void init() {
         this.toTbCore = tbQueueProvider.createTbCoreMsgProducer();
         this.toTbCore = tbQueueProvider.createTbCoreMsgProducer();
         this.toRuleEngine = tbQueueProvider.createRuleEngineMsgProducer();
         this.toRuleEngine = tbQueueProvider.createRuleEngineMsgProducer();
+        this.toUsageStats = tbQueueProvider.createToUsageStatsServiceMsgProducer();
     }
     }
 
 
     @Override
     @Override
@@ -65,4 +67,9 @@ public class TbTransportQueueProducerProvider implements TbQueueProducerProvider
     public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() {
     public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() {
         throw new RuntimeException("Not Implemented! Should not be used by Transport!");
         throw new RuntimeException("Not Implemented! Should not be used by Transport!");
     }
     }
+
+    @Override
+    public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToUsageStatsServiceMsg>> getTbUsageStatsMsgProducer() {
+        return toUsageStats;
+    }
 }
 }

+/**
+ * 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<TbProtoQueueMsg<ToUsageStatsServiceMsg>> createToUsageStatsServiceMsgProducer() {
+        //TODO: implement
+        return null;
+    }
+
+}

+/**
+ * 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 <V> ScheduledFuture<V> schedule(Callable<V> 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);
+    }
+}

+/**
+ * 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);
+
+    <V> ScheduledFuture<V> schedule(Callable<V> 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);
+
+}

+/**
+ * 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<TenantId, AtomicLong>[] values = new ConcurrentMap[UsageRecordKey.values().length];
+    private final PartitionService partitionService;
+    private final SchedulerComponent scheduler;
+    private final TbQueueProducerProvider producerProvider;
+    private TbQueueProducer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> 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<TenantId, ToUsageStatsServiceMsg.Builder> 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<TenantId, AtomicLong> map = values[key.ordinal()];
+            AtomicLong atomicValue = map.computeIfAbsent(tenantId, id -> new AtomicLong());
+            atomicValue.addAndGet(value);
+        }
+    }
+
+}

+/**
+ * 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);
+
+
+}

@@ -532,3 +532,16 @@ message ToTransportMsg {
   EntityDeleteMsg entityDeleteMsg = 9;
   EntityDeleteMsg entityDeleteMsg = 9;
   ProvisionDeviceResponseMsg provisionResponse = 10;
   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;
+}

@@ -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.ProvisionDeviceResponseMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.SessionEvent;
 import org.thingsboard.server.gen.transport.TransportProtos.SessionEvent;
 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
 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.adaptors.MqttTransportAdaptor;
 import org.thingsboard.server.transport.mqtt.session.DeviceSessionCtx;
 import org.thingsboard.server.transport.mqtt.session.DeviceSessionCtx;
 import org.thingsboard.server.transport.mqtt.session.GatewaySessionHandler;
 import org.thingsboard.server.transport.mqtt.session.GatewaySessionHandler;
@@ -97,6 +98,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
     private final UUID sessionId;
     private final UUID sessionId;
     private final MqttTransportContext context;
     private final MqttTransportContext context;
     private final TransportService transportService;
     private final TransportService transportService;
+    private final SchedulerComponent scheduler;
     private final SslHandler sslHandler;
     private final SslHandler sslHandler;
     private final ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap;
     private final ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap;
 
 
@@ -108,6 +110,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
         this.sessionId = UUID.randomUUID();
         this.sessionId = UUID.randomUUID();
         this.context = context;
         this.context = context;
         this.transportService = context.getTransportService();
         this.transportService = context.getTransportService();
+        this.scheduler = context.getScheduler();
         this.sslHandler = sslHandler;
         this.sslHandler = sslHandler;
         this.mqttQoSMap = new ConcurrentHashMap<>();
         this.mqttQoSMap = new ConcurrentHashMap<>();
         this.deviceSessionCtx = new DeviceSessionCtx(sessionId, mqttQoSMap, context);
         this.deviceSessionCtx = new DeviceSessionCtx(sessionId, mqttQoSMap, context);
@@ -333,7 +336,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
                 } else {
                 } else {
                     deviceSessionCtx.getContext().getProtoMqttAdaptor().convertToPublish(deviceSessionCtx, provisionResponseMsg).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
                     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) {
             } catch (Exception e) {
                 log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e);
                 log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e);
             }
             }

@@ -23,6 +23,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
+import org.thingsboard.server.queue.scheduler.SchedulerComponent;
 
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import javax.annotation.PreDestroy;
@@ -44,6 +45,8 @@ public abstract class TransportContext {
     private TransportService transportService;
     private TransportService transportService;
     @Autowired
     @Autowired
     private TbServiceInfoProvider serviceInfoProvider;
     private TbServiceInfoProvider serviceInfoProvider;
+    @Autowired
+    private SchedulerComponent scheduler;
 
 
     @Getter
     @Getter
     private ExecutorService executor;
     private ExecutorService executor;

@@ -17,16 +17,14 @@ package org.thingsboard.server.common.transport;
 
 
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.DeviceTransportType;
 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.GetOrCreateDeviceFromGatewayResponse;
 import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
 import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
 import org.thingsboard.server.common.transport.limits.TransportRateLimitType;
 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.ClaimDeviceMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
 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.GetEntityProfileRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileResponseMsg;
 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.PostAttributeMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.PostTelemetryMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.PostTelemetryMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceRequestMsg;
 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.ValidateDeviceTokenRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
 
 
-import java.util.concurrent.ScheduledExecutorService;
-
 /**
 /**
  * Created by ashvayka on 04.10.18.
  * Created by ashvayka on 04.10.18.
  */
  */
@@ -92,8 +88,6 @@ public interface TransportService {
 
 
     void process(SessionInfoProto sessionInfo, ClaimDeviceMsg msg, TransportServiceCallback<Void> callback);
     void process(SessionInfoProto sessionInfo, ClaimDeviceMsg msg, TransportServiceCallback<Void> callback);
 
 
-    ScheduledExecutorService getSchedulerExecutor();
-
     void registerAsyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener);
     void registerAsyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener);
 
 
     void registerSyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener, long timeout);
     void registerSyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener, long timeout);

@@ -23,7 +23,6 @@ import com.google.gson.JsonObject;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.ByteString;
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.common.data.DeviceProfile;
 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.GetOrCreateDeviceFromGatewayResponse;
 import org.thingsboard.server.common.transport.auth.TransportDeviceInfo;
 import org.thingsboard.server.common.transport.auth.TransportDeviceInfo;
 import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
 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.TransportRateLimitService;
 import org.thingsboard.server.common.transport.limits.TransportRateLimitType;
 import org.thingsboard.server.common.transport.limits.TransportRateLimitType;
 import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult;
 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.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
 import org.thingsboard.server.queue.provider.TbTransportQueueFactory;
 import org.thingsboard.server.queue.provider.TbTransportQueueFactory;
+import org.thingsboard.server.queue.scheduler.SchedulerComponent;
 import org.thingsboard.server.queue.util.TbTransportComponent;
 import org.thingsboard.server.queue.util.TbTransportComponent;
 
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PostConstruct;
@@ -94,7 +93,6 @@ import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -126,6 +124,7 @@ public class DefaultTransportService implements TransportService {
     private final TransportTenantProfileCache tenantProfileCache;
     private final TransportTenantProfileCache tenantProfileCache;
     private final TransportRateLimitService rateLimitService;
     private final TransportRateLimitService rateLimitService;
     private final DataDecodingEncodingService dataDecodingEncodingService;
     private final DataDecodingEncodingService dataDecodingEncodingService;
+    private final SchedulerComponent scheduler;
 
 
     protected TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> transportApiRequestTemplate;
     protected TbQueueRequestTemplate<TbProtoQueueMsg<TransportApiRequestMsg>, TbProtoQueueMsg<TransportApiResponseMsg>> transportApiRequestTemplate;
     protected TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> ruleEngineMsgProducer;
     protected TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> ruleEngineMsgProducer;
@@ -136,7 +135,6 @@ public class DefaultTransportService implements TransportService {
     protected MessagesStats tbCoreProducerStats;
     protected MessagesStats tbCoreProducerStats;
     protected MessagesStats transportApiStats;
     protected MessagesStats transportApiStats;
 
 
-    protected ScheduledExecutorService schedulerExecutor;
     protected ExecutorService transportCallbackExecutor;
     protected ExecutorService transportCallbackExecutor;
     private ExecutorService mainConsumerExecutor;
     private ExecutorService mainConsumerExecutor;
 
 
@@ -152,7 +150,7 @@ public class DefaultTransportService implements TransportService {
                                    StatsFactory statsFactory,
                                    StatsFactory statsFactory,
                                    TransportDeviceProfileCache deviceProfileCache,
                                    TransportDeviceProfileCache deviceProfileCache,
                                    TransportTenantProfileCache tenantProfileCache,
                                    TransportTenantProfileCache tenantProfileCache,
-                                   TransportRateLimitService rateLimitService, DataDecodingEncodingService dataDecodingEncodingService) {
+                                   TransportRateLimitService rateLimitService, DataDecodingEncodingService dataDecodingEncodingService, SchedulerComponent scheduler) {
         this.serviceInfoProvider = serviceInfoProvider;
         this.serviceInfoProvider = serviceInfoProvider;
         this.queueProvider = queueProvider;
         this.queueProvider = queueProvider;
         this.producerProvider = producerProvider;
         this.producerProvider = producerProvider;
@@ -162,6 +160,7 @@ public class DefaultTransportService implements TransportService {
         this.tenantProfileCache = tenantProfileCache;
         this.tenantProfileCache = tenantProfileCache;
         this.rateLimitService = rateLimitService;
         this.rateLimitService = rateLimitService;
         this.dataDecodingEncodingService = dataDecodingEncodingService;
         this.dataDecodingEncodingService = dataDecodingEncodingService;
+        this.scheduler = scheduler;
     }
     }
 
 
     @PostConstruct
     @PostConstruct
@@ -169,9 +168,8 @@ public class DefaultTransportService implements TransportService {
         this.ruleEngineProducerStats = statsFactory.createMessagesStats(StatsType.RULE_ENGINE.getName() + ".producer");
         this.ruleEngineProducerStats = statsFactory.createMessagesStats(StatsType.RULE_ENGINE.getName() + ".producer");
         this.tbCoreProducerStats = statsFactory.createMessagesStats(StatsType.CORE.getName() + ".producer");
         this.tbCoreProducerStats = statsFactory.createMessagesStats(StatsType.CORE.getName() + ".producer");
         this.transportApiStats = statsFactory.createMessagesStats(StatsType.TRANSPORT.getName() + ".producer");
         this.transportApiStats = statsFactory.createMessagesStats(StatsType.TRANSPORT.getName() + ".producer");
-        this.schedulerExecutor = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("transport-scheduler"));
         this.transportCallbackExecutor = Executors.newWorkStealingPool(20);
         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 = queueProvider.createTransportApiRequestTemplate();
         transportApiRequestTemplate.setMessagesStats(transportApiStats);
         transportApiRequestTemplate.setMessagesStats(transportApiStats);
         ruleEngineMsgProducer = producerProvider.getRuleEngineMsgProducer();
         ruleEngineMsgProducer = producerProvider.getRuleEngineMsgProducer();
@@ -217,9 +215,6 @@ public class DefaultTransportService implements TransportService {
         if (transportNotificationsConsumer != null) {
         if (transportNotificationsConsumer != null) {
             transportNotificationsConsumer.unsubscribe();
             transportNotificationsConsumer.unsubscribe();
         }
         }
-        if (schedulerExecutor != null) {
-            schedulerExecutor.shutdownNow();
-        }
         if (transportCallbackExecutor != null) {
         if (transportCallbackExecutor != null) {
             transportCallbackExecutor.shutdownNow();
             transportCallbackExecutor.shutdownNow();
         }
         }
@@ -232,11 +227,6 @@ public class DefaultTransportService implements TransportService {
     }
     }
 
 
     @Override
     @Override
-    public ScheduledExecutorService getSchedulerExecutor() {
-        return this.schedulerExecutor;
-    }
-
-    @Override
     public void registerAsyncSession(TransportProtos.SessionInfoProto sessionInfo, SessionMsgListener listener) {
     public void registerAsyncSession(TransportProtos.SessionInfoProto sessionInfo, SessionMsgListener listener) {
         sessions.putIfAbsent(toSessionId(sessionInfo), new SessionMetaData(sessionInfo, TransportProtos.SessionType.ASYNC, 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));
             sendToRuleEngine(tenantId, tbMsg, new TransportTbQueueCallback(callback));
             String requestId = sessionId + "-" + msg.getRequestId();
             String requestId = sessionId + "-" + msg.getRequestId();
             toServerRpcPendingMap.put(requestId, new RpcRequestMetadata(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);
         SessionMetaData currentSession = new SessionMetaData(sessionInfo, TransportProtos.SessionType.SYNC, listener);
         sessions.putIfAbsent(toSessionId(sessionInfo), currentSession);
         sessions.putIfAbsent(toSessionId(sessionInfo), currentSession);
 
 
-        ScheduledFuture executorFuture = schedulerExecutor.schedule(() -> {
+        ScheduledFuture executorFuture = scheduler.schedule(() -> {
             listener.onRemoteSessionCloseCommand(TransportProtos.SessionCloseNotificationProto.getDefaultInstance());
             listener.onRemoteSessionCloseCommand(TransportProtos.SessionCloseNotificationProto.getDefaultInstance());
             deregisterSession(sessionInfo);
             deregisterSession(sessionInfo);
         }, timeout, TimeUnit.MILLISECONDS);
         }, timeout, TimeUnit.MILLISECONDS);