Updating HTTP/2 tests to not retain buffers for validation

Motivation:

To eliminate the tests as being a cause of leaks, removing the automatic
retaining of ByteBufs in Http2TestUtil.

Modifications:

Each test that relied on retaining buffers for validation has been
modified to copy the buffer into a list of Strings that are manually
validated after the message is received.

Result:

The HTTP/2 tests should (hopefully) no longer be reporting erroneous
leaks due to the testing code, itself.

Author: nmittler

codec-http2/src/test/java/io/netty/handler/codec/http2/DataCompressionHttp2Test.java

@@ -108,7 +108,7 @@ protected void initChannel(Channel ch) throws Exception {
                 ChannelPipeline p = ch.pipeline();
                 serverAdapter = new Http2TestUtil.FrameAdapter(serverConnection,
                                 new DelegatingDecompressorFrameListener(serverConnection, serverListener),
-                                serverLatch, false);
+                                serverLatch);
                 p.addLast("reader", serverAdapter);
                 p.addLast(Http2CodecUtil.ignoreSettingsHandler());
                 serverConnectedChannel = ch;
@@ -121,7 +121,7 @@ protected void initChannel(Channel ch) throws Exception {
             @Override
             protected void initChannel(Channel ch) throws Exception {
                 ChannelPipeline p = ch.pipeline();
-                clientAdapter = new Http2TestUtil.FrameAdapter(clientListener, clientLatch, false);
+                clientAdapter = new Http2TestUtil.FrameAdapter(clientListener, clientLatch);
                 p.addLast("reader", clientAdapter);
                 p.addLast(Http2CodecUtil.ignoreSettingsHandler());
             }

codec-http2/src/test/java/io/netty/handler/codec/http2/DelegatingHttp2HttpConnectionHandlerTest.java

@@ -28,6 +28,7 @@
 import static org.mockito.Matchers.anyInt;
 import static org.mockito.Matchers.anyShort;
 import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import io.netty.bootstrap.Bootstrap;
@@ -51,25 +52,25 @@
 import io.netty.util.concurrent.Future;
 
 import java.net.InetSocketAddress;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
-import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 /**
  * Testing the {@link DelegatingHttp2HttpConnectionHandler} for {@link FullHttpRequest} objects into HTTP/2 frames
  */
 public class DelegatingHttp2HttpConnectionHandlerTest {
     private static final int CONNECTION_SETUP_READ_COUNT = 2;
 
-    private List<ByteBuf> capturedData;
-
     @Mock
     private Http2FrameListener clientListener;
 
@@ -125,12 +126,6 @@ protected void initChannel(Channel ch) throws Exception {
 
     @After
     public void teardown() throws Exception {
-        if (capturedData != null) {
-            for (int i = 0; i < capturedData.size(); ++i) {
-                capturedData.get(i).release();
-            }
-            capturedData = null;
-        }
         serverChannel.close().sync();
         Future<?> serverGroup = sb.group().shutdownGracefully(0, 0, TimeUnit.MILLISECONDS);
         Future<?> serverChildGroup = sb.childGroup().shutdownGracefully(0, 0, TimeUnit.MILLISECONDS);
@@ -180,6 +175,15 @@ public void testRequestWithBody() throws Exception {
         requestLatch(new CountDownLatch(CONNECTION_SETUP_READ_COUNT + 2));
         final String text = "foooooogoooo";
         final ByteBuf data = Unpooled.copiedBuffer(text, UTF_8);
+        final List<String> receivedBuffers = new ArrayList<String>();
+        doAnswer(new Answer<Void>() {
+            @Override
+            public Void answer(InvocationOnMock in) throws Throwable {
+                receivedBuffers.add(((ByteBuf) in.getArguments()[2]).toString(UTF_8));
+                return null;
+            }
+        }).when(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3),
+                any(ByteBuf.class), eq(0), eq(true));
         try {
             final HttpRequest request = new DefaultFullHttpRequest(HTTP_1_1, POST, "/example", data.retain());
             final HttpHeaders httpHeaders = request.headers();
@@ -200,13 +204,12 @@ public void testRequestWithBody() throws Exception {
             writeFuture.awaitUninterruptibly(2, SECONDS);
             assertTrue(writeFuture.isSuccess());
             awaitRequests();
-            final ArgumentCaptor<ByteBuf> dataCaptor = ArgumentCaptor.forClass(ByteBuf.class);
             verify(serverListener).onHeadersRead(any(ChannelHandlerContext.class), eq(3), eq(http2Headers), eq(0),
                     anyShort(), anyBoolean(), eq(0), eq(false));
-            verify(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3), dataCaptor.capture(), eq(0),
+            verify(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3), any(ByteBuf.class), eq(0),
                     eq(true));
-            capturedData = dataCaptor.getAllValues();
-            assertEquals(data, capturedData.get(0));
+            assertEquals(1, receivedBuffers.size());
+            assertEquals(text, receivedBuffers.get(0));
         } finally {
             data.release();
         }

codec-http2/src/test/java/io/netty/handler/codec/http2/Http2ConnectionRoundtripTest.java

@@ -20,11 +20,13 @@
 import static io.netty.handler.codec.http2.Http2TestUtil.runInChannel;
 import static io.netty.util.CharsetUtil.UTF_8;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyInt;
 import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import io.netty.bootstrap.Bootstrap;
@@ -41,11 +43,12 @@
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http2.Http2TestUtil.Http2Runnable;
-import io.netty.util.CharsetUtil;
 import io.netty.util.NetUtil;
 import io.netty.util.concurrent.Future;
 
+import java.io.ByteArrayOutputStream;
 import java.net.InetSocketAddress;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.CountDownLatch;
@@ -54,9 +57,11 @@
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
-import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 /**
  * Tests the full HTTP/2 framing stack including the connection and preface handlers.
@@ -144,7 +149,16 @@ public void flowControlProperlyChunksLargeMessage() throws Exception {
         final byte[] bytes = new byte[length];
         new Random().nextBytes(bytes);
         final ByteBuf data = Unpooled.wrappedBuffer(bytes);
-        List<ByteBuf> capturedData = null;
+        final ByteArrayOutputStream out = new ByteArrayOutputStream(length);
+        doAnswer(new Answer<Void>() {
+            @Override
+            public Void answer(InvocationOnMock in) throws Throwable {
+                ByteBuf buf = (ByteBuf) in.getArguments()[2];
+                buf.readBytes(out, buf.readableBytes());
+                return null;
+            }
+        }).when(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3),
+                any(ByteBuf.class), eq(0), Mockito.anyBoolean());
         try {
             // Initialize the data latch based on the number of bytes expected.
             requestLatch(new CountDownLatch(2));
@@ -163,18 +177,18 @@ public void run() {
             assertTrue(dataLatch.await(5, TimeUnit.SECONDS));
 
             // Verify that headers were received and only one DATA frame was received with endStream set.
-            final ArgumentCaptor<ByteBuf> dataCaptor = ArgumentCaptor.forClass(ByteBuf.class);
             verify(serverListener).onHeadersRead(any(ChannelHandlerContext.class), eq(3), eq(headers), eq(0),
                     eq((short) 16), eq(false), eq(0), eq(false));
-            verify(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3), dataCaptor.capture(), eq(0),
+            verify(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3), any(ByteBuf.class), eq(0),
                     eq(true));
 
             // Verify we received all the bytes.
-            capturedData = dataCaptor.getAllValues();
-            assertEquals(data, capturedData.get(0));
+            out.flush();
+            byte[] received = out.toByteArray();
+            assertArrayEquals(bytes, received);
         } finally {
             data.release();
-            release(capturedData);
+            out.close();
         }
     }
 
@@ -183,48 +197,56 @@ public void stressTest() throws Exception {
         final Http2Headers headers = dummyHeaders();
         final String text = "hello world";
         final String pingMsg = "12345678";
-        final ByteBuf data = Unpooled.copiedBuffer(text.getBytes());
-        final ByteBuf pingData = Unpooled.copiedBuffer(pingMsg.getBytes());
-        List<ByteBuf> capturedData = null;
-        List<ByteBuf> capturedPingData = null;
+        final ByteBuf data = Unpooled.copiedBuffer(text, UTF_8);
+        final ByteBuf pingData = Unpooled.copiedBuffer(pingMsg, UTF_8);
+        final List<String> receivedPingBuffers = new ArrayList<String>(NUM_STREAMS);
+        doAnswer(new Answer<Void>() {
+            @Override
+            public Void answer(InvocationOnMock in) throws Throwable {
+                receivedPingBuffers.add(((ByteBuf) in.getArguments()[1]).toString(UTF_8));
+                return null;
+            }
+        }).when(serverListener).onPingRead(any(ChannelHandlerContext.class), eq(pingData));
+        final List<String> receivedDataBuffers = new ArrayList<String>();
+        doAnswer(new Answer<Void>() {
+            @Override
+            public Void answer(InvocationOnMock in) throws Throwable {
+                receivedDataBuffers.add(((ByteBuf) in.getArguments()[2]).toString(UTF_8));
+                return null;
+            }
+        }).when(serverListener).onDataRead(any(ChannelHandlerContext.class), anyInt(), eq(data),
+                eq(0), eq(true));
         try {
             runInChannel(clientChannel, new Http2Runnable() {
                 @Override
                 public void run() {
                     for (int i = 0, nextStream = 3; i < NUM_STREAMS; ++i, nextStream += 2) {
                         http2Client.writeHeaders(ctx(), nextStream, headers, 0, (short) 16, false, 0, false,
                                 newPromise());
-                        http2Client.writePing(ctx(), pingData.retain(), newPromise());
-                        http2Client.writeData(ctx(), nextStream, data.retain(), 0, true, newPromise());
+                        http2Client.writePing(ctx(), pingData.slice().retain(), newPromise());
+                        http2Client.writeData(ctx(), nextStream, data.slice().retain(), 0, true, newPromise());
                     }
                 }
             });
             // Wait for all frames to be received.
             assertTrue(requestLatch.await(STRESS_TIMEOUT_SECONDS, SECONDS));
             verify(serverListener, times(NUM_STREAMS)).onHeadersRead(any(ChannelHandlerContext.class), anyInt(),
                     eq(headers), eq(0), eq((short) 16), eq(false), eq(0), eq(false));
-            final ArgumentCaptor<ByteBuf> dataCaptor = ArgumentCaptor.forClass(ByteBuf.class);
-            final ArgumentCaptor<ByteBuf> pingDataCaptor = ArgumentCaptor.forClass(ByteBuf.class);
             verify(serverListener, times(NUM_STREAMS)).onPingRead(any(ChannelHandlerContext.class),
-                    pingDataCaptor.capture());
-            capturedPingData = pingDataCaptor.getAllValues();
-            verify(serverListener, times(NUM_STREAMS)).onDataRead(any(ChannelHandlerContext.class), anyInt(),
-                    dataCaptor.capture(), eq(0), eq(true));
-            capturedData = dataCaptor.getAllValues();
-            data.resetReaderIndex();
-            pingData.resetReaderIndex();
-            int i;
-            for (i = 0; i < capturedPingData.size(); ++i) {
-                assertEquals(pingData, capturedPingData.get(i));
+                    any(ByteBuf.class));
+            verify(serverListener, times(NUM_STREAMS)).onDataRead(any(ChannelHandlerContext.class),
+                    anyInt(), any(ByteBuf.class), eq(0), eq(true));
+            assertEquals(NUM_STREAMS, receivedPingBuffers.size());
+            assertEquals(NUM_STREAMS, receivedDataBuffers.size());
+            for (String receivedData : receivedDataBuffers) {
+                assertEquals(text, receivedData);
             }
-            for (i = 0; i < capturedData.size(); ++i) {
-                assertEquals(capturedData.get(i).toString(CharsetUtil.UTF_8), data, capturedData.get(i));
+            for (String receivedPing : receivedPingBuffers) {
+                assertEquals(pingMsg, receivedPing);
             }
         } finally {
             data.release();
             pingData.release();
-            release(capturedData);
-            release(capturedPingData);
         }
     }
 
@@ -250,14 +272,6 @@ private ChannelPromise newPromise() {
         return ctx().newPromise();
     }
 
-    private static void release(List<ByteBuf> capturedData) {
-        if (capturedData != null) {
-            for (int i = 0; i < capturedData.size(); ++i) {
-                capturedData.get(i).release();
-            }
-        }
-    }
-
     private Http2Headers dummyHeaders() {
         return new DefaultHttp2Headers().method(as("GET")).scheme(as("https"))
         .authority(as("example.org")).path(as("/some/path/resource2")).add(randomString(), randomString());