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

using reader idle to trigger the will message.

Browse Source 
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.
This commit is contained in:
fjcyue
2017-03-12 01:03:13 +08:00
parent 46d27e3bea
commit 96fb6301a8
4 changed files with 8 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java
+1 -1
View File
@@ -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();
broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java
+4 -4
View File
@@ -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());