8296889: Race condition when cancelling a request

Reviewed-by: jpai

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

@@ -208,6 +208,11 @@ static final class Http1ResponseBodySubscriber<U> extends HttpBodySubscriberWrap
             this.exchange = exchange;
         }
 
+        @Override
+        public void onSubscribed() {
+            exchange.registerResponseSubscriber(this);
+        }
+
         @Override
         protected void complete(Throwable t) {
             try {
@@ -459,7 +464,6 @@ Http1ResponseBodySubscriber<T> createResponseSubscriber(BodyHandler<T> handler,
         BodySubscriber<T> subscriber = handler.apply(response);
         Http1ResponseBodySubscriber<T> bs =
                 new Http1ResponseBodySubscriber<T>(subscriber, this);
-        registerResponseSubscriber(bs);
         return bs;
     }
 

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

@@ -344,7 +344,6 @@ CompletableFuture<T> readBodyAsync(HttpResponse.BodyHandler<T> handler,
     Http2StreamResponseSubscriber<T> createResponseSubscriber(BodyHandler<T> handler, ResponseInfo response) {
         Http2StreamResponseSubscriber<T> subscriber =
                 new Http2StreamResponseSubscriber<>(handler.apply(response));
-        registerResponseSubscriber(subscriber);
         return subscriber;
     }
 
@@ -1543,17 +1542,22 @@ final class Http2StreamResponseSubscriber<U> extends HttpBodySubscriberWrapper<U
             super(subscriber);
         }
 
+        @Override
+        public void onSubscribed() {
+            registerResponseSubscriber(this);
+        }
+
         @Override
         protected void complete(Throwable t) {
             try {
-                Stream.this.unregisterResponseSubscriber(this);
+                unregisterResponseSubscriber(this);
             } finally {
                 super.complete(t);
             }
         }
         @Override
         protected void onCancel() {
-            Stream.this.unregisterResponseSubscriber(this);
+            unregisterResponseSubscriber(this);
         }
     }
 

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

@@ -127,6 +127,15 @@ private void propagateError(Throwable t) {
      */
     protected void onCancel() { }
 
+    /**
+     * Called right after the userSubscriber::onSubscribe is called.
+     * @apiNote
+     * This method may be used by subclasses to perform cleanup
+     * related actions after a subscription has been succesfully
+     * accepted.
+     */
+    protected void onSubscribed() { }
+
     /**
      * Complete the subscriber, either normally or exceptionally
      * ensure that the subscriber is completed only once.
@@ -169,8 +178,9 @@ public CompletionStage<T> getBody() {
     public void onSubscribe(Flow.Subscription subscription) {
         // race condition with propagateError: we need to wait until
         // subscription is finished before calling onError;
+        boolean onSubscribed;
         synchronized (this) {
-            if (subscribed.compareAndSet(false, true)) {
+            if ((onSubscribed = subscribed.compareAndSet(false, true))) {
                 SubscriptionWrapper wrapped = new SubscriptionWrapper(subscription);
                 userSubscriber.onSubscribe(this.subscription = wrapped);
             } else {
@@ -181,6 +191,7 @@ public void onSubscribe(Flow.Subscription subscription) {
                 assert completed.get();
             }
         }
+        if (onSubscribed) onSubscribed();
     }
 
     @Override

test/jdk/java/net/httpclient/CancelRequestTest.java

@@ -23,7 +23,7 @@
 
 /*
  * @test
- * @bug 8245462 8229822 8254786
+ * @bug 8245462 8229822 8254786 8296889
  * @summary Tests cancelling the request.
  * @library /test/lib http2/server
  * @key randomness