8175814: Update default HttpClient protocol version and optional request version

Reviewed-by: chegar, dfuchs

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/AbstractPushPublisher.java

@@ -54,4 +54,4 @@ abstract class AbstractPushPublisher<T> implements Flow.Publisher<T> {
         }
     }
 
-        }
+}

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/AsyncConnection.java

@@ -70,6 +70,11 @@ interface AsyncConnection {
      */
     void startReading();
 
+    /**
+     * Cancel asynchronous reading. Used to downgrade a HTTP/2 connection to HTTP/1
+     */
+    void stopAsyncReading();
+
     /**
      * In async mode, this method puts buffers at the end of the send queue.
      * When in async mode, calling this method should later be followed by
@@ -79,6 +84,11 @@ interface AsyncConnection {
      */
     void writeAsync(ByteBufferReference[] buffers) throws IOException;
 
+    /**
+     * Re-enable asynchronous reads through the callback
+     */
+    void enableCallback();
+
     /**
      * In async mode, this method may put buffers at the beginning of send queue,
      * breaking frames sequence and allowing to write these buffers before other
@@ -99,5 +109,4 @@ interface AsyncConnection {
      * and continue execution.
      */
     void flushAsync() throws IOException;
-
 }

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/AsyncSSLConnection.java

@@ -32,6 +32,7 @@ import java.nio.channels.SocketChannel;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
+import javax.net.ssl.SSLEngine;
 
 import jdk.incubator.http.internal.common.ByteBufferReference;
 import jdk.incubator.http.internal.common.ExceptionallyCloseable;
@@ -44,33 +45,48 @@ class AsyncSSLConnection extends HttpConnection
                          implements AsyncConnection, ExceptionallyCloseable {
 
     final AsyncSSLDelegate sslDelegate;
-    final PlainHttpConnection delegate;
+    final PlainHttpConnection plainConnection;
 
     AsyncSSLConnection(InetSocketAddress addr, HttpClientImpl client, String[] ap) {
         super(addr, client);
-        delegate = new PlainHttpConnection(addr, client);
+        plainConnection = new PlainHttpConnection(addr, client);
-        sslDelegate = new AsyncSSLDelegate(delegate, client, ap);
+        sslDelegate = new AsyncSSLDelegate(plainConnection, client, ap);
     }
 
     @Override
     synchronized void configureMode(Mode mode) throws IOException {
         super.configureMode(mode);
-        delegate.configureMode(mode);
+        plainConnection.configureMode(mode);
+    }
+
+    private CompletableFuture<Void> configureModeAsync(Void ignore) {
+        CompletableFuture<Void> cf = new CompletableFuture<>();
+        try {
+            configureMode(Mode.ASYNC);
+            cf.complete(null);
+        } catch (Throwable t) {
+            cf.completeExceptionally(t);
+        }
+        return cf;
     }
 
     @Override
     public void connect() throws IOException, InterruptedException {
-        delegate.connect();
+        plainConnection.connect();
+        configureMode(Mode.ASYNC);
+        startReading();
+        sslDelegate.connect();
     }
 
     @Override
     public CompletableFuture<Void> connectAsync() {
-        return delegate.connectAsync();
+        // not used currently
+        throw new InternalError();
     }
 
     @Override
     boolean connected() {
-        return delegate.connected();
+        return plainConnection.connected() && sslDelegate.connected();
     }
 
     @Override
@@ -85,7 +101,12 @@ class AsyncSSLConnection extends HttpConnection
 
     @Override
     SocketChannel channel() {
-        return delegate.channel();
+        return plainConnection.channel();
+    }
+
+    @Override
+    public void enableCallback() {
+        sslDelegate.enableCallback();
     }
 
     @Override
@@ -131,22 +152,26 @@ class AsyncSSLConnection extends HttpConnection
 
     @Override
     public void closeExceptionally(Throwable cause) {
-        Utils.close(cause, sslDelegate, delegate.channel());
+        Utils.close(cause, sslDelegate, plainConnection.channel());
     }
 
     @Override
     public void close() {
-        Utils.close(sslDelegate, delegate.channel());
+        Utils.close(sslDelegate, plainConnection.channel());
     }
 
     @Override
     void shutdownInput() throws IOException {
-        delegate.channel().shutdownInput();
+        plainConnection.channel().shutdownInput();
     }
 
     @Override
     void shutdownOutput() throws IOException {
-        delegate.channel().shutdownOutput();
+        plainConnection.channel().shutdownOutput();
+    }
+
+    SSLEngine getEngine() {
+        return sslDelegate.getEngine();
     }
 
     @Override
@@ -154,7 +179,7 @@ class AsyncSSLConnection extends HttpConnection
                                   Consumer<Throwable> errorReceiver,
                                   Supplier<ByteBufferReference> readBufferSupplier) {
         sslDelegate.setAsyncCallbacks(asyncReceiver, errorReceiver, readBufferSupplier);
-        delegate.setAsyncCallbacks(sslDelegate::asyncReceive, errorReceiver, sslDelegate::getNetBuffer);
+        plainConnection.setAsyncCallbacks(sslDelegate::asyncReceive, errorReceiver, sslDelegate::getNetBuffer);
     }
 
     // Blocking read functions not used here
