diff --git a/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java b/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java index 122f9b70..63d67151 100644 --- a/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java @@ -33,7 +33,7 @@ class MoquetteIdleTimeoutHandler extends ChannelDuplexHandler { public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { IdleState e = ((IdleStateEvent) evt).state(); - if (e == IdleState.ALL_IDLE) { + if (e == IdleState.READER_IDLE) { LOG.info("Firing channel inactive event. MqttClientId = {}.", NettyUtils.clientID(ctx.channel())); // fire a channelInactive to trigger publish of Will ctx.fireChannelInactive(); 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 67993d6d..e6b2ceed 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java @@ -184,7 +184,7 @@ public class NettyAcceptor implements ServerAcceptor { @Override void init(ChannelPipeline pipeline) { - pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, nettyChannelTimeoutSeconds)); + pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0)); pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); @@ -224,7 +224,7 @@ public class NettyAcceptor implements ServerAcceptor { new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST)); pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder()); - pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, nettyChannelTimeoutSeconds)); + pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0)); pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); pipeline.addLast("decoder", new MqttDecoder()); @@ -261,7 +261,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, nettyChannelTimeoutSeconds)); + pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0)); pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); @@ -304,7 +304,7 @@ public class NettyAcceptor implements ServerAcceptor { new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST)); pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder()); - pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, nettyChannelTimeoutSeconds)); + pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0)); pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); pipeline.addLast("decoder", new MqttDecoder()); diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index a9a8b246..d0239bd4 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -446,7 +446,7 @@ public class ProtocolProcessor { if (pipeline.names().contains("idleStateHandler")) { pipeline.remove("idleStateHandler"); } - pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime)); + pipeline.addFirst("idleStateHandler", new IdleStateHandler(idleTime, 0, 0)); } /** diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java index bb9155e3..7981c810 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java @@ -28,7 +28,7 @@ public class ProtocolDecodingServer { public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { IdleState e = ((IdleStateEvent) evt).state(); - if (e == IdleState.ALL_IDLE) { + if (e == IdleState.READER_IDLE) { //fire a channelInactive to trigger publish of Will ctx.fireChannelInactive(); ctx.close(); @@ -86,7 +86,7 @@ public class ProtocolDecodingServer { public void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); try { - pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, 2)); + pipeline.addFirst("idleStateHandler", new IdleStateHandler(2, 0, 0)); pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimeoutHandler()); pipeline.addLast("decoder", new MqttDecoder()); pipeline.addLast("encoder", MqttEncoder.INSTANCE);