Fire the IdleStateEvent and also the ReadTimeOutException / WriteTimeOutException from the Worker-Thread. See #641
This commit is contained in:
parent
8669732479
commit
2056882cfc
@ -373,6 +373,20 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
|
|||||||
return state;
|
return state;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void fireChannelIdle(
|
||||||
|
final ChannelHandlerContext ctx, final IdleState state, final long lastActivityTimeMillis) {
|
||||||
|
ctx.getPipeline().execute(new Runnable() {
|
||||||
|
|
||||||
|
public void run() {
|
||||||
|
try {
|
||||||
|
channelIdle(ctx, state, lastActivityTimeMillis);
|
||||||
|
} catch (Throwable t) {
|
||||||
|
fireExceptionCaught(ctx, t);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
protected void channelIdle(
|
protected void channelIdle(
|
||||||
ChannelHandlerContext ctx, IdleState state, long lastActivityTimeMillis) throws Exception {
|
ChannelHandlerContext ctx, IdleState state, long lastActivityTimeMillis) throws Exception {
|
||||||
ctx.sendUpstream(new DefaultIdleStateEvent(ctx.getChannel(), state, lastActivityTimeMillis));
|
ctx.sendUpstream(new DefaultIdleStateEvent(ctx.getChannel(), state, lastActivityTimeMillis));
|
||||||
@ -399,11 +413,7 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
|
|||||||
// Reader is idle - set a new timeout and notify the callback.
|
// Reader is idle - set a new timeout and notify the callback.
|
||||||
state.readerIdleTimeout =
|
state.readerIdleTimeout =
|
||||||
timer.newTimeout(this, readerIdleTimeMillis, TimeUnit.MILLISECONDS);
|
timer.newTimeout(this, readerIdleTimeMillis, TimeUnit.MILLISECONDS);
|
||||||
try {
|
fireChannelIdle(ctx, IdleState.READER_IDLE, lastReadTime);
|
||||||
channelIdle(ctx, IdleState.READER_IDLE, lastReadTime);
|
|
||||||
} catch (Throwable t) {
|
|
||||||
fireExceptionCaught(ctx, t);
|
|
||||||
}
|
|
||||||
} else {
|
} else {
|
||||||
// Read occurred before the timeout - set a new timeout with shorter delay.
|
// Read occurred before the timeout - set a new timeout with shorter delay.
|
||||||
state.readerIdleTimeout =
|
state.readerIdleTimeout =
|
||||||
@ -434,11 +444,7 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
|
|||||||
// Writer is idle - set a new timeout and notify the callback.
|
// Writer is idle - set a new timeout and notify the callback.
|
||||||
state.writerIdleTimeout =
|
state.writerIdleTimeout =
|
||||||
timer.newTimeout(this, writerIdleTimeMillis, TimeUnit.MILLISECONDS);
|
timer.newTimeout(this, writerIdleTimeMillis, TimeUnit.MILLISECONDS);
|
||||||
try {
|
fireChannelIdle(ctx, IdleState.WRITER_IDLE, lastWriteTime);
|
||||||
channelIdle(ctx, IdleState.WRITER_IDLE, lastWriteTime);
|
|
||||||
} catch (Throwable t) {
|
|
||||||
fireExceptionCaught(ctx, t);
|
|
||||||
}
|
|
||||||
} else {
|
} else {
|
||||||
// Write occurred before the timeout - set a new timeout with shorter delay.
|
// Write occurred before the timeout - set a new timeout with shorter delay.
|
||||||
state.writerIdleTimeout =
|
state.writerIdleTimeout =
|
||||||
@ -469,11 +475,7 @@ public class IdleStateHandler extends SimpleChannelUpstreamHandler
|
|||||||
// notify the callback.
|
// notify the callback.
|
||||||
state.allIdleTimeout =
|
state.allIdleTimeout =
|
||||||
timer.newTimeout(this, allIdleTimeMillis, TimeUnit.MILLISECONDS);
|
timer.newTimeout(this, allIdleTimeMillis, TimeUnit.MILLISECONDS);
|
||||||
try {
|
fireChannelIdle(ctx, IdleState.ALL_IDLE, lastIoTime);
|
||||||
channelIdle(ctx, IdleState.ALL_IDLE, lastIoTime);
|
|
||||||
} catch (Throwable t) {
|
|
||||||
fireExceptionCaught(ctx, t);
|
|
||||||
}
|
|
||||||
} else {
|
} else {
|
||||||
// Either read or write occurred before the timeout - set a new
|
// Either read or write occurred before the timeout - set a new
|
||||||
// timeout with shorter delay.
|
// timeout with shorter delay.
|
||||||
|
@ -232,6 +232,19 @@ public class ReadTimeoutHandler extends SimpleChannelUpstreamHandler
|
|||||||
return state;
|
return state;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void fireReadTimedOut(final ChannelHandlerContext ctx) throws Exception {
|
||||||
|
ctx.getPipeline().execute(new Runnable() {
|
||||||
|
|
||||||
|
public void run() {
|
||||||
|
try {
|
||||||
|
readTimedOut(ctx);
|
||||||
|
} catch (Throwable t) {
|
||||||
|
fireExceptionCaught(ctx, t);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
protected void readTimedOut(ChannelHandlerContext ctx) throws Exception {
|
protected void readTimedOut(ChannelHandlerContext ctx) throws Exception {
|
||||||
Channels.fireExceptionCaught(ctx, EXCEPTION);
|
Channels.fireExceptionCaught(ctx, EXCEPTION);
|
||||||
}
|
}
|
||||||
@ -260,13 +273,7 @@ public class ReadTimeoutHandler extends SimpleChannelUpstreamHandler
|
|||||||
// Read timed out - set a new timeout and notify the callback.
|
// Read timed out - set a new timeout and notify the callback.
|
||||||
state.timeout =
|
state.timeout =
|
||||||
timer.newTimeout(this, timeoutMillis, TimeUnit.MILLISECONDS);
|
timer.newTimeout(this, timeoutMillis, TimeUnit.MILLISECONDS);
|
||||||
try {
|
fireReadTimedOut(ctx);
|
||||||
// FIXME This should be called from an I/O thread.
|
|
||||||
// To be fixed in Netty 4.
|
|
||||||
readTimedOut(ctx);
|
|
||||||
} catch (Throwable t) {
|
|
||||||
fireExceptionCaught(ctx, t);
|
|
||||||
}
|
|
||||||
} else {
|
} else {
|
||||||
// Read occurred before the timeout - set a new timeout with shorter delay.
|
// Read occurred before the timeout - set a new timeout with shorter delay.
|
||||||
state.timeout =
|
state.timeout =
|
||||||
|
@ -159,8 +159,21 @@ public class WriteTimeoutHandler extends SimpleChannelDownstreamHandler
|
|||||||
super.writeRequested(ctx, e);
|
super.writeRequested(ctx, e);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void fireWriteTimeOut(final ChannelHandlerContext ctx) {
|
||||||
|
ctx.getPipeline().execute(new Runnable() {
|
||||||
|
|
||||||
|
public void run() {
|
||||||
|
try {
|
||||||
|
writeTimedOut(ctx);
|
||||||
|
} catch (Throwable t) {
|
||||||
|
fireExceptionCaught(ctx, t);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
protected void writeTimedOut(ChannelHandlerContext ctx) throws Exception {
|
protected void writeTimedOut(ChannelHandlerContext ctx) throws Exception {
|
||||||
Channels.fireExceptionCaughtLater(ctx, EXCEPTION);
|
Channels.fireExceptionCaught(ctx, EXCEPTION);
|
||||||
}
|
}
|
||||||
|
|
||||||
private final class WriteTimeoutTask implements TimerTask {
|
private final class WriteTimeoutTask implements TimerTask {
|
||||||
@ -185,11 +198,7 @@ public class WriteTimeoutHandler extends SimpleChannelDownstreamHandler
|
|||||||
// Mark the future as failure
|
// Mark the future as failure
|
||||||
if (future.setFailure(EXCEPTION)) {
|
if (future.setFailure(EXCEPTION)) {
|
||||||
// If succeeded to mark as failure, notify the pipeline, too.
|
// If succeeded to mark as failure, notify the pipeline, too.
|
||||||
try {
|
fireWriteTimeOut(ctx);
|
||||||
writeTimedOut(ctx);
|
|
||||||
} catch (Throwable t) {
|
|
||||||
fireExceptionCaught(ctx, t);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user