anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added default idle connection timeout in a ChannelPipeline

Browse Source
This commit is contained in:
selva.andre
2013-09-19 20:09:46 +02:00
parent 576704562e
commit ecfe3e0936
3 changed files with 25 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/server/netty/MoquetteIdleTimoutHandler.java
+19
View File
@@ -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());
}*/
}
}
}
broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java
+5 -1
View File
@@ -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);
broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java
+1 -18
View File
@@ -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) {