Fix #814 - Prevent IllegalBufferAccessException on write() and flush()
- Also fixed a incorrect port of SpdySessionHandler - Previously, it closed the connection too early when sending a GOAWAY frame - After this fix, SpdySessionHandlerTest now passes again without the previous fix
This commit is contained in:
parent
5a467b69bf
commit
310a87a51d
@ -25,7 +25,6 @@ import io.netty.channel.ChannelHandlerContext;
|
|||||||
import io.netty.channel.ChannelInboundMessageHandler;
|
import io.netty.channel.ChannelInboundMessageHandler;
|
||||||
import io.netty.channel.ChannelOutboundMessageHandler;
|
import io.netty.channel.ChannelOutboundMessageHandler;
|
||||||
|
|
||||||
import java.nio.channels.ClosedChannelException;
|
|
||||||
import java.util.concurrent.atomic.AtomicInteger;
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -440,8 +439,7 @@ public class SpdySessionHandler
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void close(ChannelHandlerContext ctx, ChannelFuture future) throws Exception {
|
public void close(ChannelHandlerContext ctx, ChannelFuture future) throws Exception {
|
||||||
sendGoAwayFrame(ctx);
|
sendGoAwayFrame(ctx, future);
|
||||||
super.close(ctx, future);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@ -878,20 +876,22 @@ public class SpdySessionHandler
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private void sendGoAwayFrame(ChannelHandlerContext ctx) {
|
private void sendGoAwayFrame(ChannelHandlerContext ctx, ChannelFuture future) {
|
||||||
// Avoid NotYetConnectedException
|
// Avoid NotYetConnectedException
|
||||||
if (!ctx.channel().isActive()) {
|
if (!ctx.channel().isActive()) {
|
||||||
|
ctx.close(future);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
sendGoAwayFrame(ctx, SpdySessionStatus.OK);
|
sendGoAwayFrame(ctx, SpdySessionStatus.OK);
|
||||||
ChannelFuture f = ctx.flush();
|
ChannelFuture f = ctx.flush();
|
||||||
if (spdySession.noActiveStreams()) {
|
if (spdySession.noActiveStreams()) {
|
||||||
f.addListener(new ClosingChannelFutureListener(ctx));
|
f.addListener(new ClosingChannelFutureListener(ctx, future));
|
||||||
} else {
|
} else {
|
||||||
closeSessionFuture = ctx.newFuture();
|
closeSessionFuture = ctx.newFuture();
|
||||||
closeSessionFuture.addListener(new ClosingChannelFutureListener(ctx));
|
closeSessionFuture.addListener(new ClosingChannelFutureListener(ctx, future));
|
||||||
}
|
}
|
||||||
|
// FIXME: Close the connection forcibly after timeout.
|
||||||
}
|
}
|
||||||
|
|
||||||
private synchronized void sendGoAwayFrame(
|
private synchronized void sendGoAwayFrame(
|
||||||
@ -905,16 +905,17 @@ public class SpdySessionHandler
|
|||||||
|
|
||||||
private static final class ClosingChannelFutureListener implements ChannelFutureListener {
|
private static final class ClosingChannelFutureListener implements ChannelFutureListener {
|
||||||
private final ChannelHandlerContext ctx;
|
private final ChannelHandlerContext ctx;
|
||||||
|
private final ChannelFuture future;
|
||||||
|
|
||||||
ClosingChannelFutureListener(ChannelHandlerContext ctx) {
|
ClosingChannelFutureListener(ChannelHandlerContext ctx, ChannelFuture future) {
|
||||||
this.ctx = ctx;
|
this.ctx = ctx;
|
||||||
|
this.future = future;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void operationComplete(ChannelFuture sentGoAwayFuture) throws Exception {
|
public void operationComplete(ChannelFuture sentGoAwayFuture) throws Exception {
|
||||||
if (!(sentGoAwayFuture.cause() instanceof ClosedChannelException)) {
|
System.err.println("ASDF");
|
||||||
ctx.close();
|
ctx.close(future);
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -233,25 +233,23 @@ public class SpdySessionHandlerTest {
|
|||||||
sessionHandler.writeInbound(remotePingFrame);
|
sessionHandler.writeInbound(remotePingFrame);
|
||||||
assertNull(sessionHandler.readOutbound());
|
assertNull(sessionHandler.readOutbound());
|
||||||
|
|
||||||
// Note that we cannot use writeInbound() here because sending closeMessage will close the channel
|
// Check if session handler sends a GOAWAY frame when closing
|
||||||
// immediately, and then we cannot test if SYN_STREAM and DATA frames are ignored after a GOAWAY frame is sent.
|
sessionHandler.writeInbound(closeMessage);
|
||||||
//// 1. Sending this message will make EchoHandler send a GOAWAY frame and close the session.
|
|
||||||
sessionHandler.inboundBuffer().add(closeMessage);
|
|
||||||
//// 2. Sending SYN_STREAM after sending closeMessage should fail with REFUSED_STREAM.
|
|
||||||
spdySynStreamFrame.setStreamId(localStreamID + 2);
|
|
||||||
sessionHandler.inboundBuffer().add(spdySynStreamFrame);
|
|
||||||
//// 3. Sending DATA after sending closeMessage should do nothing.
|
|
||||||
spdyDataFrame.setStreamId(localStreamID + 2);
|
|
||||||
sessionHandler.inboundBuffer().add(spdyDataFrame);
|
|
||||||
|
|
||||||
// At this point, we added three SPDY messages to the inbound buffer. Start testing.
|
|
||||||
sessionHandler.pipeline().fireInboundBufferUpdated();
|
|
||||||
//// 1. Check if session handler sends a GOAWAY frame when closing
|
|
||||||
assertGoAway(sessionHandler.readOutbound(), localStreamID);
|
assertGoAway(sessionHandler.readOutbound(), localStreamID);
|
||||||
//// 2. Check if session handler returns REFUSED_STREAM if it receives SYN_STREAM frames
|
assertNull(sessionHandler.readOutbound());
|
||||||
//// after sending a GOAWAY frame
|
localStreamID += 2;
|
||||||
assertRstStream(sessionHandler.readOutbound(), localStreamID + 2, SpdyStreamStatus.REFUSED_STREAM);
|
|
||||||
//// 3. Check if session handler ignores Data frames after sending a GOAWAY frame
|
// Check if session handler returns REFUSED_STREAM if it receives
|
||||||
|
// SYN_STREAM frames after sending a GOAWAY frame
|
||||||
|
spdySynStreamFrame.setStreamId(localStreamID);
|
||||||
|
sessionHandler.writeInbound(spdySynStreamFrame);
|
||||||
|
assertRstStream(sessionHandler.readOutbound(), localStreamID, SpdyStreamStatus.REFUSED_STREAM);
|
||||||
|
assertNull(sessionHandler.readOutbound());
|
||||||
|
|
||||||
|
// Check if session handler ignores Data frames after sending
|
||||||
|
// a GOAWAY frame
|
||||||
|
spdyDataFrame.setStreamId(localStreamID);
|
||||||
|
sessionHandler.writeInbound(spdyDataFrame);
|
||||||
assertNull(sessionHandler.readOutbound());
|
assertNull(sessionHandler.readOutbound());
|
||||||
|
|
||||||
sessionHandler.finish();
|
sessionHandler.finish();
|
||||||
|
@ -23,6 +23,7 @@ import io.netty.logging.InternalLogger;
|
|||||||
import io.netty.logging.InternalLoggerFactory;
|
import io.netty.logging.InternalLoggerFactory;
|
||||||
|
|
||||||
import java.net.SocketAddress;
|
import java.net.SocketAddress;
|
||||||
|
import java.nio.channels.ClosedChannelException;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.HashMap;
|
import java.util.HashMap;
|
||||||
import java.util.IdentityHashMap;
|
import java.util.IdentityHashMap;
|
||||||
@ -1238,6 +1239,11 @@ public class DefaultChannelPipeline implements ChannelPipeline {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private void flush0(final DefaultChannelHandlerContext ctx, ChannelFuture future) {
|
private void flush0(final DefaultChannelHandlerContext ctx, ChannelFuture future) {
|
||||||
|
if (!channel.isRegistered() && !channel.isActive()) {
|
||||||
|
future.setFailure(new ClosedChannelException());
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
ctx.flushBridge();
|
ctx.flushBridge();
|
||||||
((ChannelOperationHandler) ctx.handler()).flush(ctx, future);
|
((ChannelOperationHandler) ctx.handler()).flush(ctx, future);
|
||||||
@ -1320,17 +1326,16 @@ public class DefaultChannelPipeline implements ChannelPipeline {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private void write0(DefaultChannelHandlerContext ctx, Object message, ChannelFuture future, boolean msgBuf) {
|
private void write0(DefaultChannelHandlerContext ctx, Object message, ChannelFuture future, boolean msgBuf) {
|
||||||
|
if (!channel.isRegistered() && !channel.isActive()) {
|
||||||
|
future.setFailure(new ClosedChannelException());
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
if (msgBuf) {
|
if (msgBuf) {
|
||||||
MessageBuf<Object> outMsgBuf = ctx.outboundMessageBuffer();
|
ctx.outboundMessageBuffer().add(message);
|
||||||
if (!outMsgBuf.isFreed()) {
|
|
||||||
outMsgBuf.add(message);
|
|
||||||
}
|
|
||||||
} else {
|
} else {
|
||||||
ByteBuf outByteBuf = ctx.outboundByteBuffer();
|
|
||||||
if (!outByteBuf.isFreed()) {
|
|
||||||
ByteBuf buf = (ByteBuf) message;
|
ByteBuf buf = (ByteBuf) message;
|
||||||
outByteBuf.writeBytes(buf, buf.readerIndex(), buf.readableBytes());
|
ctx.outboundByteBuffer().writeBytes(buf, buf.readerIndex(), buf.readableBytes());
|
||||||
}
|
|
||||||
}
|
}
|
||||||
flush0(ctx, future);
|
flush0(ctx, future);
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user