From 96fb6301a8682e262661c390983c3952c08cb5ca Mon Sep 17 00:00:00 2001 From: fjcyue Date: Sat, 11 Mar 2017 10:13:29 +0800 Subject: [PATCH] using reader idle to trigger the will message. When client power off directly, the socket is broken and server may still push messages, which just occurred TCP re-transmission at that time, instead of raising an error immediately. This delayed the will message. --- .../moquette/server/netty/MoquetteIdleTimeoutHandler.java | 2 +- .../main/java/io/moquette/server/netty/NettyAcceptor.java | 8 ++++---- .../main/java/io/moquette/spi/impl/ProtocolProcessor.java | 2 +- .../parser/netty/performance/ProtocolDecodingServer.java | 4 ++-- 4 files changed, 8 insertions(+), 8 deletions(-) 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 3e93a9e7..d17fb7cd 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 9e307019..ee8b5c15 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 f2877ff2..b4ed9b30 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -519,7 +519,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);