Issue #372 (Data race in HttpReceiverOverHTTP2)

Fixed implementations of onData() to properly complete the callback
and to copy the data if they use it asynchronously.

jetty-http2/http2-client/src/test/java/org/eclipse/jetty/http2/client/IdleTimeoutTest.java

@@ -357,6 +357,7 @@ public class IdleTimeoutTest extends AbstractTest
             @Override
             public void onData(Stream stream, DataFrame frame, Callback callback)
             {
+                callback.succeeded();
                 dataLatch.countDown();
             }
 

jetty-http2/http2-client/src/test/java/org/eclipse/jetty/http2/client/PushCacheFilterTest.java

@@ -101,6 +101,7 @@ public class PushCacheFilterTest extends AbstractTest
                         @Override
                         public void onData(Stream stream, DataFrame frame, Callback callback)
                         {
+                            callback.succeeded();
                             warmupLatch.countDown();
                         }
                     });
@@ -188,6 +189,7 @@ public class PushCacheFilterTest extends AbstractTest
                         @Override
                         public void onData(Stream stream, DataFrame frame, Callback callback)
                         {
+                            callback.succeeded();
                             warmupLatch.countDown();
                         }
                     });
@@ -273,6 +275,7 @@ public class PushCacheFilterTest extends AbstractTest
                         @Override
                         public void onData(Stream stream, DataFrame frame, Callback callback)
                         {
+                            callback.succeeded();
                             warmupLatch.countDown();
                         }
                     });
@@ -298,6 +301,7 @@ public class PushCacheFilterTest extends AbstractTest
                     @Override
                     public void onData(Stream stream, DataFrame frame, Callback callback)
                     {
+                        callback.succeeded();
                         pushLatch.countDown();
                     }
                 };
@@ -325,6 +329,7 @@ public class PushCacheFilterTest extends AbstractTest
             @Override
             public void onData(Stream stream, DataFrame frame, Callback callback)
             {
+                callback.succeeded();
                 if (frame.isEndStream())
                     secondaryResponseLatch.countDown();
             }
@@ -372,6 +377,7 @@ public class PushCacheFilterTest extends AbstractTest
                         @Override
                         public void onData(Stream stream, DataFrame frame, Callback callback)
                         {
+                            callback.succeeded();
                             warmupLatch.countDown();
                         }
                     });
@@ -655,6 +661,7 @@ public class PushCacheFilterTest extends AbstractTest
                             @Override
                             public void onData(Stream stream, DataFrame frame, Callback callback)
                             {
+                                callback.succeeded();
                                 if (frame.isEndStream())
                                     warmupLatch.countDown();
                             }
@@ -676,6 +683,7 @@ public class PushCacheFilterTest extends AbstractTest
             @Override
             public void onData(Stream stream, DataFrame frame, Callback callback)
             {
+                callback.succeeded();
                 if (frame.isEndStream())
                     primaryResponseLatch.countDown();
             }

jetty-http2/http2-client/src/test/java/org/eclipse/jetty/http2/client/StreamCloseTest.java

@@ -129,7 +129,12 @@ public class StreamCloseTest extends AbstractTest
                     public void onData(final Stream stream, DataFrame frame, final Callback callback)
                     {
                         Assert.assertTrue(((HTTP2Stream)stream).isRemotelyClosed());
-                        stream.data(frame, new Callback()
+
+                        // We must copy the data that we send asynchronously.
+                        ByteBuffer data = frame.getData();
+                        ByteBuffer copy = ByteBuffer.allocate(data.remaining());
+                        copy.put(data).flip();
+                        stream.data(new DataFrame(stream.getId(), copy, frame.isEndStream()), new Callback()
                         {
                             @Override
                             public void succeeded()
@@ -155,6 +160,7 @@ public class StreamCloseTest extends AbstractTest
             public void onData(Stream stream, DataFrame frame, Callback callback)
             {
                 // The sent data callback may not be notified yet here.
+                callback.succeeded();
                 completeLatch.countDown();
             }
         });

jetty-http2/http2-client/src/test/java/org/eclipse/jetty/http2/client/StreamCountTest.java

@@ -70,6 +70,10 @@ public class StreamCountTest extends AbstractTest
                             MetaData.Response metaData = new MetaData.Response(HttpVersion.HTTP_2, 200, fields);
                             stream.headers(new HeadersFrame(stream.getId(), metaData, null, true), callback);
                         }
+                        else
+                        {
+                            callback.succeeded();
+                        }
                     }
                 };
             }
@@ -144,6 +148,10 @@ public class StreamCountTest extends AbstractTest
                             MetaData.Response metaData = new MetaData.Response(HttpVersion.HTTP_2, 200, fields);
                             stream.headers(new HeadersFrame(stream.getId(), metaData, null, true), callback);
                         }
+                        else
+                        {
+                            callback.succeeded();
+                        }
                     }
                 };
             }

jetty-http2/http2-client/src/test/java/org/eclipse/jetty/http2/client/StreamResetTest.java

@@ -171,6 +171,7 @@ public class StreamResetTest extends AbstractTest
             @Override
             public void onData(Stream stream, DataFrame frame, Callback callback)
             {
+                callback.succeeded();
                 stream1DataLatch.countDown();
             }
         });
@@ -186,6 +187,7 @@ public class StreamResetTest extends AbstractTest
             @Override
             public void onData(Stream stream, DataFrame frame, Callback callback)
             {
+                callback.succeeded();
                 stream2DataLatch.countDown();
             }
         });