Introduce HTTP connection SPI

Author: vietj

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

@@ -20,7 +20,7 @@
 import io.vertx.codegen.annotations.VertxGen;
 import io.vertx.core.MultiMap;
 import io.vertx.core.http.impl.headers.HeadersMultiMap;
-import io.vertx.core.http.impl.http2.Http2HeadersMultiMap;
+import io.vertx.core.http.impl.spi.Http2HeadersMultiMap;
 
 /**
  * Contains a bunch of useful HTTP headers stuff:

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

@@ -75,7 +75,6 @@ public class Http1xClientConnection extends Http1xConnection implements HttpClie
   private final HostAndPort authority;
   public final ClientMetrics metrics;
   private final HttpVersion version;
-  private final boolean pooled;
   private final long lifetimeEvictionTimestamp;
 
   private final Deque<Stream> requests = new ArrayDeque<>();
@@ -102,7 +101,6 @@ public class Http1xClientConnection extends Http1xConnection implements HttpClie
                          HostAndPort authority,
                          ContextInternal context,
                          ClientMetrics metrics,
-                         boolean pooled,
                          long maxLifetime) {
     super(context, chctx);
     this.client = client;
@@ -115,7 +113,6 @@ public class Http1xClientConnection extends Http1xConnection implements HttpClie
     this.lifetimeEvictionTimestamp = maxLifetime > 0 ? System.currentTimeMillis() + maxLifetime : Long.MAX_VALUE;
     this.keepAliveTimeout = options.getKeepAliveTimeout();
     this.expirationTimestamp = expirationTimestampOf(keepAliveTimeout);
-    this.pooled = pooled;
   }
 
   @Override
@@ -151,11 +148,6 @@ public synchronized long activeStreams() {
     return requests.isEmpty() && responses.isEmpty() ? 0 : 1;
   }
 
-  @Override
-  public boolean pooled() {
-    return pooled;
-  }
-
   /**
    * @return a raw {@code NetSocket} - for internal use - must be called from event-loop
    */

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

@@ -129,6 +129,12 @@ TracingPolicy tracingPolicy() {
     return tracingPolicy;
   }
 
+  @Override
+  public HttpServerConnection streamHandler(Handler<HttpServerStream> handler) {
+    // This feature is not available for HTTP/1.1
+    return this;
+  }
+
   public HttpServerConnection handler(Handler<HttpServerRequest> handler) {
     requestHandler = handler;
     return this;

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

@@ -79,11 +79,6 @@ public long concurrency() {
     return upgradeProcessed ? current.concurrency() : 1L;
   }
 
-  @Override
-  public boolean pooled() {
-    return current.pooled();
-  }
-
   @Override
   public long activeStreams() {
     return current.concurrency();
@@ -397,7 +392,7 @@ public void upgradeFailure(Throwable cause) {
       };
       upgrade.upgrade(upgradingStream, request, buf, end,
         upgradingConnection.channelHandlerContext().channel(),
-        upgradingConnection.pooled(), blah
+        blah
       );
       PromiseInternal<Void> promise = upgradingStream.context().promise();
       writeHead(request, chunked, buf, end, priority, connect, promise);
@@ -904,7 +899,6 @@ void upgrade(HttpClientStream upgradingStream,
                  Buffer content,
                  boolean end,
                  Channel channel,
-                 boolean pooled,
                  UpgradeResult result);
   }
 }

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

