From ecfe3e09366c1dea5c6e1b207941e44a5b066122 Mon Sep 17 00:00:00 2001
From: "selva.andre" <selva.andre@gmail.com>
Date: Thu, 19 Sep 2013 20:09:46 +0200
Subject: [PATCH] Added default idle connection timeout in a ChannelPipeline

---
 .../netty/MoquetteIdleTimoutHandler.java      | 19 +++++++++++++++++++
 .../moquette/server/netty/NettyAcceptor.java  |  6 +++++-
 .../moquette/server/netty/NettyChannel.java   | 19 +------------------
 3 files changed, 25 insertions(+), 19 deletions(-)
 create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java

diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java
new file mode 100644
index 00000000..a02de9d4
--- /dev/null
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java
@@ -0,0 +1,19 @@
+package org.dna.mqtt.moquette.server.netty;
+
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.timeout.IdleState;
+
+class MoquetteIdleTimoutHandler extends ChannelDuplexHandler {
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof IdleState) {
+            IdleState e = (IdleState) evt;
+            if (e == IdleState.ALL_IDLE) {
+                ctx.close();
+            } /*else if (e.getState() == IdleState.WRITER_IDLE) {
+                    ctx.writeAndFlush(new PingMessage());
+                }*/
+        }
+    }
+}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java
index b10e576d..dab9fc73 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java
@@ -12,6 +12,8 @@ import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.logging.LogLevel;
 import io.netty.handler.logging.LoggingHandler;
 import java.io.IOException;
+
+import io.netty.handler.timeout.IdleStateHandler;
 import org.dna.mqtt.commons.Constants;
 import org.dna.mqtt.moquette.messaging.spi.IMessaging;
 import org.dna.mqtt.moquette.parser.netty.MQTTDecoder;
@@ -46,7 +48,9 @@ public class NettyAcceptor implements ServerAcceptor {
                  @Override
                  public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline pipeline = ch.pipeline();
-                    pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
+                    pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
+                    pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
+                    //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
                     pipeline.addLast("decoder", new MQTTDecoder());
                     pipeline.addLast("encoder", new MQTTEncoder());
                     pipeline.addLast("handler", handler);
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java
index e83aa609..b69d3d8c 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java
@@ -1,8 +1,6 @@
 package org.dna.mqtt.moquette.server.netty;
 
-import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.timeout.IdleState;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
@@ -17,21 +15,6 @@ import org.dna.mqtt.moquette.server.ServerChannel;
  */
 public class NettyChannel implements ServerChannel {
     
-    private static class IdleTimoutHandler extends ChannelDuplexHandler {
-        @Override
-        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
-            if (evt instanceof IdleState) {
-                IdleState e = (IdleState) evt;
-                if (e == IdleState.ALL_IDLE) {
-                    ctx.close();
-                } /*else if (e.getState() == IdleState.WRITER_IDLE) {
-                    ctx.writeAndFlush(new PingMessage());
-                }*/
-            }
-        }
-    }
-    
-    
     private ChannelHandlerContext m_channel;
     
     private Map<Object, AttributeKey<Object>> m_attributesKeys = new HashMap<Object, AttributeKey<Object>>();
@@ -72,7 +55,7 @@ public class NettyChannel implements ServerChannel {
             m_channel.pipeline().remove("idleEventHandler");
         }
         m_channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
-        m_channel.pipeline().addAfter("idleStateHandler", "idleEventHandler", new IdleTimoutHandler());
+        m_channel.pipeline().addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
     }
 
     public void close(boolean immediately) {