8207966: HttpClient response without content-length does not return body

Reviewed-by: chegar

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

@@ -40,6 +40,7 @@ import java.util.function.Function;
 import java.net.http.HttpHeaders;
 import java.net.http.HttpResponse;
 import jdk.internal.net.http.ResponseContent.BodyParser;
+import jdk.internal.net.http.ResponseContent.UnknownLengthBodyParser;
 import jdk.internal.net.http.common.Log;
 import jdk.internal.net.http.common.Logger;
 import jdk.internal.net.http.common.MinimalFuture;
@@ -67,6 +68,7 @@ class Http1Response<T> {
     private final BodyReader bodyReader; // used to read the body
     private final Http1AsyncReceiver asyncReceiver;
     private volatile EOFException eof;
+    private volatile BodyParser bodyParser;
     // max number of bytes of (fixed length) body to ignore on redirect
     private final static int MAX_IGNORE = 1024;
 
@@ -230,6 +232,10 @@ class Http1Response<T> {
         return finished;
     }
 
+    /**
+     * Return known fixed content length or -1 if chunked, or -2 if no content-length
+     * information in which case, connection termination delimits the response body
+     */
     int fixupContentLen(int clen) {
         if (request.method().equalsIgnoreCase("HEAD") || responseCode == HTTP_NOT_MODIFIED) {
             return 0;
@@ -239,7 +245,11 @@ class Http1Response<T> {
                        .equalsIgnoreCase("chunked")) {
                 return -1;
             }
-            return 0;
+            if (responseCode == 101) {
+                // this is a h2c or websocket upgrade, contentlength must be zero
+                return 0;
+            }
+            return -2;
         }
         return clen;
     }
@@ -401,7 +411,7 @@ class Http1Response<T> {
                 // to prevent the SelectorManager thread from exiting until
                 // the body is fully read.
                 refCountTracker.acquire();
-                bodyReader.start(content.getBodyParser(
+                bodyParser = content.getBodyParser(
                     (t) -> {
                         try {
                             if (t != null) {
@@ -417,7 +427,8 @@ class Http1Response<T> {
                                 connection.close();
                             }
                         }
-                    }));
+                    });
+                bodyReader.start(bodyParser);
                 CompletableFuture<State> bodyReaderCF = bodyReader.completion();
                 asyncReceiver.subscribe(bodyReader);
                 assert bodyReaderCF != null : "parsing not started";
@@ -723,6 +734,11 @@ class Http1Response<T> {
 
         @Override
         public final void onReadError(Throwable t) {
+            if (t instanceof EOFException && bodyParser != null &&
+                    bodyParser instanceof UnknownLengthBodyParser) {
+                ((UnknownLengthBodyParser)bodyParser).complete();
+                return;
+            }
             t = wrapWithExtraDetail(t, parser::currentStateMessage);
             Http1Response.this.onReadError(t);
         }

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

@@ -75,6 +75,12 @@ class ResponseContent {
         if (chunkedContentInitialized) {
             return chunkedContent;
         }
+        if (contentLength == -2) {
+            // HTTP/1.0 content
+            chunkedContentInitialized = true;
+            chunkedContent = false;
+            return chunkedContent;
+        }
         if (contentLength == -1) {
             String tc = headers.firstValue("Transfer-Encoding")
                                .orElse("");
@@ -111,7 +117,9 @@ class ResponseContent {
         if (contentChunked()) {
             return new ChunkedBodyParser(onComplete);
         } else {
-            return new FixedLengthBodyParser(contentLength, onComplete);
+            return contentLength == -2
+                ? new UnknownLengthBodyParser(onComplete)
+                : new FixedLengthBodyParser(contentLength, onComplete);
         }
     }
 
@@ -392,6 +400,79 @@ class ResponseContent {
 
     }
 
+    class UnknownLengthBodyParser implements BodyParser {
+        final Consumer<Throwable> onComplete;
+        final Logger debug = Utils.getDebugLogger(this::dbgString, Utils.DEBUG);
+        final String dbgTag = ResponseContent.this.dbgTag + "/UnknownLengthBodyParser";
+        volatile Throwable closedExceptionally;
+        volatile AbstractSubscription sub;
+        volatile int breceived = 0;
+
+        UnknownLengthBodyParser(Consumer<Throwable> onComplete) {
+            this.onComplete = onComplete;
+        }
+
+        String dbgString() {
+            return dbgTag;
+        }
+
+        @Override
+        public void onSubscribe(AbstractSubscription sub) {
+            if (debug.on())
+                debug.log("onSubscribe: " + pusher.getClass().getName());
+            pusher.onSubscribe(this.sub = sub);
+        }
+
+        @Override
+        public String currentStateMessage() {
+            return format("http1_0 content, bytes received: %d", breceived);
+        }
+
+        @Override
+        public void accept(ByteBuffer b) {
+            if (closedExceptionally != null) {
+                if (debug.on())
+                    debug.log("already closed: " + closedExceptionally);
+                return;
+            }
+            boolean completed = false;
+            try {
+                if (debug.on())
+                    debug.log("Parser got %d bytes ", b.remaining());
+
+                if (b.hasRemaining()) {
+                    // only reduce demand if we actually push something.
+                    // we would not have come here if there was no
+                    // demand.
+                    boolean hasDemand = sub.demand().tryDecrement();
+                    assert hasDemand;
+                    breceived += b.remaining();
+                    pusher.onNext(List.of(b.asReadOnlyBuffer()));
+                }
+            } catch (Throwable t) {
+                if (debug.on()) debug.log("Unexpected exception", t);
+                closedExceptionally = t;
+                if (!completed) {
+                    onComplete.accept(t);
+                }
+            }
+        }
+
+        /**
+         * Must be called externally when connection has closed
+         * and therefore no more bytes can be read
+         */
+        public void complete() {
+            // We're done! All data has been received.
+            if (debug.on())
+                debug.log("Parser got all expected bytes: completing");
+            assert closedExceptionally == null;
+            onFinished.run();
+            pusher.onComplete();
+            onComplete.accept(closedExceptionally); // should be null
+        }
+    }
+
     class FixedLengthBodyParser implements BodyParser {
         final int contentLength;
         final Consumer<Throwable> onComplete;