diff --git a/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java b/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java index a802de09..92db1c59 100644 --- a/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java @@ -13,33 +13,31 @@ * * You may elect to redistribute this code under either of these licenses. */ + package io.moquette.server.netty; import io.netty.channel.ChannelDuplexHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.util.concurrent.EventExecutor; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; /** - * Auto-flush data on channel after a read timeout. - * It's inspired by IdleStateHandler but it's specialized version, just flushing data after - * no read is done on the channel after a period. It's used to avoid aggressively flushing from - * the ProtocolProcessor. + * Auto-flush data on channel after a read timeout. It's inspired by IdleStateHandler but it's + * specialized version, just flushing data after no read is done on the channel after a period. It's + * used to avoid aggressively flushing from the ProtocolProcessor. */ public class AutoFlushHandler extends ChannelDuplexHandler { - private static final Logger LOG = LoggerFactory.getLogger(AutoFlushHandler.class); + private static final Logger LOG = LoggerFactory.getLogger(AutoFlushHandler.class); private static final long MIN_TIMEOUT_NANOS = TimeUnit.MILLISECONDS.toNanos(1); private final long writerIdleTimeNanos; volatile ScheduledFuture writerIdleTimeout; volatile long lastWriteTime; -// private boolean firstWriterIdleEvent = true; + // private boolean firstWriterIdleEvent = true; private volatile int state; // 0 - none, 1 - initialized, 2 - destroyed @@ -57,7 +55,7 @@ public class AutoFlushHandler extends ChannelDuplexHandler { // not be invoked. We have to initialize here instead. initialize(ctx); } else { - // channelActive() event has not been fired yet. this.channelActive() will be invoked + // channelActive() event has not been fired yet. this.channelActive() will be invoked // and initialization will occur there. } } @@ -79,7 +77,7 @@ public class AutoFlushHandler extends ChannelDuplexHandler { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { // This method will be invoked only if this handler was added - // before channelActive() event is fired. If a user adds this handler + // before channelActive() event is fired. If a user adds this handler // after the channelActive() event, initialize() will be called by beforeAdd(). initialize(ctx); super.channelActive(ctx); @@ -91,30 +89,30 @@ public class AutoFlushHandler extends ChannelDuplexHandler { super.channelInactive(ctx); } -// @Override -// public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { -// if (writerrIdleTimeNanos > 0) { -// reading = true; -// firstReaderIdleEvent = true; -// } -// ctx.fireChannelRead(msg); -// } -// -// @Override -// public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { -// if (writerrIdleTimeNanos > 0) { -// lastReadTime = System.nanoTime(); -// reading = false; -// } -// ctx.fireChannelReadComplete(); -// } + // @Override + // public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + // if (writerrIdleTimeNanos > 0) { + // reading = true; + // firstReaderIdleEvent = true; + // } + // ctx.fireChannelRead(msg); + // } + // + // @Override + // public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { + // if (writerrIdleTimeNanos > 0) { + // lastReadTime = System.nanoTime(); + // reading = false; + // } + // ctx.fireChannelReadComplete(); + // } private void initialize(ChannelHandlerContext ctx) { // Avoid the case where destroy() is called before scheduling timeouts. // See: https://github.com/netty/netty/issues/143 - if (LOG.isDebugEnabled()) { - LOG.debug("Initializing autoflush handler. MqttClientId = {}.", NettyUtils.clientID(ctx.channel())); - } + if (LOG.isDebugEnabled()) { + LOG.debug("Initializing autoflush handler. MqttClientId = {}.", NettyUtils.clientID(ctx.channel())); + } switch (state) { case 1: case 2: @@ -126,9 +124,7 @@ public class AutoFlushHandler extends ChannelDuplexHandler { EventExecutor loop = ctx.executor(); lastWriteTime = System.nanoTime(); - writerIdleTimeout = loop.schedule( - new WriterIdleTimeoutTask(ctx), - writerIdleTimeNanos, TimeUnit.NANOSECONDS); + writerIdleTimeout = loop.schedule(new WriterIdleTimeoutTask(ctx), writerIdleTimeNanos, TimeUnit.NANOSECONDS); } private void destroy() { @@ -143,14 +139,16 @@ public class AutoFlushHandler extends ChannelDuplexHandler { /** * Is called when the write timeout expire. * - * @param ctx the channel context. - * @throws Exception in case of any IO error. + * @param ctx + * the channel context. + * @throws Exception + * in case of any IO error. */ - protected void channelIdle(ChannelHandlerContext ctx/*, IdleStateEvent evt*/) throws Exception { -// ctx.fireUserEventTriggered(evt); - if (LOG.isDebugEnabled()) { - LOG.debug("Flushing idle Netty channel. MqttClientId = {}.", NettyUtils.clientID(ctx.channel())); - } + protected void channelIdle(ChannelHandlerContext ctx/* , IdleStateEvent evt */) throws Exception { + // ctx.fireUserEventTriggered(evt); + if (LOG.isDebugEnabled()) { + LOG.debug("Flushing idle Netty channel. MqttClientId = {}.", NettyUtils.clientID(ctx.channel())); + } ctx.channel().flush(); } @@ -168,22 +166,19 @@ public class AutoFlushHandler extends ChannelDuplexHandler { return; } -// long lastWriteTime = IdleStateHandler.this.lastWriteTime; -// long lastWriteTime = lastWriteTime; + // long lastWriteTime = IdleStateHandler.this.lastWriteTime; + // long lastWriteTime = lastWriteTime; long nextDelay = writerIdleTimeNanos - (System.nanoTime() - lastWriteTime); if (nextDelay <= 0) { // Writer is idle - set a new timeout and notify the callback. - writerIdleTimeout = ctx.executor().schedule( - this, writerIdleTimeNanos, TimeUnit.NANOSECONDS); + writerIdleTimeout = ctx.executor().schedule(this, writerIdleTimeNanos, TimeUnit.NANOSECONDS); try { - /*IdleStateEvent event; - if (firstWriterIdleEvent) { - firstWriterIdleEvent = false; - event = IdleStateEvent.FIRST_WRITER_IDLE_STATE_EVENT; - } else { - event = IdleStateEvent.WRITER_IDLE_STATE_EVENT; - }*/ - channelIdle(ctx/*, event*/); + /* + * IdleStateEvent event; if (firstWriterIdleEvent) { firstWriterIdleEvent = + * false; event = IdleStateEvent.FIRST_WRITER_IDLE_STATE_EVENT; } else { event = + * IdleStateEvent.WRITER_IDLE_STATE_EVENT; } + */ + channelIdle(ctx/* , event */); } catch (Throwable t) { ctx.fireExceptionCaught(t); }