8225583: Examine the HttpResponse.BodySubscribers for null handling and multiple subscriptions

Backport-of: af2a46b

Author: zhengyu123

src/java.net.http/share/classes/java/net/http/HttpResponse.java

@@ -1252,7 +1252,7 @@ public static BodySubscriber<Void> discarding() {
         /**
          * Returns a {@code BodySubscriber} which buffers data before delivering
          * it to the given downstream subscriber. The subscriber guarantees to
-         * deliver {@code buffersize} bytes of data to each invocation of the
+         * deliver {@code bufferSize} bytes of data to each invocation of the
          * downstream's {@link BodySubscriber#onNext(Object) onNext} method,
          * except for the final invocation, just before
          * {@link BodySubscriber#onComplete() onComplete} is invoked. The final

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2018, 2019, 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
@@ -40,6 +40,7 @@
 import java.util.concurrent.Flow;
 import java.util.concurrent.Flow.Subscriber;
 import java.util.concurrent.Flow.Subscription;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
@@ -56,6 +57,7 @@ public final class LineSubscriberAdapter<S extends Subscriber<? super String>,R>
     private final Function<? super S, ? extends R> finisher;
     private final Charset charset;
     private final String eol;
+    private final AtomicBoolean subscribed = new AtomicBoolean();
     private volatile LineSubscription downstream;
 
     private LineSubscriberAdapter(S subscriber,
@@ -72,6 +74,12 @@ private LineSubscriberAdapter(S subscriber,
 
     @Override
     public void onSubscribe(Subscription subscription) {
+        Objects.requireNonNull(subscription);
+        if (!subscribed.compareAndSet(false, true)) {
+            subscription.cancel();
+            return;
+        }
+
         downstream = LineSubscription.create(subscription,
                                              charset,
                                              eol,
@@ -82,6 +90,7 @@ public void onSubscribe(Subscription subscription) {
 
     @Override
     public void onNext(List<ByteBuffer> item) {
+        Objects.requireNonNull(item);
         try {
             downstream.submit(item);
         } catch (Throwable t) {
@@ -91,6 +100,7 @@ public void onNext(List<ByteBuffer> item) {
 
     @Override
     public void onError(Throwable throwable) {
+        Objects.requireNonNull(throwable);
         try {
             downstream.signalError(throwable);
         } finally {

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

@@ -84,6 +84,7 @@ public CompletionStage<Void> getBody() {
 
         @Override
         public void onSubscribe(Flow.Subscription subscription) {
+            Objects.requireNonNull(subscription);
             if (!subscribed.compareAndSet(false, true)) {
                 subscription.cancel();
             } else {
@@ -94,6 +95,7 @@ public void onSubscribe(Flow.Subscription subscription) {
 
         @Override
         public void onNext(List<ByteBuffer> items) {
+            Objects.requireNonNull(items);
             for (ByteBuffer item : items) {
                 byte[] buf = new byte[item.remaining()];
                 item.get(buf);
@@ -104,6 +106,7 @@ public void onNext(List<ByteBuffer> items) {
 
         @Override
         public void onError(Throwable throwable) {
+            Objects.requireNonNull(throwable);
             result.completeExceptionally(throwable);
         }
 
@@ -131,6 +134,7 @@ public static class PathSubscriber implements BodySubscriber<Path> {
         private final FilePermission[] filePermissions;
         private final CompletableFuture<Path> result = new MinimalFuture<>();
 
+        private final AtomicBoolean subscribed = new AtomicBoolean();
         private volatile Flow.Subscription subscription;
         private volatile FileChannel out;
 
@@ -170,6 +174,12 @@ public static PathSubscriber create(Path file,
 
         @Override
         public void onSubscribe(Flow.Subscription subscription) {
+            Objects.requireNonNull(subscription);
+            if (!subscribed.compareAndSet(false, true)) {
+                subscription.cancel();
+                return;
+            }
+
             this.subscription = subscription;
             if (System.getSecurityManager() == null) {
                 try {
@@ -411,6 +421,7 @@ public int read() throws IOException {
 
         @Override
         public void onSubscribe(Flow.Subscription s) {
+            Objects.requireNonNull(s);
             try {
                 if (!subscribed.compareAndSet(false, true)) {
                     s.cancel();
@@ -539,6 +550,7 @@ public NullSubscriber(Optional<T> result) {
 
         @Override
         public void onSubscribe(Flow.Subscription subscription) {
+            Objects.requireNonNull(subscription);
             if (!subscribed.compareAndSet(false, true)) {
                 subscription.cancel();
             } else {
@@ -553,6 +565,7 @@ public void onNext(List<ByteBuffer> items) {
 
         @Override
         public void onError(Throwable throwable) {
+            Objects.requireNonNull(throwable);
             cf.completeExceptionally(throwable);
         }
 
@@ -819,13 +832,21 @@ private void subscribe(Flow.Subscriber<? super List<ByteBuffer>> subscriber) {
             }
         }
 
+        private final AtomicBoolean subscribed = new AtomicBoolean();
+
         @Override
         public void onSubscribe(Flow.Subscription subscription) {
-            subscriptionCF.complete(subscription);
+            Objects.requireNonNull(subscription);
+            if (!subscribed.compareAndSet(false, true)) {
+                subscription.cancel();
+            } else {
+                subscriptionCF.complete(subscription);
+            }
         }
 
         @Override
         public void onNext(List<ByteBuffer> item) {
+            Objects.requireNonNull(item);
             try {
                 // cannot be called before onSubscribe()
                 assert subscriptionCF.isDone();
@@ -853,6 +874,7 @@ assert suppress(subscriptionCF.isDone(),
             // onError can be called before request(1), and therefore can
             // be called before subscriberRef is set.
             signalError(throwable);
+            Objects.requireNonNull(throwable);
         }
 
         @Override