Merge pull request #5760 from eclipse-vertx/http-client-decoupling

Decouple HttpClient from HTTP/1.x and H2

Author: vietj

vertx-core/src/main/java/io/vertx/core/http/WebSocketFrame.java

@@ -13,9 +13,8 @@
 
 import io.vertx.codegen.annotations.CacheReturn;
 import io.vertx.codegen.annotations.DataObject;
-import io.vertx.codegen.annotations.VertxGen;
 import io.vertx.core.buffer.Buffer;
-import io.vertx.core.http.impl.ws.WebSocketFrameImpl;
+import io.vertx.core.http.impl.websocket.WebSocketFrameImpl;
 
 /**
  * A WebSocket frame that represents either text or binary data.

vertx-core/src/main/java/io/vertx/core/http/impl/CleanableHttpClient.java

@@ -91,11 +91,6 @@ public VertxInternal vertx() {
     return delegate.vertx();
   }
 
-  @Override
-  public HttpClientOptions options() {
-    return delegate.options();
-  }
-
   @Override
   public boolean isMetricsEnabled() {
     return delegate.isMetricsEnabled();
@@ -112,8 +107,8 @@ public Function<HttpClientResponse, Future<RequestOptions>> redirectHandler() {
   }
 
   @Override
-  public NetClientInternal netClient() {
-    return delegate.netClient();
+  public HttpChannelConnector channelConnector() {
+    return delegate.channelConnector();
   }
 
   @Override

vertx-core/src/main/java/io/vertx/core/http/impl/Http1xOrH2ChannelConnector.java

@@ -24,6 +24,8 @@
 import io.vertx.core.http.HttpClientOptions;
 import io.vertx.core.http.HttpHeaders;
 import io.vertx.core.http.HttpVersion;
+import io.vertx.core.http.impl.http1x.Http1xClientConnection;
+import io.vertx.core.http.impl.http1x.Http2UpgradeClientConnection;
 import io.vertx.core.http.impl.http2.Http2ClientChannelInitializer;
 import io.vertx.core.http.impl.http2.codec.Http2CodecClientChannelInitializer;
 import io.vertx.core.http.impl.http2.multiplex.Http2MultiplexClientChannelInitializer;
@@ -37,6 +39,7 @@
 import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.core.spi.metrics.HttpClientMetrics;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -51,16 +54,36 @@
  */
 public class Http1xOrH2ChannelConnector implements HttpChannelConnector {
 
+  private final HttpClientOptions options;
   private final HttpClientMetrics clientMetrics;
   private final NetClientInternal netClient;
 
   public Http1xOrH2ChannelConnector(NetClientInternal netClient,
+                                    HttpClientOptions options,
                                     HttpClientMetrics clientMetrics) {
+
+    if (!options.isKeepAlive() && options.isPipelining()) {
+      throw new IllegalStateException("Cannot have pipelining with no keep alive");
+    }
+    List<HttpVersion> alpnVersions = options.getAlpnVersions();
+    if (alpnVersions == null || alpnVersions.isEmpty()) {
+      if (options.getProtocolVersion() == HttpVersion.HTTP_2) {
+        options.setAlpnVersions(List.of(HttpVersion.HTTP_2, HttpVersion.HTTP_1_1));
+      } else {
+        options.setAlpnVersions(List.of(options.getProtocolVersion()));
+      }
+    }
+
     this.clientMetrics = clientMetrics;
+    this.options = options;
     this.netClient = netClient;
   }
 
-  private Http2ClientChannelInitializer http2Initializer(HttpClientOptions options) {
+  public NetClientInternal netClient() {
+    return netClient;
+  }
+
+  private Http2ClientChannelInitializer http2Initializer() {
     if (options.getHttp2MultiplexImplementation()) {
       return new Http2MultiplexClientChannelInitializer(
         HttpUtils.fromVertxSettings(options.getInitialSettings()),
@@ -80,14 +103,14 @@ private void connect(ContextInternal context, HttpConnectParams params, HostAndP
     if (authority != null) {
       connectOptions.setHost(authority.host());
       connectOptions.setPort(authority.port());
-      if (params.ssl && params.options.isForceSni()) {
+      if (params.ssl && options.isForceSni()) {
         connectOptions.setSniServerName(authority.host());
       }
     }
     connectOptions.setSsl(params.ssl);
     if (params.ssl) {
       if (params.sslOptions != null) {
-        connectOptions.setSslOptions(params.sslOptions.copy().setUseAlpn(params.useAlpn));
+        connectOptions.setSslOptions(params.sslOptions.copy().setUseAlpn(options.isUseAlpn()));
       } else {
         connectOptions.setSslOptions(new ClientSSLOptions().setHostnameVerificationAlgorithm("HTTPS"));
       }
@@ -116,40 +139,45 @@ public Future<HttpClientConnection> wrap(ContextInternal context, HttpConnectPar
     Channel ch = so.channelHandlerContext().channel();
     if (params.ssl) {
       String protocol = so.applicationLayerProtocol();
-      if (params.useAlpn) {
+      if (options.isUseAlpn()) {
         if ("h2".equals(protocol)) {
-          applyHttp2ConnectionOptions(ch.pipeline(), params.options);
-          Http2ClientChannelInitializer http2ChannelInitializer = http2Initializer(params.options);
+          applyHttp2ConnectionOptions(ch.pipeline());
+          Http2ClientChannelInitializer http2ChannelInitializer = http2Initializer();
           http2ChannelInitializer.http2Connected(context, authority, metric, maxLifetimeMillis, ch, metrics, promise);
         } else {
-          applyHttp1xConnectionOptions(ch.pipeline(), params.options);
+          applyHttp1xConnectionOptions(ch.pipeline());
           HttpVersion fallbackProtocol = "http/1.0".equals(protocol) ?
             HttpVersion.HTTP_1_0 : HttpVersion.HTTP_1_1;
-          http1xConnected(params.options, fallbackProtocol, server, authority, true, context, metric, maxLifetimeMillis, ch, metrics, promise);
+          http1xConnected(fallbackProtocol, server, authority, true, context, metric, maxLifetimeMillis, ch, metrics, promise);
         }
       } else {
-        applyHttp1xConnectionOptions(ch.pipeline(), params.options);
-        http1xConnected(params.options, params.version, server, authority, true, context, metric, maxLifetimeMillis, ch, metrics, promise);
+        applyHttp1xConnectionOptions(ch.pipeline());
+        http1xConnected(options.getProtocolVersion(), server, authority, true, context, metric, maxLifetimeMillis, ch, metrics, promise);
       }
     } else {
-      if (params.version == HttpVersion.HTTP_2) {
-        if (params.options.isHttp2ClearTextUpgrade()) {
-          applyHttp1xConnectionOptions(pipeline, params.options);
-          http1xConnected(params.options, params.version, server, authority, false, context, metric, maxLifetimeMillis, ch, metrics, promise);
+      if (options.getProtocolVersion() == HttpVersion.HTTP_2) {
+        if (options.isHttp2ClearTextUpgrade()) {
+          applyHttp1xConnectionOptions(pipeline);
+          http1xConnected(options.getProtocolVersion(), server, authority, false, context, metric, maxLifetimeMillis, ch, metrics, promise);
         } else {
-          applyHttp2ConnectionOptions(pipeline, params.options);
-          Http2ClientChannelInitializer http2ChannelInitializer = http2Initializer(params.options);
+          applyHttp2ConnectionOptions(pipeline);
+          Http2ClientChannelInitializer http2ChannelInitializer = http2Initializer();
           http2ChannelInitializer.http2Connected(context, authority, metric, maxLifetimeMillis, ch, metrics, promise);
         }
       } else {
-        applyHttp1xConnectionOptions(pipeline, params.options);
-        http1xConnected(params.options, params.version, server, authority, false, context, metric, maxLifetimeMillis, ch, metrics, promise);
+        applyHttp1xConnectionOptions(pipeline);
+        http1xConnected(options.getProtocolVersion(), server, authority, false, context, metric, maxLifetimeMillis, ch, metrics, promise);
       }
     }
     return promise.future();
   }
 
   public Future<HttpClientConnection> httpConnect(ContextInternal context, SocketAddress server, HostAndPort authority, HttpConnectParams params, long maxLifetimeMillis, ClientMetrics<?, ?, ?> metrics) {
+
+    if (!options.isUseAlpn() && params.ssl && this.options.getProtocolVersion() == HttpVersion.HTTP_2) {
+      return context.failedFuture("Must enable ALPN when using H2");
+    }
+
     Promise<NetSocket> promise = context.promise();
     Future<NetSocket> future = promise.future();
     // We perform the compose operation before calling connect to be sure that the composition happens
@@ -159,7 +187,7 @@ public Future<HttpClientConnection> httpConnect(ContextInternal context, SocketA
     return ret;
   }
 
-  private void applyHttp2ConnectionOptions(ChannelPipeline pipeline, HttpClientOptions options) {
+  private void applyHttp2ConnectionOptions(ChannelPipeline pipeline) {
     int idleTimeout = options.getIdleTimeout();
     int readIdleTimeout = options.getReadIdleTimeout();
     int writeIdleTimeout = options.getWriteIdleTimeout();
@@ -168,7 +196,7 @@ private void applyHttp2ConnectionOptions(ChannelPipeline pipeline, HttpClientOpt
     }
   }
 
-  private void applyHttp1xConnectionOptions(ChannelPipeline pipeline, HttpClientOptions options) {
+  private void applyHttp1xConnectionOptions(ChannelPipeline pipeline) {
     int idleTimeout = options.getIdleTimeout();
     int readIdleTimeout = options.getReadIdleTimeout();
     int writeIdleTimeout = options.getWriteIdleTimeout();
@@ -190,8 +218,7 @@ private void applyHttp1xConnectionOptions(ChannelPipeline pipeline, HttpClientOp
     }
   }
 
-  private void http1xConnected(HttpClientOptions options,
-                               HttpVersion version,
+  private void http1xConnected(HttpVersion version,
                                SocketAddress server,
                                HostAndPort authority,
                                boolean ssl,
@@ -211,7 +238,7 @@ private void http1xConnected(HttpClientOptions options,
     });
     clientHandler.addHandler(conn -> {
       if (upgrade) {
-        Http2ClientChannelInitializer http2ChannelInitializer = http2Initializer(options);
+        Http2ClientChannelInitializer http2ChannelInitializer = http2Initializer();
         Http2UpgradeClientConnection.Http2ChannelUpgrade channelUpgrade= http2ChannelInitializer.channelUpgrade(conn, maxLifetimeMillis, metrics);
         boolean preflightRequest = options.isHttp2ClearTextUpgradeWithPreflightRequest();
         if (preflightRequest) {
@@ -244,4 +271,14 @@ private void http1xConnected(HttpClientOptions options,
     });
     ch.pipeline().addLast("handler", clientHandler);
   }
+
+  @Override
+  public Future<Void> shutdown(Duration timeout) {
+    return netClient.shutdown(timeout.toMillis(), TimeUnit.MILLISECONDS);
+  }
+
+  @Override
+  public Future<Void> close() {
+    return netClient.close();
+  }
 }

vertx-core/src/main/java/io/vertx/core/http/impl/HttpChannelConnector.java

@@ -17,6 +17,8 @@
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.spi.metrics.ClientMetrics;
 
+import java.time.Duration;
+
 /**
  * Performs the channel configuration and connection according to the client options and the protocol version.
  *
@@ -31,4 +33,8 @@ Future<HttpClientConnection> httpConnect(ContextInternal context,
                                            long maxLifetimeMillis,
                                            ClientMetrics<?, ?, ?> metrics);
 
+  Future<Void> shutdown(Duration timeout);
+
+  Future<Void> close();
+
 }