diff --git a/ChangeLog.txt b/ChangeLog.txt
index 3b9acaa0..a4b5f9b5 100644
--- a/ChangeLog.txt
+++ b/ChangeLog.txt
@@ -1,4 +1,5 @@
Version 0.10:
+ Netty epoll system added
Separated in memory and MapDB storage implementations
Cleanly separated inflight part (inbound and outbound)
diff --git a/broker/pom.xml b/broker/pom.xml
index a0a256a6..3a67af93 100644
--- a/broker/pom.xml
+++ b/broker/pom.xml
@@ -73,6 +73,16 @@
${netty.version}
+
+ io.netty
+ netty-transport-native-epoll
+ ${netty.version}
+ linux-x86_64
+
+
+
commons-codec
commons-codec
diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java
index 47edb890..220dd4a8 100644
--- a/broker/src/main/java/io/moquette/BrokerConstants.java
+++ b/broker/src/main/java/io/moquette/BrokerConstants.java
@@ -56,6 +56,7 @@ public final class BrokerConstants {
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";
+ public static final String NETTY_EPOLL_PROPERTY_NAME = "netty.epoll";
public static final String STORAGE_CLASS_NAME = "storage_class";
private BrokerConstants() {
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 c438c1b7..71bf48dc 100644
--- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java
+++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java
@@ -25,7 +25,10 @@ import io.moquette.spi.security.ISslContextCreator;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.MessageToMessageDecoder;
@@ -99,6 +102,8 @@ public class NettyAcceptor implements ServerAcceptor {
private boolean nettySoKeepalive;
private int nettyChannelTimeoutSeconds;
+ private Class extends ServerSocketChannel> channelClass;
+
@Override
public void initialize(ProtocolProcessor processor, IConfig props, ISslContextCreator sslCtxCreator)
throws IOException {
@@ -114,8 +119,19 @@ public class NettyAcceptor implements ServerAcceptor {
nettyChannelTimeoutSeconds = Integer
.parseInt(props.getProperty(BrokerConstants.NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME, "10"));
- m_bossGroup = new NioEventLoopGroup();
- m_workerGroup = new NioEventLoopGroup();
+ boolean epoll = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_EPOLL_PROPERTY_NAME, "false"));
+ if (epoll) {
+ LOG.info("Netty is using epoll.");
+ m_bossGroup = new EpollEventLoopGroup();
+ m_workerGroup = new EpollEventLoopGroup();
+ channelClass = EpollServerSocketChannel.class;
+ } else {
+ LOG.info("Netty is using nio.");
+ m_bossGroup = new NioEventLoopGroup();
+ m_workerGroup = new NioEventLoopGroup();
+ channelClass = NioServerSocketChannel.class;
+ }
+
final NettyMQTTHandler handler = new NettyMQTTHandler(processor);
initializePlainTCPTransport(handler, props);
@@ -136,7 +152,7 @@ public class NettyAcceptor implements ServerAcceptor {
private void initFactory(String host, int port, String protocol, final PipelineInitializer pipeliner) {
LOG.info("Initializing server. Protocol={}", protocol);
ServerBootstrap b = new ServerBootstrap();
- b.group(m_bossGroup, m_workerGroup).channel(NioServerSocketChannel.class)
+ b.group(m_bossGroup, m_workerGroup).channel(channelClass)
.childHandler(new ChannelInitializer() {
@Override
diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf
index b1eaea90..c3fd7b9c 100644
--- a/distribution/src/main/resources/moquette.conf
+++ b/distribution/src/main/resources/moquette.conf
@@ -99,4 +99,14 @@ password_file config/password_file.conf
# interval between flushes of MapDB storage to disk. It's in
# seconds, if not specified defaults is 30 s.
#*********************************************************************
-# autosave_interval 120
\ No newline at end of file
+# autosave_interval 120
+
+#*********************************************************************
+# Netty Configuration
+#*********************************************************************
+#
+# Linux systems can use epoll instead of nio. To get a performance
+# gain and reduced GC.
+# http://netty.io/wiki/native-transports.html for more information
+#
+# netty.epoll=true