Make sure the ChannelFuture's of the MessageEvent's are notified on channelClosed(..) event and on removal of the handler from the ChannelPipeline. See #308
This commit is contained in:
parent
9ebff3abf2
commit
b5706d54f7
@ -15,6 +15,8 @@
|
||||
*/
|
||||
package org.jboss.netty.handler.queue;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.channels.ClosedChannelException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Queue;
|
||||
@ -29,6 +31,7 @@ import org.jboss.netty.channel.ChannelFutureListener;
|
||||
import org.jboss.netty.channel.ChannelHandlerContext;
|
||||
import org.jboss.netty.channel.ChannelStateEvent;
|
||||
import org.jboss.netty.channel.Channels;
|
||||
import org.jboss.netty.channel.LifeCycleAwareChannelHandler;
|
||||
import org.jboss.netty.channel.MessageEvent;
|
||||
import org.jboss.netty.channel.SimpleChannelHandler;
|
||||
import org.jboss.netty.channel.socket.nio.NioSocketChannelConfig;
|
||||
@ -156,7 +159,7 @@ import org.jboss.netty.util.internal.QueueFactory;
|
||||
* {@link HashedWheelTimer} every second.
|
||||
* @apiviz.landmark
|
||||
*/
|
||||
public class BufferedWriteHandler extends SimpleChannelHandler {
|
||||
public class BufferedWriteHandler extends SimpleChannelHandler implements LifeCycleAwareChannelHandler {
|
||||
|
||||
private final Queue<MessageEvent> queue;
|
||||
private final boolean consolidateOnFlush;
|
||||
@ -350,4 +353,70 @@ public class BufferedWriteHandler extends SimpleChannelHandler {
|
||||
ctx.sendDownstream(e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Fail all buffered writes that are left. See <a href="https://github.com/netty/netty/issues/308>#308</a> for more details
|
||||
*/
|
||||
@Override
|
||||
public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
|
||||
Throwable cause = null;
|
||||
for (;;) {
|
||||
MessageEvent ev = queue.poll();
|
||||
|
||||
if (ev == null) {
|
||||
break;
|
||||
}
|
||||
|
||||
if (cause == null) {
|
||||
cause = new ClosedChannelException();
|
||||
}
|
||||
ev.getFuture().setFailure(cause);
|
||||
|
||||
}
|
||||
if (cause != null) {
|
||||
Channels.fireExceptionCaught(ctx.getChannel(), cause);
|
||||
}
|
||||
|
||||
super.channelClosed(ctx, e);
|
||||
}
|
||||
|
||||
public void beforeAdd(ChannelHandlerContext ctx) throws Exception {
|
||||
// Nothing to do
|
||||
|
||||
}
|
||||
|
||||
public void afterAdd(ChannelHandlerContext ctx) throws Exception {
|
||||
// Nothing to do
|
||||
|
||||
}
|
||||
|
||||
public void beforeRemove(ChannelHandlerContext ctx) throws Exception {
|
||||
// flush a last time before remove the handler
|
||||
flush(consolidateOnFlush);
|
||||
}
|
||||
|
||||
/**
|
||||
* Fail all buffered writes that are left. See <a href="https://github.com/netty/netty/issues/308>#308</a> for more details
|
||||
*/
|
||||
public void afterRemove(ChannelHandlerContext ctx) throws Exception {
|
||||
Throwable cause = null;
|
||||
for (;;) {
|
||||
MessageEvent ev = queue.poll();
|
||||
|
||||
if (ev == null) {
|
||||
break;
|
||||
}
|
||||
|
||||
if (cause == null) {
|
||||
cause = new IOException("Unable to flush message");
|
||||
}
|
||||
ev.getFuture().setFailure(cause);
|
||||
|
||||
}
|
||||
if (cause != null) {
|
||||
Channels.fireExceptionCaughtLater(ctx.getChannel(), cause);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user