Merge pull request #1662 from yefimov-andrey/mqtt-keep-alive-fix

Added MQTT keep alive parameter

@@ -469,6 +469,7 @@ transport:
       boss_group_thread_count: "${NETTY_BOSS_GROUP_THREADS:1}"
       boss_group_thread_count: "${NETTY_BOSS_GROUP_THREADS:1}"
       worker_group_thread_count: "${NETTY_WORKER_GROUP_THREADS:12}"
       worker_group_thread_count: "${NETTY_WORKER_GROUP_THREADS:12}"
       max_payload_size: "${NETTY_MAX_PAYLOAD_SIZE:65536}"
       max_payload_size: "${NETTY_MAX_PAYLOAD_SIZE:65536}"
+      so_keep_alive: "${NETTY_SO_KEEPALIVE:true}"
     # MQTT SSL configuration
     # MQTT SSL configuration
     ssl:
     ssl:
       # Enable/disable SSL support
       # Enable/disable SSL support

@@ -17,6 +17,7 @@ package org.thingsboard.server.transport.mqtt;
 
 
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelOption;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
@@ -50,6 +51,8 @@ public class MqttTransportService {
     private Integer bossGroupThreadCount;
     private Integer bossGroupThreadCount;
     @Value("${transport.mqtt.netty.worker_group_thread_count}")
     @Value("${transport.mqtt.netty.worker_group_thread_count}")
     private Integer workerGroupThreadCount;
     private Integer workerGroupThreadCount;
+    @Value("${transport.mqtt.netty.so_keep_alive}")
+    private boolean keepAlive;
 
 
     @Autowired
     @Autowired
     private MqttTransportContext context;
     private MqttTransportContext context;
@@ -69,7 +72,8 @@ public class MqttTransportService {
         ServerBootstrap b = new ServerBootstrap();
         ServerBootstrap b = new ServerBootstrap();
         b.group(bossGroup, workerGroup)
         b.group(bossGroup, workerGroup)
                 .channel(NioServerSocketChannel.class)
                 .channel(NioServerSocketChannel.class)
-                .childHandler(new MqttTransportServerInitializer(context));
+                .childHandler(new MqttTransportServerInitializer(context))
+                .childOption(ChannelOption.SO_KEEPALIVE, keepAlive);
 
 
         serverChannel = b.bind(host, port).sync().channel();
         serverChannel = b.bind(host, port).sync().channel();
         log.info("Mqtt transport started!");
         log.info("Mqtt transport started!");