@@ -62,7 +62,6 @@ public class HttpChannelConnector {
   private final HttpVersion version;
   private final HostAndPort authority;
   private final SocketAddress server;
-  private final boolean pooled;
   private final long maxLifetime;
   private final Http2ClientChannelInitializer http2ChannelInitializer;
 
@@ -76,7 +75,6 @@ public HttpChannelConnector(HttpClientBase client,
                               boolean useAlpn,
                               HostAndPort authority,
                               SocketAddress server,
-                              boolean pooled,
                               long maxLifetimeMillis) {
 
     Http2ClientChannelInitializer http2ChannelInitializer;
@@ -92,7 +90,7 @@ public HttpChannelConnector(HttpClientBase client,
         client.options.isDecompressionSupported(),
         client.options.getLogActivity());
     } else {
-      http2ChannelInitializer = new Http2CodecClientChannelInitializer(client, metrics, pooled, maxLifetimeMillis, authority);
+      http2ChannelInitializer = new Http2CodecClientChannelInitializer(client, metrics, maxLifetimeMillis, authority);
     }
 
     this.client = client;
@@ -106,7 +104,6 @@ public HttpChannelConnector(HttpClientBase client,
     this.version = version;
     this.authority = authority;
     this.server = server;
-    this.pooled = pooled;
     this.maxLifetime = maxLifetimeMillis;
     this.http2ChannelInitializer = http2ChannelInitializer;
   }
@@ -239,7 +236,7 @@ private void http1xConnected(HttpVersion version,
     boolean upgrade = version == HttpVersion.HTTP_2 && options.isHttp2ClearTextUpgrade();
     VertxHandler<Http1xClientConnection> clientHandler = VertxHandler.create(chctx -> {
       HttpClientMetrics met = client.metrics();
-      Http1xClientConnection conn = new Http1xClientConnection(upgrade ? HttpVersion.HTTP_1_1 : version, client, chctx, ssl, server, authority, context, metrics, pooled, maxLifetime);
+      Http1xClientConnection conn = new Http1xClientConnection(upgrade ? HttpVersion.HTTP_1_1 : version, client, chctx, ssl, server, authority, context, metrics, maxLifetime);
       if (met != null) {
         conn.metric(socketMetric);
         met.endpointConnected(metrics);

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

@@ -73,11 +73,6 @@ public interface HttpClientConnection extends HttpConnection {
    */
   HttpClientConnection concurrencyChangeHandler(Handler<Long> handler);
 
-  /**
-   * @return whether the connection is pooled
-   */
-  boolean pooled();
-
   /**
    * @return the {@link ChannelHandlerContext} of the handler managing the connection
    */

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

@@ -137,7 +137,7 @@ private Function<EndpointKey, SharedHttpClientConnectionGroup> httpEndpointProvi
         key = new EndpointKey(key.ssl, key.sslOptions, proxyOptions, server, key.authority);
         proxyOptions = null;
       }
-      HttpChannelConnector connector = new HttpChannelConnector(HttpClientImpl.this, netClient, key.sslOptions, proxyOptions, clientMetrics, options.getProtocolVersion(), key.ssl, options.isUseAlpn(), key.authority, key.server, true, maxLifetime);
+      HttpChannelConnector connector = new HttpChannelConnector(HttpClientImpl.this, netClient, key.sslOptions, proxyOptions, clientMetrics, options.getProtocolVersion(), key.ssl, options.isUseAlpn(), key.authority, key.server, maxLifetime);
       return new SharedHttpClientConnectionGroup(
         vertx,
         HttpClientImpl.this,
@@ -244,7 +244,6 @@ public Future<io.vertx.core.http.HttpClientConnection> connect(HttpConnectOption
       useAlpn,
       authority,
       server,
-      false,
       0);
     return (Future) connector.httpConnect(vertx.getOrCreateContext()).map(conn -> new UnpooledHttpClientConnection(conn).init());
   }

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

@@ -14,15 +14,15 @@
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
 import io.vertx.core.http.HttpConnection;
-import io.vertx.core.http.HttpServerRequest;
-import io.vertx.core.http.impl.http2.Http2HeadersMultiMap;
 import io.vertx.core.internal.ContextInternal;
 
 /**
  * @author <a href="mailto:[email protected]">Julien Viet</a>
  */
 public interface HttpServerConnection extends HttpConnection {
 
+  HttpServerConnection streamHandler(Handler<HttpServerStream> handler);
+
   MultiMap newHeaders();
 
   boolean supportsSendFile();

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

@@ -23,7 +23,7 @@
 import io.vertx.core.http.HttpServerRequest;
 import io.vertx.core.http.ServerWebSocket;
 import io.vertx.core.http.ServerWebSocketHandshake;
-import io.vertx.core.http.impl.http2.Http2ServerConnection;
+import io.vertx.core.http.impl.spi.HttpServerConnectionProvider;
 import io.vertx.core.internal.ContextInternal;
 
 import java.util.ArrayList;
@@ -96,7 +96,7 @@ public void handle(HttpServerConnection conn) {
       http1Conn.handler(requestHandler);
       http1Conn.invalidRequestHandler(invalidRequestHandler);
     } else {
-      Http2ServerConnection http2Conn = (Http2ServerConnection) conn;
+      HttpServerConnectionProvider http2Conn = (HttpServerConnectionProvider) conn;
       http2Conn.streamHandler(stream -> {
         HttpServerOptions options = server.options;
         HttpServerRequestImpl request = new HttpServerRequestImpl(stream, stream.context(), options.isHandle100ContinueAutomatically(),

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

@@ -103,7 +103,6 @@ public class HttpServerConnectionInitializer {
         compressionManager,
         streamContextSupplier,
         connectionHandler,
-        serverOrigin,
         metric,
         options.getLogActivity()
       );