diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectDecoder.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectDecoder.java index 8f92fe03e2..3db9f843c1 100644 --- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectDecoder.java +++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectDecoder.java @@ -19,6 +19,7 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPipeline; +import io.netty.channel.socket.ChannelInputShutdownEvent; import io.netty.handler.codec.ByteToMessageDecoder; import io.netty.handler.codec.DecoderResult; import io.netty.handler.codec.TooLongFrameException; @@ -239,6 +240,12 @@ public abstract class HttpObjectDecoder extends ByteToMessageDecoder { out.add(message); return; default: + /** + * RFC 7230, 3.3.3 states that if a + * request does not have either a transfer-encoding or a content-length header then the message body + * length is 0. However for a response the body length is the number of octets received prior to the + * server closing the connection. So we treat this as variable length chunked encoding. + */ long contentLength = contentLength(); if (contentLength == 0 || contentLength == -1 && isDecodingRequest()) { out.add(message); @@ -417,6 +424,17 @@ public abstract class HttpObjectDecoder extends ByteToMessageDecoder { } } + @Override + public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { + if (evt instanceof ChannelInputShutdownEvent) { + // The decodeLast method is invoked when a channelInactive event is encountered. + // This method is responsible for ending requests in some situations and must be called + // when the input has been shutdown. + super.channelInactive(ctx); + } + super.userEventTriggered(ctx, evt); + } + protected boolean isContentAlwaysEmpty(HttpMessage msg) { if (msg instanceof HttpResponse) { HttpResponse res = (HttpResponse) msg; diff --git a/codec-http/src/test/java/io/netty/handler/codec/http/HttpClientCodecTest.java b/codec-http/src/test/java/io/netty/handler/codec/http/HttpClientCodecTest.java index 2fa49f443b..3ba6e5f1d7 100644 --- a/codec-http/src/test/java/io/netty/handler/codec/http/HttpClientCodecTest.java +++ b/codec-http/src/test/java/io/netty/handler/codec/http/HttpClientCodecTest.java @@ -15,16 +15,39 @@ */ package io.netty.handler.codec.http; +import io.netty.bootstrap.Bootstrap; +import io.netty.bootstrap.ServerBootstrap; import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.ChannelOption; +import io.netty.channel.SimpleChannelInboundHandler; import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.SocketChannel; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.CodecException; import io.netty.handler.codec.PrematureChannelClosureException; import io.netty.util.CharsetUtil; +import io.netty.util.NetUtil; import org.junit.Test; -import static io.netty.util.ReferenceCountUtil.*; -import static org.hamcrest.CoreMatchers.*; -import static org.junit.Assert.*; +import java.net.InetSocketAddress; +import java.util.concurrent.CountDownLatch; + +import static io.netty.util.ReferenceCountUtil.release; +import static io.netty.util.ReferenceCountUtil.releaseLater; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.hamcrest.CoreMatchers.instanceOf; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; public class HttpClientCodecTest { @@ -124,4 +147,88 @@ public class HttpClientCodecTest { assertTrue(e instanceof PrematureChannelClosureException); } } + + @Test + public void testServerCloseSocketInputProvidesData() throws InterruptedException { + ServerBootstrap sb = new ServerBootstrap(); + Bootstrap cb = new Bootstrap(); + final CountDownLatch serverChannelLatch = new CountDownLatch(1); + final CountDownLatch responseRecievedLatch = new CountDownLatch(1); + try { + sb.group(new NioEventLoopGroup(2)); + sb.channel(NioServerSocketChannel.class); + sb.childHandler(new ChannelInitializer() { + @Override + protected void initChannel(Channel ch) throws Exception { + // Don't use the HttpServerCodec, because we don't want to have content-length or anything added. + ch.pipeline().addLast(new HttpRequestDecoder(4096, 8192, 8192, true)); + ch.pipeline().addLast(new HttpObjectAggregator(4096)); + ch.pipeline().addLast(new SimpleChannelInboundHandler() { + @Override + protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) { + // This is just a simple demo...don't block in IO + assertTrue(ctx.channel() instanceof SocketChannel); + final SocketChannel sChannel = (SocketChannel) ctx.channel(); + /** + * The point of this test is to not add any content-length or content-encoding headers + * and the client should still handle this. + * See RFC 7230, 3.3.3. + */ + sChannel.writeAndFlush(Unpooled.wrappedBuffer(("HTTP/1.0 200 OK\r\n" + + "Date: Fri, 31 Dec 1999 23:59:59 GMT\r\n" + + "Content-Type: text/html\r\n\r\n").getBytes(CharsetUtil.ISO_8859_1))) + .addListener(new ChannelFutureListener() { + @Override + public void operationComplete(ChannelFuture future) throws Exception { + assertTrue(future.isSuccess()); + sChannel.writeAndFlush(Unpooled.wrappedBuffer( + "hello half closed!\r\n" + .getBytes(CharsetUtil.ISO_8859_1))) + .addListener(new ChannelFutureListener() { + @Override + public void operationComplete(ChannelFuture future) throws Exception { + assertTrue(future.isSuccess()); + sChannel.shutdownOutput(); + } + }); + } + }); + } + }); + serverChannelLatch.countDown(); + } + }); + + cb.group(new NioEventLoopGroup(1)); + cb.channel(NioSocketChannel.class); + cb.option(ChannelOption.ALLOW_HALF_CLOSURE, true); + cb.handler(new ChannelInitializer() { + @Override + protected void initChannel(Channel ch) throws Exception { + ch.pipeline().addLast(new HttpClientCodec(4096, 8192, 8192, true, true)); + ch.pipeline().addLast(new HttpObjectAggregator(4096)); + ch.pipeline().addLast(new SimpleChannelInboundHandler() { + @Override + protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) { + responseRecievedLatch.countDown(); + } + }); + } + }); + + Channel serverChannel = sb.bind(new InetSocketAddress(0)).sync().channel(); + int port = ((InetSocketAddress) serverChannel.localAddress()).getPort(); + + ChannelFuture ccf = cb.connect(new InetSocketAddress(NetUtil.LOCALHOST, port)); + assertTrue(ccf.awaitUninterruptibly().isSuccess()); + Channel clientChannel = ccf.channel(); + assertTrue(serverChannelLatch.await(5, SECONDS)); + clientChannel.writeAndFlush(new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/")); + assertTrue(responseRecievedLatch.await(5, SECONDS)); + } finally { + sb.group().shutdownGracefully(); + sb.childGroup().shutdownGracefully(); + cb.group().shutdownGracefully(); + } + } }