Fixed handling of negative request. Added BodyPublishers::concat to the TCK tests.

Author: dfuch

src/java.net.http/share/classes/jdk/internal/net/http/PullPublisher.java

@@ -102,7 +102,7 @@ protected void run() {
                     return;
                 }
 
-                while (demand.tryDecrement() && !cancelled) {
+                while (demand.tryDecrement() && !cancelled && error == null) {
                     T next;
                     try {
                         if (!iter.hasNext()) {
@@ -121,6 +121,11 @@ protected void run() {
                     completed = true;
                     pullScheduler.stop();
                     subscriber.onComplete();
+                } else if ((t = error) != null && !completed) {
+                    completed = true;
+                    pullScheduler.stop();
+                    subscriber.onError(t);
+                    return;
                 }
             }
         }

src/java.net.http/share/classes/jdk/internal/net/http/RequestPublishers.java

@@ -560,6 +560,7 @@ private static final class AggregateSubscription
         final Demand demand = new Demand(); // from upstream
         final Demand demanded = new Demand(); // requested downstream
         final AtomicReference<Throwable> error = new AtomicReference<>();
+        volatile Throwable illegalRequest;
         volatile BodyPublisher publisher; // downstream
         volatile Flow.Subscription subscription; // downstream
         volatile boolean cancelled;
@@ -574,7 +575,11 @@ public void request(long n) {
             if (cancelled || publisher == null && bodies.isEmpty()) {
                 return;
             }
-            demand.increase(n);
+            try {
+                demand.increase(n);
+            } catch (IllegalArgumentException x) {
+                illegalRequest = x;
+            }
             scheduler.runOrSchedule();
         }
 
@@ -584,6 +589,17 @@ public void cancel() {
             scheduler.runOrSchedule();
         }
 
+        private boolean cancelSubscription() {
+            Flow.Subscription subscription = this.subscription;
+            if (subscription != null) {
+                this.subscription = null;
+                this.publisher = null;
+                subscription.cancel();
+            }
+            scheduler.stop();
+            return subscription != null;
+        }
+
         public void run() {
             try {
                 while (error.get() == null
@@ -592,13 +608,10 @@ public void run() {
                     boolean cancelled = this.cancelled;
                     BodyPublisher publisher = this.publisher;
                     Flow.Subscription subscription = this.subscription;
+                    Throwable illegalRequest = this.illegalRequest;
                     if (cancelled) {
-                        this.publisher = null;
-                        bodies.removeIf((b) -> true);
-                        if (subscription != null) {
-                            subscription.cancel();
-                        }
-                        scheduler.stop();
+                        bodies.clear();
+                        cancelSubscription();
                         return;
                     }
                     if (publisher == null && !bodies.isEmpty()) {
@@ -608,6 +621,10 @@ public void run() {
                     } else if (publisher == null) {
                         return;
                     }
+                    if (illegalRequest != null) {
+                        onError(illegalRequest);
+                        return;
+                    }
                     if (subscription == null) return;
                     if (!demand.isFulfilled()) {
                         long n = demand.decreaseAndGet(demand.get());
@@ -629,14 +646,24 @@ public void onSubscribe(Flow.Subscription subscription) {
 
         @Override
         public void onNext(ByteBuffer item) {
+            // make sure to cancel the subscription if we receive
+            // an item after the subscription was cancelled or
+            // an error was reported.
+            if (cancelled || error.get() != null) {
+                cancelSubscription();
+                return;
+            }
             demanded.tryDecrement();
             subscriber.onNext(item);
         }
 
         @Override
         public void onError(Throwable throwable) {
             if (error.compareAndSet(null, throwable)) {
+                publisher = null;
+                subscription = null;
                 subscriber.onError(throwable);
+                scheduler.stop();
             }
         }
 
@@ -652,7 +679,9 @@ public void onComplete() {
             } else {
                 publisher = null;
                 subscription = null;
-                subscriber.onComplete();
+                if (!cancelled) {
+                    subscriber.onComplete();
+                }
                 scheduler.stop();
             }
         }

src/java.net.http/share/classes/jdk/internal/net/http/common/Demand.java

@@ -47,7 +47,7 @@ public final class Demand {
      */
     public boolean increase(long n) {
         if (n <= 0) {
-            throw new IllegalArgumentException(String.valueOf(n));
+            throw new IllegalArgumentException("non-positive subscription request: " + String.valueOf(n));
         }
         long prev = val.getAndAccumulate(n, (p, i) -> p + i < 0 ? Long.MAX_VALUE : p + i);
         return prev == 0;

test/jdk/java/net/httpclient/AggregateRequestBodyTest.java

@@ -103,6 +103,7 @@ public class AggregateRequestBodyTest implements HttpServerAdapters {
     static final int RESPONSE_CODE = 200;
     static final int ITERATION_COUNT = 4;
     static final Class<IllegalArgumentException> IAE = IllegalArgumentException.class;
+    static final Class<CompletionException> CE = CompletionException.class;
     // a shared executor helps reduce the amount of threads created by the test
     static final Executor executor = new TestExecutor(Executors.newCachedThreadPool());
     static final ConcurrentMap<String, Throwable> FAILURES = new ConcurrentHashMap<>();
@@ -543,6 +544,13 @@ public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
                 description.replace("null", "length(-1)"));
     }
 
+    private static final Throwable completionCause(CompletionException x) {
+        while (x.getCause() instanceof CompletionException) {
+            x = (CompletionException)x.getCause();
+        }
+        return x.getCause();
+    }
+
     @Test(dataProvider = "negativeRequests")
     public void testNegativeRequest(long n) {
         assert n <= 0 : "test for negative request called with n > 0 : " + n;
@@ -551,8 +559,15 @@ public void testNegativeRequest(long n) {
         RequestSubscriber subscriber = new RequestSubscriber();
         publisher.subscribe(subscriber);
         Subscription subscription = subscriber.subscriptionCF.join();
-        IllegalArgumentException iae = expectThrows(IAE, () -> subscription.request(n));
-        System.out.printf("Got expected IAE for %d: %s%n", n, iae);
+        subscription.request(n);
+        CompletionException expected = expectThrows(CE, () -> subscriber.resultCF.join());
+        Throwable cause = completionCause(expected);
+        if (cause instanceof IllegalArgumentException) {
+            System.out.printf("Got expected IAE for %d: %s%n", n, cause);
+        } else {
+            throw new AssertionError("Unexpected exception: " + cause,
+                    (cause == null) ? expected : cause);
+        }
     }
 
     @Test(dataProvider = "contentLengths")

test/jdk/java/net/httpclient/reactivestreams-tck-tests/BodyPublishersConcat.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright (c) 2020, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+import org.reactivestreams.tck.TestEnvironment;
+import org.reactivestreams.tck.flow.FlowPublisherVerification;
+
+import java.net.http.HttpRequest.BodyPublisher;
+import java.net.http.HttpRequest.BodyPublishers;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Flow.Publisher;
+
+/* See TckDriver.java for more information */
+public class BodyPublishersConcat
+        extends FlowPublisherVerification<ByteBuffer> {
+
+    private static final int ELEMENT_SIZE = 16 * 1024;
+
+    public BodyPublishersConcat() {
+        super(new TestEnvironment(450L));
+    }
+
+    private static BodyPublisher ofByteArrays(int n, byte[] bytes) {
+        return BodyPublishers.ofByteArrays(Collections.nCopies((int) n, bytes));
+    }
+
+    @Override
+    public Publisher<ByteBuffer> createFlowPublisher(long nElements) {
+        System.out.println("BodyPublishersConcat: %d elements requested"
+                .formatted(nElements));
+        byte[] bytes = S.arrayOfNRandomBytes(ELEMENT_SIZE);
+        if (nElements == 0) {
+            System.out.println("BodyPublishersConcat: empty publisher");
+            return BodyPublishers.concat();
+        } else if (nElements == 1) {
+            System.out.println("BodyPublishersConcat: singleton publisher");
+            return BodyPublishers.concat(ofByteArrays(1, bytes));
+        } else if (nElements < 4) {
+            int left = (int)nElements/2;
+            int right = (int)nElements - left;
+            System.out.println("BodyPublishersConcat: dual publisher (%d, %d)".formatted(left, right));
+            return BodyPublishers.concat(ofByteArrays(left, bytes),
+                    ofByteArrays(right, bytes));
+        } else {
+            List<BodyPublisher> publishers = new ArrayList<>();
+            List<Integer> sizes = new ArrayList<>();
+            long remaining = nElements;
+            int max = (int) Math.min((long)Integer.MAX_VALUE, nElements/2L);
+            while (remaining > 0) {
+                int length = S.randomIntUpTo(max);
+                if (length == 0) length = 1;
+                sizes.add(length);
+                if (remaining > length) {
+                    publishers.add(ofByteArrays(length, bytes));
+                    remaining = remaining - length;
+                } else {
+                    publishers.add(ofByteArrays((int)remaining, bytes));
+                    remaining = 0;
+                }
+            }
+            System.out.println("BodyPublishersConcat: multi publisher " + sizes);
+            return BodyPublishers.concat(publishers.toArray(BodyPublisher[]::new));
+        }
+    }
+
+    @Override
+    public Publisher<ByteBuffer> createFailedFlowPublisher() {
+        return null;
+    }
+}

test/jdk/java/net/httpclient/reactivestreams-tck-tests/TckDriver.java

@@ -38,6 +38,7 @@
  * @compile -encoding UTF-8 BodyPublishersOfFile.java
  * @compile -encoding UTF-8 BodyPublishersOfInputStream.java
  * @compile -encoding UTF-8 BodyPublishersOfSubByteArray.java
+ * @compile -encoding UTF-8 BodyPublishersConcat.java
  *
  * @compile -encoding UTF-8 BodySubscribersBuffering.java
  * @compile -encoding UTF-8 BodySubscribersDiscarding.java
@@ -65,6 +66,7 @@
  * @run testng/othervm BodyPublishersOfFile
  * @run testng/othervm BodyPublishersOfInputStream
  * @run testng/othervm BodyPublishersOfSubByteArray
+ * @run testng/othervm BodyPublishersConcat
  *
  * @run testng/othervm BodySubscribersBuffering
  * @run testng/othervm BodySubscribersDiscarding