Resolves

Author: GoeLin

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

@@ -120,12 +120,12 @@ public String toString() {
      * concrete implementations: {@link Http1Request.StreamSubscriber}, and
      * {@link Http1Request.FixedContentSubscriber}, for receiving chunked and
      * fixed length bodies, respectively. */
-    static abstract class Http1BodySubscriber implements Flow.Subscriber<ByteBuffer> {
+    abstract static class Http1RequestBodySubscriber implements Flow.Subscriber<ByteBuffer> {
         final MinimalFuture<Flow.Subscription> whenSubscribed = new MinimalFuture<>();
         private volatile Flow.Subscription subscription;
         volatile boolean complete;
         private final Logger debug;
-        Http1BodySubscriber(Logger debug) {
+        Http1RequestBodySubscriber(Logger debug) {
             assert debug != null;
             this.debug = debug;
         }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2019, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2022, 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

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2020, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2022, 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
@@ -275,119 +275,6 @@ public void nullBody(HttpResponse<T> resp, Throwable t) {
         }
     }
 
-    static final Flow.Subscription NOP = new Flow.Subscription() {
-        @Override
-        public void request(long n) { }
-        public void cancel() { }
-    };
-
-    /**
-     * The Http1AsyncReceiver ensures that all calls to
-     * the subscriber, including onSubscribe, occur sequentially.
-     * There could however be some race conditions that could happen
-     * in case of unexpected errors thrown at unexpected places, which
-     * may cause onError to be called multiple times.
-     * The Http1BodySubscriber will ensure that the user subscriber
-     * is actually completed only once - and only after it is
-     * subscribed.
-     * @param <U> The type of response.
-     */
-    final static class Http1BodySubscriber<U> implements TrustedSubscriber<U> {
-        final HttpResponse.BodySubscriber<U> userSubscriber;
-        final AtomicBoolean completed = new AtomicBoolean();
-        volatile Throwable withError;
-        volatile boolean subscribed;
-        Http1BodySubscriber(HttpResponse.BodySubscriber<U> userSubscriber) {
-            this.userSubscriber = userSubscriber;
-        }
-
-        @Override
-        public boolean needsExecutor() {
-            return TrustedSubscriber.needsExecutor(userSubscriber);
-        }
-
-        // propagate the error to the user subscriber, even if not
-        // subscribed yet.
-        private void propagateError(Throwable t) {
-            assert t != null;
-            try {
-                // if unsubscribed at this point, it will not
-                // get subscribed later - so do it now and
-                // propagate the error
-                if (subscribed == false) {
-                    subscribed = true;
-                    userSubscriber.onSubscribe(NOP);
-                }
-            } finally  {
-                // if onError throws then there is nothing to do
-                // here: let the caller deal with it by logging
-                // and closing the connection.
-                userSubscriber.onError(t);
-            }
-        }
-
-        // complete the subscriber, either normally or exceptionally
-        // ensure that the subscriber is completed only once.
-        private void complete(Throwable t) {
-            if (completed.compareAndSet(false, true)) {
-                t  = withError = Utils.getCompletionCause(t);
-                if (t == null) {
-                    assert subscribed;
-                    try {
-                        userSubscriber.onComplete();
-                    } catch (Throwable x) {
-                        // Simply propagate the error by calling
-                        // onError on the user subscriber, and let the
-                        // connection be reused since we should have received
-                        // and parsed all the bytes when we reach here.
-                        // If onError throws in turn, then we will simply
-                        // let that new exception flow up to the caller
-                        // and let it deal with it.
-                        // (i.e: log and close the connection)
-                        // Note that rethrowing here could introduce a
-                        // race that might cause the next send() operation to
-                        // fail as the connection has already been put back
-                        // into the cache when we reach here.
-                        propagateError(t = withError = Utils.getCompletionCause(x));
-                    }
-                } else {
-                    propagateError(t);
-                }
-            }
-        }
-
-        @Override
-        public CompletionStage<U> getBody() {
-            return userSubscriber.getBody();
-        }
-
-        @Override
-        public void onSubscribe(Flow.Subscription subscription) {
-            if (!subscribed) {
-                subscribed = true;
-                userSubscriber.onSubscribe(subscription);
-            } else {
-                // could be already subscribed and completed
-                // if an unexpected error occurred before the actual
-                // subscription - though that's not supposed
-                // happen.
-                assert completed.get();
-            }
-        }
-        @Override
-        public void onNext(List<ByteBuffer> item) {
-            assert !completed.get();
-            userSubscriber.onNext(item);
-        }
-        @Override
-        public void onError(Throwable throwable) {
-            complete(throwable);
-        }
-        @Override
-        public void onComplete() {
-            complete(null);
-        }
-    }
 
     public <U> CompletableFuture<U> readBody(HttpResponse.BodySubscriber<U> p,
                                          boolean return2Cache,

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

@@ -65,14 +65,19 @@
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentSkipListSet;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
 import java.util.function.BooleanSupplier;
 import java.util.stream.Stream;
 import java.net.http.HttpClient;
@@ -659,28 +664,41 @@ final long webSocketClose() {
     }
 
     // Returns the pendingOperationCount.
+    // Incremented with any operation, whether it's HTTP/1.1, HTTP/2, or WebSocket
     final long referenceCount() {
         return pendingOperationCount.get();
     }
 
+    // Trackers are used in test to verify that an instance of
+    // HttpClient has shutdown correctly, and that all operations
+    // have terminated.
     final static class HttpClientTracker implements Tracker {
+        final AtomicLong requestCount;
         final AtomicLong httpCount;
         final AtomicLong http2Count;
         final AtomicLong websocketCount;
         final AtomicLong operationsCount;
+        final AtomicLong connnectionsCount;
         final Reference<?> reference;
+        final AtomicBoolean isAlive;
         final String name;
-        HttpClientTracker(AtomicLong http,
+        HttpClientTracker(AtomicLong request,
+                          AtomicLong http,
                           AtomicLong http2,
                           AtomicLong ws,
                           AtomicLong ops,
+                          AtomicLong conns,
                           Reference<?> ref,
+                          AtomicBoolean isAlive,
                           String name) {
+            this.requestCount = request;
             this.httpCount = http;
             this.http2Count = http2;
             this.websocketCount = ws;
             this.operationsCount = ops;
+            this.connnectionsCount = conns;
             this.reference = ref;
+            this.isAlive = isAlive;
             this.name = name;
         }
         @Override

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2020, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2022, 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