@@ -176,7 +201,12 @@ class AsyncSSLConnection extends HttpConnection
 
     @Override
     public void startReading() {
-        delegate.startReading();
+        plainConnection.startReading();
         sslDelegate.startReading();
     }
+
+    @Override
+    public void stopAsyncReading() {
+        plainConnection.stopAsyncReading();
+    }
 }

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/AsyncSSLDelegate.java

@@ -28,8 +28,10 @@ package jdk.incubator.http;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
@@ -72,13 +74,13 @@ import jdk.incubator.http.internal.common.ExceptionallyCloseable;
  *     channelInputQ
  *        /\
  *        ||
- * "lowerRead" method puts buffers into channelInputQ. It is invoked from
+ * "asyncReceive" method puts buffers into channelInputQ. It is invoked from
  * OP_READ events from the selector.
  *
  * Whenever handshaking is required, the doHandshaking() method is called
  * which creates a thread to complete the handshake. It takes over the
  * channelInputQ from upperRead, and puts outgoing packets on channelOutputQ.
- * Selector events are delivered to lowerRead and lowerWrite as normal.
+ * Selector events are delivered to asyncReceive and lowerWrite as normal.
  *
  * Errors
  *
@@ -92,9 +94,6 @@ class AsyncSSLDelegate implements ExceptionallyCloseable, AsyncConnection {
     // while SSL handshaking happening.
     final AsyncWriteQueue appOutputQ = new AsyncWriteQueue(this::upperWrite);
 
-    // queue of wrapped ByteBuffers waiting to be sent on socket channel
-    //final Queue<ByteBuffer> channelOutputQ;
-
     // Bytes read into this queue before being unwrapped. Backup on this
     // Q should only happen when the engine is stalled due to delegated tasks
     final Queue<ByteBufferReference> channelInputQ;
@@ -107,35 +106,34 @@ class AsyncSSLDelegate implements ExceptionallyCloseable, AsyncConnection {
 
     final SSLEngine engine;
     final SSLParameters sslParameters;
-    //final SocketChannel chan;
     final HttpConnection lowerOutput;
     final HttpClientImpl client;
     // should be volatile to provide proper synchronization(visibility) action
     volatile Consumer<ByteBufferReference> asyncReceiver;
     volatile Consumer<Throwable> errorHandler;
+    volatile boolean connected = false;
 
     // Locks.
     final Object reader = new Object();
     // synchronizing handshake state
     final Semaphore handshaker = new Semaphore(1);
-    // flag set when frame or writer is blocked waiting for handshake to finish
+    final String[] alpn;
-    //boolean writerBlocked;
-    //boolean readerBlocked;
 
     // alpn[] may be null. upcall is callback which receives incoming decoded bytes off socket
 
     AsyncSSLDelegate(HttpConnection lowerOutput, HttpClientImpl client, String[] alpn)
     {
         SSLContext context = client.sslContext();
-        //channelOutputQ = new Queue<>();
-        //channelOutputQ.registerPutCallback(this::lowerWrite);
         engine = context.createSSLEngine();
         engine.setUseClientMode(true);
         SSLParameters sslp = client.sslParameters()
                                    .orElseGet(context::getSupportedSSLParameters);
         sslParameters = Utils.copySSLParameters(sslp);
         if (alpn != null) {
+            Log.logSSL("AsyncSSLDelegate: Setting application protocols: " + Arrays.toString(alpn));
             sslParameters.setApplicationProtocols(alpn);
+        } else {
+            Log.logSSL("AsyncSSLDelegate: no applications set!");
         }
         logParams(sslParameters);
         engine.setSSLParameters(sslParameters);
@@ -143,6 +141,7 @@ class AsyncSSLDelegate implements ExceptionallyCloseable, AsyncConnection {
         this.client = client;
         this.channelInputQ = new Queue<>();
         this.channelInputQ.registerPutCallback(this::upperRead);
+        this.alpn = alpn;
     }
 
     @Override
@@ -162,6 +161,10 @@ class AsyncSSLDelegate implements ExceptionallyCloseable, AsyncConnection {
         }
     }
 
+    SSLEngine getEngine() {
+        return engine;
+    }
+
     @Override
     public void closeExceptionally(Throwable t) {
         Utils.close(t, appOutputQ, channelInputQ, lowerOutput);
@@ -223,6 +226,18 @@ class AsyncSSLDelegate implements ExceptionallyCloseable, AsyncConnection {
         }
     }
 
+    // Connecting at this level means the initial handshake has completed.
+    // This means that the initial SSL parameters are available including
+    // ALPN result.
+    void connect() throws IOException, InterruptedException {
+        doHandshakeNow("Init");
+        connected = true;
+    }
+
+    boolean connected() {
+        return connected;
+    }
+
     private void startHandshake(String tag) {
         Runnable run = () -> {
             try {
@@ -241,22 +256,28 @@ class AsyncSSLDelegate implements ExceptionallyCloseable, AsyncConnection {
     {
         handshaker.acquire();
         try {
-            channelInputQ.registerPutCallback(null);
+            channelInputQ.disableCallback();
             lowerOutput.flushAsync();
             Log.logTrace("{0}: Starting handshake...", tag);
             doHandshakeImpl();
             Log.logTrace("{0}: Handshake completed", tag);
-            channelInputQ.registerPutCallback(this::upperRead);
+            // don't unblock the channel here, as we aren't sure yet, whether ALPN
+            // negotiation succeeded. Caller will call enableCallback() externally
         } finally {
             handshaker.release();
         }
     }
 
+    public void enableCallback() {
+        channelInputQ.enableCallback();
+    }
+
      /**
      * Executes entire handshake in calling thread.
      * Returns after handshake is completed or error occurs
      */
     private void doHandshakeImpl() throws IOException {
+        engine.beginHandshake();
         while (true) {
             SSLEngineResult.HandshakeStatus status = engine.getHandshakeStatus();
             switch(status) {
@@ -272,7 +293,9 @@ class AsyncSSLDelegate implements ExceptionallyCloseable, AsyncConnection {
                 case NEED_UNWRAP: case NEED_UNWRAP_AGAIN:
                     handshakeReceiveAndUnWrap();
                     break;
-                case FINISHED: case NOT_HANDSHAKING:
+                case FINISHED:
+                    return;
+                case NOT_HANDSHAKING:
                     return;
                 default:
                     throw new InternalError("Unexpected Handshake Status: "
@@ -311,6 +334,12 @@ class AsyncSSLDelegate implements ExceptionallyCloseable, AsyncConnection {
         // maybe this class does not need to implement AsyncConnection
     }
 
+    @Override
+    public void stopAsyncReading() {
+        // maybe this class does not need to implement AsyncConnection
+    }
+
+
     static class EngineResult {
         final SSLEngineResult result;
         final ByteBufferReference destBuffer;

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/ConnectionPool.java

@@ -111,7 +111,8 @@ final class ConnectionPool {
     }
 
     static CacheKey cacheKey(InetSocketAddress destination,
-                             InetSocketAddress proxy) {
+                             InetSocketAddress proxy)
+    {
         return new CacheKey(destination, proxy);
     }
 

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/Exchange.java

@@ -549,7 +549,7 @@ final class Exchange<T> {
     }
 
     HttpClient.Version version() {
-        return client.version();
+        return multi.version();
     }
 
     private static SocketPermission getSocketPermissionFor(URI url) {

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/ExchangeImpl.java

@@ -81,7 +81,17 @@ abstract class ExchangeImpl<T> {
         } else {
             Http2ClientImpl c2 = exchange.client().client2(); // TODO: improve
             HttpRequestImpl request = exchange.request();
-            Http2Connection c = c2.getConnectionFor(request);
+            Http2Connection c;
+            try {
+                c = c2.getConnectionFor(request);
+            } catch (Http2Connection.ALPNException e) {
+                // failed to negotiate "h2"
+                AsyncSSLConnection as = e.getConnection();
+                as.stopAsyncReading();
+                SSLConnection sslc = new SSLConnection(as);
+                ExchangeImpl<U> ex = new Http1Exchange<>(exchange, sslc);
+                return ex;
+            }
             if (c == null) {
                 // no existing connection. Send request with HTTP 1 and then
                 // upgrade if successful

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/Http2ClientImpl.java

@@ -104,7 +104,15 @@ class Http2ClientImpl {
             return connection;
         }
         // we are opening the connection here blocking until it is done.
-        connection = new Http2Connection(req, this);
+        try {
+            connection = new Http2Connection(req, this);
+        } catch (Throwable t) {
+            synchronized (opening) {
+                opening.remove(key);
+                opening.notifyAll();
+            }
+            throw t;
+        }
         synchronized (opening) {
             connections.put(key, connection);
             opening.remove(key);

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/Http2Connection.java

@@ -43,6 +43,7 @@ import java.util.Formatter;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.stream.Collectors;
+import javax.net.ssl.SSLEngine;
 import jdk.incubator.http.internal.common.*;
 import jdk.incubator.http.internal.frame.*;
 import jdk.incubator.http.internal.hpack.Encoder;
@@ -82,8 +83,6 @@ import static jdk.incubator.http.internal.frame.SettingsFrame.*;
  * stream are provided by calling Stream.incoming().
  */
 class Http2Connection  {
-
-
     /*
      *  ByteBuffer pooling strategy for HTTP/2 protocol:
      *
@@ -258,15 +257,46 @@ class Http2Connection  {
                 keyFor(request.uri(), request.proxy(h2client.client())));
         Log.logTrace("Connection send window size {0} ", windowController.connectionWindowSize());
 
-        connection.connect();
         // start reading
         AsyncConnection asyncConn = (AsyncConnection)connection;
         asyncConn.setAsyncCallbacks(this::asyncReceive, this::shutdown, this::getReadBuffer);
-        connection.configureMode(Mode.ASYNC); // set mode only AFTER setAsyncCallbacks to provide visibility.
+        connection.connect();
-        asyncConn.startReading();
+        checkSSLConfig();
+        // safe to resume async reading now.
+        asyncConn.enableCallback();
         sendConnectionPreface();
     }
 
+    /**
+     * Throws an IOException if h2 was not negotiated
+     */
+    private void checkSSLConfig() throws IOException {
+        AsyncSSLConnection aconn = (AsyncSSLConnection)connection;
+        SSLEngine engine = aconn.getEngine();
+        String alpn = engine.getApplicationProtocol();
+        if (alpn == null || !alpn.equals("h2")) {
+            String msg;
+            if (alpn == null) {
+                Log.logSSL("ALPN not supported");
+                msg = "ALPN not supported";
+            } else switch (alpn) {
+              case "":
+                Log.logSSL("No ALPN returned");
+                msg = "No ALPN negotiated";
+                break;
+              case "http/1.1":
+                Log.logSSL("HTTP/1.1 ALPN returned");
+                msg = "HTTP/1.1 ALPN returned";
+                break;
+              default:
+                Log.logSSL("unknown ALPN returned");
+                msg = "Unexpected ALPN: " + alpn;
+                throw new IOException(msg);
+            }
+            throw new ALPNException(msg, aconn);
+        }
+    }
+
     static String keyFor(HttpConnection connection) {
         boolean isProxy = connection.isProxied();
         boolean isSecure = connection.isSecure();
@@ -858,4 +888,21 @@ class Http2Connection  {
             return 0;
         }
     }
+
+    /**
+     * Thrown when https handshake negotiates http/1.1 alpn instead of h2
+     */
+    static final class ALPNException extends IOException {
+        private static final long serialVersionUID = 23138275393635783L;
+        final AsyncSSLConnection connection;
+
+        ALPNException(String msg, AsyncSSLConnection connection) {
+            super(msg);
+            this.connection = connection;
+        }
+
+        AsyncSSLConnection getConnection() {
+            return connection;
+        }
+    }
 }

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/HttpClient.java

@@ -154,9 +154,9 @@ public abstract class HttpClient {
 
         /**
          * Requests a specific HTTP protocol version where possible. If not set,
-         * the version defaults to {@link HttpClient.Version#HTTP_1_1}. If
+         * the version defaults to {@link HttpClient.Version#HTTP_2}. If
          * {@link HttpClient.Version#HTTP_2} is set, then each request will
-         * attempt to upgrade to HTTP/2.  If the upgrade succeeds, then the
+         * attempt to upgrade to HTTP/2. If the upgrade succeeds, then the
          * response to this request will use HTTP/2 and all subsequent requests
          * and responses to the same
          * <a href="https://tools.ietf.org/html/rfc6454#section-4">origin server</a>
@@ -267,7 +267,7 @@ public abstract class HttpClient {
 
     /**
      * Returns the HTTP protocol version requested for this client. The default
-     * value is {@link HttpClient.Version#HTTP_1_1}
+     * value is {@link HttpClient.Version#HTTP_2}
      *
      * @return the HTTP protocol version requested
      */

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/HttpClientBuilderImpl.java

@@ -40,7 +40,7 @@ class HttpClientBuilderImpl extends HttpClient.Builder {
     HttpClient.Redirect followRedirects;
     ProxySelector proxy;
     Authenticator authenticator;
-    HttpClient.Version version = HttpClient.Version.HTTP_1_1;
+    HttpClient.Version version;
     Executor executor;
     // Security parameters
     SSLContext sslContext;

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/HttpClientImpl.java

@@ -125,7 +125,11 @@ class HttpClientImpl extends HttpClient {
                 Redirect.NEVER : builder.followRedirects;
         this.proxySelector = builder.proxy;
         authenticator = builder.authenticator;
-        version = builder.version;
+        if (builder.version == null) {
+            version = HttpClient.Version.HTTP_2;
+        } else {
+            version = builder.version;
+        }
         if (builder.sslParams == null) {
             sslParams = getDefaultParams(sslContext);
         } else {

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/HttpConnection.java

@@ -152,15 +152,16 @@ abstract class HttpConnection implements Closeable {
             HttpClientImpl client,
             HttpRequestImpl request, boolean isHttp2)
     {
-        HttpConnection c;
+        HttpConnection c = null;
         InetSocketAddress proxy = request.proxy(client);
         boolean secure = request.secure();
         ConnectionPool pool = client.connectionPool();
         String[] alpn =  null;
 
-        if (secure && client.version() == HttpClient.Version.HTTP_2) {
+        if (secure && isHttp2) {
-            alpn = new String[1];
+            alpn = new String[2];
             alpn[0] = "h2";
+            alpn[1] = "http/1.1";
         }
 
         if (!secure) {
@@ -171,7 +172,9 @@ abstract class HttpConnection implements Closeable {
                 return getPlainConnection(addr, proxy, request, client);
             }
         } else {
-            c = pool.getConnection(true, addr, proxy);
+            if (!isHttp2) { // if http2 we don't cache connections
+                c = pool.getConnection(true, addr, proxy);
+            }
             if (c != null) {
                 return c;
             } else {

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/HttpRequest.java

@@ -303,10 +303,11 @@ public abstract class HttpRequest {
         public abstract Builder expectContinue(boolean enable);
 
         /**
-         * Overrides the {@link HttpClient#version()  } setting for this
+         * Sets the preferred {@link HttpClient.Version} for this
-         * request. This sets the version requested. The corresponding
+         * request. The corresponding {@link HttpResponse} should be checked
-         * {@link HttpResponse} should be checked for the version that was
+         * for the version that was used. If the version is not set
-         * used.
+         * in a request, then the version requested will be that of the
+         * sending {@link HttpClient}.
          *
          * @param version the HTTP protocol version requested
          * @return this request builder
@@ -497,13 +498,16 @@ public abstract class HttpRequest {
     public abstract URI uri();
 
     /**
-     * Returns the HTTP protocol version that will be requested for this
+     * Returns an {@code Optional} containing the HTTP protocol version that
-     * {@code HttpRequest}. The corresponding {@link HttpResponse} should be
+     * will be requested for this {@code HttpRequest}. If the version was not
+     * set in the request's builder, then the {@code Optional} is empty.
+     * In that case, the version requested will be that of the sending
+     * {@link HttpClient}. The corresponding {@link HttpResponse} should be
      * queried to determine the version that was actually used.
      *
      * @return HTTP protocol version
      */
-    public abstract HttpClient.Version version();
+    public abstract Optional<HttpClient.Version> version();
 
     /**
      * The (user-accessible) request headers that this request was (or will be)

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/HttpRequestBuilderImpl.java

@@ -28,6 +28,7 @@ package jdk.incubator.http;
 import java.net.URI;
 import jdk.incubator.http.HttpRequest.BodyProcessor;
 import java.time.Duration;
+import java.util.Optional;
 import static java.util.Objects.requireNonNull;
 import jdk.incubator.http.internal.common.HttpHeadersImpl;
 import static jdk.incubator.http.internal.common.Utils.isValidName;
@@ -41,7 +42,7 @@ class HttpRequestBuilderImpl extends HttpRequest.Builder {
     //private HttpClient.Redirect followRedirects;
     private boolean expectContinue;
     private HttpRequest.BodyProcessor body;
-    private HttpClient.Version version;
+    private volatile Optional<HttpClient.Version> version;
     //private final HttpClientImpl client;
     //private ProxySelector proxy;
     private Duration duration;
@@ -52,10 +53,12 @@ class HttpRequestBuilderImpl extends HttpRequest.Builder {
         this.uri = uri;
         this.userHeaders = new HttpHeadersImpl();
         this.method = "GET"; // default, as per spec
+        this.version = Optional.empty();
     }
 
     public HttpRequestBuilderImpl() {
         this.userHeaders = new HttpHeadersImpl();
+        this.version = Optional.empty();
     }
 
     @Override
@@ -149,7 +152,7 @@ class HttpRequestBuilderImpl extends HttpRequest.Builder {
     @Override
     public HttpRequestBuilderImpl version(HttpClient.Version version) {
         requireNonNull(version);
-        this.version = version;
+        this.version = Optional.of(version);
         return this;
     }
 
@@ -169,7 +172,7 @@ class HttpRequestBuilderImpl extends HttpRequest.Builder {
 
     public HttpRequest.BodyProcessor body() { return body; }
 
-    HttpClient.Version version() { return version; }
+    Optional<HttpClient.Version> version() { return version; }
 
     @Override
     public HttpRequest.Builder GET() { return method("GET", null); }

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/HttpRequestImpl.java

@@ -52,7 +52,7 @@ class HttpRequestImpl extends HttpRequest implements WebSocketRequest {
     private boolean isWebSocket;
     private AccessControlContext acc;
     private final Duration duration;
-    private final HttpClient.Version version;
+    private final Optional<HttpClient.Version> version;
 
     /**
      * Creates an HttpRequestImpl from the given builder.
@@ -128,8 +128,8 @@ class HttpRequestImpl extends HttpRequest implements WebSocketRequest {
         this.authority = authority;
         this.secure = false;
         this.expectContinue = false;
-        this.duration = null; // block TODO: fix
+        this.duration = null;
-        this.version = client.version(); // TODO: ??
+        this.version = Optional.of(client.version());
     }
 
     /**
@@ -191,12 +191,6 @@ class HttpRequestImpl extends HttpRequest implements WebSocketRequest {
     @Override
     public boolean expectContinue() { return expectContinue; }
 
-    public boolean requestHttp2() {
-        return version.equals(HttpClient.Version.HTTP_2);
-    }
-
-//    AccessControlContext getAccessControlContext() { return acc; }
-
     InetSocketAddress proxy(HttpClientImpl client) {
         ProxySelector ps = client.proxy().orElse(null);
         if (ps == null) {
@@ -254,7 +248,7 @@ class HttpRequestImpl extends HttpRequest implements WebSocketRequest {
     HttpHeadersImpl getSystemHeaders() { return systemHeaders; }
 
     @Override
-    public HttpClient.Version version() { return version; }
+    public Optional<HttpClient.Version> version() { return version; }
 
     void addSystemHeader(String name, String value) {
         systemHeaders.addHeader(name, value);

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/MultiExchange.java

@@ -192,7 +192,7 @@ class MultiExchange<U,T> {
     }
 
     HttpClient.Version version() {
-        return client.version();
+        return request.version().orElse(client.version());
     }
 
     private synchronized void setExchange(Exchange<T> exchange) {

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/PlainHttpConnection.java

@@ -76,6 +76,11 @@ class PlainHttpConnection extends HttpConnection implements AsyncConnection {
         }
     }
 
+    @Override
+    public void stopAsyncReading() {
+        client.cancelRegistration(chan);
+    }
+
     class ConnectEvent extends AsyncEvent {
         CompletableFuture<Void> cf;
 
@@ -213,6 +218,12 @@ class PlainHttpConnection extends HttpConnection implements AsyncConnection {
         }
     }
 
+    @Override
+    public void enableCallback() {
+        // not used
+        assert false;
+    }
+
     void asyncOutput(ByteBufferReference[] refs, AsyncWriteQueue delayCallback) {
         try {
             ByteBuffer[] bufs = ByteBufferReference.toBuffers(refs);
@@ -246,7 +257,6 @@ class PlainHttpConnection extends HttpConnection implements AsyncConnection {
         closed = true;
         try {
             Log.logError("Closing: " + toString());
-            //System.out.println("Closing: " + this);
             chan.close();
         } catch (IOException e) {}
     }
@@ -272,7 +282,6 @@ class PlainHttpConnection extends HttpConnection implements AsyncConnection {
                 while (true) {
                     ByteBufferReference buf = readBufferSupplier.get();
                     int n = chan.read(buf.get());
-                    //System.err.printf("Read %d bytes from chan\n", n);
                     if (n == -1) {
                         throw new IOException();
                     }

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/SSLConnection.java

@@ -69,6 +69,18 @@ class SSLConnection extends HttpConnection {
         delegate = new PlainHttpConnection(addr, client);
     }
 
+    /**
+     * Create an SSLConnection from an existing connected AsyncSSLConnection.
+     * Used when downgrading from HTTP/2 to HTTP/1.1
+     */
+    SSLConnection(AsyncSSLConnection c) {
+        super(c.address, c.client);
+        this.delegate = c.plainConnection;
+        AsyncSSLDelegate adel = c.sslDelegate;
+        this.sslDelegate = new SSLDelegate(adel.engine, delegate.channel(), client);
+        this.alpn = adel.alpn;
+    }
+
     @Override
     SSLParameters sslParameters() {
         return sslDelegate.getSSLParameters();

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/SSLDelegate.java

@@ -51,6 +51,15 @@ class SSLDelegate {
     final SocketChannel chan;
     final HttpClientImpl client;
 
+    SSLDelegate(SSLEngine eng, SocketChannel chan, HttpClientImpl client)
+    {
+        this.engine = eng;
+        this.chan = chan;
+        this.client = client;
+        this.wrapper = new EngineWrapper(chan, engine);
+        this.sslParameters = engine.getSSLParameters();
+    }
+
     // alpn[] may be null
     SSLDelegate(SocketChannel chan, HttpClientImpl client, String[] alpn)
         throws IOException
@@ -63,9 +72,9 @@ class SSLDelegate {
         sslParameters = Utils.copySSLParameters(sslp);
         if (alpn != null) {
             sslParameters.setApplicationProtocols(alpn);
-            Log.logSSL(() -> "Setting application protocols: " + Arrays.toString(alpn));
+            Log.logSSL("SSLDelegate: Setting application protocols: {0}" + Arrays.toString(alpn));
         } else {
-            Log.logSSL("No application protocols proposed");
+            Log.logSSL("SSLDelegate: No application protocols proposed");
         }
         engine.setSSLParameters(sslParameters);
         wrapper = new EngineWrapper(chan, engine);
@@ -181,7 +190,7 @@ class SSLDelegate {
         boolean closed = false;
         int u_remaining; // the number of bytes left in unwrap_src after an unwrap()
 
-        EngineWrapper (SocketChannel chan, SSLEngine engine) throws IOException {
+        EngineWrapper (SocketChannel chan, SSLEngine engine) {
             this.chan = chan;
             this.engine = engine;
             wrapLock = new Object();

jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/internal/common/Queue.java

@@ -28,6 +28,7 @@ package jdk.incubator.http.internal.common;
 import java.io.IOException;
 import java.util.LinkedList;
 import java.util.stream.Stream;
+import java.util.Objects;
 
 // Each stream has one of these for input. Each Http2Connection has one
 // for output. Can be used blocking or asynchronously.
@@ -38,33 +39,9 @@ public class Queue<T> implements ExceptionallyCloseable  {
     private volatile boolean closed = false;
     private volatile Throwable exception = null;
     private Runnable callback;
-    private boolean forceCallback;
+    private boolean callbackDisabled = false;
     private int waiters; // true if someone waiting
 
-    public synchronized void putAll(T[] objs) throws IOException {
-        if (closed) {
-            throw new IOException("stream closed");
-        }
-        boolean wasEmpty = q.isEmpty();
-
-        for (T obj : objs) {
-            q.add(obj);
-        }
-
-        if (waiters > 0) {
-            notifyAll();
-        }
-
-        if (wasEmpty || forceCallback) {
-            forceCallback = false;
-            if (callback != null) {
-                // Note: calling callback while holding the lock is
-                // dangerous and may lead to deadlocks.
-                callback.run();
-           }
-        }
-    }
-
     public synchronized int size() {
         return q.size();
     }
@@ -81,17 +58,30 @@ public class Queue<T> implements ExceptionallyCloseable  {
         }
 
         q.add(obj);
+
         if (waiters > 0) {
             notifyAll();
         }
 
-        if (q.size() == 1 || forceCallback) {
+        if (callbackDisabled) {
-            forceCallback = false;
+            return;
-            if (callback != null) {
+        }
-                // Note: calling callback while holding the lock is
+
-                // dangerous and may lead to deadlocks.
+        if (q.size() > 0 && callback != null) {
-                callback.run();
+            // Note: calling callback while holding the lock is
-            }
+            // dangerous and may lead to deadlocks.
+            callback.run();
+        }
+    }
+
+    public synchronized void disableCallback() {
+        callbackDisabled = true;
+    }
+
+    public synchronized void enableCallback() {
+        callbackDisabled = false;
+        while (q.size() > 0) {
+            callback.run();
         }
     }
 
@@ -100,8 +90,9 @@ public class Queue<T> implements ExceptionallyCloseable  {
      * the Queue was empty.
      */
     public synchronized void registerPutCallback(Runnable callback) {
+        Objects.requireNonNull(callback);
         this.callback = callback;
-        if (callback != null && q.size() > 0) {
+        if (q.size() > 0) {
             // Note: calling callback while holding the lock is
             // dangerous and may lead to deadlocks.
             callback.run();
@@ -167,12 +158,10 @@ public class Queue<T> implements ExceptionallyCloseable  {
     }
 
     public synchronized void pushback(T v) {
-        forceCallback = true;
         q.addFirst(v);
     }
 
     public synchronized void pushbackAll(T[] v) {
-        forceCallback = true;
         for (int i=v.length-1; i>=0; i--) {
             q.addFirst(v[i]);
         }

jdk/test/java/net/httpclient/RequestBodyTest.java

@@ -104,6 +104,7 @@ public class RequestBodyTest {
         SSLContext ctx = LightWeightHttpServer.ctx;
         client = HttpClient.newBuilder()
                            .sslContext(ctx)
+                           .version(HttpClient.Version.HTTP_1_1)
                            .followRedirects(HttpClient.Redirect.ALWAYS)
                            .executor(exec)
                            .build();

jdk/test/java/net/httpclient/SmokeTest.java

@@ -165,6 +165,7 @@ public class SmokeTest {
         client = HttpClient.newBuilder()
                            .sslContext(ctx)
                            .executor(e)
+                           .version(HttpClient.Version.HTTP_1_1)
                            .sslParameters(sslparams)
                            .followRedirects(HttpClient.Redirect.ALWAYS)
                            .build();

jdk/test/java/net/httpclient/VersionTest.java

@@ -0,0 +1,140 @@
+/*
+ * Copyright (c) 2017, 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.
+ */
+
+/*
+ * @test
+ * @bug 8175814
+ * @modules jdk.incubator.httpclient java.logging jdk.httpserver
+ * @run main/othervm -Djdk.httpclient.HttpClient.log=errors,requests,headers,trace VersionTest
+ */
+
+import com.sun.net.httpserver.Headers;
+import com.sun.net.httpserver.HttpContext;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.URI;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutorService;
+import java.net.InetSocketAddress;
+import jdk.incubator.http.HttpClient;
+import jdk.incubator.http.HttpRequest;
+import jdk.incubator.http.HttpResponse;
+import static jdk.incubator.http.HttpRequest.BodyProcessor.fromString;
+import static jdk.incubator.http.HttpResponse.*;
+import static jdk.incubator.http.HttpResponse.BodyHandler.asString;
+import static jdk.incubator.http.HttpResponse.BodyHandler.discard;
+import static jdk.incubator.http.HttpClient.Version.HTTP_1_1;
+import static jdk.incubator.http.HttpClient.Version.HTTP_2;
+
+/**
+ */
+public class VersionTest {
+    static HttpServer s1 ;
+    static ExecutorService executor;
+    static int port;
+    static HttpClient client;
+    static URI uri;
+    static volatile boolean error = false;
+
+    public static void main(String[] args) throws Exception {
+        initServer();
+
+        client = HttpClient.newBuilder()
+                           .executor(executor)
+                           .build();
+        // first check that the version is HTTP/2
+        if (client.version() != HttpClient.Version.HTTP_2) {
+            throw new RuntimeException("Default version not HTTP_2");
+        }
+        try {
+            test(HTTP_1_1);
+            test(HTTP_2);
+        } finally {
+            s1.stop(0);
+            executor.shutdownNow();
+        }
+        if (error)
+            throw new RuntimeException();
+    }
+
+    public static void test(HttpClient.Version version) throws Exception {
+        HttpRequest r = HttpRequest.newBuilder(uri)
+                .version(version)
+                .GET()
+                .build();
+        HttpResponse<Void> resp = client.send(r, discard(null));
+        System.out.printf("Client: response is %d\n", resp.statusCode());
+        if (resp.version() != HTTP_1_1) {
+            throw new RuntimeException();
+        }
+        //System.out.printf("Client: response body is %s\n", resp.body());
+    }
+
+    static void initServer() throws Exception {
+        InetSocketAddress addr = new InetSocketAddress (0);
+        s1 = HttpServer.create (addr, 0);
+        HttpHandler h = new Handler();
+
+        HttpContext c1 = s1.createContext("/", h);
+
+        executor = Executors.newCachedThreadPool();
+        s1.setExecutor(executor);
+        s1.start();
+
+        port = s1.getAddress().getPort();
+        uri = new URI("http://127.0.0.1:" + Integer.toString(port) + "/foo");
+        System.out.println("HTTP server port = " + port);
+    }
+}
+
+class Handler implements HttpHandler {
+    int counter = 0;
+
+    void checkHeader(Headers h) {
+        counter++;
+        if (counter == 1 && h.containsKey("Upgrade")) {
+            VersionTest.error = true;
+        }
+        if (counter > 1 && !h.containsKey("Upgrade")) {
+            VersionTest.error = true;
+        }
+    }
+
+    @Override
+    public synchronized void handle(HttpExchange t)
+        throws IOException
+    {
+        String reply = "Hello world";
+        int len = reply.length();
+        Headers h = t.getRequestHeaders();
+        checkHeader(h);
+        System.out.printf("Sending response 200\n");
+        t.sendResponseHeaders(200, len);
+        OutputStream o = t.getResponseBody();
+        o.write(reply.getBytes());
+        t.close();
+    }
+}

jdk/test/java/net/httpclient/http2/ErrorTest.java

@@ -31,7 +31,7 @@
  *          jdk.incubator.httpclient/jdk.incubator.http.internal.frame
  *          jdk.incubator.httpclient/jdk.incubator.http.internal.hpack
  *          java.security.jgss
- * @run testng/othervm -Djdk.httpclient.HttpClient.log=ssl,errors ErrorTest
+ * @run testng/othervm/timeout=60 -Djavax.net.debug=ssl -Djdk.httpclient.HttpClient.log=all ErrorTest
  * @summary check exception thrown when bad TLS parameters selected
  */
 
@@ -76,10 +76,13 @@ public class ErrorTest {
 
         Http2TestServer httpsServer = null;
         try {
+            SSLContext serverContext = (new SimpleSSLContext()).get();
+            SSLParameters p = serverContext.getSupportedSSLParameters();
+            p.setApplicationProtocols(new String[]{"h2"});
             httpsServer = new Http2TestServer(true,
                                               0,
                                               exec,
-                                              sslContext);
+                                              serverContext);
             httpsServer.addHandler(new EchoHandler(), "/");
             int httpsPort = httpsServer.getAddress().getPort();
             String httpsURIString = "https://127.0.0.1:" + httpsPort + "/bar/";

jdk/test/java/net/httpclient/http2/Timeout.java

@@ -32,6 +32,7 @@ import java.time.Duration;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.CompletionException;
 import javax.net.ssl.SSLServerSocket;
+import javax.net.ssl.SSLParameters;
 import javax.net.ssl.SSLServerSocketFactory;
 import javax.net.ssl.SSLSocket;
 import static jdk.incubator.http.HttpRequest.BodyProcessor.fromString;
@@ -75,6 +76,9 @@ public class Timeout {
             Thread server = new Thread(() -> {
                 while (true) {
                     System.out.println("server: ready");
+                    SSLParameters params = ssocket.getSSLParameters();
+                    params.setApplicationProtocols(new String[]{"h2"});
+                    ssocket.setSSLParameters(params);
                     ready = true;
                     try (SSLSocket socket = (SSLSocket) ssocket.accept()) {