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

Merge pull request #280 from fjcyue/master

Browse Source 
using reader idle to trigger the will message.
This commit is contained in:
Andrea Selva
2017-03-22 22:20:22 +01:00
committed by GitHub
parent c659a54158 96fb6301a8
commit 451b5db941
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());
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+1 -1
View File
@@ -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));
}
/**
perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java
+2 -2
View File
@@ -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);