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

broker/AutoFlushHandler: code formatter used

Browse Source
This commit is contained in:
Andreas Mager
2017-02-01 17:13:37 +01:00
parent 4483e554e9
commit 2adee4054c
1 changed files with 47 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/netty/AutoFlushHandler.java
+47 -52
View File
@@ -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);
}