Move channel connector construction outside of HttpClient

Author: vietj

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

@@ -37,6 +37,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;
@@ -244,4 +245,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();
+
 }

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

@@ -23,6 +23,7 @@
 import io.vertx.core.spi.metrics.Metrics;
 import io.vertx.core.spi.metrics.MetricsProvider;
 
+import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
@@ -34,15 +35,15 @@ public class HttpClientBase implements MetricsProvider, Closeable {
 
   protected final VertxInternal vertx;
   public final HttpClientOptions options;
-  protected final NetClientInternal netClient;
-  protected final HttpClientMetrics metrics;
+  protected final HttpChannelConnector connector;
+  protected final HttpClientMetrics<?, ?, ?> metrics;
   protected final CloseSequence closeSequence;
   private volatile ClientSSLOptions defaultSslOptions;
   private long closeTimeout = 0L;
   private TimeUnit closeTimeoutUnit = TimeUnit.SECONDS;
   private Predicate<SocketAddress> proxyFilter;
 
-  public HttpClientBase(VertxInternal vertx, HttpClientOptions options) {
+  public HttpClientBase(VertxInternal vertx, HttpClientOptions options, HttpChannelConnector connector, HttpClientMetrics<?, ?, ?> metrics) {
     if (!options.isKeepAlive() && options.isPipelining()) {
       throw new IllegalStateException("Cannot have pipelining with no keep alive");
     }
@@ -60,7 +61,8 @@ public HttpClientBase(VertxInternal vertx, HttpClientOptions options) {
     this.options = options;
     this.closeSequence = new CloseSequence(p -> doClose(p), p1 -> doShutdown(p1));
     this.proxyFilter = options.getNonProxyHosts() != null ? ProxyFilter.nonProxyHosts(options.getNonProxyHosts()) : ProxyFilter.DEFAULT_PROXY_FILTER;
-    this.netClient = new NetClientBuilder(vertx, new NetClientOptions(options).setProxyOptions(null)).metrics(metrics).build();
+    this.connector = connector;
+//    this.netClient = new NetClientBuilder(vertx, new NetClientOptions(options).setProxyOptions(null)).metrics(metrics).build();
     this.defaultSslOptions = options.getSslOptions();
 
     ClientSSLOptions sslOptions = options.getSslOptions();
@@ -76,7 +78,7 @@ private void configureSSLOptions(ClientSSLOptions sslOptions) {
   }
 
   public NetClientInternal netClient() {
-    return netClient;
+    return null;
   }
 
   public Future<Void> closeFuture() {
@@ -144,11 +146,11 @@ public HttpClientMetrics metrics() {
   }
 
   protected void doShutdown(Completable<Void> p) {
-    netClient.shutdown(closeTimeout, closeTimeoutUnit).onComplete(p);
+    connector.shutdown(Duration.ofMillis(closeTimeoutUnit.toMillis(closeTimeout))).onComplete(p);
   }
 
   protected void doClose(Completable<Void> p) {
-    netClient.close().onComplete(p);
+    connector.close().onComplete(p);
   }
 
   public Future<Void> shutdown(long timeout, TimeUnit unit) {

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

@@ -9,10 +9,14 @@
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.VertxInternal;
 import io.vertx.core.internal.http.HttpClientInternal;
+import io.vertx.core.internal.net.NetClientInternal;
+import io.vertx.core.net.NetClientOptions;
 import io.vertx.core.net.endpoint.LoadBalancer;
 import io.vertx.core.net.AddressResolver;
 import io.vertx.core.net.endpoint.impl.EndpointResolverImpl;
 import io.vertx.core.net.endpoint.EndpointResolver;
+import io.vertx.core.net.impl.tcp.NetClientBuilder;
+import io.vertx.core.spi.metrics.HttpClientMetrics;
 
 import java.util.function.Function;
 
@@ -89,6 +93,13 @@ private EndpointResolver endpointResolver(HttpClientOptions co) {
     return null;
   }
 
+  private HttpClientImpl createHttpClientImpl(EndpointResolver resolver, HttpClientOptions co, PoolOptions po) {
+    HttpClientMetrics<?, ?, ?> metrics = vertx.metrics() != null ? vertx.metrics().createHttpClientMetrics(co) : null;
+    NetClientInternal tcpClient = new NetClientBuilder(vertx, new NetClientOptions(co).setProxyOptions(null)).metrics(metrics).build();
+    HttpChannelConnector channelConnector = new Http1xOrH2ChannelConnector(tcpClient, metrics);
+    return new HttpClientImpl(vertx, channelConnector, metrics, resolver, co, po);
+  }
+
   @Override
   public HttpClientAgent build() {
     HttpClientOptions co = clientOptions != null ? clientOptions : new HttpClientOptions();
@@ -100,15 +111,14 @@ public HttpClientAgent build() {
     if (co.isShared()) {
       CloseFuture closeFuture = new CloseFuture();
       client = vertx.createSharedResource("__vertx.shared.httpClients", co.getName(), closeFuture, cf_ -> {
-
-        HttpClientImpl impl = new HttpClientImpl(vertx, resolver, co, po);
+        HttpClientImpl impl = createHttpClientImpl(resolver, co, po);
         cf_.add(completion -> impl.close().onComplete(completion));
         return impl;
       });
       client = new CleanableHttpClient((HttpClientInternal) client, vertx.cleaner(), (timeout, timeunit) -> closeFuture.close());
       closeable = closeFuture;
     } else {
-      HttpClientImpl impl = new HttpClientImpl(vertx, resolver, co, po);
+      HttpClientImpl impl = createHttpClientImpl(resolver, co, po);
       closeable = impl;
       client = new CleanableHttpClient(impl, vertx.cleaner(), impl::shutdown);
     }

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

@@ -27,6 +27,7 @@
 import io.vertx.core.net.endpoint.ServerInteraction;
 import io.vertx.core.net.endpoint.impl.EndpointResolverImpl;
 import io.vertx.core.spi.metrics.ClientMetrics;
+import io.vertx.core.spi.metrics.HttpClientMetrics;
 import io.vertx.core.spi.metrics.MetricsProvider;
 import io.vertx.core.spi.metrics.PoolMetrics;
 
@@ -49,22 +50,22 @@ public class HttpClientImpl extends HttpClientBase implements HttpClientInternal
   private final PoolOptions poolOptions;
   private final ResourceManager<EndpointKey, SharedHttpClientConnectionGroup> httpCM;
   private final EndpointResolverInternal endpointResolver;
-  private final HttpChannelConnector connector;
   private volatile Function<HttpClientResponse, Future<RequestOptions>> redirectHandler = DEFAULT_REDIRECT_HANDLER;
   private long timerID;
   volatile Handler<HttpConnection> connectionHandler;
   private final Function<ContextInternal, ContextInternal> contextProvider;
   private final long maxLifetime;
 
   public HttpClientImpl(VertxInternal vertx,
+                        HttpChannelConnector connector,
+                        HttpClientMetrics<?, ?, ?> metrics,
                         EndpointResolver endpointResolver,
                         HttpClientOptions options,
                         PoolOptions poolOptions) {
-    super(vertx, options);
+    super(vertx, options, connector, metrics);
 
     this.endpointResolver = (EndpointResolverImpl) endpointResolver;
     this.poolOptions = poolOptions;
-    this.connector = new Http1xOrH2ChannelConnector(netClient, metrics);
     this.httpCM = new ResourceManager<>();
     if (poolCheckerIsNeeded(options, poolOptions)) {
       PoolChecker checker = new PoolChecker(this);

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

@@ -24,6 +24,7 @@
 import io.vertx.core.net.ProxyOptions;
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.spi.metrics.ClientMetrics;
+import io.vertx.core.spi.metrics.HttpClientMetrics;
 import io.vertx.core.spi.metrics.PoolMetrics;
 
 import java.net.URI;
@@ -34,15 +35,13 @@
 public class WebSocketClientImpl extends HttpClientBase implements WebSocketClient {
 
   private final WebSocketClientOptions options;
-  private final HttpChannelConnector connector;
   private final ResourceManager<EndpointKey, WebSocketGroup> webSocketCM;
 
-  public WebSocketClientImpl(VertxInternal vertx, HttpClientOptions options, WebSocketClientOptions wsOptions) {
-    super(vertx, options);
+  public WebSocketClientImpl(VertxInternal vertx, HttpClientOptions options, WebSocketClientOptions wsOptions, HttpChannelConnector connector, HttpClientMetrics<?, ?, ?> metrics) {
+    super(vertx, options, connector, metrics);
 
     this.options = wsOptions;
     this.webSocketCM = new ResourceManager<>();
-    this.connector = new Http1xOrH2ChannelConnector(netClient, metrics);
   }
 
   protected void doShutdown(Completable<Void> p) {

vertx-core/src/main/java/io/vertx/core/impl/VertxImpl.java

@@ -411,21 +411,28 @@ public WebSocketClient createWebSocketClient(WebSocketClientOptions options) {
     if (options.isShared()) {
       CloseFuture closeFuture = new CloseFuture();
       client = createSharedResource("__vertx.shared.webSocketClients", options.getName(), closeFuture, cf_ -> {
-        WebSocketClientImpl impl = new WebSocketClientImpl(this, o, options);
+        WebSocketClientImpl impl = createWebSocketClientImpl(o, options);
         cf_.add(completion -> impl.close().onComplete(completion));
         return impl;
       });
       client = new CleanableWebSocketClient(client, cleaner, (timeout, timeunit) -> closeFuture.close());
       closeable = closeFuture;
     } else {
-      WebSocketClientImpl impl = new WebSocketClientImpl(this, o, options);
+      WebSocketClientImpl impl = createWebSocketClientImpl(o, options);
       closeable = impl;
       client = new CleanableWebSocketClient(impl, cleaner, impl::shutdown);
     }
     cf.add(closeable);
     return client;
   }
 
+  private WebSocketClientImpl createWebSocketClientImpl(HttpClientOptions o, WebSocketClientOptions options) {
+    HttpClientMetrics<?, ?, ?> metrics = metrics() != null ? metrics().createHttpClientMetrics(o) : null;
+    NetClientInternal tcpClient = new NetClientBuilder(this, new NetClientOptions(o).setProxyOptions(null)).metrics(metrics).build();
+    Http1xOrH2ChannelConnector channelConnector = new Http1xOrH2ChannelConnector(tcpClient, metrics);
+    return new WebSocketClientImpl(this, o, options, channelConnector, metrics);
+  }
+
   @Override
   public HttpClientBuilder httpClientBuilder() {
     return new HttpClientBuilderInternal(this);