From ebdee0bb4483b9c3b2332e8be1fe9835e332bc50 Mon Sep 17 00:00:00 2001 From: Luis Barrios Date: Sun, 15 Jan 2017 14:40:08 +0100 Subject: [PATCH] Make netty channel timeout configurable --- .../java/io/moquette/BrokerConstants.java | 9 ++++---- .../moquette/server/netty/NettyAcceptor.java | 22 ++++++++++--------- 2 files changed, 17 insertions(+), 14 deletions(-) diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 591c07bf..acbdd234 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -51,8 +51,9 @@ public class BrokerConstants { public static final String HOST = "0.0.0.0"; public static final String NEED_CLIENT_AUTH = "need_client_auth"; public static final String HAZELCAST_CONFIGURATION = "hazelcast.configuration"; - public static final String NETTY_SO_BACKLOG = "netty.so_backlog"; - public static final String NETTY_SO_REUSEADDR = "netty.so_reuseaddr"; - public static final String NETTY_TCP_NODELAY = "netty.tcp_nodelay"; - public static final String NETTY_SO_KEEPALIVE = "netty.so_keepalive"; + public static final String NETTY_SO_BACKLOG_PROPERTY_NAME = "netty.so_backlog"; + public static final String NETTY_SO_REUSEADDR_PROPERTY_NAME = "netty.so_reuseaddr"; + public static final String NETTY_TCP_NODELAY_PROPERTY_NAME = "netty.tcp_nodelay"; + public static final String NETTY_SO_KEEPALIVE_PROPERTY_NAME = "netty.so_keepalive"; + public static final String NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME = "netty.channel_timeout.seconds"; } diff --git a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java index e8d2843f..eeda08dc 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java @@ -17,7 +17,6 @@ package io.moquette.server.netty; import io.moquette.BrokerConstants; import static io.moquette.BrokerConstants.*; -import io.moquette.parser.commons.Constants; import io.moquette.parser.netty.MQTTDecoder; import io.moquette.parser.netty.MQTTEncoder; import io.moquette.spi.security.ISslContextCreator; @@ -99,16 +98,19 @@ public class NettyAcceptor implements ServerAcceptor { private boolean nettySoReuseaddr; private boolean nettyTcpNodelay; private boolean nettySoKeepalive; + private int nettyChannelTimeoutSeconds; @Override public void initialize(ProtocolProcessor processor, IConfig props, ISslContextCreator sslCtxCreator) throws IOException { LOG.info("Initializing Netty acceptor..."); - nettySoBacklog = Integer.parseInt(props.getProperty(BrokerConstants.NETTY_SO_BACKLOG, "128")); - nettySoReuseaddr = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_SO_REUSEADDR, "true")); - nettyTcpNodelay = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_TCP_NODELAY, "true")); - nettySoKeepalive = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_SO_KEEPALIVE, "true")); - + nettySoBacklog = Integer.parseInt(props.getProperty(BrokerConstants.NETTY_SO_BACKLOG_PROPERTY_NAME, "128")); + nettySoReuseaddr = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_SO_REUSEADDR_PROPERTY_NAME, "true")); + nettyTcpNodelay = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_TCP_NODELAY_PROPERTY_NAME, "true")); + nettySoKeepalive = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_SO_KEEPALIVE_PROPERTY_NAME, "true")); + nettyChannelTimeoutSeconds = Integer + .parseInt(props.getProperty(BrokerConstants.NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME, "10")); + m_bossGroup = new NioEventLoopGroup(); m_workerGroup = new NioEventLoopGroup(); final NettyMQTTHandler handler = new NettyMQTTHandler(processor); @@ -173,7 +175,7 @@ public class NettyAcceptor implements ServerAcceptor { initFactory(host, port, "TCP MQTT", new PipelineInitializer() { @Override void init(ChannelPipeline pipeline) { - pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); + pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, nettyChannelTimeoutSeconds)); pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); @@ -208,7 +210,7 @@ public class NettyAcceptor implements ServerAcceptor { pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST)); pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder()); - pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); + pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, nettyChannelTimeoutSeconds)); pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); pipeline.addLast("decoder", new MQTTDecoder()); @@ -241,7 +243,7 @@ public class NettyAcceptor implements ServerAcceptor { @Override void init(ChannelPipeline pipeline) throws Exception { pipeline.addLast("ssl", createSslHandler(sslContext, needsClientAuth)); - pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); + pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, nettyChannelTimeoutSeconds)); pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); @@ -278,7 +280,7 @@ public class NettyAcceptor implements ServerAcceptor { pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST)); pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder()); - pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); + pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, nettyChannelTimeoutSeconds)); pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); pipeline.addLast("decoder", new MQTTDecoder());